mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-10 08:28:25 +00:00
6c76f5f9c6
Every ruff-strict rule that sat above its budget limit (FURB188, RUF022, SIM118, UP007, UP032, UP037) is now at zero, LIT001 and LIT006 are back under their ceilings, and the freed headroom is ratcheted out of ruff-strict-budget.json, type-discipline-budget.json, and basedpyright-code-budget.json so the gates take the fast path again
1830 lines
72 KiB
Python
1830 lines
72 KiB
Python
"""
|
|
Redis Cache implementation
|
|
|
|
Has 4 primary methods:
|
|
- set_cache
|
|
- get_cache
|
|
- async_set_cache
|
|
- async_get_cache
|
|
"""
|
|
|
|
import ast
|
|
import asyncio
|
|
import functools
|
|
import hashlib
|
|
import inspect
|
|
import json
|
|
import time
|
|
from collections.abc import Awaitable, Callable, Sequence
|
|
from contextvars import ContextVar
|
|
from datetime import timedelta
|
|
from typing import TYPE_CHECKING, Any, Final, TypeVar, cast
|
|
|
|
import litellm
|
|
from litellm._logging import print_verbose, verbose_logger
|
|
from litellm.constants import (
|
|
DEFAULT_REDIS_MAJOR_VERSION,
|
|
REDIS_CIRCUIT_BREAKER_ENABLED,
|
|
REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD,
|
|
REDIS_CIRCUIT_BREAKER_RECOVERY_TIMEOUT,
|
|
)
|
|
from litellm.litellm_core_utils.core_helpers import _get_parent_otel_span_from_kwargs
|
|
from litellm.litellm_core_utils.coroutine_checker import coroutine_checker
|
|
from litellm.types.caching import (
|
|
RedisPipelineIncrementOperation,
|
|
RedisPipelineLpopOperation,
|
|
RedisPipelineRpushOperation,
|
|
)
|
|
from litellm.types.services import ServiceTypes
|
|
|
|
from .base_cache import BaseCache
|
|
|
|
if TYPE_CHECKING:
|
|
from opentelemetry.trace import Span as _Span
|
|
from redis.asyncio import Redis, RedisCluster
|
|
from redis.asyncio.client import Pipeline
|
|
from redis.asyncio.cluster import ClusterPipeline
|
|
|
|
pipeline = Pipeline
|
|
cluster_pipeline = ClusterPipeline
|
|
async_redis_client = Redis
|
|
async_redis_cluster_client = RedisCluster
|
|
Span = _Span | Any
|
|
else:
|
|
pipeline = Any
|
|
cluster_pipeline = Any
|
|
async_redis_client = Any
|
|
async_redis_cluster_client = Any
|
|
Span = Any
|
|
|
|
|
|
def _get_call_stack_info(num_frames: int = 2) -> str:
|
|
"""
|
|
Get the function names from the previous 1-2 functions in the call stack.
|
|
|
|
Args:
|
|
num_frames: Number of previous frames to include (default: 2)
|
|
|
|
Returns:
|
|
A string with format "current_function <- caller_function [<- grandparent_function]"
|
|
"""
|
|
try:
|
|
current_frame: Final = inspect.currentframe()
|
|
if current_frame is None:
|
|
return "unknown"
|
|
|
|
# Skip this function and the immediate caller (which sets call_type)
|
|
f_back: Final = current_frame.f_back
|
|
if f_back is None:
|
|
return "unknown"
|
|
frame = f_back.f_back
|
|
if frame is None:
|
|
return "unknown"
|
|
function_names: Final = []
|
|
|
|
for _ in range(num_frames):
|
|
if frame is None:
|
|
break
|
|
func_name = frame.f_code.co_name
|
|
function_names.append(func_name)
|
|
frame = frame.f_back
|
|
|
|
if not function_names:
|
|
return "unknown"
|
|
|
|
return " <- ".join(function_names)
|
|
except Exception:
|
|
return "unknown"
|
|
|
|
|
|
class RedisCircuitBreaker:
|
|
"""
|
|
Tracks Redis health for a RedisCache instance.
|
|
|
|
States:
|
|
CLOSED - normal, Redis is called
|
|
OPEN - Redis is down, raise immediately (no network call)
|
|
HALF_OPEN - recovery probe: allow one request through
|
|
|
|
Transitions:
|
|
CLOSED -> OPEN after failure_threshold consecutive failures
|
|
OPEN -> HALF_OPEN after recovery_timeout seconds
|
|
HALF_OPEN -> CLOSED on success
|
|
HALF_OPEN -> OPEN on failure (resets timer)
|
|
"""
|
|
|
|
CLOSED = "closed"
|
|
OPEN = "open"
|
|
HALF_OPEN = "half_open"
|
|
|
|
def __init__(
|
|
self,
|
|
failure_threshold: int,
|
|
recovery_timeout: int,
|
|
enabled: bool = True,
|
|
) -> None:
|
|
self.failure_threshold = failure_threshold
|
|
self.recovery_timeout = recovery_timeout
|
|
self.enabled = enabled
|
|
self._failure_count = 0
|
|
self._opened_at: float | None = None
|
|
self._state = self.CLOSED
|
|
|
|
def is_open(self) -> bool:
|
|
"""Returns True if Redis calls should be skipped."""
|
|
if not self.enabled:
|
|
return False
|
|
if self._state == self.HALF_OPEN:
|
|
# Probe already in flight — fast-fail all concurrent requests.
|
|
# Only the one call that caused the OPEN→HALF_OPEN transition
|
|
# (which returned False) is the designated probe.
|
|
return True
|
|
if self._state == self.OPEN:
|
|
if time.time() - (self._opened_at or 0) > self.recovery_timeout:
|
|
self._state = self.HALF_OPEN
|
|
return False # this caller is the designated probe
|
|
return True
|
|
return False
|
|
|
|
def record_failure(self) -> None:
|
|
if not self.enabled:
|
|
return
|
|
self._failure_count += 1
|
|
self._opened_at = time.time()
|
|
if self._failure_count >= self.failure_threshold:
|
|
if self._state != self.OPEN:
|
|
verbose_logger.warning(
|
|
"Redis circuit breaker OPENED after %d consecutive failures — fast-failing Redis calls for %ds",
|
|
self._failure_count,
|
|
self.recovery_timeout,
|
|
)
|
|
self._state = self.OPEN
|
|
|
|
def record_success(self) -> None:
|
|
if not self.enabled:
|
|
return
|
|
if self._state == self.HALF_OPEN:
|
|
verbose_logger.info("Redis circuit breaker CLOSED — Redis recovered")
|
|
self._failure_count = 0
|
|
self._state = self.CLOSED
|
|
|
|
|
|
_RedisCallResult = TypeVar("_RedisCallResult")
|
|
|
|
|
|
_swallowed_redis_failures: Final[ContextVar[int]] = ContextVar("litellm_swallowed_redis_failures", default=0)
|
|
|
|
|
|
@functools.lru_cache(maxsize=1)
|
|
def _redis_health_error_types() -> tuple[type, ...]:
|
|
"""Exception types that mean the Redis backend itself is unhealthy.
|
|
|
|
Command and data errors say nothing about connectivity: an INCR against a non-numeric
|
|
value or an undecodable cached entry is a request problem, and counting those would let
|
|
a caller trip the shared breaker on demand, dropping rate limiting to per-process
|
|
counters that spreading traffic across replicas can outrun.
|
|
|
|
Imported lazily because this module is reachable from a base ``import litellm`` while
|
|
redis is not a base dependency.
|
|
"""
|
|
from redis.exceptions import BusyLoadingError, ClusterDownError
|
|
from redis.exceptions import ConnectionError as RedisConnectionError
|
|
from redis.exceptions import TimeoutError as RedisTimeoutError
|
|
|
|
return (RedisConnectionError, RedisTimeoutError, BusyLoadingError, ClusterDownError, OSError, asyncio.TimeoutError)
|
|
|
|
|
|
def _is_redis_health_failure(exc: BaseException) -> bool:
|
|
"""True when ``exc`` indicates Redis is unreachable rather than the request being bad."""
|
|
try:
|
|
return isinstance(exc, _redis_health_error_types())
|
|
except ImportError:
|
|
return True
|
|
|
|
|
|
def _record_swallowed_redis_failure(breaker: RedisCircuitBreaker, exc: BaseException) -> None:
|
|
"""Record a Redis failure that the calling method is about to swallow.
|
|
|
|
The marker is a ContextVar rather than a counter on the breaker because breakers are
|
|
shared by every concurrent caller. A plain shared counter cannot tell "my call failed"
|
|
from "some other in-flight call failed", so a success overlapping someone else's
|
|
failure would be discarded and a Redis that is answering would still be evicted.
|
|
asyncio gives each task its own copy of the context, so this is per-call.
|
|
"""
|
|
if not _is_redis_health_failure(exc):
|
|
return
|
|
breaker.record_failure()
|
|
_swallowed_redis_failures.set(_swallowed_redis_failures.get() + 1)
|
|
|
|
|
|
async def _run_under_circuit_breaker(
|
|
breaker: RedisCircuitBreaker,
|
|
name: str,
|
|
call: Callable[[], Awaitable[_RedisCallResult]],
|
|
) -> _RedisCallResult:
|
|
"""Run one Redis coroutine under a circuit breaker.
|
|
|
|
Shared by the method decorator and the Lua script executor so both feed the same
|
|
health signal. Success is recorded only when nothing failed while ``call`` ran,
|
|
because several Redis methods catch their own connection errors and return a default.
|
|
"""
|
|
if breaker.is_open():
|
|
raise Exception(f"Redis circuit breaker is open — skipping {name}")
|
|
swallowed_before: Final = _swallowed_redis_failures.get()
|
|
try:
|
|
result: Final = await call()
|
|
except Exception as e:
|
|
if _is_redis_health_failure(e):
|
|
breaker.record_failure()
|
|
raise
|
|
if _swallowed_redis_failures.get() == swallowed_before:
|
|
breaker.record_success()
|
|
return result
|
|
|
|
|
|
def _redis_circuit_breaker_guard(method):
|
|
"""
|
|
Decorator for RedisCache async methods.
|
|
Checks the circuit breaker before each call; records success/failure after.
|
|
Does not apply to ping/disconnect/test_connection (health/teardown must always run).
|
|
|
|
A returning method is not proof of a healthy Redis: several methods catch their own
|
|
connection errors and return a default so callers degrade rather than fail. Counting
|
|
those as successes reset the failure streak on every request, so the breaker could
|
|
never open and Redis was never taken out of the pool. Success is therefore recorded
|
|
only when no failure was registered while the method ran.
|
|
"""
|
|
|
|
@functools.wraps(method)
|
|
async def wrapper(self, *args, **kwargs):
|
|
return await _run_under_circuit_breaker(
|
|
self._circuit_breaker, method.__name__, lambda: method(self, *args, **kwargs)
|
|
)
|
|
|
|
return wrapper
|
|
|
|
|
|
class RedisCache(BaseCache):
|
|
# if users don't provider one, use the default litellm cache
|
|
|
|
def __init__(
|
|
self,
|
|
host=None,
|
|
port=None,
|
|
password=None,
|
|
redis_flush_size: int | None = 100,
|
|
namespace: str | None = None,
|
|
startup_nodes: list | None = None, # for redis-cluster
|
|
socket_timeout: float | None = 5.0, # default 5 second timeout
|
|
**kwargs,
|
|
):
|
|
from litellm._service_logger import ServiceLogging
|
|
|
|
from .._redis import get_redis_client, get_redis_connection_pool
|
|
|
|
redis_kwargs: Final = {}
|
|
if host is not None:
|
|
redis_kwargs["host"] = host
|
|
if port is not None:
|
|
redis_kwargs["port"] = port
|
|
if password is not None:
|
|
redis_kwargs["password"] = password
|
|
if startup_nodes is not None:
|
|
redis_kwargs["startup_nodes"] = startup_nodes
|
|
if socket_timeout is not None:
|
|
redis_kwargs["socket_timeout"] = socket_timeout
|
|
|
|
### HEALTH MONITORING OBJECT ###
|
|
if kwargs.get("service_logger_obj", None) is not None and isinstance(
|
|
kwargs["service_logger_obj"], ServiceLogging
|
|
):
|
|
self.service_logger_obj = kwargs.pop("service_logger_obj")
|
|
else:
|
|
self.service_logger_obj = ServiceLogging()
|
|
|
|
redis_kwargs.update(kwargs)
|
|
self.redis_client = get_redis_client(**redis_kwargs)
|
|
self.redis_async_client: async_redis_client | async_redis_cluster_client | None = None
|
|
self.redis_kwargs = redis_kwargs
|
|
self.async_redis_conn_pool = get_redis_connection_pool(**redis_kwargs)
|
|
|
|
# redis namespaces
|
|
self.namespace = namespace
|
|
# for high traffic, we store the redis results in memory and then batch write to redis
|
|
self.redis_batch_writing_buffer: list = []
|
|
if redis_flush_size is None:
|
|
self.redis_flush_size: int = 100
|
|
else:
|
|
self.redis_flush_size = redis_flush_size
|
|
self.redis_version = "Unknown"
|
|
try:
|
|
if not coroutine_checker.is_async_callable(self.redis_client):
|
|
self.redis_version = self.redis_client.info()["redis_version"]
|
|
except Exception:
|
|
pass
|
|
|
|
self._circuit_breaker = RedisCircuitBreaker(
|
|
failure_threshold=REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD,
|
|
recovery_timeout=REDIS_CIRCUIT_BREAKER_RECOVERY_TIMEOUT,
|
|
enabled=REDIS_CIRCUIT_BREAKER_ENABLED,
|
|
)
|
|
|
|
self._setup_health_pings()
|
|
|
|
if litellm.default_redis_ttl is not None:
|
|
super().__init__(default_ttl=int(litellm.default_redis_ttl))
|
|
else:
|
|
super().__init__() # defaults to 60s
|
|
|
|
def _setup_health_pings(self):
|
|
"""Setup async and sync health pings for Redis."""
|
|
# ASYNC HEALTH PING
|
|
try:
|
|
_ = asyncio.get_running_loop().create_task(self.ping())
|
|
except Exception as e:
|
|
if "no running event loop" in str(e):
|
|
verbose_logger.debug("Ignoring async redis ping. No running event loop.")
|
|
else:
|
|
verbose_logger.error(
|
|
"Error connecting to Async Redis client - %s",
|
|
e,
|
|
extra={"error": str(e)},
|
|
)
|
|
self._handle_async_ping_error(e)
|
|
|
|
# SYNC HEALTH PING
|
|
try:
|
|
if hasattr(self.redis_client, "ping"):
|
|
self.redis_client.ping()
|
|
except Exception as e:
|
|
verbose_logger.error("Error connecting to Sync Redis client", extra={"error": str(e)})
|
|
self._handle_sync_ping_error(e)
|
|
|
|
def _handle_async_ping_error(self, e: Exception):
|
|
"""Handle async ping error with service failure hook."""
|
|
try:
|
|
loop: Final = asyncio.get_running_loop()
|
|
start_time: Final = time.time()
|
|
end_time: Final = start_time
|
|
loop.create_task(
|
|
self.service_logger_obj.async_service_failure_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=end_time - start_time,
|
|
error=e,
|
|
call_type="redis_async_ping",
|
|
)
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
def _handle_sync_ping_error(self, e: Exception):
|
|
"""Handle sync ping error with service failure hook."""
|
|
try:
|
|
loop: Final = asyncio.get_running_loop()
|
|
start_time: Final = time.time()
|
|
end_time: Final = start_time
|
|
loop.create_task(
|
|
self.service_logger_obj.async_service_failure_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=end_time - start_time,
|
|
error=e,
|
|
call_type="redis_sync_ping",
|
|
)
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
def _get_async_client_cache_key(self) -> str:
|
|
"""
|
|
Generate a cache key for the async Redis client based on connection parameters.
|
|
This ensures different Redis configurations use different cached clients.
|
|
"""
|
|
# Create a stable representation of redis_kwargs for hashing
|
|
# Sort keys to ensure consistent hash regardless of parameter order
|
|
sorted_kwargs: Final = sorted(self.redis_kwargs.items())
|
|
kwargs_str: Final = json.dumps(sorted_kwargs, sort_keys=True)
|
|
kwargs_hash: Final = hashlib.sha256(kwargs_str.encode()).hexdigest()[:16]
|
|
return f"async-redis-client-{kwargs_hash}"
|
|
|
|
def init_async_client(
|
|
self,
|
|
) -> async_redis_client | async_redis_cluster_client:
|
|
from litellm import in_memory_llm_clients_cache
|
|
|
|
from .._redis import get_redis_async_client, get_redis_connection_pool
|
|
|
|
cache_key: Final = self._get_async_client_cache_key()
|
|
cached_client: Final = in_memory_llm_clients_cache.get_cache(key=cache_key)
|
|
if cached_client is not None:
|
|
redis_async_client = cast(async_redis_client | async_redis_cluster_client, cached_client)
|
|
else:
|
|
# Create new connection pool and client for current event loop
|
|
self.async_redis_conn_pool = get_redis_connection_pool(**self.redis_kwargs)
|
|
redis_async_client = get_redis_async_client(connection_pool=self.async_redis_conn_pool, **self.redis_kwargs)
|
|
in_memory_llm_clients_cache.set_cache(key=cache_key, value=redis_async_client)
|
|
|
|
self.redis_async_client = redis_async_client
|
|
return redis_async_client
|
|
|
|
def check_and_fix_namespace(self, key: str) -> str:
|
|
"""
|
|
Make sure each key starts with the given namespace
|
|
"""
|
|
if key is None:
|
|
return key
|
|
if self.namespace is not None and not key.startswith(self.namespace):
|
|
key = self.namespace + ":" + key
|
|
|
|
return key
|
|
|
|
def _parse_redis_major_version(self) -> int:
|
|
"""
|
|
Parse Redis version to extract the major version number.
|
|
|
|
Handles multiple version formats:
|
|
- Strings: "7.0.0", "6", "7.0.0-rc1", " 7.0.0 "
|
|
- Floats: 7.0 (e.g., from AWS ElastiCache Valkey)
|
|
- Integers: 7
|
|
- Malformed: "latest", "", "Unknown" (defaults to DEFAULT_REDIS_MAJOR_VERSION)
|
|
|
|
Returns:
|
|
int: The major version number (defaults to DEFAULT_REDIS_MAJOR_VERSION if unparseable)
|
|
"""
|
|
if self.redis_version == "Unknown":
|
|
return DEFAULT_REDIS_MAJOR_VERSION
|
|
|
|
try:
|
|
version_str: Final = str(self.redis_version).strip()
|
|
# Handle cases where there's no dot (e.g., "7" or 7)
|
|
if "." in version_str:
|
|
major_version = int(version_str.split(".")[0])
|
|
else:
|
|
# Direct integer or single-digit string
|
|
major_version = int(float(version_str))
|
|
return major_version
|
|
except (ValueError, AttributeError):
|
|
# Fallback for unparseable versions (e.g., "v7.0.0", "latest")
|
|
return DEFAULT_REDIS_MAJOR_VERSION
|
|
|
|
def set_cache(self, key, value, **kwargs):
|
|
ttl: Final = self.get_ttl(**kwargs)
|
|
print_verbose(f"Set Redis Cache: key: {key}\nValue {value}\nttl={ttl}, redis_version={self.redis_version}")
|
|
key = self.check_and_fix_namespace(key=key)
|
|
try:
|
|
start_time: Final = time.time()
|
|
self.redis_client.set(name=key, value=str(value), ex=ttl)
|
|
end_time: Final = time.time()
|
|
_duration: Final = end_time - start_time
|
|
self.service_logger_obj.service_success_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
call_type=f"set_cache <- {_get_call_stack_info()}",
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
)
|
|
except Exception as e:
|
|
# NON blocking - notify users Redis is throwing an exception
|
|
print_verbose(f"litellm.caching.caching: set() - Got exception from REDIS : {e}")
|
|
|
|
def increment_cache(self, key, value: int, ttl: float | None = None, **kwargs) -> int:
|
|
_redis_client: Final = self.redis_client
|
|
start_time = time.time()
|
|
set_ttl: Final = self.get_ttl(ttl=ttl)
|
|
key = self.check_and_fix_namespace(key=key)
|
|
try:
|
|
start_time = time.time()
|
|
result: Final[int] = _redis_client.incr(name=key, amount=value)
|
|
end_time = time.time()
|
|
_duration = end_time - start_time
|
|
self.service_logger_obj.service_success_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
call_type=f"increment_cache <- {_get_call_stack_info()}",
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
)
|
|
|
|
if set_ttl is not None:
|
|
# check if key already has ttl, if not -> set ttl
|
|
start_time = time.time()
|
|
current_ttl: Final = _redis_client.ttl(key)
|
|
end_time = time.time()
|
|
_duration = end_time - start_time
|
|
self.service_logger_obj.service_success_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
call_type=f"increment_cache_ttl <- {_get_call_stack_info()}",
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
)
|
|
if current_ttl == -1:
|
|
# Key has no expiration
|
|
start_time = time.time()
|
|
_redis_client.expire(key, set_ttl)
|
|
end_time = time.time()
|
|
_duration = end_time - start_time
|
|
self.service_logger_obj.service_success_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
call_type=f"increment_cache_expire <- {_get_call_stack_info()}",
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
)
|
|
return result
|
|
except Exception as e:
|
|
## LOGGING ##
|
|
end_time = time.time()
|
|
_duration = end_time - start_time
|
|
verbose_logger.error(
|
|
"LiteLLM Redis Caching: increment_cache() - Got exception from REDIS %s, Writing value=%s",
|
|
str(e),
|
|
value,
|
|
)
|
|
raise e
|
|
|
|
@_redis_circuit_breaker_guard
|
|
async def async_scan_iter(self, pattern: str, count: int = 100) -> list:
|
|
start_time: Final = time.time()
|
|
try:
|
|
keys: Final = []
|
|
_redis_client: Final = self.init_async_client()
|
|
if not hasattr(_redis_client, "scan_iter"):
|
|
verbose_logger.debug(
|
|
"Redis client does not support scan_iter, potentially using Redis Cluster. Returning empty list."
|
|
)
|
|
return []
|
|
|
|
pattern = self.check_and_fix_namespace(key=pattern)
|
|
async for key in _redis_client.scan_iter(match=pattern + "*", count=count):
|
|
keys.append(key)
|
|
if len(keys) >= count:
|
|
break
|
|
|
|
## LOGGING ##
|
|
end_time = time.time()
|
|
_duration = end_time - start_time
|
|
asyncio.create_task(
|
|
self.service_logger_obj.async_service_success_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
call_type=f"async_scan_iter <- {_get_call_stack_info()}",
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
)
|
|
) # DO NOT SLOW DOWN CALL B/C OF THIS
|
|
return keys
|
|
except Exception as e:
|
|
# NON blocking - notify users Redis is throwing an exception
|
|
## LOGGING ##
|
|
end_time = time.time()
|
|
_duration = end_time - start_time
|
|
asyncio.create_task(
|
|
self.service_logger_obj.async_service_failure_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
error=e,
|
|
call_type=f"async_scan_iter <- {_get_call_stack_info()}",
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
)
|
|
)
|
|
raise e
|
|
|
|
def async_register_script(self, script: str) -> Callable[..., Awaitable[Any]]:
|
|
"""
|
|
Register a Lua script with Redis asynchronously.
|
|
Works with both standalone Redis and Redis Cluster.
|
|
|
|
The returned callable namespaces every key it is invoked with, so Lua
|
|
scripts hit the same prefixed keys as get/set/increment. Without this,
|
|
scripts would operate on raw keys while the rest of the cache uses the
|
|
namespace, leaving rate-limit and lock keys outside the configured prefix.
|
|
|
|
Registration is deferred to call time and cached per running event loop
|
|
(via in_memory_llm_clients_cache, which keys its entries on the loop). A
|
|
registered script is bound to the connection of the loop it was created
|
|
on; awaiting it from another loop raises "got Future attached to a
|
|
different loop". Binding lazily on the calling loop gives the script the
|
|
same per-loop scoping init_async_client already gives the clients, so a
|
|
script registered once at startup is never reused across loops.
|
|
|
|
Args:
|
|
script (str): The Lua script to register
|
|
|
|
Returns:
|
|
A callable ``(keys, args, client=None)`` that runs the script
|
|
against the calling loop's Redis client.
|
|
"""
|
|
# Keyed by connection params and namespace as well as the script, so
|
|
# two RedisCache instances pointing at different servers or using
|
|
# different key prefixes never share an executor; in_memory_llm_clients_cache
|
|
# then adds the running loop, completing the per-(client, namespace, loop)
|
|
# scoping.
|
|
script_cache_key: Final = (
|
|
f"redis-registered-script-{self._get_async_client_cache_key()}-"
|
|
f"{self.namespace}-{hashlib.sha256(script.encode()).hexdigest()[:16]}"
|
|
)
|
|
|
|
async def run_script(keys: Sequence[str], args: Sequence[Any], client: Any = None) -> Any:
|
|
async def execute() -> object:
|
|
executor: Callable[..., Awaitable[Any]] | None = litellm.in_memory_llm_clients_cache.get_cache(
|
|
key=script_cache_key
|
|
)
|
|
if executor is None:
|
|
executor = self._register_script_for_current_loop(script)
|
|
litellm.in_memory_llm_clients_cache.set_cache(key=script_cache_key, value=executor)
|
|
return await executor(keys=keys, args=args, client=client)
|
|
|
|
return await _run_under_circuit_breaker(self._circuit_breaker, "run_script", execute)
|
|
|
|
return run_script
|
|
|
|
def _register_script_for_current_loop(self, script: str) -> Callable[..., Awaitable[Any]]:
|
|
"""
|
|
Register the script against the current event loop's Redis client.
|
|
|
|
Kept separate from async_register_script so each loop caches its own
|
|
executor; see that method for why the binding must be per loop.
|
|
"""
|
|
_redis_client: Final[Any] = self.init_async_client()
|
|
if hasattr(_redis_client, "register_script"):
|
|
registered_script: Final = _redis_client.register_script(script)
|
|
|
|
async def standalone_executor(keys: Sequence[str], args: Sequence[Any], client: Any = None) -> Any:
|
|
namespaced_keys: Final = tuple(self.check_and_fix_namespace(key=key) for key in keys)
|
|
return await registered_script(keys=namespaced_keys, args=args, client=client)
|
|
|
|
return standalone_executor
|
|
|
|
if hasattr(_redis_client, "script_load"):
|
|
script_sha: Final = _redis_client.script_load(script)
|
|
|
|
async def cluster_executor(keys: Sequence[str], args: Sequence[Any], client: Any = None) -> Any:
|
|
namespaced_keys: Final = tuple(self.check_and_fix_namespace(key=key) for key in keys)
|
|
return await _redis_client.evalsha(script_sha, len(namespaced_keys), *namespaced_keys, *args)
|
|
|
|
return cluster_executor
|
|
|
|
raise ValueError("Redis client does not support Lua script registration")
|
|
|
|
@_redis_circuit_breaker_guard
|
|
async def async_set_cache(self, key, value, **kwargs):
|
|
from redis.asyncio import Redis
|
|
|
|
if key is None:
|
|
verbose_logger.debug(
|
|
"LiteLLM Redis Caching: async set() skipped — key is None, value=%r",
|
|
value,
|
|
)
|
|
return None
|
|
|
|
start_time: Final = time.time()
|
|
try:
|
|
_redis_client: Final[Redis] = self.init_async_client()
|
|
except Exception as e:
|
|
end_time = time.time()
|
|
_duration = end_time - start_time
|
|
asyncio.create_task(
|
|
self.service_logger_obj.async_service_failure_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
error=e,
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs),
|
|
call_type=f"async_set_cache <- {_get_call_stack_info()}",
|
|
)
|
|
)
|
|
verbose_logger.error(
|
|
"LiteLLM Redis Caching: async set() - Got exception from REDIS %s, key=%r, value=%r",
|
|
str(e),
|
|
key,
|
|
value,
|
|
)
|
|
raise e
|
|
|
|
key = self.check_and_fix_namespace(key=key)
|
|
ttl: Final = self.get_ttl(**kwargs)
|
|
nx: Final = kwargs.get("nx", False)
|
|
print_verbose(f"Set ASYNC Redis Cache: key: {key}\nValue {value}\nttl={ttl}")
|
|
|
|
try:
|
|
if not hasattr(_redis_client, "set"):
|
|
raise Exception("Redis client cannot set cache. Attribute not found.")
|
|
result: Final = await _redis_client.set(
|
|
name=key,
|
|
value=json.dumps(value),
|
|
nx=nx,
|
|
ex=ttl,
|
|
)
|
|
print_verbose(f"Successfully Set ASYNC Redis Cache: key: {key}\nValue {value}\nttl={ttl}")
|
|
end_time = time.time()
|
|
_duration = end_time - start_time
|
|
asyncio.create_task(
|
|
self.service_logger_obj.async_service_success_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
call_type=f"async_set_cache <- {_get_call_stack_info()}",
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs),
|
|
event_metadata={"key": key},
|
|
)
|
|
)
|
|
return result
|
|
except Exception as e:
|
|
end_time = time.time()
|
|
_duration = end_time - start_time
|
|
asyncio.create_task(
|
|
self.service_logger_obj.async_service_failure_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
error=e,
|
|
call_type=f"async_set_cache <- {_get_call_stack_info()}",
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs),
|
|
event_metadata={"key": key},
|
|
)
|
|
)
|
|
verbose_logger.error(
|
|
"LiteLLM Redis Caching: async set() - Got exception from REDIS %s, Writing value=%s",
|
|
str(e),
|
|
value,
|
|
)
|
|
_record_swallowed_redis_failure(self._circuit_breaker, e)
|
|
|
|
async def _pipeline_helper(
|
|
self,
|
|
pipe: pipeline | cluster_pipeline,
|
|
cache_list: list[tuple[Any, Any]],
|
|
ttl: float | None,
|
|
) -> list:
|
|
"""
|
|
Helper function for executing a pipeline of set operations on Redis
|
|
"""
|
|
ttl = self.get_ttl(ttl=ttl)
|
|
# Iterate through each key-value pair in the cache_list and set them in the pipeline.
|
|
for cache_key, cache_value in cache_list:
|
|
cache_key = self.check_and_fix_namespace(key=cache_key)
|
|
print_verbose(f"Set ASYNC Redis Cache PIPELINE: key: {cache_key}\nValue {cache_value}\nttl={ttl}")
|
|
json_cache_value = json.dumps(cache_value)
|
|
# Set the value with a TTL if it's provided.
|
|
_td: timedelta | None = None
|
|
if ttl is not None:
|
|
_td = timedelta(seconds=ttl)
|
|
pipe.set(
|
|
name=cache_key,
|
|
value=json_cache_value,
|
|
ex=_td,
|
|
)
|
|
# Execute the pipeline and return the results.
|
|
results: Final = await pipe.execute()
|
|
return results
|
|
|
|
@_redis_circuit_breaker_guard
|
|
async def async_set_cache_pipeline(self, cache_list: list[tuple[Any, Any]], ttl: float | None = None, **kwargs):
|
|
"""
|
|
Use Redis Pipelines for bulk write operations
|
|
"""
|
|
# don't waste a network request if there's nothing to set
|
|
if len(cache_list) == 0:
|
|
return
|
|
|
|
_redis_client: Final = self.init_async_client()
|
|
start_time: Final = time.time()
|
|
|
|
print_verbose(f"Set Async Redis Cache: key list: {cache_list}\nttl={ttl}, redis_version={self.redis_version}")
|
|
cache_value: Final[Any] = None
|
|
try:
|
|
async with _redis_client.pipeline(transaction=False) as pipe:
|
|
results: Final = await self._pipeline_helper(pipe, cache_list, ttl)
|
|
|
|
print_verbose(f"pipeline results: {results}")
|
|
# Optionally, you could process 'results' to make sure that all set operations were successful.
|
|
## LOGGING ##
|
|
end_time = time.time()
|
|
_duration = end_time - start_time
|
|
asyncio.create_task(
|
|
self.service_logger_obj.async_service_success_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
call_type=f"async_set_cache_pipeline <- {_get_call_stack_info()}",
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs),
|
|
)
|
|
)
|
|
return
|
|
except Exception as e:
|
|
## LOGGING ##
|
|
end_time = time.time()
|
|
_duration = end_time - start_time
|
|
asyncio.create_task(
|
|
self.service_logger_obj.async_service_failure_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
error=e,
|
|
call_type=f"async_set_cache_pipeline <- {_get_call_stack_info()}",
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs),
|
|
)
|
|
)
|
|
|
|
verbose_logger.error(
|
|
"LiteLLM Redis Caching: async set_cache_pipeline() - Got exception from REDIS %s, Writing value=%s",
|
|
str(e),
|
|
cache_value,
|
|
)
|
|
_record_swallowed_redis_failure(self._circuit_breaker, e)
|
|
|
|
async def _set_cache_sadd_helper(
|
|
self,
|
|
redis_client: async_redis_client,
|
|
key: str,
|
|
value: list,
|
|
ttl: float | None,
|
|
) -> None:
|
|
"""Helper function for async_set_cache_sadd. Separated for testing."""
|
|
ttl = self.get_ttl(ttl=ttl)
|
|
try:
|
|
await redis_client.sadd(key, *value)
|
|
if ttl is not None:
|
|
_td: Final = timedelta(seconds=ttl)
|
|
await redis_client.expire(key, _td)
|
|
except Exception:
|
|
raise
|
|
|
|
@_redis_circuit_breaker_guard
|
|
async def async_set_cache_sadd(self, key, value: list, ttl: float | None, **kwargs):
|
|
from redis.asyncio import Redis
|
|
|
|
start_time: Final = time.time()
|
|
try:
|
|
_redis_client: Final[Redis] = self.init_async_client()
|
|
except Exception as e:
|
|
end_time = time.time()
|
|
_duration = end_time - start_time
|
|
asyncio.create_task(
|
|
self.service_logger_obj.async_service_failure_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
error=e,
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs),
|
|
call_type=f"async_set_cache_sadd <- {_get_call_stack_info()}",
|
|
)
|
|
)
|
|
# NON blocking - notify users Redis is throwing an exception
|
|
verbose_logger.error(
|
|
"LiteLLM Redis Caching: async set() - Got exception from REDIS %s, Writing value=%s",
|
|
str(e),
|
|
value,
|
|
)
|
|
raise e
|
|
|
|
key = self.check_and_fix_namespace(key=key)
|
|
print_verbose(f"Set ASYNC Redis Cache: key: {key}\nValue {value}\nttl={ttl}")
|
|
try:
|
|
await self._set_cache_sadd_helper(redis_client=_redis_client, key=key, value=value, ttl=ttl)
|
|
print_verbose(f"Successfully Set ASYNC Redis Cache SADD: key: {key}\nValue {value}\nttl={ttl}")
|
|
end_time = time.time()
|
|
_duration = end_time - start_time
|
|
asyncio.create_task(
|
|
self.service_logger_obj.async_service_success_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
call_type=f"async_set_cache_sadd <- {_get_call_stack_info()}",
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs),
|
|
)
|
|
)
|
|
except Exception as e:
|
|
end_time = time.time()
|
|
_duration = end_time - start_time
|
|
asyncio.create_task(
|
|
self.service_logger_obj.async_service_failure_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
error=e,
|
|
call_type=f"async_set_cache_sadd <- {_get_call_stack_info()}",
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs),
|
|
)
|
|
)
|
|
# NON blocking - notify users Redis is throwing an exception
|
|
verbose_logger.error(
|
|
"LiteLLM Redis Caching: async set_cache_sadd() - Got exception from REDIS %s, Writing value=%s",
|
|
str(e),
|
|
value,
|
|
)
|
|
_record_swallowed_redis_failure(self._circuit_breaker, e)
|
|
|
|
@_redis_circuit_breaker_guard
|
|
async def batch_cache_write(self, key, value, **kwargs):
|
|
print_verbose(
|
|
f"in batch cache writing for redis buffer size={len(self.redis_batch_writing_buffer)}",
|
|
)
|
|
key = self.check_and_fix_namespace(key=key)
|
|
self.redis_batch_writing_buffer.append((key, value))
|
|
if len(self.redis_batch_writing_buffer) >= self.redis_flush_size:
|
|
await self.flush_cache_buffer() # logging done in here
|
|
|
|
@_redis_circuit_breaker_guard
|
|
async def async_increment(
|
|
self,
|
|
key,
|
|
value: float,
|
|
ttl: int | None = None,
|
|
parent_otel_span: Span | None = None,
|
|
refresh_ttl: bool = False,
|
|
) -> float:
|
|
from redis.asyncio import Redis
|
|
|
|
_redis_client: Final[Redis] = self.init_async_client()
|
|
start_time: Final = time.time()
|
|
_used_ttl: Final = self.get_ttl(ttl=ttl)
|
|
key = self.check_and_fix_namespace(key=key)
|
|
try:
|
|
result: Final = await _redis_client.incrbyfloat(name=key, amount=value)
|
|
if _used_ttl is not None:
|
|
if refresh_ttl:
|
|
await _redis_client.expire(key, _used_ttl)
|
|
else:
|
|
current_ttl: Final = await _redis_client.ttl(key)
|
|
if current_ttl == -1:
|
|
await _redis_client.expire(key, _used_ttl)
|
|
|
|
## LOGGING ##
|
|
end_time = time.time()
|
|
_duration = end_time - start_time
|
|
|
|
asyncio.create_task(
|
|
self.service_logger_obj.async_service_success_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
call_type=f"async_increment <- {_get_call_stack_info()}",
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
parent_otel_span=parent_otel_span,
|
|
)
|
|
)
|
|
return result
|
|
except Exception as e:
|
|
## LOGGING ##
|
|
end_time = time.time()
|
|
_duration = end_time - start_time
|
|
asyncio.create_task(
|
|
self.service_logger_obj.async_service_failure_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
error=e,
|
|
call_type=f"async_increment <- {_get_call_stack_info()}",
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
parent_otel_span=parent_otel_span,
|
|
)
|
|
)
|
|
verbose_logger.error(
|
|
"LiteLLM Redis Caching: async async_increment() - Got exception from REDIS %s, Writing value=%s",
|
|
str(e),
|
|
value,
|
|
)
|
|
raise e
|
|
|
|
@_redis_circuit_breaker_guard
|
|
async def async_set_max(
|
|
self,
|
|
key: str,
|
|
value: float,
|
|
ttl: int | None = None,
|
|
) -> float | None:
|
|
"""Atomically set ``key`` to ``value`` only when ``value`` is greater
|
|
than the stored value (or the key is unset), refreshing the TTL.
|
|
|
|
Monotonic by construction: it never lowers the stored value, so a repair
|
|
that writes an authoritative-but-slightly-stale total cannot clobber a
|
|
concurrent increment that has already pushed the counter higher. The
|
|
GET/compare/SET runs in a single Lua call, so it is also atomic across
|
|
racing callers and pods. Returns the resulting value.
|
|
"""
|
|
_redis_client: Final = self.init_async_client()
|
|
_used_ttl: Final = self.get_ttl(ttl=ttl)
|
|
key = self.check_and_fix_namespace(key=key)
|
|
lua: Final = (
|
|
"local cur = redis.call('GET', KEYS[1]) "
|
|
"if cur == false or tonumber(cur) < tonumber(ARGV[1]) then "
|
|
"redis.call('SET', KEYS[1], ARGV[1]) "
|
|
"if tonumber(ARGV[2]) > 0 then redis.call('EXPIRE', KEYS[1], ARGV[2]) end "
|
|
"return ARGV[1] end "
|
|
"return cur"
|
|
)
|
|
result = cast(
|
|
"str | bytes | int | float | None",
|
|
await _redis_client.eval(lua, 1, key, str(value), str(int(_used_ttl or 0))),
|
|
)
|
|
if result is None:
|
|
return None
|
|
if isinstance(result, bytes):
|
|
result = result.decode()
|
|
return float(result)
|
|
|
|
async def flush_cache_buffer(self):
|
|
print_verbose(f"flushing to redis....reached size of buffer {len(self.redis_batch_writing_buffer)}")
|
|
await self.async_set_cache_pipeline(self.redis_batch_writing_buffer)
|
|
self.redis_batch_writing_buffer = []
|
|
|
|
def _get_cache_logic(self, cached_response: Any):
|
|
"""
|
|
Common 'get_cache_logic' across sync + async redis client implementations
|
|
"""
|
|
if cached_response is None:
|
|
return cached_response
|
|
# cached_response is in `b{} convert it to ModelResponse
|
|
cached_response = cached_response.decode("utf-8") # Convert bytes to string
|
|
try:
|
|
cached_response = json.loads(cached_response) # Convert string to dictionary
|
|
except Exception:
|
|
cached_response = ast.literal_eval(cached_response)
|
|
return cached_response
|
|
|
|
def get_cache(self, key, parent_otel_span: Span | None = None, **kwargs):
|
|
try:
|
|
key = self.check_and_fix_namespace(key=key)
|
|
print_verbose(f"Get Redis Cache: key: {key}")
|
|
start_time: Final = time.time()
|
|
cached_response: Final = self.redis_client.get(key)
|
|
end_time: Final = time.time()
|
|
_duration: Final = end_time - start_time
|
|
self.service_logger_obj.service_success_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
call_type=f"get_cache <- {_get_call_stack_info()}",
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
parent_otel_span=parent_otel_span,
|
|
)
|
|
print_verbose(f"Got Redis Cache: key: {key}, cached_response {cached_response}")
|
|
return self._get_cache_logic(cached_response=cached_response)
|
|
except Exception as e:
|
|
# NON blocking - notify users Redis is throwing an exception
|
|
verbose_logger.error("litellm.caching.caching: get() - Got exception from REDIS: ", e)
|
|
|
|
def _run_redis_mget_operation(self, keys: list[str]) -> list[Any]:
|
|
"""
|
|
Wrapper to call `mget` on the redis client
|
|
|
|
We use a wrapper so RedisCluster can override this method
|
|
"""
|
|
return self.redis_client.mget(keys=keys)
|
|
|
|
async def _async_run_redis_mget_operation(self, keys: list[str]) -> list[Any]:
|
|
"""
|
|
Wrapper to call `mget` on the redis client
|
|
|
|
We use a wrapper so RedisCluster can override this method
|
|
"""
|
|
async_redis_client: Final = self.init_async_client()
|
|
return await async_redis_client.mget(keys=keys)
|
|
|
|
def batch_get_cache(
|
|
self,
|
|
key_list: list[str] | list[str | None],
|
|
parent_otel_span: Span | None = None,
|
|
) -> dict:
|
|
"""
|
|
Use Redis for bulk read operations
|
|
|
|
Args:
|
|
key_list: List of keys to get from Redis
|
|
parent_otel_span: Optional parent OpenTelemetry span
|
|
|
|
Returns:
|
|
dict: A dictionary mapping keys to their cached values
|
|
"""
|
|
key_value_dict = {}
|
|
_key_list: Final = [key for key in key_list if key is not None]
|
|
|
|
try:
|
|
_keys: Final = []
|
|
for cache_key in _key_list:
|
|
cache_key = self.check_and_fix_namespace(key=cache_key or "")
|
|
_keys.append(cache_key)
|
|
start_time: Final = time.time()
|
|
results: Final[list] = self._run_redis_mget_operation(keys=_keys)
|
|
end_time: Final = time.time()
|
|
_duration: Final = end_time - start_time
|
|
self.service_logger_obj.service_success_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
call_type=f"batch_get_cache <- {_get_call_stack_info()}",
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
parent_otel_span=parent_otel_span,
|
|
)
|
|
|
|
# Associate the results back with their keys.
|
|
# 'results' is a list of values corresponding to the order of keys in '_key_list'.
|
|
key_value_dict = dict(zip(_key_list, results))
|
|
|
|
decoded_results: Final = {}
|
|
for k, v in key_value_dict.items():
|
|
if isinstance(k, bytes):
|
|
k = k.decode("utf-8")
|
|
v = self._get_cache_logic(v)
|
|
decoded_results[k] = v
|
|
|
|
return decoded_results
|
|
except Exception as e:
|
|
verbose_logger.error("Error occurred in batch get cache - %s", e)
|
|
return key_value_dict
|
|
|
|
@_redis_circuit_breaker_guard
|
|
async def async_get_cache(self, key, parent_otel_span: Span | None = None, **kwargs):
|
|
from redis.asyncio import Redis
|
|
|
|
_redis_client: Final[Redis] = self.init_async_client()
|
|
key = self.check_and_fix_namespace(key=key)
|
|
start_time: Final = time.time()
|
|
|
|
try:
|
|
print_verbose(f"Get Async Redis Cache: key: {key}")
|
|
cached_response: Final = await _redis_client.get(key)
|
|
print_verbose(f"Got Async Redis Cache: key: {key}, cached_response {cached_response}")
|
|
response: Final = self._get_cache_logic(cached_response=cached_response)
|
|
|
|
end_time = time.time()
|
|
_duration = end_time - start_time
|
|
asyncio.create_task(
|
|
self.service_logger_obj.async_service_success_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
call_type=f"async_get_cache <- {_get_call_stack_info()}",
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
parent_otel_span=parent_otel_span,
|
|
event_metadata={"key": key},
|
|
)
|
|
)
|
|
return response
|
|
except Exception as e:
|
|
end_time = time.time()
|
|
_duration = end_time - start_time
|
|
asyncio.create_task(
|
|
self.service_logger_obj.async_service_failure_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
error=e,
|
|
call_type=f"async_get_cache <- {_get_call_stack_info()}",
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
parent_otel_span=parent_otel_span,
|
|
event_metadata={"key": key},
|
|
)
|
|
)
|
|
print_verbose(f"litellm.caching.caching: async get() - Got exception from REDIS: {e}")
|
|
_record_swallowed_redis_failure(self._circuit_breaker, e)
|
|
|
|
@_redis_circuit_breaker_guard
|
|
async def async_batch_get_cache(
|
|
self,
|
|
key_list: list[str] | list[str | None],
|
|
parent_otel_span: Span | None = None,
|
|
) -> dict:
|
|
"""
|
|
Use Redis for bulk read operations
|
|
|
|
Args:
|
|
key_list: List of keys to get from Redis
|
|
parent_otel_span: Optional parent OpenTelemetry span
|
|
|
|
Returns:
|
|
dict: A dictionary mapping keys to their cached values
|
|
|
|
`.mget` does not support None keys. This will filter out None keys.
|
|
"""
|
|
# typed as Any, redis python lib has incomplete type stubs for RedisCluster and does not include `mget`
|
|
key_value_dict = {}
|
|
start_time: Final = time.time()
|
|
_key_list: Final = [key for key in key_list if key is not None]
|
|
try:
|
|
_keys: Final = []
|
|
for cache_key in _key_list:
|
|
cache_key = self.check_and_fix_namespace(key=cache_key)
|
|
_keys.append(cache_key)
|
|
results: Final = await self._async_run_redis_mget_operation(keys=_keys)
|
|
## LOGGING ##
|
|
end_time = time.time()
|
|
_duration = end_time - start_time
|
|
asyncio.create_task(
|
|
self.service_logger_obj.async_service_success_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
call_type=f"async_batch_get_cache <- {_get_call_stack_info()}",
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
parent_otel_span=parent_otel_span,
|
|
)
|
|
)
|
|
|
|
# Associate the results back with their keys.
|
|
# 'results' is a list of values corresponding to the order of keys in 'key_list'.
|
|
key_value_dict = dict(zip(_key_list, results))
|
|
|
|
decoded_results: Final = {}
|
|
for k, v in key_value_dict.items():
|
|
if isinstance(k, bytes):
|
|
k = k.decode("utf-8")
|
|
v = self._get_cache_logic(v)
|
|
decoded_results[k] = v
|
|
|
|
return decoded_results
|
|
except Exception as e:
|
|
## LOGGING ##
|
|
end_time = time.time()
|
|
_duration = end_time - start_time
|
|
asyncio.create_task(
|
|
self.service_logger_obj.async_service_failure_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
error=e,
|
|
call_type=f"async_batch_get_cache <- {_get_call_stack_info()}",
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
parent_otel_span=parent_otel_span,
|
|
)
|
|
)
|
|
verbose_logger.error("Error occurred in async batch get cache - %s", e)
|
|
_record_swallowed_redis_failure(self._circuit_breaker, e)
|
|
return key_value_dict
|
|
|
|
def sync_ping(self) -> bool:
|
|
"""
|
|
Tests if the sync redis client is correctly setup.
|
|
"""
|
|
print_verbose("Pinging Sync Redis Cache")
|
|
start_time: Final = time.time()
|
|
try:
|
|
response: Final[bool] = self.redis_client.ping()
|
|
print_verbose(f"Redis Cache PING: {response}")
|
|
## LOGGING ##
|
|
end_time = time.time()
|
|
_duration = end_time - start_time
|
|
self.service_logger_obj.service_success_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
call_type=f"sync_ping <- {_get_call_stack_info()}",
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
)
|
|
return response
|
|
except Exception as e:
|
|
# NON blocking - notify users Redis is throwing an exception
|
|
## LOGGING ##
|
|
end_time = time.time()
|
|
_duration = end_time - start_time
|
|
self.service_logger_obj.service_failure_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
error=e,
|
|
call_type=f"sync_ping <- {_get_call_stack_info()}",
|
|
)
|
|
verbose_logger.error("LiteLLM Redis Cache PING: - Got exception from REDIS : %s", e)
|
|
raise e
|
|
|
|
async def ping(self) -> bool:
|
|
# typed as Any, redis python lib has incomplete type stubs for RedisCluster and does not include `ping`
|
|
_redis_client: Final[Any] = self.init_async_client()
|
|
start_time: Final = time.time()
|
|
print_verbose("Pinging Async Redis Cache")
|
|
try:
|
|
response: Final = await _redis_client.ping()
|
|
## LOGGING ##
|
|
end_time = time.time()
|
|
_duration = end_time - start_time
|
|
asyncio.create_task(
|
|
self.service_logger_obj.async_service_success_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
call_type=f"async_ping <- {_get_call_stack_info()}",
|
|
)
|
|
)
|
|
return response
|
|
except Exception as e:
|
|
# NON blocking - notify users Redis is throwing an exception
|
|
## LOGGING ##
|
|
end_time = time.time()
|
|
_duration = end_time - start_time
|
|
asyncio.create_task(
|
|
self.service_logger_obj.async_service_failure_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
error=e,
|
|
call_type=f"async_ping <- {_get_call_stack_info()}",
|
|
)
|
|
)
|
|
verbose_logger.error("LiteLLM Redis Cache PING: - Got exception from REDIS : %s", e)
|
|
raise e
|
|
|
|
@_redis_circuit_breaker_guard
|
|
async def delete_cache_keys(self, keys):
|
|
# typed as Any, redis python lib has incomplete type stubs for RedisCluster and does not include `delete`
|
|
_redis_client: Final[Any] = self.init_async_client()
|
|
keys = [self.check_and_fix_namespace(key=key) for key in keys]
|
|
# keys is a list, unpack it so it gets passed as individual elements to delete
|
|
await _redis_client.delete(*keys)
|
|
|
|
def client_list(self) -> list:
|
|
client_list: Final[list] = self.redis_client.client_list()
|
|
return client_list
|
|
|
|
def info(self):
|
|
info: Final = self.redis_client.info()
|
|
return info
|
|
|
|
def flush_cache(self):
|
|
self.redis_client.flushall()
|
|
|
|
def flushall(self):
|
|
self.redis_client.flushall()
|
|
|
|
async def disconnect(self):
|
|
await self.async_redis_conn_pool.disconnect(inuse_connections=True)
|
|
try:
|
|
self.redis_client.close()
|
|
except Exception as e:
|
|
verbose_logger.debug("Error closing sync Redis client: %s", e)
|
|
|
|
async def test_connection(self) -> dict:
|
|
"""
|
|
Test the Redis connection by creating a new client and pinging it.
|
|
|
|
This creates a fresh connection without using cached clients or connection pools
|
|
to ensure the credentials are actually valid.
|
|
|
|
Returns:
|
|
dict: {"status": "success" | "failed", "message": str, "error": Optional[str]}
|
|
"""
|
|
try:
|
|
import redis.asyncio as redis_async
|
|
|
|
# Create a fresh Redis client with current settings
|
|
redis_client: Final = redis_async.Redis(**self.redis_kwargs)
|
|
|
|
# Test the connection
|
|
ping_result: Final = await redis_client.ping()
|
|
|
|
# Close the connection
|
|
await redis_client.aclose()
|
|
|
|
if ping_result:
|
|
return {
|
|
"status": "success",
|
|
"message": "Redis connection test successful",
|
|
}
|
|
else:
|
|
return {"status": "failed", "message": "Redis ping returned False"}
|
|
except Exception as e:
|
|
verbose_logger.error("Redis connection test failed: %s", e)
|
|
return {
|
|
"status": "failed",
|
|
"message": f"Redis connection failed: {e}",
|
|
"error": str(e),
|
|
}
|
|
|
|
@_redis_circuit_breaker_guard
|
|
async def async_delete_cache(self, key: str):
|
|
# typed as Any, redis python lib has incomplete type stubs for RedisCluster and does not include `delete`
|
|
_redis_client: Final[Any] = self.init_async_client()
|
|
key = self.check_and_fix_namespace(key=key)
|
|
# keys is str
|
|
return await _redis_client.delete(key)
|
|
|
|
def delete_cache(self, key):
|
|
key = self.check_and_fix_namespace(key=key)
|
|
self.redis_client.delete(key)
|
|
|
|
async def _pipeline_increment_helper(
|
|
self,
|
|
pipe: pipeline,
|
|
increment_list: list[RedisPipelineIncrementOperation],
|
|
) -> list[float] | None:
|
|
"""Helper function for pipeline increment operations"""
|
|
# Iterate through each increment operation and add commands to pipeline
|
|
for increment_op in increment_list:
|
|
cache_key = self.check_and_fix_namespace(key=increment_op["key"])
|
|
print_verbose(
|
|
f"Increment ASYNC Redis Cache PIPELINE: key: {cache_key}\nValue {increment_op['increment_value']}\nttl={increment_op['ttl']}"
|
|
)
|
|
pipe.incrbyfloat(cache_key, increment_op["increment_value"])
|
|
if increment_op["ttl"] is not None:
|
|
_td = timedelta(seconds=increment_op["ttl"])
|
|
pipe.expire(cache_key, _td)
|
|
# Execute the pipeline and return results
|
|
results: Final = await pipe.execute()
|
|
# only return float values
|
|
verbose_logger.debug("Increment ASYNC Redis Cache PIPELINE: results: %s", results)
|
|
return [r for r in results if isinstance(r, float)]
|
|
|
|
@_redis_circuit_breaker_guard
|
|
async def async_increment_pipeline(
|
|
self, increment_list: list[RedisPipelineIncrementOperation], **kwargs
|
|
) -> list[float] | None:
|
|
"""
|
|
Use Redis Pipelines for bulk increment operations
|
|
Args:
|
|
increment_list: List of RedisPipelineIncrementOperation dicts containing:
|
|
- key: str
|
|
- increment_value: float
|
|
- ttl_seconds: int
|
|
"""
|
|
# don't waste a network request if there's nothing to increment
|
|
if len(increment_list) == 0:
|
|
return None
|
|
|
|
from redis.asyncio import Redis
|
|
|
|
_redis_client: Final[Redis] = self.init_async_client()
|
|
start_time: Final = time.time()
|
|
|
|
print_verbose(f"Increment Async Redis Cache Pipeline: increment list: {increment_list}")
|
|
|
|
try:
|
|
async with _redis_client.pipeline(transaction=False) as pipe:
|
|
results: Final = await self._pipeline_increment_helper(pipe, increment_list)
|
|
|
|
## LOGGING ##
|
|
end_time = time.time()
|
|
_duration = end_time - start_time
|
|
asyncio.create_task(
|
|
self.service_logger_obj.async_service_success_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
call_type=f"async_increment_pipeline <- {_get_call_stack_info()}",
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs),
|
|
)
|
|
)
|
|
return results
|
|
except Exception as e:
|
|
## LOGGING ##
|
|
end_time = time.time()
|
|
_duration = end_time - start_time
|
|
asyncio.create_task(
|
|
self.service_logger_obj.async_service_failure_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
error=e,
|
|
call_type=f"async_increment_pipeline <- {_get_call_stack_info()}",
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs),
|
|
)
|
|
)
|
|
verbose_logger.error(
|
|
"LiteLLM Redis Caching: async increment_pipeline() - Got exception from REDIS %s",
|
|
str(e),
|
|
)
|
|
raise e
|
|
|
|
@_redis_circuit_breaker_guard
|
|
async def async_get_ttl(self, key: str) -> int | None:
|
|
"""
|
|
Get the remaining TTL of a key in Redis
|
|
|
|
Args:
|
|
key (str): The key to get TTL for
|
|
|
|
Returns:
|
|
Optional[int]: The remaining TTL in seconds, or None if key doesn't exist
|
|
|
|
Redis ref: https://redis.io/docs/latest/commands/ttl/
|
|
"""
|
|
try:
|
|
# typed as Any, redis python lib has incomplete type stubs for RedisCluster and does not include `ttl`
|
|
_redis_client: Final[Any] = self.init_async_client()
|
|
key = self.check_and_fix_namespace(key=key)
|
|
ttl: Final = await _redis_client.ttl(key)
|
|
if ttl <= -1: # -1 means the key does not exist, -2 key does not exist
|
|
return None
|
|
return ttl
|
|
except Exception as e:
|
|
verbose_logger.debug("Redis TTL Error: %s", e)
|
|
_record_swallowed_redis_failure(self._circuit_breaker, e)
|
|
return None
|
|
|
|
@_redis_circuit_breaker_guard
|
|
async def async_rpush(
|
|
self,
|
|
key: str,
|
|
values: list[Any],
|
|
parent_otel_span: Span | None = None,
|
|
**kwargs,
|
|
) -> int:
|
|
"""
|
|
Append one or multiple values to a list stored at key
|
|
|
|
Args:
|
|
key: The Redis key of the list
|
|
values: One or more values to append to the list
|
|
parent_otel_span: Optional parent OpenTelemetry span
|
|
|
|
Returns:
|
|
int: The length of the list after the push operation
|
|
"""
|
|
_redis_client: Final[Any] = self.init_async_client()
|
|
key = self.check_and_fix_namespace(key=key)
|
|
start_time: Final = time.time()
|
|
try:
|
|
response: Final = await _redis_client.rpush(key, *values)
|
|
## LOGGING ##
|
|
end_time = time.time()
|
|
_duration = end_time - start_time
|
|
asyncio.create_task(
|
|
self.service_logger_obj.async_service_success_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
call_type=f"async_rpush <- {_get_call_stack_info()}",
|
|
)
|
|
)
|
|
return response
|
|
except Exception as e:
|
|
# NON blocking - notify users Redis is throwing an exception
|
|
## LOGGING ##
|
|
end_time = time.time()
|
|
_duration = end_time - start_time
|
|
asyncio.create_task(
|
|
self.service_logger_obj.async_service_failure_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
error=e,
|
|
call_type=f"async_rpush <- {_get_call_stack_info()}",
|
|
)
|
|
)
|
|
verbose_logger.error("LiteLLM Redis Cache RPUSH: - Got exception from REDIS : %s", e)
|
|
raise e
|
|
|
|
async def _pipeline_rpush_helper(
|
|
self,
|
|
pipe: pipeline,
|
|
rpush_list: list[RedisPipelineRpushOperation],
|
|
) -> list[int]:
|
|
"""Helper function for pipeline rpush operations"""
|
|
for rpush_op in rpush_list:
|
|
key = self.check_and_fix_namespace(key=rpush_op["key"])
|
|
pipe.rpush(key, *rpush_op["values"])
|
|
results: Final = await pipe.execute()
|
|
# Preserve positional correspondence — raise on per-command errors
|
|
for r in results:
|
|
if isinstance(r, Exception):
|
|
raise r
|
|
return results
|
|
|
|
@_redis_circuit_breaker_guard
|
|
async def async_rpush_pipeline(
|
|
self,
|
|
rpush_list: list[RedisPipelineRpushOperation],
|
|
) -> list[int]:
|
|
"""
|
|
Use Redis Pipelines for bulk RPUSH operations
|
|
|
|
Args:
|
|
rpush_list: List of RedisPipelineRpushOperation dicts containing:
|
|
- key: str
|
|
- values: List[Any]
|
|
|
|
Returns:
|
|
List[int]: List lengths after each push
|
|
"""
|
|
if len(rpush_list) == 0:
|
|
return []
|
|
|
|
_redis_client: Final[Any] = self.init_async_client()
|
|
start_time: Final = time.time()
|
|
|
|
try:
|
|
async with _redis_client.pipeline(transaction=False) as pipe:
|
|
results: Final = await self._pipeline_rpush_helper(pipe, rpush_list)
|
|
|
|
## LOGGING ##
|
|
end_time = time.time()
|
|
_duration = end_time - start_time
|
|
asyncio.create_task(
|
|
self.service_logger_obj.async_service_success_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
call_type=f"async_rpush_pipeline <- {_get_call_stack_info()}",
|
|
)
|
|
)
|
|
return results
|
|
except Exception as e:
|
|
## LOGGING ##
|
|
end_time = time.time()
|
|
_duration = end_time - start_time
|
|
asyncio.create_task(
|
|
self.service_logger_obj.async_service_failure_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
error=e,
|
|
call_type=f"async_rpush_pipeline <- {_get_call_stack_info()}",
|
|
)
|
|
)
|
|
verbose_logger.error(
|
|
"LiteLLM Redis Caching: async_rpush_pipeline() - Got exception from REDIS %s",
|
|
str(e),
|
|
)
|
|
raise e
|
|
|
|
async def handle_lpop_count_for_older_redis_versions(self, pipe: pipeline, key: str, count: int) -> list[bytes]:
|
|
result: Final[list[bytes]] = []
|
|
for _ in range(count):
|
|
pipe.lpop(key)
|
|
results = await pipe.execute()
|
|
|
|
# Filter out None values and decode bytes
|
|
for r in results:
|
|
if r is not None:
|
|
result.append(r)
|
|
|
|
return result
|
|
|
|
@_redis_circuit_breaker_guard
|
|
async def async_lpop(
|
|
self,
|
|
key: str,
|
|
count: int | None = None,
|
|
parent_otel_span: Span | None = None,
|
|
**kwargs,
|
|
) -> Any | list[Any]:
|
|
_redis_client: Final[Any] = self.init_async_client()
|
|
key = self.check_and_fix_namespace(key=key)
|
|
start_time: Final = time.time()
|
|
print_verbose(f"LPOP from Redis list: key: {key}, count: {count}")
|
|
try:
|
|
major_version: Final = self._parse_redis_major_version()
|
|
|
|
if count is not None and major_version < 7:
|
|
# For Redis < 7.0, use pipeline to execute multiple LPOP commands
|
|
async with _redis_client.pipeline(transaction=False) as pipe:
|
|
result = await self.handle_lpop_count_for_older_redis_versions(pipe, key, count)
|
|
else:
|
|
# For Redis >= 7.0 or when count is None, use native LPOP with count
|
|
result = await _redis_client.lpop(key, count)
|
|
|
|
## LOGGING ##
|
|
end_time = time.time()
|
|
_duration = end_time - start_time
|
|
asyncio.create_task(
|
|
self.service_logger_obj.async_service_success_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
call_type=f"async_lpop <- {_get_call_stack_info()}",
|
|
)
|
|
)
|
|
|
|
# Handle result parsing if needed
|
|
if isinstance(result, bytes):
|
|
try:
|
|
return result.decode("utf-8")
|
|
except Exception:
|
|
return result
|
|
elif isinstance(result, list) and all(isinstance(item, bytes) for item in result):
|
|
try:
|
|
return [item.decode("utf-8") for item in result]
|
|
except Exception:
|
|
return result
|
|
return result
|
|
except Exception as e:
|
|
# NON blocking - notify users Redis is throwing an exception
|
|
## LOGGING ##
|
|
end_time = time.time()
|
|
_duration = end_time - start_time
|
|
asyncio.create_task(
|
|
self.service_logger_obj.async_service_failure_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
error=e,
|
|
call_type=f"async_lpop <- {_get_call_stack_info()}",
|
|
)
|
|
)
|
|
verbose_logger.error("LiteLLM Redis Cache LPOP: - Got exception from REDIS : %s", e)
|
|
raise e
|
|
|
|
async def _pipeline_lpop_helper(
|
|
self,
|
|
pipe: pipeline,
|
|
lpop_list: list[RedisPipelineLpopOperation],
|
|
) -> list[list[str] | None]:
|
|
"""Helper function for pipeline lpop operations.
|
|
|
|
For Redis >= 7, queues one LPOP(key, count) per operation.
|
|
For Redis < 7, queues `count` individual LPOP(key) commands per operation.
|
|
"""
|
|
major_version: Final = self._parse_redis_major_version()
|
|
|
|
if major_version >= 7:
|
|
for lpop_op in lpop_list:
|
|
key = self.check_and_fix_namespace(key=lpop_op["key"])
|
|
pipe.lpop(key, lpop_op["count"])
|
|
raw_results = await pipe.execute()
|
|
else:
|
|
# For Redis < 7, LPOP doesn't support count param.
|
|
# Issue `count` individual LPOP commands per key, all in one pipeline.
|
|
counts: Final[list[int]] = []
|
|
for lpop_op in lpop_list:
|
|
key = self.check_and_fix_namespace(key=lpop_op["key"])
|
|
count = lpop_op["count"] or 1
|
|
counts.append(count)
|
|
for _ in range(count):
|
|
pipe.lpop(key)
|
|
flat_results: Final = await pipe.execute()
|
|
|
|
# Re-group the flat results back into per-key lists
|
|
raw_results = []
|
|
offset = 0
|
|
for count in counts:
|
|
key_results = [r for r in flat_results[offset : offset + count] if r is not None]
|
|
raw_results.append(key_results if key_results else None)
|
|
offset += count
|
|
|
|
# Raise on per-command errors (matches _pipeline_rpush_helper behavior)
|
|
for r in raw_results:
|
|
if isinstance(r, Exception):
|
|
raise r
|
|
|
|
# Decode bytes -> str for each result set
|
|
decoded_results: Final[list[list[str] | None]] = []
|
|
for r in raw_results:
|
|
if r is None:
|
|
decoded_results.append(None)
|
|
elif isinstance(r, list):
|
|
try:
|
|
decoded_results.append(
|
|
[item.decode("utf-8") if isinstance(item, bytes) else item for item in r if item is not None]
|
|
or None
|
|
)
|
|
except Exception:
|
|
decoded_results.append(r)
|
|
else:
|
|
decoded_results.append(None)
|
|
return decoded_results
|
|
|
|
@_redis_circuit_breaker_guard
|
|
async def async_lpop_pipeline(
|
|
self,
|
|
lpop_list: list[RedisPipelineLpopOperation],
|
|
) -> list[list[str] | None]:
|
|
"""
|
|
Use Redis Pipelines for bulk LPOP operations
|
|
|
|
Args:
|
|
lpop_list: List of RedisPipelineLpopOperation dicts containing:
|
|
- key: str
|
|
- count: Optional[int]
|
|
|
|
Returns:
|
|
List[Optional[List[str]]]: Decoded results per key, None if key was empty
|
|
"""
|
|
if len(lpop_list) == 0:
|
|
return []
|
|
|
|
_redis_client: Final[Any] = self.init_async_client()
|
|
start_time: Final = time.time()
|
|
|
|
try:
|
|
async with _redis_client.pipeline(transaction=False) as pipe:
|
|
results: Final = await self._pipeline_lpop_helper(pipe, lpop_list)
|
|
|
|
## LOGGING ##
|
|
end_time = time.time()
|
|
_duration = end_time - start_time
|
|
asyncio.create_task(
|
|
self.service_logger_obj.async_service_success_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
call_type=f"async_lpop_pipeline <- {_get_call_stack_info()}",
|
|
)
|
|
)
|
|
return results
|
|
except Exception as e:
|
|
## LOGGING ##
|
|
end_time = time.time()
|
|
_duration = end_time - start_time
|
|
asyncio.create_task(
|
|
self.service_logger_obj.async_service_failure_hook(
|
|
service=ServiceTypes.REDIS,
|
|
duration=_duration,
|
|
error=e,
|
|
call_type=f"async_lpop_pipeline <- {_get_call_stack_info()}",
|
|
)
|
|
)
|
|
verbose_logger.error(
|
|
"LiteLLM Redis Caching: async_lpop_pipeline() - Got exception from REDIS %s",
|
|
str(e),
|
|
)
|
|
raise e
|