跳至内容

Cache

CacheInterface

基础: ABC

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

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

get abstractmethod 

get(key: str) -> Any

根据键从缓存中检索一个值。

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

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

Source code in ragas/src/ragas/cache.py 
22
23
24
25
26
27
28
29
30
31
32
@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

设置 abstractmethod 

set(key: str, value) -> None

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

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

Source code in ragas/src/ragas/cache.py 
34
35
36
37
38
39
40
41
42
@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（如果 key 存在于缓存中），否则为 False。

Source code in ragas/src/ragas/cache.py 
44
45
46
47
48
49
50
51
52
53
54
@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, optional): 存放缓存文件的目录。默认值为 ".cache".

Source code in ragas/src/ragas/cache.py 
79
80
81
82
83
84
85
86
87
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(key: str) -> Any

根据键从磁盘缓存中检索值。

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

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

Source code in ragas/src/ragas/cache.py 
89
90
91
92
93
94
95
96
97
98
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(key: str, value) -> None

使用给定的 key 将一个值存储到磁盘缓存中。

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

Source code in ragas/src/ragas/cache.py 
100
101
102
103
104
105
106
107
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。

Source code in ragas/src/ragas/cache.py 
109
110
111
112
113
114
115
116
117
118
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(cache_backend: Optional[CacheInterface] = None)

装饰器，用于为函数添加缓存功能。

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

Args: cache_backend (Optional[CacheInterface]): 用于存储结果的缓存后端。 如果 None，缓存被禁用。

Returns: Callable: 一个被装饰的函数，实现缓存行为。

Source code in ragas/src/ragas/cache.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
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
优云智算