33차시. Modular RAG 최적화

6. 최적화 ④ — 캐싱

출처: 원본 PDF 33차시_ModularRAG최적화.pdf 13~14쪽 · 쪽 그림 13쪽 14쪽

같은 질문이 반복되면 전체 파이프라인을 건너뛸 수 있습니다.

6-1) 완전 일치 캐시

Python

from functools import lru_cache
import hashlib, json, time

_cache = {}
CACHE_TTL = 3600      # 1시간

def _key(question: str) -> str:
    return hashlib.md5(question.strip().lower().encode()).hexdigest()

def ask(question: str) -> dict:
    k = _key(question)
    hit = _cache.get(k)
    if hit and time.time() - hit["at"] < CACHE_TTL:
        result = dict(hit["result"])
        result["cached"] = True
        return result

    final = app.invoke(init_state(question),
                       {"recursion_limit": config.RECURSION_LIMIT})
    result = {...}                      # 기존 반환 형태

    # 검증 통과한 답변만 캐시 (실패는 캐시하지 않음)
    if final.get("grade") == "pass":
        _cache[k] = {"result": result, "at": time.time()}
    return result

6-2) 시맨틱 캐시 (심화)

🔍 표현이 달라도 같은 질문이라면
"환불은 며칠 이내인가요?"
"환불 기한이 어떻게 되나요?"
완전 일치 캐시는 이 둘을 다른 질문으로 봅니다.
시맨틱 캐시는 질문의 임베딩을 저장하고,
새 질문과 유사도가 0.95 이상이면
이전 답변을 재사용합니다.
임베딩 1회(아주 저렴) vs 전체 파이프라인(비쌈)
→ 이득이 큽니다.

Python

_cache_vecs = []      # [(vector, key), ...]
def _find_similar(question, threshold=0.95):
    if not _cache_vecs:
        return None
    v = np.array(_emb.embed_query(question))
    for vec, key in _cache_vecs:
        # 코사인 유사도 = 내적 ÷ (길이 × 길이)
        dot_product = np.dot(v, vec)
        length1 = np.linalg.norm(v)
        length2 = np.linalg.norm(vec)
        sim = dot_product / (length1 * length2)
        if sim >= threshold:
            return key
    return None

⚠️ 캐시 주의사항
① 문서가 바뀌면 캐시를 비워야 합니다
인덱스 재생성 시 _cache.clear()
② 실패한 답변은 캐시하지 마세요
fallback 응답이 계속 재사용되면 최악입니다
③ 유사도 임계값을 높게 잡으세요
0.95 미만이면 다른 질문에 같은 답이 나갑니다
④ TTL을 두세요
영구 캐시는 문서 변경을 반영하지 못합니다