Implementation:BerriAI Litellm S3 Cache

Attribute	Value
Sources	litellm/caching/s3_cache.py
Domains	Caching, AWS S3, Performance
Last Updated	2026-02-15 16:00 GMT

Overview

S3Cache is a cache backend implementation that stores and retrieves LiteLLM completion responses in Amazon S3, supporting both synchronous and asynchronous operations with TTL-based expiration.

Description

The S3Cache class extends BaseCache and provides four core methods: set_cache, get_cache, async_set_cache, and async_get_cache. Cache keys are converted to S3 object keys by replacing colons with forward slashes and prepending an optional path prefix. Values are serialized to JSON for storage.

When setting cache entries, an optional TTL (time-to-live) can be specified, which sets both the Expires header and Cache-Control directive on the S3 object. When retrieving entries, the class checks the Expires timestamp and returns None for expired entries. The S3 response body is deserialized from JSON (with a fallback to ast.literal_eval) back to a dictionary.

Async operations use asyncio.get_event_loop().run_in_executor to delegate to the synchronous methods in a thread pool, avoiding blocking the event loop. The class also supports batch cache setting via async_set_cache_pipeline using asyncio.gather.

Usage

Import and configure S3Cache when you want to use Amazon S3 as a persistent cache backend for LiteLLM responses. This is useful for caching expensive LLM completions across distributed systems.

Code Reference

Source Location

litellm/caching/s3_cache.py

Signature

class S3Cache(BaseCache):
    def __init__(self, s3_bucket_name, s3_region_name=None, s3_api_version=None,
                 s3_use_ssl: Optional[bool] = True, s3_verify=None, s3_endpoint_url=None,
                 s3_aws_access_key_id=None, s3_aws_secret_access_key=None,
                 s3_aws_session_token=None, s3_config=None, s3_path=None, **kwargs)
    def set_cache(self, key, value, **kwargs)
    def get_cache(self, key, **kwargs)
    async def async_set_cache(self, key, value, **kwargs)
    async def async_get_cache(self, key, **kwargs)
    def flush_cache(self)
    async def disconnect(self)
    async def async_set_cache_pipeline(self, cache_list, **kwargs)

Import

from litellm.caching.s3_cache import S3Cache

I/O Contract

Inputs

Parameter	Type	Description
s3_bucket_name	str	Name of the S3 bucket for caching.
s3_region_name	Optional[str]	AWS region name.
s3_endpoint_url	Optional[str]	Custom endpoint URL (for S3-compatible services like MinIO).
s3_path	Optional[str]	Optional key prefix within the bucket.
key	str	The cache key (colons are converted to forward slashes in S3).
value	Any	The value to cache (JSON-serializable).
ttl (kwarg)	Optional[int]	Time-to-live in seconds for the cache entry.
cache_list	list	List of (key, value) tuples for batch setting.

Outputs

Method	Return Type	Description
set_cache / async_set_cache	None	Stores the value in S3; no return value.
get_cache / async_get_cache	Optional[dict]	The cached value as a dictionary, or None if not found or expired.

Usage Examples

import litellm
from litellm import Cache

# Configure S3 cache
litellm.cache = Cache(
    type="s3",
    s3_bucket_name="my-litellm-cache",
    s3_region_name="us-east-1",
    s3_aws_access_key_id="YOUR_KEY",
    s3_aws_secret_access_key="YOUR_SECRET",
    s3_path="cache/",
)

# Completions are now automatically cached in S3
response = litellm.completion(
    model="gpt-3.5-turbo",
    messages=[{"role": "user", "content": "Hello!"}],
    caching=True,
)

# Subsequent identical calls will return the cached response
response2 = litellm.completion(
    model="gpt-3.5-turbo",
    messages=[{"role": "user", "content": "Hello!"}],
    caching=True,
)

Related Pages

• BerriAI_Litellm_S3_Logger - S3-based logging (different from caching)