LLM 캐싱과 시맨틱 중복 제거: AI 비용을 절반으로 줄이는 전략

이 글은 누구를 위한 것인가

• LLM API 비용이 월 수백만 원을 넘어가는 팀
• 같은 질문이 반복되는 FAQ·고객지원 서비스를 운영하는 팀
• 프롬프트 캐싱(prompt caching)을 실제로 구현하고 싶은 개발자

---

들어가며

LLM API 비용의 30-50%는 중복 요청에서 온다. "배송 기간이 얼마나 걸리나요?"는 매일 수천 번 온다. 이 답은 캐시에서 주면 비용이 0이다.

이 글은 bluefoxdev.kr의 LLM 비용 최적화 를 참고하여 작성했습니다.

---

1. 캐싱 전략 비교

[LLM 캐싱 방식]

1. 정확 일치 캐시 (Exact Match):
   입력 해시 → 출력 저장
   히트율: 5-20% (완전히 같은 쿼리만)
   구현: 간단 (Redis GET/SET)
   적합: FAQ, 정형화된 프롬프트

2. 시맨틱 유사도 캐시 (Semantic Cache):
   입력 임베딩 → 유사 캐시 검색 (cosine > 0.95)
   히트율: 20-50% (의미가 비슷한 쿼리 포함)
   구현: 임베딩 + 벡터 DB
   적합: 고객 상담, 자연어 검색

3. 프롬프트 캐싱 (API 레벨):
   Claude prompt caching: 시스템 프롬프트 재사용
   비용 절감: 입력 토큰 90% 할인
   적합: 긴 시스템 프롬프트, RAG 문서

[캐시 설계 원칙]
  TTL 설정:
    정적 지식 (FAQ): 24시간
    반정적 (가격): 1시간
    실시간 (재고): 캐시 금지

  캐시 무효화:
    데이터 변경 이벤트 구독
    버전 태그 (v1:, v2:)
    수동 purge API

[비용 절감 시뮬레이션]
  일 요청: 10,000건
  평균 입력 토큰: 500
  정확 일치 캐시 히트 20%: 비용 20% 절감
  시맨틱 캐시 추가 30%: 총 50% 절감
  프롬프트 캐싱: 긴 컨텍스트 70-90% 절감

---

2. 시맨틱 캐시 구현

import anthropic
import hashlib
import json
import numpy as np
import redis
from dataclasses import dataclass

client = anthropic.Anthropic()

@dataclass
class CacheEntry:
    query: str
    response: str
    embedding: list[float]
    hit_count: int = 0

class SemanticCache:
    """시맨틱 유사도 기반 LLM 응답 캐시"""
    
    def __init__(
        self,
        redis_client: redis.Redis,
        similarity_threshold: float = 0.95,
        ttl: int = 3600,
    ):
        self.redis = redis_client
        self.threshold = similarity_threshold
        self.ttl = ttl
    
    def _embed(self, text: str) -> list[float]:
        import openai
        oai = openai.OpenAI()
        resp = oai.embeddings.create(
            input=text,
            model="text-embedding-3-small",
        )
        return resp.data[0].embedding
    
    def _exact_key(self, text: str) -> str:
        return f"llm:exact:{hashlib.sha256(text.encode()).hexdigest()}"
    
    async def get(self, query: str) -> str | None:
        # 1단계: 정확 일치 확인
        exact_key = self._exact_key(query)
        exact_hit = self.redis.get(exact_key)
        if exact_hit:
            self.redis.incr(f"{exact_key}:hits")
            return exact_hit.decode()
        
        # 2단계: 시맨틱 유사도 검색
        query_emb = np.array(self._embed(query))
        
        cache_keys = self.redis.keys("llm:semantic:*:emb")
        
        best_similarity = 0
        best_response = None
        
        for key in cache_keys:
            emb_data = self.redis.get(key)
            if not emb_data:
                continue
            
            stored_emb = np.array(json.loads(emb_data))
            similarity = np.dot(query_emb, stored_emb) / (
                np.linalg.norm(query_emb) * np.linalg.norm(stored_emb) + 1e-8
            )
            
            if similarity > best_similarity:
                best_similarity = similarity
                response_key = key.decode().replace(":emb", ":resp")
                best_response = self.redis.get(response_key)
        
        if best_similarity >= self.threshold and best_response:
            return best_response.decode()
        
        return None
    
    async def set(self, query: str, response: str):
        # 정확 일치 캐시
        exact_key = self._exact_key(query)
        self.redis.setex(exact_key, self.ttl, response)
        
        # 시맨틱 캐시
        emb = self._embed(query)
        cache_id = hashlib.md5(query.encode()).hexdigest()[:8]
        
        self.redis.setex(f"llm:semantic:{cache_id}:emb", self.ttl, json.dumps(emb))
        self.redis.setex(f"llm:semantic:{cache_id}:resp", self.ttl, response)
        self.redis.setex(f"llm:semantic:{cache_id}:query", self.ttl, query)

def make_cached_claude_call(
    messages: list[dict],
    system: str,
    cache: SemanticCache,
    model: str = "claude-haiku-4-5-20251001",
) -> str:
    """캐시 통합 Claude 호출"""
    
    user_query = messages[-1]["content"] if messages else ""
    
    import asyncio
    cached = asyncio.run(cache.get(user_query))
    if cached:
        return cached
    
    # 프롬프트 캐싱 활성화 (긴 시스템 프롬프트)
    response = client.messages.create(
        model=model,
        max_tokens=1000,
        system=[
            {
                "type": "text",
                "text": system,
                "cache_control": {"type": "ephemeral"},
            }
        ],
        messages=messages,
    )
    
    result = response.content[0].text
    asyncio.run(cache.set(user_query, result))
    
    return result

def estimate_cache_savings(
    daily_requests: int,
    avg_input_tokens: int,
    hit_rate: float,
    price_per_million_input: float = 0.8,
) -> dict:
    """캐시 절감액 추정"""
    
    monthly_requests = daily_requests * 30
    total_cost_no_cache = (monthly_requests * avg_input_tokens / 1_000_000) * price_per_million_input
    
    cached_requests = int(monthly_requests * hit_rate)
    uncached_requests = monthly_requests - cached_requests
    total_cost_with_cache = (uncached_requests * avg_input_tokens / 1_000_000) * price_per_million_input
    
    return {
        "monthly_cost_no_cache": round(total_cost_no_cache, 2),
        "monthly_cost_with_cache": round(total_cost_with_cache, 2),
        "savings_usd": round(total_cost_no_cache - total_cost_with_cache, 2),
        "savings_pct": round(hit_rate * 100, 1),
    }

---

마무리

시맨틱 캐시는 "배송 얼마나 걸려요?"와 "배송 기간이 며칠인가요?"를 같은 질문으로 처리한다. 구현 순서: 정확 일치 캐시 먼저 → 시맨틱 캐시 추가 → Claude 프롬프트 캐싱 적용. 세 단계로 LLM 비용 50-80%를 절감할 수 있다.