DSPy 캐시 사용 및 커스터마이징
DSPy 캐시 사용 및 커스터마이징 (Use and Customize DSPy Cache)
이 튜토리얼에서는 DSPy의 캐싱 메커니즘의 설계를 살펴보고, 이를 효과적으로 사용하고 커스터마이징하는 방법을 알아볼게요.
출처: 문서
본문
DSPy 캐시 구조 (DSPy Cache Structure)
DSPy의 캐싱 시스템은 세 개의 뚜렷한 계층으로 구성돼요:
- 인메모리 캐시 (In-memory cache):
cachetools.LRUCache로 구현되며, 자주 사용하는 데이터에 빠르게 접근할 수 있게 해줘요. - 온디스크 캐시 (On-disk cache):
diskcache.FanoutCache를 활용하며, 캐시된 항목의 영구 저장을 제공해요. - 프롬프트 캐시 (서버 측 캐시): LLM 서비스 제공자(예: OpenAI, Anthropic)가 관리하는 계층이에요.
DSPy는 인메모리와 온디스크 답변 캐시를 제어해요. 제공자 측 프롬프트 캐싱은 별개로, DSPy가 캐싱 지침을 전달할 수는 있지만 적격성, 보존 기간, 비용 청구는 제공자가 결정합니다.
DSPy 캐시 사용하기 (Using DSPy Cache)
기본적으로 인메모리와 온디스크 캐싱은 DSPy에서 모두 자동으로 활성화돼요. 캐시를 사용하기 위해 특별한 조치가 필요하지 않아요. 캐시 히트(cache hit)가 발생하면 모듈 호출의 실행 시간이 크게 줄어드는 것을 관찰할 수 있어요. 또한 사용량 추적(usage tracking)이 활성화된 경우, 캐시된 호출의 사용량 지표는 None이 됩니다.
다음 예제를 살펴볼게요:
import dspy
import os
import time
os.environ["OPENAI_API_KEY"] = "{your_openai_key}"
dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"), track_usage=True)
predict = dspy.Predict("question->answer")
start = time.time()
result1 = predict(question="Who is the GOAT of basketball?")
print(f"Time elapse: {time.time() - start: 2f}\n\nTotal usage: {result1.get_lm_usage()}")
start = time.time()
result2 = predict(question="Who is the GOAT of basketball?")
print(f"Time elapse: {time.time() - start: 2f}\n\nTotal usage: {result2.get_lm_usage()}")
샘플 출력은 다음과 같아요:
Time elapse: 4.384113
Total usage: {'openai/gpt-4o-mini': {'completion_tokens': 97, 'prompt_tokens': 144, 'total_tokens': 241, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0, 'text_tokens': None}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0, 'text_tokens': None, 'image_tokens': None}}}
Time elapse: 0.000529
Total usage: {}
첫 번째 호출은 4.38초가 걸렸지만, 두 번째 호출은 캐시 히트 덕분에 0.0005초밖에 걸리지 않았고 사용량도 비어 있는 걸 확인할 수 있어요.
제공자 측 프롬프트 캐싱 사용하기 (Using Provider-Side Prompt Caching)
DSPy의 내장 캐싱 메커니즘에 더해, Anthropic과 OpenAI 같은 LLM 제공자가 제공하는 제공자 측 프롬프트 캐싱을 활용할 수 있어요. 이 기능은 dspy.ReAct()처럼 유사한 프롬프트를 반복적으로 보내는 모듈에서 특히 유용한데, 제공자 서버에 프롬프트 프리픽스를 캐싱해 지연 시간과 비용을 모두 줄여줘요.
네이티브 lm15 엔진 (3.4 개발 API) (Native lm15 engines)
prompt_cache로 실제 dspy.lm15.CacheConfig를 전달해요. 이 작은 브리지는 고급 사용자와 어댑터 작성자를 위한 것이에요:
import dspy
lm = dspy.LM(
"anthropic/claude-haiku-4-5",
engine="lm15",
cache=False, # disable DSPy's answer cache, not the provider's prompt cache
prompt_cache=dspy.lm15.CacheConfig(prefix="stable"),
)
dspy.configure(lm=lm, track_usage=True)
program = dspy.Predict("question -> answer")
first = program(question="What is the capital of France?")
second = program(question="What is the capital of Germany?")
second.get_lm_usage()
prefix="stable"은 lm15에게 지원되는 곳에서 재사용 가능한 시스템/도구 프리픽스를 표시하도록 요청해요. 자동 캐싱이 있는 제공자는 마커가 필요 없을 수도 있어요. 이 기능은 데모를 자동으로 선택하지 않으며, 이 짧은 예제는 제공자의 캐시 최소 기준보다 낮을 수 있어요. 절약 효과를 얻으려면 요청 간에 적격하고 동일한 프리픽스가 있어야 하며, 각 변화하는 입력은 여전히 새로운 답변을 받습니다.
일반 호출에서는 호출 시점의 값이 LM 기본값을 덮어써요:
lm("A new question", prompt_cache=dspy.lm15.CacheConfig(prefix="history"))
lm("Another question", prompt_cache=None) # remove the configured caching hint
lm_without_hint = lm.copy(prompt_cache=None)
어댑터 작성자는 lm_kwargs["prompt_cache"]에 같은 객체를 넣을 수 있어요. DSPy는 일반 messages/options를 읽은 후에 이를 표준 요청에 첨부합니다. prefix_until_index를 사용할 때는 lm15의 표준 메시지 목록을 가리키며, 원래의 OpenAI 형태 행을 가리키지 않아요. 그 경계를 정밀하게 제어해야 한다면 명시적 lm15 요청을 사용하세요.
cache=True/False는 여전히 DSPy의 답변 캐시를 제어해요.prompt_cache=None은 이 브리지의 힌트만 제거하며, 제공자의 자동 캐싱을 비활성화한다는 보장은 아니에요.- 명시적 타입화된
Request호출은LM.prompt_cache기본값을 상속하거나 호출 시점의prompt_cache덮어쓰기를 받지 않고 자신의Config.cache만 사용해요. - 이 옵션은 네이티브 lm15와 표준 커스텀 엔진에서 동작해요. LiteLLM이 필요한 일반 요청은 정책을 조용히 버리거나 번역하는 대신 오류를 발생시킵니다.
prompt_cache_key같은 제공자 형태의 옵션과 결합하지 말고, 대신CacheConfig.key를 사용하세요. - 정책은
copy()와 JSON LM-state 저장/로드에서도 유지돼요. 서로 다른 정책은 서로 다른 DSPy 답변 캐시 키를 가지며, 정책이 없는 호출은 기존 키를 유지해요. - 캐시 읽기/쓰기는 제공자가 보고하는 곳에서 usage/history에 나타나요. 캐시 쓰기와 긴 보존은 추가 비용이 들 수 있고, 가격 메타데이터가 불완전하면 비용 추정치를 알 수 없을 수도 있어요.
- 저장된 캐시가 자동으로 생성·갱신·삭제되지는 않아요.
CacheConfig.resource를 제공하는 경우, 직접 관리하고 있는 리소스를 식별해야 합니다.
LiteLLM 호환 엔진 (LiteLLM compatibility engine)
engine="litellm"에서는 prompt_cache 대신 cache_control_injection_points 같은 LiteLLM 자체 옵션을 사용해요. 제공자별 지원은 LiteLLM 프롬프트 캐싱 문서를 참고하세요.
import dspy
import os
os.environ["ANTHROPIC_API_KEY"] = "{your_anthropic_key}"
lm = dspy.LM(
"anthropic/claude-sonnet-4-5-20250929",
engine="litellm",
cache_control_injection_points=[
{
"location": "message",
"role": "system",
}
],
)
dspy.configure(lm=lm)
# Use with any DSPy module
predict = dspy.Predict("question->answer")
result = predict(question="What is the capital of France?")
이 기능은 특히 다음 상황에서 유용해요:
- 동일한 지침으로
dspy.ReAct()를 사용할 때 - 일정하게 유지되는 긴 시스템 프롬프트를 다룰 때
- 유사한 컨텍스트로 여러 요청을 만들 때
Pickle 역직렬화 제한하기 (Restricting Pickle Deserialization)
기본적으로 DSPy의 온디스크 캐시는 직렬화에 Python의 pickle을 사용해요. 이는 임의의 Python 객체를 처리하지만, pickle.load는 임의의 코드를 실행할 수 있어서 손상되거나 악의적인 캐시 파일이 위험할 수 있어요.
DSPy는 캐시가 역직렬화할 수 있는 타입을 제한하는 옵트인 restrict_pickle 모드를 제공해요:
dspy.configure_cache(restrict_pickle=True)
활성화하면 캐시는 다음만 허용해요:
- LiteLLM 및 OpenAI 응답 타입 (
litellm.types.*,openai.types.*) — DSPy가 LM 호출, 임베딩, Responses API에 대해 캐시하는 pydantic 데이터 모델이에요. - NumPy 배열 재구성 헬퍼 —
numpy.ndarray(임베딩 캐시에서 사용)를 역직렬화하는 데 필요한 특정 내부 함수들이에요. safe_types를 통한 사용자 등록 타입 — 명시적으로 신뢰하는 추가 타입이에요.
커스텀 타입(dataclass, pydantic 모델 등)을 캐시한다면 이를 등록하세요:
from dataclasses import dataclass
@dataclass
class MyResult:
score: float
label: str
dspy.configure_cache(restrict_pickle=True, safe_types=[MyResult])
허용 목록에 없는 타입이면 캐시는 이를 miss로 처리하고 None을 반환해요. 로그 메시지는 거부된 정확한 타입을 알려줍니다:
WARNING dspy.clients.cache: Failed to deserialize disk cache entry <key>
중첩 타입 (Nested types)
등록한 타입에 중첩된 커스텀 타입이 포함되어 있다면 모두 등록해야 해요. 예를 들어 MyResult에 Metadata 필드가 있다면 둘 다 등록하세요:
dspy.configure_cache(restrict_pickle=True, safe_types=[MyResult, Metadata])
오류 메시지가 정확히 어떤 중첩 타입이 빠졌는지 알려줄 거예요.
DSPy 캐시 비활성화/활성화 (Disabling/Enabling DSPy Cache)
캐싱을 비활성화해야 하는 시나리오가 있을 수 있는데, 전체 또는 인메모리/온디스크 캐시를 선택적으로 비활성화할 수 있어요. 예를 들어:
- 동일한 LM 요청에 대해 서로 다른 응답이 필요한 경우
- 디스크 쓰기 권한이 없어 온디스크 캐시를 비활성화해야 하는 경우
- 메모리 자원이 제한되어 인메모리 캐시를 비활성화하고 싶은 경우
DSPy는 이를 위해 dspy.configure_cache() 유틸리티 함수를 제공해요. 각 캐시 타입의 활성화/비활성화 상태를 제어하는 플래그를 사용할 수 있어요:
dspy.configure_cache(
enable_disk_cache=False,
enable_memory_cache=False,
)
추가로 인메모리와 온디스크 캐시의 용량도 관리할 수 있어요:
dspy.configure_cache(
enable_disk_cache=True,
enable_memory_cache=True,
disk_size_limit_bytes=YOUR_DESIRED_VALUE,
memory_max_entries=YOUR_DESIRED_VALUE,
)
disk_size_limit_bytes는 온디스크 캐시의 최대 크기(바이트)를 정의하고, memory_max_entries는 인메모리 캐시의 최대 항목 수를 지정한다는 점에 유의하세요.
캐시 이해 및 커스터마이징 (Understanding and Customizing the Cache)
특정 상황에서는 커스텀 캐시를 구현하고 싶을 수도 있는데, 예를 들어 캐시 키가 생성되는 방식을 더 세밀하게 제어하고 싶을 때가 그렇죠. 기본적으로 캐시 키는 api_key 같은 자격 증명을 제외한 litellm에 보내지는 모든 요청 인자의 해시에서 파생돼요.
커스텀 캐시를 만들려면 dspy.clients.Cache를 서브클래싱하고 관련 메서드를 오버라이드하면 돼요:
class CustomCache(dspy.clients.Cache):
def __init__(self, **kwargs):
{write your own constructor}
def cache_key(self, request: dict[str, Any], ignored_args_for_cache_key: Optional[list[str]] = None) -> str:
{write your logic of computing cache key}
def get(self, request: dict[str, Any], ignored_args_for_cache_key: Optional[list[str]] = None) -> Any:
{write your cache read logic}
def put(
self,
request: dict[str, Any],
value: Any,
ignored_args_for_cache_key: Optional[list[str]] = None,
enable_memory_cache: bool = True,
) -> None:
{write your cache write logic}
DSPy의 나머지 부분과 매끄럽게 통합되도록, 기본 클래스와 같은 메서드 시그니처로 커스텀 캐시를 구현하거나 최소한 메서드 정의에 **kwargs를 포함해 캐시 읽기/쓰기 중 런타임 오류를 방지하는 것을 권장해요.
커스텀 캐시 클래스를 정의했다면 DSPy가 이를 사용하도록 지시할 수 있어요:
dspy.cache = CustomCache()
실용적인 예제로 살펴볼게요. 캐시 키 계산이 호출되는 특정 LM 같은 다른 매개변수는 무시하고 요청 메시지 내용에만 의존하길 원한다고 가정해 봐요. 다음과 같이 커스텀 캐시를 만들 수 있어요:
class CustomCache(dspy.clients.Cache):
def cache_key(self, request: dict[str, Any], ignored_args_for_cache_key: Optional[list[str]] = None) -> str:
messages = request.get("messages", [])
return sha256(orjson.dumps(messages, option=orjson.OPT_SORT_KEYS)).hexdigest()
dspy.cache = CustomCache(enable_disk_cache=True, enable_memory_cache=True, disk_cache_dir=dspy.clients.DISK_CACHE_DIR)
비교를 위해 커스텀 캐시 없이 아래 코드를 실행해 볼게요:
import dspy
import os
import time
os.environ["OPENAI_API_KEY"] = "{your_openai_key}"
dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"))
predict = dspy.Predict("question->answer")
start = time.time()
result1 = predict(question="Who is the GOAT of soccer?")
print(f"Time elapse: {time.time() - start: 2f}")
start = time.time()
with dspy.context(lm=dspy.LM("openai/gpt-4.1-mini")):
result2 = predict(question="Who is the GOAT of soccer?")
print(f"Time elapse: {time.time() - start: 2f}")
경과 시간을 보면 두 번째 호출에서 캐시가 히트되지 않는 걸 확인할 수 있어요. 하지만 커스텀 캐시를 사용하면:
import dspy
import os
import time
from typing import Dict, Any, Optional
import orjson
from hashlib import sha256
os.environ["OPENAI_API_KEY"] = "{your_openai_key}"
dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"))
class CustomCache(dspy.clients.Cache):
def cache_key(self, request: dict[str, Any], ignored_args_for_cache_key: Optional[list[str]] = None) -> str:
messages = request.get("messages", [])
return sha256(orjson.dumps(messages, option=orjson.OPT_SORT_KEYS)).hexdigest()
dspy.cache = CustomCache(enable_disk_cache=True, enable_memory_cache=True, disk_cache_dir=dspy.clients.DISK_CACHE_DIR)
predict = dspy.Predict("question->answer")
start = time.time()
result1 = predict(question="Who is the GOAT of volleyball?")
print(f"Time elapse: {time.time() - start: 2f}")
start = time.time()
with dspy.context(lm=dspy.LM("openai/gpt-4.1-mini")):
result2 = predict(question="Who is the GOAT of volleyball?")
print(f"Time elapse: {time.time() - start: 2f}")
두 번째 호출에서 캐시가 히트되는 것을 관찰할 수 있어요. 이는 커스텀 캐시 키 로직의 효과를 보여줘요 — 두 호출이 다른 LM을 사용했지만 메시지 내용이 같으므로 같은 캐시 키를 갖게 된 거죠.