Ragas에서의 캐싱

Ragas에서의 캐싱 (Caching in Ragas)

캐싱을 사용하면 중복 계산을 피해서 평가와 테스트셋 생성을 가속화할 수 있어요. LLM과 임베딩 모델의 응답을 캐시하기 위해 Exact Match Caching을 사용해요.

출처: 문서

본문

캐싱을 사용하면 중복 계산을 피해서 평가와 테스트셋 생성을 가속화할 수 있어요. LLM과 Embedding 모델의 응답을 캐시하기 위해 Exact Match Caching을 사용해요.

캐시된 응답을 로컬 디스크 캐시에 저장하는 DiskCacheBackend를 사용할 수 있어요. CacheInterface를 구현해서 자신만의 커스텀 캐셔를 구현할 수도 있어요.

최신 LLM과 임베딩에서 캐싱 사용하기

새로운 메트릭 컬렉션과 실험은 간단한 인터페이스를 통해 캐싱을 지원해요.

빠른 시작

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

# 캐시를 한 번 생성
cache = DiskCacheBackend()

# LLM 팩토리와 함께 사용
client = OpenAI(api_key="...")
llm = llm_factory("gpt-4o-mini", client=client, cache=cache)

# 이제 모든 LLM 호출이 캐시됨!
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

# 캐시 인스턴스 생성
cache = DiskCacheBackend()

# 캐싱으로 LLM 생성
client = OpenAI(api_key="...")
llm = llm_factory("gpt-4o-mini", client=client, cache=cache)

# 첫 호출 - API 요청하고 결과 캐시
response1 = llm.generate("Evaluate this text", Response)

# 두 번째 호출 - 캐시된 결과를 즉시 반환
response2 = llm.generate("Evaluate this text", Response)

# 결과: 동일한 출력, 60배 빠름, $0 비용

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)

# 첫 호출 - API 요청
vector1 = embeddings.embed_text("Some text to embed")

# 두 번째 호출 - 즉시 캐시 히트
vector2 = embeddings.embed_text("Some text to embed")

assert vector1 == vector2  # 동일한 결과

실험에서 캐싱

캐싱은 동일한 평가를 여러 번 실행하는 실험에서 특히 강력해요.

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

# 캐시된 LLM을 한 번 설정
cache = DiskCacheBackend()
llm = llm_factory("gpt-4o-mini", client=client, cache=cache)

# 메트릭에 사용
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
    }

# 데이터셋 로드
dataset = Dataset.from_list([
    {"response": "Paris is the capital of France", "reference": "Paris"},
    {"response": "London is the capital of UK", "reference": "London"},
])

# 첫 실행 - API 호출하고 결과 캐시
print("First run (populating cache)...")
results1 = await evaluate_model.arun(dataset)
# 2개 샘플에 약 2초 소요

# 두 번째 실행 - 캐시 사용, 거의 즉시!
print("Second run (using cache)...")
results2 = await evaluate_model.arun(dataset)
# 2개 샘플에 약 0.1초 소요

# 결과는 동일하지만 20배 더 빠름!

캐시 관리

캐시 지우기

# 캐시된 모든 데이터 지우기
cache = DiskCacheBackend()
cache.cache.clear()

크기 제한 설정

# 캐시를 1GB로 제한
cache = DiskCacheBackend()
cache.cache.reset('size_limit', 1e9)  # 1GB
cache.cache.reset('cull_limit', 10)   # 가득 차면 10% 제거

캐시 위치

기본적으로 캐시는 .cache/ 디렉토리에 저장돼요. 이를 변경할 수 있어요.

cache = DiskCacheBackend(cache_dir="my_custom_cache")

캐싱의 이점

비용 절감: 동일한 입력에 대한 반복 API 호출 방지 (50-60% 절감)
속도: 캐시된 호출은 거의 즉시 반환 (60배 이상 빠름)
개발: API 호출을 기다리지 않고 빠르게 반복
재현성: 동일한 입력은 항상 동일한 결과 반환

캐시 히트 발생 조건:

✅ 동일한 프롬프트/텍스트 (정확 일치)
✅ 동일한 모델 파라미터 (temperature, max_tokens 등)
✅ 동일한 응답 모델/구조 (LLM의 경우)

캐시 미스 발생 조건:

❌ 다른 프롬프트/텍스트
❌ 다른 파라미터
❌ 다른 응답 모델

안티 패턴 (캐시하지 말아야 할 때)

❌ 비결정적 프롬프트: 프롬프트가 무작위 요소나 타임스탬프를 포함하는 경우
❌ 높은 temperature: temperature > 0.7인 경우 (응답 변동이 너무 큼)
❌ 스트리밍 응답: 캐싱은 스트리밍에서 작동하지 않음
❌ 실시간 데이터: 응답이 현재 상태를 반영해야 하는 경우

환경별 참고 사항

노트북: 캐시는 셀 실행과 커널 재시작 사이에 유지돼요.

웹 애플리케이션: 더 나은 성능을 위해 요청 간 캐시를 공유해요.

서버리스 함수: /tmp 디렉토리 사용:

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

분산 워커: 캐시는 프로세스 안전하지만, 고처리량 시스템에서는 CacheInterface로 Redis 백엔드 구현을 고려해요.

성능 기대치

Scenario	Time	Cost
First run (100 samples)	~2 minutes	$0.50
Second run (cached)	~2 seconds	$0.00
Speedup	60x faster	100% savings

레거시 캐싱 (Deprecated)

Deprecated: LangchainLLMWrapper를 사용하는 이 접근 방식은 deprecated이며 v1.0에서 제거될 예정이에요. 위에서 보여준 대로 llm_factory()와 embedding_factory()를 사용하는 현대적인 접근 방식을 사용해요.

LangchainLLMWrapper로 레거시 캐싱 사용

레거시 LLM과 임베딩 모델에서 DiskCacheBackend를 사용하는 방법을 볼게요.

from ragas.cache import DiskCacheBackend

cacher = DiskCacheBackend()

# 캐시가 비어있는지 확인하고 지우기
print(len(cacher.cache))
cacher.cache.clear()
print(len(cacher.cache))

캐셔로 LLM과 임베딩 모델을 만들어요. 여기서는 예시로 langchain-openai의 ChatOpenAI를 사용해요.

from langchain_openai import ChatOpenAI
from ragas.llms import LangchainLLMWrapper

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

# 캐시가 작동하는 것을 보려면 로깅 수준을 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

# AspectCritic으로 Answer Correctness 정의
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)]

# 데이터셋 로드
dataset = load_dataset(
    "vibrantlabsai/amnesty_qa", "english_v3", trust_remote_code=True
)
eval_dataset = EvaluationDataset.from_hf_dataset(dataset["eval"])

# 데이터셋 평가
results = evaluate(
    dataset=eval_dataset,
    metrics=metrics,
)

results

이것은 로컬 머신에서 거의 2분 걸렸어요. 이제 다시 실행해서 캐시가 작동하는 것을 확인해요.

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

results

거의 즉시 실행돼요.

이것을 테스트셋 생성에서도 사용할 수 있어요. generator_llm을 캐시된 버전으로 교체하면 돼요. 자세한 내용은 testset generation 섹션을 참고해요.

Ragas에서의 캐싱