Ragas 中的缓存

你可以使用缓存来加速评测和测试集生成，避免重复计算。我们使用 Exact Match Caching 来缓存 LLM 和 Embedding 模型的响应。

你可以使用 DiskCacheBackend，它用本地磁盘缓存来存储缓存的响应。你也可以通过实现 CacheInterface 来实现自己的自定义缓存器。

在现代 LLM 和 Embeddings 中使用缓存

新的 metrics collections 和 experiments 通过简单接口支持缓存。

快速开始

from ragas.cache import DiskCacheBackend
from ragas.llms import llm_factory
from openai import OpenAI

# Create cache once
cache = DiskCacheBackend()

# Use with LLM factory
client = OpenAI(api_key="...")
llm = llm_factory("gpt-4o-mini", client=client, cache=cache)

# All LLM calls are now cached!
from pydantic import BaseModel

class Response(BaseModel):
    answer: str

response = llm.generate("Evaluate this...", Response)

与 llm_factory 一起缓存

from ragas.cache import DiskCacheBackend
from ragas.llms import llm_factory
from openai import OpenAI

# Create cache instance
cache = DiskCacheBackend()

# Create LLM with caching
client = OpenAI(api_key="...")
llm = llm_factory("gpt-4o-mini", client=client, cache=cache)

# First call - makes API request and caches result
response1 = llm.generate("Evaluate this text", Response)

# Second call - returns cached result instantly
response2 = llm.generate("Evaluate this text", Response)

# Result: Same output, 60x faster, $0 cost

与 embedding_factory 一起缓存

from ragas.cache import DiskCacheBackend
from ragas.embeddings import embedding_factory
from openai import OpenAI

cache = DiskCacheBackend()
client = OpenAI(api_key="...")

embeddings = embedding_factory("openai", client=client, cache=cache)

# First call - makes API request
vector1 = embeddings.embed_text("Some text to embed")

# Second call - instant cache hit
vector2 = embeddings.embed_text("Some text to embed")

assert vector1 == vector2  # Identical results

在 Experiments 中缓存

在多次运行同一评测的 experiments 中，缓存尤其有用：

from ragas import experiment, Dataset
from ragas.cache import DiskCacheBackend
from ragas.llms import llm_factory
from ragas.metrics.collections import FactualCorrectness

# Setup cached LLM once
cache = DiskCacheBackend()
llm = llm_factory("gpt-4o-mini", client=client, cache=cache)

# Use in metric
metric = FactualCorrectness(llm=llm)

@experiment()
async def evaluate_model(row):
    score = metric.score(
        response=row["response"],
        reference=row["reference"]
    )
    return {
        **row,
        "factual_correctness": score.value,
        "reason": score.reason
    }

# Load your dataset
dataset = Dataset.from_list([
    {"response": "Paris is the capital of France", "reference": "Paris"},
    {"response": "London is the capital of UK", "reference": "London"},
])

# First run - makes API calls and caches results
print("First run (populating cache)...")
results1 = await evaluate_model.arun(dataset)
# Takes ~2 seconds for 2 samples

# Second run - uses cache, nearly instant!
print("Second run (using cache)...")
results2 = await evaluate_model.arun(dataset)
# Takes ~0.1 seconds for 2 samples

# Results are identical, but 20x faster!

缓存管理

清空缓存

# Clear all cached data
cache = DiskCacheBackend()
cache.cache.clear()

设置大小限制

# Limit cache to 1GB
cache = DiskCacheBackend()
cache.cache.reset('size_limit', 1e9)  # 1GB
cache.cache.reset('cull_limit', 10)   # Remove 10% when full

缓存位置

默认情况下，缓存在 .cache/ 目录中存储。你可以更改它：

cache = DiskCacheBackend(cache_dir="my_custom_cache")

缓存的好处

1. 节省成本：避免对相同输入重复 API 调用（节省 50-60%）
2. 速度：缓存命中几乎即时返回（快 60 倍以上）
3. 开发：无需等待 API 调用即可快速迭代
4. 可复现性：相同输入始终返回相同结果

缓存命中发生在：

• ✅ 相同 prompt/文本（精确匹配）
• ✅ 相同模型参数（temperature、max_tokens 等）
• ✅ 相同响应模型/结构（对 LLM）

缓存未命中发生在：

• ❌ 不同的 prompt/文本
• ❌ 不同的参数
• ❌ 不同的响应模型

反模式（何时不要缓存）

• ❌ 非确定性 prompt：如果 prompt 包含随机元素或时间戳
• ❌ 高 temperature：如果 temperature > 0.7（响应变化太大）
• ❌ 流式响应：缓存不适用于 streaming
• ❌ 实时数据：如果响应需要反映当前状态

特定环境说明

Notebooks：缓存在 cell 执行和内核重启之间会持久保留

Web 应用：在请求之间共享缓存以获得更好性能

Serverless Functions：使用 /tmp 目录：

cache = DiskCacheBackend(cache_dir="/tmp/.cache")

分布式 Workers：缓存是进程安全的，但对于高吞吐系统，可考虑通过 CacheInterface 实现 Redis backend

性能预期

旧版缓存（已弃用）

已弃用

这种使用 LangchainLLMWrapper 的方法已弃用，并将在 v1.0 中移除。请使用上文所示的现代方法 llm_factory() 和 embedding_factory()。

与 LangchainLLMWrapper 一起使用旧版缓存

来看如何将 DiskCacheBackend 与旧版 LLM 和 Embedding 模型一起使用。

from ragas.cache import DiskCacheBackend

cacher = DiskCacheBackend()

# check if the cache is empty and clear it
print(len(cacher.cache))
cacher.cache.clear()
print(len(cacher.cache))

创建带缓存器的 LLM 和 Embedding 模型，这里我以 langchain-openai 的 ChatOpenAI 为例。

from langchain_openai import ChatOpenAI
from ragas.llms import LangchainLLMWrapper

cached_llm = LangchainLLMWrapper(ChatOpenAI(model="gpt-4o"), cache=cacher)

# if you want to see the cache in action, set the logging level to debug
import logging
from ragas.utils import set_logging_level

set_logging_level("ragas.cache", logging.DEBUG)

现在运行一次简单评测。

from ragas import evaluate
from ragas import EvaluationDataset

from ragas.metrics import FactualCorrectness, AspectCritic
from datasets import load_dataset

# Define Answer Correctness with AspectCritic
answer_correctness = AspectCritic(
    name="answer_correctness",
    definition="Is the answer correct? Does it match the reference answer?",
    llm=cached_llm,
)

metrics = [answer_correctness, FactualCorrectness(llm=cached_llm)]

# load the dataset
dataset = load_dataset(
    "vibrantlabsai/amnesty_qa", "english_v3", trust_remote_code=True
)
eval_dataset = EvaluationDataset.from_hf_dataset(dataset["eval"])

# evaluate the dataset
results = evaluate(
    dataset=eval_dataset,
    metrics=metrics,
)

results

这在我们本地机器上跑了将近 2 分钟。现在再跑一次，看看缓存的效果。

results = evaluate(
    dataset=eval_dataset,
    metrics=metrics,
)

results

几乎是瞬时完成。

你也可以在测试集生成中使用它，把 generator_llm 替换为它的缓存版本。更多细节参见 testset generation 部分。