LLM 캐싱 전략 가이드
개요
LLM 서빙에서 캐싱은 비용 절감과 응답 속도 개선의 핵심이다. 적절한 캐싱 전략으로 70-90%의 비용 절감과 10-100배의 속도 향상이 가능하다.
왜 LLM 캐싱이 중요한가?
| 문제 |
영향 |
캐싱 효과 |
| API 비용 |
토큰당 과금 |
중복 호출 제거 |
| 응답 지연 |
1-10초 소요 |
밀리초 수준으로 |
| Rate Limit |
동시 요청 제한 |
요청 수 감소 |
| 일관성 |
동일 질문 다른 답 |
동일 답변 보장 |
캐싱 계층
1. Exact Match Cache
구현
import hashlib
import redis
import json
from typing import Optional, Dict, Any
class ExactMatchCache:
"""
Exact Match Cache
동일한 프롬프트 + 모델 + 파라미터 조합에 대해 캐싱
"""
def __init__(
self,
redis_client: redis.Redis,
default_ttl: int = 3600 # 1시간
):
self.redis = redis_client
self.default_ttl = default_ttl
def _hash_key(
self,
prompt: str,
model: str,
params: Dict[str, Any]
) -> str:
"""
캐시 키 생성
모델, 프롬프트, 파라미터를 조합하여 고유 해시 생성
"""
# 파라미터 정렬하여 일관된 해시 보장
sorted_params = sorted(params.items())
content = f"{model}:{prompt}:{sorted_params}"
return f"llm_cache:{hashlib.sha256(content.encode()).hexdigest()}"
def get(
self,
prompt: str,
model: str,
params: Dict[str, Any]
) -> Optional[str]:
"""캐시에서 응답 조회"""
key = self._hash_key(prompt, model, params)
cached = self.redis.get(key)
if cached:
# 메트릭 기록
self.redis.incr("llm_cache:hits")
return json.loads(cached)
self.redis.incr("llm_cache:misses")
return None
def set(
self,
prompt: str,
model: str,
params: Dict[str, Any],
response: str,
ttl: Optional[int] = None
) -> None:
"""응답을 캐시에 저장"""
key = self._hash_key(prompt, model, params)
self.redis.setex(
key,
ttl or self.default_ttl,
json.dumps(response)
)
def get_hit_rate(self) -> float:
"""캐시 적중률 계산"""
hits = int(self.redis.get("llm_cache:hits") or 0)
misses = int(self.redis.get("llm_cache:misses") or 0)
total = hits + misses
return hits / total if total > 0 else 0.0
# 사용 예시
redis_client = redis.Redis(host='localhost', port=6379)
cache = ExactMatchCache(redis_client, default_ttl=3600)
# 캐시 조회
prompt = "Python에서 리스트 정렬하는 방법"
model = "gpt-4"
params = {"temperature": 0.7, "max_tokens": 500}
cached_response = cache.get(prompt, model, params)
if cached_response:
print("Cache hit!")
response = cached_response
else:
print("Cache miss - calling LLM")
response = llm_client.generate(prompt, **params)
cache.set(prompt, model, params, response)
적용 시나리오별 설정
| 시나리오 |
예상 히트율 |
권장 TTL |
설명 |
| FAQ 봇 |
60-80% |
24시간 |
질문이 반복됨 |
| 코드 자동완성 |
10-20% |
1시간 |
컨텍스트 다양 |
| 문서 요약 |
5-10% |
무제한 |
문서별 캐싱 |
| 자유 대화 |
1-5% |
미사용 |
거의 중복 없음 |
2. Semantic Cache
아키텍처
구현
import numpy as np
from typing import Optional, Dict, List, Tuple
from dataclasses import dataclass
@dataclass
class CacheEntry:
"""캐시 엔트리"""
query: str
embedding: np.ndarray
response: str
metadata: Dict
class SemanticCache:
"""
Semantic Cache
임베딩 유사도 기반 캐싱
"""
def __init__(
self,
embedding_model,
vector_store, # Pinecone, Redis Vector, Chroma 등
similarity_threshold: float = 0.90
):
self.embedding_model = embedding_model
self.vector_store = vector_store
self.threshold = similarity_threshold
def _get_embedding(self, text: str) -> np.ndarray:
"""텍스트 임베딩 생성"""
return self.embedding_model.encode(text)
def search(
self,
query: str,
top_k: int = 1
) -> Optional[Tuple[str, float]]:
"""
유사한 캐시 항목 검색
Returns:
(response, similarity) 또는 None
"""
query_embedding = self._get_embedding(query)
# 벡터 DB에서 유사 항목 검색
results = self.vector_store.search(
vector=query_embedding,
top_k=top_k
)
if not results:
return None
top_result = results[0]
similarity = top_result['score']
if similarity >= self.threshold:
return (top_result['response'], similarity)
return None
def store(
self,
query: str,
response: str,
metadata: Optional[Dict] = None
) -> None:
"""캐시에 저장"""
embedding = self._get_embedding(query)
self.vector_store.upsert(
id=hashlib.md5(query.encode()).hexdigest(),
vector=embedding,
metadata={
"query": query,
"response": response,
**(metadata or {})
}
)
# 사용 예시
from sentence_transformers import SentenceTransformer
embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
# vector_store = PineconeIndex(...) 또는 RedisVector(...)
cache = SemanticCache(
embedding_model=embedding_model,
vector_store=vector_store,
similarity_threshold=0.90
)
# 조회
result = cache.search("Python 리스트 정렬 방법")
if result:
response, similarity = result
print(f"Cache hit! (similarity: {similarity:.2f})")
else:
response = llm_client.generate(query)
cache.store(query, response)
구현 고려사항
| 항목 |
권장값 |
비고 |
| 유사도 임계값 |
0.90-0.95 |
낮으면 오탐, 높으면 히트율 감소 |
| 임베딩 모델 |
text-embedding-3-small |
속도/품질 균형 |
| 벡터 DB |
Redis Vector, Pinecone |
저지연 필수 |
| 인덱스 타입 |
HNSW |
근사 검색, 빠름 |
Semantic Cache 주의사항
한계점과 해결책:
| 문제 |
예시 |
해결책 |
| 시간 민감 쿼리 |
"오늘 날씨" ≈ "어제 날씨" |
시간 표현 정규화 |
| 맥락 의존 쿼리 |
"그거 다시 설명해줘" |
이전 N턴 포함 해싱 |
| 개인화 쿼리 |
"내 주문 상태" |
사용자별 캐시 분리 |
class ContextAwareSemanticCache(SemanticCache):
"""
컨텍스트를 고려한 Semantic Cache
"""
def normalize_query(self, query: str, context: Dict) -> str:
"""
쿼리 정규화
- 시간 표현 상대화
- 대명사 해소
"""
import re
from datetime import datetime
normalized = query
# "오늘" → 날짜로 변환
if "오늘" in query:
today = context.get('date', datetime.now().strftime('%Y-%m-%d'))
normalized = query.replace("오늘", f"({today})")
# 이전 대화에서 참조 해소
if "그것" in query or "그거" in query:
last_topic = context.get('last_topic', '')
if last_topic:
normalized = re.sub(r'그것|그거', last_topic, normalized)
return normalized
def get_cache_key(
self,
query: str,
user_id: str,
context: Dict
) -> str:
"""사용자별, 컨텍스트별 캐시 키"""
normalized = self.normalize_query(query, context)
return f"{user_id}:{normalized}"
3. KV Cache (Prefix Caching)
원리
vLLM에서의 활용
from vllm import LLM, SamplingParams
# Prefix caching 활성화
llm = LLM(
model="meta-llama/Llama-2-7b-hf",
enable_prefix_caching=True,
max_num_batched_tokens=8192,
gpu_memory_utilization=0.9
)
# 공통 시스템 프롬프트
SYSTEM_PROMPT = """당신은 전문적인 고객 상담 AI입니다.
다음 규칙을 반드시 따르세요:
1. 항상 공손하고 친절하게 응대합니다.
2. 정확한 정보만 제공합니다.
3. 불확실한 경우 확인 요청을 합니다.
4. 개인정보 보호에 유의합니다.
응답 형식:
- 먼저 고객의 문의를 요약합니다.
- 해결책이나 정보를 제공합니다.
- 추가 도움이 필요한지 확인합니다.
"""
# 여러 사용자 쿼리 - SYSTEM_PROMPT의 KV 캐시 공유
queries = [
f"{SYSTEM_PROMPT}\n\nUser: 환불 절차가 어떻게 되나요?\nAssistant:",
f"{SYSTEM_PROMPT}\n\nUser: 배송은 보통 얼마나 걸리나요?\nAssistant:",
f"{SYSTEM_PROMPT}\n\nUser: 상품 교환이 가능한가요?\nAssistant:",
]
# 배치 처리
sampling_params = SamplingParams(
temperature=0.7,
max_tokens=500
)
outputs = llm.generate(queries, sampling_params)
for output in outputs:
print(output.outputs[0].text)
KV Cache 효과
| 시스템 프롬프트 길이 |
KV 캐시 절감 |
TTFT 개선 |
| 500 토큰 |
15-20% |
100-200ms |
| 2000 토큰 |
40-50% |
400-800ms |
| 8000 토큰 |
60-70% |
1-2초 |
4. 캐시 무효화 전략
시간 기반
from enum import Enum
from typing import Dict
class ContentType(Enum):
FACTUAL = "factual" # 상식, 불변 지식
DOCUMENTATION = "documentation" # 제품/서비스 문서
NEWS = "news" # 뉴스, 시사
REALTIME = "realtime" # 실시간 데이터
# 콘텐츠 유형별 TTL 정책
CACHE_POLICIES: Dict[ContentType, int] = {
ContentType.FACTUAL: 86400 * 30, # 30일
ContentType.DOCUMENTATION: 86400 * 7, # 7일
ContentType.NEWS: 3600, # 1시간
ContentType.REALTIME: 0, # 캐시 안함
}
def get_ttl_for_query(query: str, content_classifier) -> int:
"""쿼리 내용에 따른 TTL 결정"""
content_type = content_classifier.classify(query)
return CACHE_POLICIES.get(content_type, 3600)
이벤트 기반
from typing import Set, Optional
from datetime import datetime
import redis
class CacheInvalidator:
"""
캐시 무효화 관리자
"""
def __init__(self, redis_client: redis.Redis):
self.redis = redis_client
def invalidate_by_tag(self, tag: str) -> int:
"""
태그 기반 무효화
예: 특정 문서 관련 캐시 모두 삭제
"""
pattern = f"llm_cache:*:tag:{tag}"
keys = self.redis.keys(pattern)
if keys:
return self.redis.delete(*keys)
return 0
def invalidate_by_prefix(self, prefix: str) -> int:
"""
프리픽스 기반 무효화
예: 특정 프롬프트 버전 관련 캐시 삭제
"""
pattern = f"llm_cache:{prefix}:*"
keys = self.redis.keys(pattern)
if keys:
return self.redis.delete(*keys)
return 0
def invalidate_by_timestamp(
self,
before: datetime
) -> int:
"""
시간 기반 무효화
특정 시점 이전 캐시 삭제
"""
# ZRANGEBYSCORE로 타임스탬프 인덱스 조회
cutoff = before.timestamp()
old_keys = self.redis.zrangebyscore(
"llm_cache:timestamps",
"-inf",
cutoff
)
if old_keys:
# 캐시 항목 삭제
self.redis.delete(*old_keys)
# 인덱스에서도 제거
self.redis.zremrangebyscore(
"llm_cache:timestamps",
"-inf",
cutoff
)
return len(old_keys)
return 0
def flush_all(self) -> None:
"""전체 LLM 캐시 삭제"""
keys = self.redis.keys("llm_cache:*")
if keys:
self.redis.delete(*keys)
# Webhook 핸들러 예시
from flask import Flask, request
app = Flask(__name__)
invalidator = CacheInvalidator(redis_client)
@app.route("/webhook/document-updated", methods=["POST"])
def handle_document_update():
"""문서 업데이트 시 캐시 무효화"""
doc_id = request.json.get("document_id")
count = invalidator.invalidate_by_tag(f"doc:{doc_id}")
return {"invalidated": count}
@app.route("/webhook/model-changed", methods=["POST"])
def handle_model_change():
"""모델 변경 시 전체 캐시 무효화"""
invalidator.flush_all()
return {"status": "flushed"}
5. 모니터링 메트릭
| 메트릭 |
계산 |
목표값 |
알림 조건 |
| Hit Rate |
히트 / 전체 요청 |
> 30% |
< 10% |
| Cost Savings |
(1 - 실제비용/캐시없음비용) × 100 |
> 50% |
< 20% |
| Latency P50 |
캐시 히트 지연 |
< 100ms |
> 500ms |
| Staleness Rate |
오래된 캐시 반환 비율 |
< 5% |
> 10% |
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Optional
import time
@dataclass
class CacheMetrics:
"""캐시 메트릭"""
hits: int = 0
misses: int = 0
latency_sum_ms: float = 0
latency_count: int = 0
cost_saved: float = 0
stale_hits: int = 0
class CacheMonitor:
"""
캐시 모니터링
"""
def __init__(self, redis_client):
self.redis = redis_client
self.metrics = CacheMetrics()
def record_hit(self, latency_ms: float, is_stale: bool = False):
"""캐시 히트 기록"""
self.metrics.hits += 1
self.metrics.latency_sum_ms += latency_ms
self.metrics.latency_count += 1
if is_stale:
self.metrics.stale_hits += 1
# Redis에도 기록 (분산 환경용)
self.redis.incr("llm_cache:metrics:hits")
def record_miss(self, llm_cost: float):
"""캐시 미스 기록"""
self.metrics.misses += 1
self.redis.incr("llm_cache:metrics:misses")
def record_cost_saved(self, amount: float):
"""절감 비용 기록"""
self.metrics.cost_saved += amount
self.redis.incrbyfloat("llm_cache:metrics:cost_saved", amount)
@property
def hit_rate(self) -> float:
"""히트율"""
total = self.metrics.hits + self.metrics.misses
return self.metrics.hits / total if total > 0 else 0
@property
def avg_latency_ms(self) -> float:
"""평균 지연시간"""
if self.metrics.latency_count == 0:
return 0
return self.metrics.latency_sum_ms / self.metrics.latency_count
@property
def staleness_rate(self) -> float:
"""Stale 히트 비율"""
if self.metrics.hits == 0:
return 0
return self.metrics.stale_hits / self.metrics.hits
def get_dashboard_data(self) -> dict:
"""대시보드용 데이터"""
return {
"hit_rate": f"{self.hit_rate:.1%}",
"avg_latency_ms": f"{self.avg_latency_ms:.1f}ms",
"cost_saved": f"${self.metrics.cost_saved:.2f}",
"staleness_rate": f"{self.staleness_rate:.1%}",
"total_requests": self.metrics.hits + self.metrics.misses
}
# 사용 예시
monitor = CacheMonitor(redis_client)
# 캐시 조회 시
start = time.time()
result = cache.get(query, model, params)
latency = (time.time() - start) * 1000
if result:
monitor.record_hit(latency)
monitor.record_cost_saved(0.003) # 예상 LLM 호출 비용
else:
response = llm_client.generate(query)
monitor.record_miss(0.003)
cache.set(query, model, params, response)
# 메트릭 확인
print(monitor.get_dashboard_data())
구현 체크리스트
Exact Match Cache
[ ] Redis 또는 인메모리 캐시 설정
[ ] 해시 키 생성 로직 (모델 + 파라미터 포함)
[ ] TTL 정책 정의
[ ] 히트/미스 메트릭 수집
Semantic Cache
[ ] 임베딩 모델 선택 (속도 vs 품질)
[ ] 벡터 DB 설정 (Redis Vector, Pinecone 등)
[ ] 유사도 임계값 튜닝
[ ] 예외 쿼리 필터링 (시간, 맥락, 개인화)
[ ] 정기적 벤치마크 및 임계값 조정
KV Cache
[ ] vLLM prefix_caching 활성화
[ ] 시스템 프롬프트 표준화
[ ] 배치 처리 최적화
[ ] GPU 메모리 모니터링
모니터링
[ ] 히트율 대시보드
[ ] 비용 절감 추적
[ ] 캐시 품질 검증 (샘플링)
[ ] 알림 설정 (히트율 급락 등)
트러블슈팅
흔한 문제와 해결책
| 문제 |
원인 |
해결책 |
| 낮은 히트율 |
쿼리 다양성 높음 |
Semantic cache 도입 |
| 잘못된 캐시 반환 |
임계값 너무 낮음 |
유사도 임계값 상향 (0.92→0.95) |
| 캐시 오염 |
할루시네이션 캐싱 |
응답 품질 검증 후 캐싱 |
| 메모리 부족 |
캐시 크기 증가 |
TTL 단축, LRU 정책 |
| 지연시간 증가 |
벡터 검색 느림 |
인덱스 최적화, 더 빠른 DB |
참고 자료