缓存

CacheInterface

基类: ABC

定义缓存实现接口的抽象基类。

此类提供了一个标准接口，所有缓存实现都必须遵循。它支持基本的缓存操作，如获取(get)、设置(set)和键检查(key checking)。

get abstractmethod

get(key: str) -> Any

按键从缓存中检索值。

参数: key: 要在缓存中查找的键。

返回值: 与键关联的缓存值。

源代码位于 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

使用给定的键在缓存中存储一个值。

参数: key: 用于存储值的键。 value: 要缓存的值。

源代码位于 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

检查缓存中是否存在一个键。

参数: key: 要检查的键。

返回值: 如果缓存中存在该键，则为 True；否则为 False。

源代码位于 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')

基类: CacheInterface

使用 diskcache 库将数据存储在磁盘上的缓存实现。

此缓存后端将数据持久化到磁盘，使其能够在程序运行之间保留。它实现了 CacheInterface，用于 Ragas 缓存功能。

参数: cache_dir (str, 可选): 缓存文件将存储的目录。默认为 ".cache"。

源代码位于 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

按键从磁盘缓存中检索值。

参数: key: 要在缓存中查找的键。

返回值: 与键关联的缓存值；如果未找到，则为 None。

源代码位于 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

使用给定的键在磁盘缓存中存储一个值。

参数: key: 用于存储值的键。 value: 要缓存的值。

源代码位于 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

检查磁盘缓存中是否存在一个键。

参数: key: 要检查的键。

返回值: 如果缓存中存在该键，则为 True；否则为 False。

源代码位于 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)

为函数添加缓存功能的装饰器。

此装饰器可应用于同步和异步函数以缓存其结果。如果未提供缓存后端，则返回原始函数而不进行更改。

参数: cache_backend (Optional[CacheInterface]): 用于存储结果的缓存后端。如果为 None，则禁用缓存。

返回值: Callable: 一个实现缓存行为的装饰函数。

源代码位于 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