CacheInterface

基类：ABC

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

该类提供所有缓存实现必须遵循的标准接口。它支持 get、set 以及键检查等基本缓存操作。

get

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

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

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)
            picklable_result = _make_pydantic_picklable(result)
            backend.set(cache_key, picklable_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)
            picklable_result = _make_pydantic_picklable(result)
            backend.set(cache_key, picklable_result)
            return result

        return async_wrapper if is_async else sync_wrapper

    return decorator