mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-06 08:16:43 +00:00
1847 lines
72 KiB
Python
1847 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
|
|
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)
|
|
|
|
|
|
def _opaque_kwarg_key(value: object) -> str:
|
|
return f"{type(value).__name__}-{id(value)}"
|
|
|
|
|
|
@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.
|
|
"""
|
|
# 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, default=_opaque_kwarg_key)
|
|
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 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[str | bytes | int | float],
|
|
client: object = None,
|
|
) -> object:
|
|
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[str | bytes | int | float],
|
|
client: object = None,
|
|
) -> object:
|
|
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[str | bytes | int | float],
|
|
client: object = None,
|
|
) -> object:
|
|
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: Sequence[tuple[str, object]],
|
|
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: Sequence[tuple[str, object]], 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 = 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]) -> Sequence[bytes | str | None]:
|
|
"""
|
|
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]) -> Sequence[bytes | str | None]:
|
|
"""
|
|
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 = 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:
|
|
from .._redis import get_redis_async_client
|
|
|
|
# Create a fresh Redis client with current settings
|
|
redis_client: Final = get_redis_async_client(**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: Sequence[str | bytes | int | float],
|
|
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: Sequence[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: Sequence[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
|