Overview
The Ragas Cache Framework provides a pluggable caching layer for memoizing function results, supporting both synchronous and asynchronous functions with disk-based persistence through the diskcache library.
Description
The cache framework defines an abstract base class (CacheInterface) and a concrete disk-based implementation (DiskCacheBackend) for caching expensive computations in Ragas. It also provides a decorator (cacher) that transparently adds caching behavior to any function.
The framework works by generating SHA-256 hash keys from function arguments (including Pydantic model serialization), checking for cached results before executing the wrapped function, and storing results after execution. It handles Pydantic model pickling issues caused by the instructor library's dynamic class creation by detecting class identity mismatches and recreating objects with the correct class from the module namespace.
Key components include:
- CacheInterface - Abstract base class defining get/set/has_key operations
- DiskCacheBackend - Persistent disk-based cache using the diskcache library
- cacher - Decorator that adds transparent caching to sync and async functions
- _generate_cache_key - Utility that creates deterministic cache keys from function signatures and arguments
- _make_pydantic_picklable - Utility that fixes pickling issues with instructor-generated Pydantic models
Usage
Use this framework when you want to cache expensive LLM or embedding API calls to avoid redundant computation. The DiskCacheBackend persists results between program runs, making it ideal for iterative evaluation workflows. Apply the cacher decorator to any function that makes costly external calls, or use the CacheInterface directly in embedding and LLM wrapper classes.
Code Reference
Source Location
Signature
class CacheInterface(ABC):
@abstractmethod
def get(self, key: str) -> Any: ...
@abstractmethod
def set(self, key: str, value) -> None: ...
@abstractmethod
def has_key(self, key: str) -> bool: ...
class DiskCacheBackend(CacheInterface):
def __init__(self, cache_dir: str = ".cache"): ...
def cacher(cache_backend: Optional[CacheInterface] = None): ...
Import
from ragas.cache import CacheInterface, DiskCacheBackend, cacher
I/O Contract
CacheInterface Methods
get
| Name |
Type |
Required |
Description
|
| key |
str |
Yes |
The key to look up in the cache
|
| Name |
Type |
Description
|
| return |
Any |
The cached value associated with the key, or None if not found
|
set
| Name |
Type |
Required |
Description
|
| key |
str |
Yes |
The key to store the value under
|
| value |
Any |
Yes |
The value to cache
|
| Name |
Type |
Description
|
| return |
None |
No return value
|
has_key
| Name |
Type |
Required |
Description
|
| key |
str |
Yes |
The key to check for existence
|
| Name |
Type |
Description
|
| return |
bool |
True if the key exists in the cache, False otherwise
|
DiskCacheBackend Inputs
| Name |
Type |
Required |
Description
|
| cache_dir |
str |
No |
Directory where cache files will be stored (defaults to ".cache")
|
cacher Decorator Inputs
| Name |
Type |
Required |
Description
|
| cache_backend |
Optional[CacheInterface] |
No |
The cache backend to use; if None, caching is disabled and the original function is returned unchanged
|
cacher Decorator Outputs
| Name |
Type |
Description
|
| return |
Callable |
A decorated function that checks cache before execution and stores results after execution
|
Usage Examples
Basic Usage with DiskCacheBackend
from ragas.cache import DiskCacheBackend
# Create a disk-based cache
cache = DiskCacheBackend(cache_dir=".my_cache")
# Store and retrieve values
cache.set("my_key", {"result": 42})
if cache.has_key("my_key"):
value = cache.get("my_key")
print(value) # {'result': 42}
Using the cacher Decorator
from ragas.cache import DiskCacheBackend, cacher
cache = DiskCacheBackend(cache_dir=".llm_cache")
@cacher(cache_backend=cache)
def expensive_llm_call(prompt: str) -> str:
# This result will be cached on first call
# and returned from cache on subsequent calls with the same prompt
return call_llm_api(prompt)
@cacher(cache_backend=cache)
async def async_expensive_call(prompt: str) -> str:
# Works with async functions too
return await async_call_llm_api(prompt)
Using Cache with Embeddings
from ragas.cache import DiskCacheBackend
from ragas.embeddings import OpenAIEmbeddings
from openai import OpenAI
cache = DiskCacheBackend(cache_dir=".embedding_cache")
client = OpenAI()
embeddings = OpenAIEmbeddings(client=client, cache=cache)
Related Pages