Cache

CacheInterface

Bases: ABC

Abstract base class defining the interface for cache implementations.

This class provides a standard interface that all cache implementations must follow. It supports basic cache operations like get, set and key checking.

get abstractmethod

get(key: str) -> Any

Retrieve a value from the cache by key.

Args: key: The key to look up in the cache.

Returns: The cached value associated with the key.

Source code in src/ragas/cache.py

@abstractmethod
def get(self, key: str) -> Any:
    """Retrieve a value from the cache by key.

    Args:
        key: The key to look up in the cache.

    Returns:
        The cached value associated with the key.
    """
    pass

set abstractmethod

set(key: str, value) -> None

Store a value in the cache with the given key.

Args: key: The key to store the value under. value: The value to cache.

Source code in src/ragas/cache.py

@abstractmethod
def set(self, key: str, value) -> None:
    """Store a value in the cache with the given key.

    Args:
        key: The key to store the value under.
        value: The value to cache.
    """
    pass

has_key abstractmethod

has_key(key: str) -> bool

Check if a key exists in the cache.

Args: key: The key to check for.

Returns: True if the key exists in the cache, False otherwise.

Source code in src/ragas/cache.py

@abstractmethod
def has_key(self, key: str) -> bool:
    """Check if a key exists in the cache.

    Args:
        key: The key to check for.

    Returns:
        True if the key exists in the cache, False otherwise.
    """
    pass

DiskCacheBackend

DiskCacheBackend(cache_dir: str = '.cache')

Bases: CacheInterface

A cache implementation that stores data on disk using the diskcache library.

This cache backend persists data to disk, allowing it to survive between program runs. It implements the CacheInterface for use with Ragas caching functionality.

Args: cache_dir (str, optional): Directory where cache files will be stored. Defaults to ".cache".

Source code in src/ragas/cache.py

def __init__(self, cache_dir: str = ".cache"):
    try:
        from diskcache import Cache
    except ImportError:
        raise ImportError(
            "For using the diskcache backend, please install it with `pip install diskcache`."
        )

    self.cache = Cache(cache_dir)

get

get(key: str) -> Any

Retrieve a value from the disk cache by key.

Args: key: The key to look up in the cache.

Returns: The cached value associated with the key, or None if not found.

Source code in src/ragas/cache.py

def get(self, key: str) -> Any:
    """Retrieve a value from the disk cache by key.

    Args:
        key: The key to look up in the cache.

    Returns:
        The cached value associated with the key, or None if not found.
    """
    return self.cache.get(key)

set

set(key: str, value) -> None

Store a value in the disk cache with the given key.

Args: key: The key to store the value under. value: The value to cache.

Source code in src/ragas/cache.py

def set(self, key: str, value) -> None:
    """Store a value in the disk cache with the given key.

    Args:
        key: The key to store the value under.
        value: The value to cache.
    """
    self.cache.set(key, value)

has_key

has_key(key: str) -> bool

Check if a key exists in the disk cache.

Args: key: The key to check for.

Returns: True if the key exists in the cache, False otherwise.

Source code in src/ragas/cache.py

def has_key(self, key: str) -> bool:
    """Check if a key exists in the disk cache.

    Args:
        key: The key to check for.

    Returns:
        True if the key exists in the cache, False otherwise.
    """
    return key in self.cache

cacher

cacher(cache_backend: Optional[CacheInterface] = None)

Decorator that adds caching functionality to a function.

This decorator can be applied to both synchronous and asynchronous functions to cache their results. If no cache backend is provided, the original function is returned unchanged.

Args: cache_backend (Optional[CacheInterface]): The cache backend to use for storing results. If None, caching is disabled.

Returns: Callable: A decorated function that implements caching behavior.

Source code in src/ragas/cache.py

def cacher(cache_backend: Optional[CacheInterface] = None):
    """Decorator that adds caching functionality to a function.

    This decorator can be applied to both synchronous and asynchronous functions to cache their results.
    If no cache backend is provided, the original function is returned unchanged.

    Args:
        cache_backend (Optional[CacheInterface]): The cache backend to use for storing results.
            If None, caching is disabled.

    Returns:
        Callable: A decorated function that implements caching behavior.
    """

    def decorator(func):
        if cache_backend is None:
            return func

        # hack to make pyright happy
        backend: CacheInterface = cache_backend

        is_async = inspect.iscoroutinefunction(func)

        @functools.wraps(func)
        async def async_wrapper(*args, **kwargs):
            cache_key = _generate_cache_key(func, args, kwargs)

            if backend.has_key(cache_key):
                logger.debug(f"Cache hit for {cache_key}")
                return backend.get(cache_key)

            result = await func(*args, **kwargs)
            backend.set(cache_key, result)
            return result

        @functools.wraps(func)
        def sync_wrapper(*args, **kwargs):
            cache_key = _generate_cache_key(func, args, kwargs)

            if backend.has_key(cache_key):
                logger.debug(f"Cache hit for {cache_key}")
                return backend.get(cache_key)

            result = func(*args, **kwargs)
            backend.set(cache_key, result)
            return result

        return async_wrapper if is_async else sync_wrapper

    return decorator