RAG 캐싱 전략 완벽 가이드

RAG 시스템의 성능을 획기적으로 개선하는 캐싱 전략을 배웁니다. 쿼리 캐싱부터 임베딩 캐싱, Redis 통합까지 실무에서 바로 적용할 수 있는 최적화 기법을 다룹니다.

---

목차

1. 쿼리 캐싱
2. 임베딩 캐싱
3. Redis 통합
4. 실습: 캐시 시스템 구축
5. 실습: 응답 속도 개선

---

1. 쿼리 캐싱

어느 날 김개발 씨는 RAG 시스템을 운영하면서 이상한 점을 발견했습니다. "왜 같은 질문인데 매번 몇 초씩 걸리지?" 사용자들이 비슷한 질문을 반복하는데, 시스템은 매번 처음부터 다시 검색하고 있었습니다.

박시니어 씨가 말했습니다. "캐싱을 적용해보는 게 어때요?"

쿼리 캐싱은 동일하거나 유사한 질문에 대한 답변을 저장해두었다가 재사용하는 기법입니다. 마치 도서관 사서가 자주 묻는 질문의 답을 메모해두는 것과 같습니다.

이를 통해 응답 시간을 수십 배 단축할 수 있으며, API 비용도 크게 절감할 수 있습니다.

다음 코드를 살펴봅시다.

import hashlib
from datetime import datetime, timedelta
from typing import Optional, Dict

class QueryCache:
    def __init__(self, ttl_minutes: int = 60):
        self.cache: Dict[str, dict] = {}
        self.ttl = timedelta(minutes=ttl_minutes)

    def _get_cache_key(self, query: str) -> str:
        # 쿼리를 해시값으로 변환하여 캐시 키 생성
        return hashlib.md5(query.lower().strip().encode()).hexdigest()

    def get(self, query: str) -> Optional[str]:
        key = self._get_cache_key(query)
        if key in self.cache:
            cached_item = self.cache[key]
            # TTL 체크: 만료되지 않았다면 반환
            if datetime.now() - cached_item['timestamp'] < self.ttl:
                return cached_item['response']
            # 만료된 캐시는 삭제
            del self.cache[key]
        return None

    def set(self, query: str, response: str):
        key = self._get_cache_key(query)
        self.cache[key] = {
            'response': response,
            'timestamp': datetime.now()
        }

# 실제 사용 예제
cache = QueryCache(ttl_minutes=30)

def rag_query_with_cache(query: str) -> str:
    # 캐시 확인
    cached_response = cache.get(query)
    if cached_response:
        print("캐시에서 응답 반환!")
        return cached_response

    # 캐시 미스: 실제 RAG 처리
    response = perform_rag_search(query)  # 실제 RAG 로직
    cache.set(query, response)
    return response

김개발 씨는 입사 6개월 차 주니어 개발자입니다. 회사에서 고객 지원용 RAG 챗봇을 운영하고 있는데, 매일 비슷한 패턴의 질문들이 쏟아집니다.

"배송은 언제 되나요?", "환불 정책이 어떻게 되나요?" 같은 질문들이 하루에도 수백 번씩 반복됩니다. 문제는 이런 반복적인 질문에도 시스템이 매번 처음부터 벡터 검색을 수행하고, LLM API를 호출한다는 점이었습니다.

응답 시간도 느리고, API 비용도 만만치 않았습니다. 선배 개발자 박시니어 씨가 모니터링 화면을 보며 말했습니다.

"김개발 씨, 이 질문들 보세요. 거의 똑같은 내용이 반복되고 있어요.

캐싱을 적용하면 훨씬 효율적일 거예요." 그렇다면 쿼리 캐싱이란 정확히 무엇일까요? 쉽게 비유하자면, 쿼리 캐싱은 마치 패스트푸드점의 미리 만들어둔 메뉴와 같습니다.

햄버거를 주문할 때마다 처음부터 만들지 않고, 미리 만들어둔 것을 바로 제공하는 것처럼, 자주 받는 질문의 답변을 미리 저장해두었다가 즉시 제공하는 것입니다. 캐싱이 없던 시절에는 어땠을까요?

사용자가 "환불 정책이 뭔가요?"라고 물으면, 시스템은 매번 벡터 데이터베이스에서 관련 문서를 검색하고, LLM에게 답변을 생성하도록 요청했습니다. 이 과정은 통상 2-5초가 걸렸습니다.

같은 질문이 하루에 100번 반복되면, 불필요한 처리가 100번 발생하는 셈이었습니다. 더 큰 문제는 비용이었습니다.

LLM API 호출은 건당 비용이 발생하는데, 똑같은 질문에 대해 매번 비용을 지불하는 것은 비효율의 극치였습니다. 또한 벡터 데이터베이스에도 부하가 계속 걸렸습니다.

바로 이런 문제를 해결하기 위해 쿼리 캐싱이 등장했습니다. 쿼리 캐싱을 사용하면 동일한 질문에 대해 즉시 응답이 가능해집니다.

첫 번째 사용자가 질문하면 정상적으로 RAG 처리를 하지만, 그 결과를 저장해둡니다. 두 번째 사용자가 같은 질문을 하면 저장된 답변을 바로 반환합니다.

응답 시간이 5초에서 0.01초로 줄어드는 것입니다. 위의 코드를 한 줄씩 살펴보겠습니다.

먼저 QueryCache 클래스는 캐시의 핵심 로직을 담고 있습니다. 초기화할 때 ttl_minutes 파라미터를 받는데, 이것은 캐시가 얼마나 오래 유효한지를 정하는 값입니다.

30분으로 설정하면 30분 후에는 캐시가 만료되어 새로 검색합니다. _get_cache_key 메서드는 질문 문자열을 해시값으로 변환합니다.

"환불 정책이 뭔가요?"와 "환불 정책이 뭔가요? " 처럼 공백만 다른 질문도 같은 것으로 인식하기 위해 **lower().strip()**을 적용합니다.

MD5 해시를 사용하면 긴 질문도 짧은 키로 저장할 수 있습니다. get 메서드는 캐시에서 답변을 가져옵니다.

중요한 점은 TTL 체크를 한다는 것입니다. 캐시가 너무 오래되면 정보가 구식일 수 있으므로, 일정 시간이 지나면 자동으로 삭제됩니다.

set 메서드는 질문과 답변을 저장하면서 현재 시간도 함께 기록합니다. 이 시간 정보가 나중에 TTL 체크에 사용됩니다.

실제 현업에서는 어떻게 활용할까요? 예를 들어 전자상거래 고객센터 챗봇을 운영한다고 가정해봅시다.

"배송 조회는 어떻게 하나요?"라는 질문이 하루에 500번 들어옵니다. 첫 번째 질문에만 RAG 처리를 하고, 나머지 499번은 캐시에서 응답하면 API 비용이 99.8% 절감됩니다.

네이버, 카카오 같은 대형 서비스에서는 이런 방식으로 수천만 원의 비용을 절감하고 있습니다. 하지만 주의할 점도 있습니다.

초보 개발자들이 흔히 하는 실수 중 하나는 TTL을 너무 길게 설정하는 것입니다. 만약 상품 정보가 자주 변경되는데 캐시를 24시간 유지하면, 사용자가 잘못된 정보를 받을 수 있습니다.

따라서 데이터의 특성에 맞게 적절한 TTL을 설정해야 합니다. 또 다른 실수는 모든 질문을 캐싱하려는 것입니다.

"내 주문번호 12345의 배송 상태는?"처럼 개인화된 질문은 캐싱하면 안 됩니다. 다른 사용자에게 엉뚱한 정보가 노출될 수 있기 때문입니다.

다시 김개발 씨의 이야기로 돌아가 봅시다. 박시니어 씨의 조언대로 캐싱을 적용한 김개발 씨는 놀라운 결과를 얻었습니다.

평균 응답 시간이 3.2초에서 0.5초로 줄었고, API 비용은 70% 감소했습니다. 쿼리 캐싱을 제대로 이해하면 RAG 시스템의 성능과 비용을 동시에 최적화할 수 있습니다.

여러분도 오늘 배운 내용을 실제 프로젝트에 적용해 보세요.

실전 팁

💡 - TTL은 데이터 특성에 맞게 설정하세요 (정적 정보는 길게, 동적 정보는 짧게)

• 개인화된 질문은 캐싱 대상에서 제외하세요
• 캐시 적중률을 모니터링하여 효과를 측정하세요

---

2. 임베딩 캐싱

쿼리 캐싱으로 성과를 본 김개발 씨는 또 다른 문제를 발견했습니다. "질문이 조금씩 다른데 임베딩 API를 매번 호출하네?" 벡터 데이터베이스 검색 비용도 만만치 않았습니다.

박시니어 씨가 조언했습니다. "임베딩 자체를 캐싱해보는 건 어때요?"

임베딩 캐싱은 텍스트를 벡터로 변환한 결과를 저장해두는 기법입니다. 마치 번역가가 자주 쓰는 문장의 번역을 노트에 적어두는 것과 같습니다.

임베딩 API 호출 비용을 줄이고, 벡터 변환 시간을 단축하여 전체 시스템 성능을 크게 향상시킵니다.

다음 코드를 살펴봅시다.

import hashlib
import numpy as np
from typing import Optional, List
import pickle

class EmbeddingCache:
    def __init__(self, cache_file: str = "embedding_cache.pkl"):
        self.cache_file = cache_file
        self.cache = self._load_cache()

    def _load_cache(self) -> dict:
        # 파일에서 캐시 로드
        try:
            with open(self.cache_file, 'rb') as f:
                return pickle.load(f)
        except FileNotFoundError:
            return {}

    def _save_cache(self):
        # 캐시를 파일에 저장 (지속성 보장)
        with open(self.cache_file, 'wb') as f:
            pickle.dump(self.cache, f)

    def _get_key(self, text: str) -> str:
        return hashlib.sha256(text.encode()).hexdigest()

    def get_embedding(self, text: str) -> Optional[np.ndarray]:
        key = self._get_key(text)
        return self.cache.get(key)

    def set_embedding(self, text: str, embedding: np.ndarray):
        key = self._get_key(text)
        self.cache[key] = embedding
        # 주기적으로 디스크에 저장
        if len(self.cache) % 100 == 0:
            self._save_cache()

# 실제 사용 예제
embedding_cache = EmbeddingCache()

def get_cached_embedding(text: str) -> np.ndarray:
    # 캐시 확인
    cached = embedding_cache.get_embedding(text)
    if cached is not None:
        print(f"임베딩 캐시 히트: {text[:30]}...")
        return cached

    # 캐시 미스: 실제 API 호출
    print(f"임베딩 생성 중: {text[:30]}...")
    embedding = call_embedding_api(text)  # OpenAI, Cohere 등
    embedding_cache.set_embedding(text, embedding)
    return embedding

김개발 씨는 쿼리 캐싱을 적용한 후 모니터링 대시보드를 들여다보고 있었습니다. 응답 시간은 많이 개선되었지만, 여전히 비용 항목 중 하나가 눈에 띄었습니다.

바로 임베딩 API 호출 비용이었습니다. RAG 시스템에서는 사용자의 질문을 벡터로 변환해야 유사 문서를 검색할 수 있습니다.

문제는 "배송 조회 방법 알려줘"와 "배송 조회는 어떻게 해?"처럼 표현만 살짝 다른 질문도 각각 임베딩 API를 호출한다는 점이었습니다. 박시니어 씨가 비용 리포트를 보며 말했습니다.

"김개발 씨, 한 달에 임베딩 API 호출이 50만 건이네요. 이 중 상당수는 중복될 텐데, 임베딩 자체를 캐싱하면 어떨까요?" 그렇다면 임베딩 캐싱이란 정확히 무엇일까요?

쉽게 비유하자면, 임베딩 캐싱은 마치 외국어 공부할 때 단어장을 만드는 것과 같습니다. "apple"을 처음 볼 때는 사전을 찾지만, 단어장에 적어두면 다음부터는 사전 없이 바로 뜻을 알 수 있습니다.

임베딩도 마찬가지로, 한 번 변환한 결과를 저장해두면 같은 텍스트에 대해 다시 API를 호출할 필요가 없습니다. 임베딩 캐싱이 없던 시절에는 어땠을까요?

사용자가 "환불 정책"에 대해 물으면 시스템은 OpenAI나 Cohere의 임베딩 API를 호출했습니다. 이 과정은 보통 100-200ms가 걸립니다.

10분 후 다른 사용자가 똑같이 "환불 정책"에 대해 물으면, 또다시 API를 호출했습니다. 같은 텍스트인데도 말입니다.

더 큰 문제는 비용이었습니다. OpenAI의 임베딩 API는 1000토큰당 약 0.0001달러입니다.

적은 금액 같지만, 하루 10만 건의 질문이 들어오면 한 달에 수백 달러가 나갑니다. 게다가 문서를 벡터화할 때도 같은 비용이 발생합니다.

바로 이런 문제를 해결하기 위해 임베딩 캐싱이 등장했습니다. 임베딩 캐싱을 사용하면 동일한 텍스트에 대해 API를 단 한 번만 호출합니다.

결과를 저장해두었다가 다음번에 재사용하는 것입니다. 이를 통해 API 호출 횟수를 70-90% 줄일 수 있으며, 비용도 그만큼 절감됩니다.

위의 코드를 한 줄씩 살펴보겠습니다. 먼저 EmbeddingCache 클래스는 임베딩 결과를 파일에 저장합니다.

메모리만 사용하는 쿼리 캐싱과 달리, 임베딩은 서버가 재시작되어도 유지되어야 하기 때문에 pickle을 사용해 디스크에 저장합니다. _load_cache 메서드는 프로그램 시작 시 기존 캐시를 불러옵니다.

만약 파일이 없다면 빈 딕셔너리로 시작합니다. 이렇게 하면 서버를 껐다 켜도 캐시가 보존됩니다.

_get_key 메서드는 텍스트를 SHA256 해시로 변환합니다. MD5보다 조금 느리지만 더 안전하며, 충돌 가능성이 극히 낮습니다.

같은 텍스트는 항상 같은 키를 생성합니다. set_embedding 메서드에서 주목할 점은 주기적 저장입니다.

매번 디스크에 쓰면 성능이 떨어지므로, 100개마다 한 번씩 저장합니다. 이는 성능과 안정성의 균형을 맞춘 설계입니다.

실제 현업에서는 어떻게 활용할까요? 예를 들어 법률 자문 RAG 시스템을 운영한다고 가정해봅시다.

"민법 제750조"라는 텍스트는 수많은 질문과 문서에 반복적으로 등장합니다. 첫 번째 임베딩만 API를 호출하고, 이후에는 캐시에서 가져오면 됩니다.

실제로 한 법률 스타트업은 이 방식으로 월 임베딩 비용을 1,200달러에서 200달러로 줄였습니다. 또 다른 활용 사례는 문서 청크의 임베딩입니다.

RAG 시스템은 보통 문서를 작은 조각으로 나눠서 벡터화합니다. 문서가 업데이트되지 않는 한, 이 청크들의 임베딩은 변하지 않습니다.

따라서 한 번 임베딩한 청크는 계속 재사용할 수 있습니다. 하지만 주의할 점도 있습니다.

초보 개발자들이 흔히 하는 실수 중 하나는 캐시 크기를 관리하지 않는 것입니다. 임베딩은 보통 1536차원의 벡터이므로, 10만 개를 저장하면 수 GB의 메모리를 차지합니다.

따라서 LRU 같은 정책으로 오래된 캐시를 제거해야 합니다. 또 다른 실수는 임베딩 모델을 변경했는데 캐시를 그대로 사용하는 것입니다.

OpenAI의 text-embedding-3-small과 text-embedding-3-large는 차원이 다릅니다. 모델을 바꾸면 캐시를 초기화해야 합니다.

다시 김개발 씨의 이야기로 돌아가 봅시다. 임베딩 캐싱을 적용한 후 모니터링 결과가 인상적이었습니다.

임베딩 API 호출이 월 50만 건에서 8만 건으로 줄었습니다. 비용은 84% 감소했고, 응답 속도도 평균 150ms 빨라졌습니다.

임베딩 캐싱을 제대로 이해하면 RAG 시스템의 비용과 성능을 동시에 최적화할 수 있습니다. 특히 같은 개념이 반복적으로 등장하는 도메인에서는 필수적인 기법입니다.

실전 팁

💡 - 캐시 크기를 모니터링하고 LRU 정책으로 관리하세요

• 임베딩 모델 변경 시 캐시를 초기화하세요
• 파일 저장 주기를 적절히 조절하여 성능과 안정성의 균형을 맞추세요

---

3. Redis 통합

임베딩 캐싱까지 적용한 김개발 씨는 새로운 고민에 빠졌습니다. "서버를 2대로 늘렸는데 캐시가 공유되지 않네?" 각 서버가 별도의 캐시를 유지하다 보니 효율이 떨어졌습니다.

박시니어 씨가 해결책을 제시했습니다. "Redis를 사용하면 어떨까요?"

Redis 통합은 분산 환경에서 여러 서버가 캐시를 공유할 수 있게 해주는 솔루션입니다. 마치 여러 점원이 하나의 재고 시스템을 공유하는 것과 같습니다.

인메모리 데이터베이스인 Redis를 활용하면 매우 빠른 읽기/쓰기가 가능하며, 확장성과 안정성도 확보할 수 있습니다.

다음 코드를 살펴봅시다.

import redis
import json
import numpy as np
from typing import Optional

class RedisCache:
    def __init__(self, host: str = 'localhost', port: int = 6379, db: int = 0):
        self.client = redis.Redis(
            host=host,
            port=port,
            db=db,
            decode_responses=False  # 바이너리 데이터 처리
        )
        self.query_prefix = "query:"
        self.embedding_prefix = "embedding:"

    def get_query_cache(self, query: str) -> Optional[str]:
        # 쿼리 캐시 조회
        key = self.query_prefix + query
        result = self.client.get(key)
        return result.decode('utf-8') if result else None

    def set_query_cache(self, query: str, response: str, ttl: int = 1800):
        # 쿼리 캐시 저장 (TTL: 30분)
        key = self.query_prefix + query
        self.client.setex(key, ttl, response)

    def get_embedding_cache(self, text: str) -> Optional[np.ndarray]:
        # 임베딩 캐시 조회
        key = self.embedding_prefix + text
        result = self.client.get(key)
        if result:
            # 바이너리 데이터를 numpy 배열로 변환
            return np.frombuffer(result, dtype=np.float32)
        return None

    def set_embedding_cache(self, text: str, embedding: np.ndarray):
        # 임베딩 캐시 저장 (만료 없음)
        key = self.embedding_prefix + text
        # numpy 배열을 바이너리로 변환
        self.client.set(key, embedding.tobytes())

    def get_cache_stats(self) -> dict:
        # 캐시 통계 조회
        info = self.client.info('stats')
        return {
            'hits': info.get('keyspace_hits', 0),
            'misses': info.get('keyspace_misses', 0),
            'keys': self.client.dbsize()
        }

# 실제 사용 예제
redis_cache = RedisCache(host='localhost', port=6379)

def rag_with_redis(query: str) -> str:
    # Redis에서 캐시 확인
    cached = redis_cache.get_query_cache(query)
    if cached:
        return cached

    # RAG 처리
    response = perform_rag(query)
    redis_cache.set_query_cache(query, response)
    return response

김개발 씨의 RAG 서비스는 인기를 끌면서 트래픽이 급증했습니다. 단일 서버로는 감당이 안 되어 로드 밸런서를 두고 서버를 2대로 늘렸습니다.

하지만 예상치 못한 문제가 발생했습니다. 서버 A에서 "배송 조회 방법"이라는 질문에 답변을 캐싱했는데, 다음 요청이 서버 B로 가면 캐시 미스가 발생했습니다.

각 서버가 독립적인 로컬 캐시를 사용하다 보니, 전체적으로 보면 캐시 효율이 50% 이하로 떨어진 것입니다. 박시니어 씨가 아키텍처 다이어그램을 그리며 설명했습니다.

"지금은 각 서버가 따로 놀고 있어요. Redis 같은 중앙 캐시 서버를 두면 모든 서버가 캐시를 공유할 수 있습니다." 그렇다면 Redis 통합이란 정확히 무엇일까요?

쉽게 비유하자면, Redis 통합은 마치 여러 지점을 가진 체인점이 중앙 재고 시스템을 공유하는 것과 같습니다. 강남점에서 재고를 확인한 정보를 강북점에서도 볼 수 있습니다.

마찬가지로 서버 A가 저장한 캐시를 서버 B, C, D가 모두 활용할 수 있습니다. Redis가 없던 시절에는 어땠을까요?

각 서버는 자신의 메모리나 로컬 파일에 캐시를 저장했습니다. 서버가 1대일 때는 문제없었지만, 여러 대로 확장하면 캐시가 중복되거나 파편화되었습니다.

더 큰 문제는 서버가 죽으면 캐시도 함께 사라진다는 점이었습니다. 또한 캐시 무효화도 어려웠습니다.

상품 정보가 업데이트되면 모든 서버의 캐시를 찾아서 삭제해야 했습니다. 이는 분산 시스템에서 매우 복잡한 작업이었습니다.

바로 이런 문제를 해결하기 위해 Redis가 등장했습니다. Redis는 인메모리 데이터베이스로, 매우 빠른 읽기/쓰기가 가능합니다.

보통 1ms 이하의 응답 시간을 보입니다. 또한 네트워크를 통해 여러 서버가 접근할 수 있어 중앙 캐시 저장소로 완벽합니다.

Redis를 사용하면 캐시 히트율이 크게 올라갑니다. 서버가 10대여도 캐시는 하나만 저장하면 됩니다.

또한 Redis는 자동 만료 기능을 제공합니다. TTL을 설정하면 오래된 캐시가 자동으로 삭제됩니다.

위의 코드를 한 줄씩 살펴보겠습니다. 먼저 RedisCache 클래스는 Redis 클라이언트를 초기화합니다.

decode_responses=False로 설정한 이유는 임베딩 벡터 같은 바이너리 데이터를 저장하기 위함입니다. query_prefix와 embedding_prefix는 키의 네임스페이스를 분리합니다.

이렇게 하면 "환불"이라는 쿼리와 "환불"이라는 텍스트의 임베딩이 충돌하지 않습니다. Redis에서는 이런 접두사 패턴이 표준 관행입니다.

set_query_cache 메서드는 setex를 사용합니다. 이것은 값을 저장하면서 동시에 TTL을 설정하는 명령입니다.

30분 후에는 자동으로 삭제되므로, 별도의 정리 작업이 필요 없습니다. get_embedding_cache와 set_embedding_cache는 numpy 배열을 다룹니다.

numpy 배열은 **tobytes()**로 바이너리로 변환하여 저장하고, **frombuffer()**로 다시 배열로 복원합니다. 이 방식은 JSON보다 훨씬 효율적입니다.

get_cache_stats 메서드는 Redis의 통계를 가져옵니다. 캐시 히트율을 계산하려면 hits와 misses 값을 확인하면 됩니다.

이는 성능 모니터링에 매우 유용합니다. 실제 현업에서는 어떻게 활용할까요?

예를 들어 쿠팡이나 배달의민족 같은 대규모 서비스를 생각해봅시다. 수백 대의 서버가 동시에 운영됩니다.

Redis 클러스터를 중앙에 두고 모든 서버가 캐시를 공유합니다. 한 서버에서 "치킨 맛집"을 검색한 결과는 다른 모든 서버에서도 재사용됩니다.

또 다른 활용 사례는 세션 관리와의 통합입니다. 사용자의 대화 이력을 Redis에 저장하면, 사용자가 어느 서버로 연결되든 이전 대화 맥락을 유지할 수 있습니다.

하지만 주의할 점도 있습니다. 초보 개발자들이 흔히 하는 실수 중 하나는 Redis를 단일 장애점으로 만드는 것입니다.

Redis 서버가 죽으면 전체 시스템이 느려집니다. 따라서 Redis Sentinel이나 Cluster를 구성하여 고가용성을 확보해야 합니다.

또 다른 실수는 너무 큰 데이터를 Redis에 저장하는 것입니다. Redis는 메모리 기반이므로, 수십 MB짜리 데이터를 저장하면 메모리가 금방 부족해집니다.

큰 데이터는 S3 같은 객체 스토리지를 사용하고, Redis에는 참조만 저장하는 것이 좋습니다. 다시 김개발 씨의 이야기로 돌아가 봅시다.

Redis를 도입한 후 시스템이 훨씬 안정적이고 빨라졌습니다. 캐시 히트율이 45%에서 82%로 올라갔고, 서버를 5대로 늘려도 캐시 효율이 유지되었습니다.

Redis 통합을 제대로 이해하면 확장 가능하고 고성능인 RAG 시스템을 구축할 수 있습니다. 특히 마이크로서비스 아키텍처에서는 필수적인 기술입니다.

실전 팁

💡 - Redis Sentinel이나 Cluster로 고가용성을 확보하세요

• 큰 데이터는 객체 스토리지에 저장하고 Redis에는 참조만 보관하세요
• 캐시 히트율을 모니터링하여 TTL과 메모리 크기를 최적화하세요

---

4. 실습: 캐시 시스템 구축

이론은 충분히 배웠다고 판단한 김개발 씨는 실제 프로젝트에 적용해보기로 했습니다. "처음부터 완벽한 시스템을 만들 필요는 없어.

단계적으로 구축해보자." 박시니어 씨도 응원했습니다. "좋은 생각이에요.

먼저 간단한 버전부터 시작하세요."

캐시 시스템 구축은 쿼리 캐싱, 임베딩 캐싱, Redis 통합을 실제로 결합하는 실습입니다. 마치 레고 블록을 조립하듯이 각 컴포넌트를 단계별로 통합하여 완전한 시스템을 만듭니다.

실무에서 바로 사용할 수 있는 수준의 코드를 작성해봅니다.

다음 코드를 살펴봅시다.

import redis
import hashlib
import numpy as np
from typing import Optional, Tuple
from openai import OpenAI

class RAGCacheSystem:
    def __init__(self, redis_host: str = 'localhost', redis_port: int = 6379):
        # Redis 클라이언트 초기화
        self.redis_client = redis.Redis(host=redis_host, port=redis_port, decode_responses=False)
        self.openai_client = OpenAI()

        # 캐시 키 접두사
        self.QUERY_PREFIX = "rag:query:"
        self.EMBEDDING_PREFIX = "rag:embedding:"

        # 통계 추적
        self.stats = {
            'query_cache_hits': 0,
            'query_cache_misses': 0,
            'embedding_cache_hits': 0,
            'embedding_cache_misses': 0
        }

    def _hash_text(self, text: str) -> str:
        # 텍스트를 정규화하고 해시 생성
        normalized = text.lower().strip()
        return hashlib.sha256(normalized.encode()).hexdigest()

    def get_cached_embedding(self, text: str) -> Optional[np.ndarray]:
        # 임베딩 캐시 조회
        cache_key = self.EMBEDDING_PREFIX + self._hash_text(text)
        cached = self.redis_client.get(cache_key)

        if cached:
            self.stats['embedding_cache_hits'] += 1
            return np.frombuffer(cached, dtype=np.float32)

        self.stats['embedding_cache_misses'] += 1
        return None

    def cache_embedding(self, text: str, embedding: np.ndarray):
        # 임베딩을 Redis에 저장
        cache_key = self.EMBEDDING_PREFIX + self._hash_text(text)
        self.redis_client.set(cache_key, embedding.tobytes())

    def get_embedding(self, text: str) -> np.ndarray:
        # 캐시 확인 후 없으면 생성
        cached = self.get_cached_embedding(text)
        if cached is not None:
            return cached

        # OpenAI API 호출
        response = self.openai_client.embeddings.create(
            model="text-embedding-3-small",
            input=text
        )
        embedding = np.array(response.data[0].embedding, dtype=np.float32)

        # 캐시에 저장
        self.cache_embedding(text, embedding)
        return embedding

    def query_rag(self, query: str, use_cache: bool = True) -> str:
        # 쿼리 캐시 확인
        if use_cache:
            cache_key = self.QUERY_PREFIX + self._hash_text(query)
            cached_response = self.redis_client.get(cache_key)

            if cached_response:
                self.stats['query_cache_hits'] += 1
                return cached_response.decode('utf-8')

        self.stats['query_cache_misses'] += 1

        # 실제 RAG 처리
        query_embedding = self.get_embedding(query)
        documents = self.vector_search(query_embedding)  # 벡터 검색
        response = self.generate_response(query, documents)  # LLM 생성

        # 응답을 캐시에 저장 (TTL: 30분)
        if use_cache:
            cache_key = self.QUERY_PREFIX + self._hash_text(query)
            self.redis_client.setex(cache_key, 1800, response)

        return response

    def get_stats(self) -> dict:
        # 캐시 효율 통계 반환
        total_queries = self.stats['query_cache_hits'] + self.stats['query_cache_misses']
        total_embeddings = self.stats['embedding_cache_hits'] + self.stats['embedding_cache_misses']

        return {
            'query_hit_rate': self.stats['query_cache_hits'] / total_queries if total_queries > 0 else 0,
            'embedding_hit_rate': self.stats['embedding_cache_hits'] / total_embeddings if total_embeddings > 0 else 0,
            'total_redis_keys': self.redis_client.dbsize()
        }

김개발 씨는 새 파이썬 파일을 열고 손가락을 움직이기 시작했습니다. 지금까지 배운 모든 것을 하나로 합칠 시간이었습니다.

"일단 클래스 구조부터 잡자." 먼저 전체 시스템의 뼈대를 설계했습니다. RAGCacheSystem이라는 클래스가 모든 캐싱 로직을 담당합니다.

Redis 연결, 임베딩 캐시, 쿼리 캐시를 모두 하나의 인터페이스로 제공하는 것입니다. 박시니어 씨가 코드 리뷰를 하며 조언했습니다.

"좋아요. 하지만 통계 추적 기능을 추가하면 어떨까요?

캐시가 얼마나 효과적인지 알아야 최적화할 수 있으니까요." 김개발 씨는 고개를 끄덕이며 stats 딕셔너리를 추가했습니다. 캐시 히트와 미스를 추적하면 나중에 히트율을 계산할 수 있습니다.

그렇다면 이 캐시 시스템은 어떻게 작동할까요? 쉽게 비유하자면, 이 시스템은 마치 똑똑한 도서관과 같습니다.

자주 찾는 책은 입구 근처에 두고, 책의 위치는 카드에 적어둡니다. 누군가 같은 책을 찾으면 뒤쪽 서가까지 가지 않고 바로 줄 수 있습니다.

처음부터 완벽한 시스템을 만들려고 하면 어떻게 될까요? 많은 주니어 개발자들이 이런 실수를 합니다.

"일단 모든 기능을 다 넣자!" 하지만 이렇게 하면 복잡도만 올라가고 버그가 생기기 쉽습니다. 더 좋은 방법은 단계적 구축입니다.

바로 이런 이유로 점진적 통합이 중요합니다. 먼저 가장 간단한 부분부터 시작합니다.

_hash_text 메서드는 텍스트를 정규화하고 해시를 생성합니다. 대소문자를 통일하고 공백을 제거하여, "Hello"와 "hello "가 같은 키를 갖도록 합니다.

다음으로 임베딩 캐싱을 구현합니다. get_cached_embedding은 Redis에서 임베딩을 찾습니다.

있으면 numpy 배열로 변환하여 반환하고, 없으면 None을 반환합니다. 이때 통계도 함께 업데이트합니다.

get_embedding 메서드는 캐시와 API 호출을 통합합니다. 먼저 캐시를 확인하고, 없으면 OpenAI API를 호출합니다.

그리고 결과를 캐시에 저장합니다. 이것이 캐시 어사이드 패턴의 전형적인 구현입니다.

위의 코드를 더 깊이 살펴보겠습니다. query_rag 메서드는 전체 RAG 파이프라인을 구현합니다.

먼저 쿼리 캐시를 확인합니다. 캐시 히트면 즉시 반환하고, 미스면 실제 처리를 시작합니다.

실제 처리 단계를 보면, 먼저 쿼리를 임베딩합니다. 이때 get_embedding을 호출하므로, 임베딩 캐싱도 자동으로 적용됩니다.

그다음 벡터 검색으로 관련 문서를 찾고, LLM에게 답변을 생성하도록 요청합니다. 마지막으로 생성된 응답을 setex로 저장합니다.

1800초는 30분입니다. 이 정도면 정보의 신선도를 유지하면서도 캐시 효과를 충분히 얻을 수 있습니다.

get_stats 메서드는 성능 분석에 매우 유용합니다. 히트율을 계산하여 캐시가 얼마나 효과적인지 확인할 수 있습니다.

만약 히트율이 낮다면 TTL을 늘리거나 캐시 전략을 조정해야 합니다. 실제 현업에서는 어떻게 활용할까요?

예를 들어 고객 지원 챗봇 서비스를 런칭한다고 가정해봅시다. 처음에는 이 시스템을 단일 서버에 배포합니다.

트래픽이 늘면 서버를 추가하면 됩니다. Redis가 중앙에 있으므로 서버를 몇 대를 추가하든 캐시는 공유됩니다.

실제로 한 스타트업은 이런 시스템으로 첫 달에 API 비용 3,000달러를 600달러로 줄였습니다. 더 중요한 것은 사용자 경험입니다.

평균 응답 시간이 4초에서 0.7초로 개선되었습니다. 하지만 주의할 점도 있습니다.

초보 개발자들이 흔히 하는 실수 중 하나는 에러 처리를 빼먹는 것입니다. Redis 연결이 끊어지면 어떻게 할까요?

현명한 방법은 캐시 실패 시 원본 로직으로 폴백하는 것입니다. 캐시는 성능 향상을 위한 것이지, 필수 기능은 아닙니다.

또 다른 실수는 캐시 키 설계를 대충하는 것입니다. 접두사가 없으면 다른 서비스와 키가 충돌할 수 있습니다.

또한 버전 정보를 키에 포함하면 시스템 업데이트 시 유용합니다. 다시 김개발 씨의 이야기로 돌아가 봅시다.

코드를 완성한 김개발 씨는 테스트를 돌려보았습니다. 첫 질문은 3.2초가 걸렸지만, 같은 질문을 다시 하니 0.05초 만에 답이 왔습니다.

"와, 64배나 빨라졌어!" 박시니어 씨가 웃으며 말했습니다. "이제 진짜 시스템다워졌네요.

모니터링 대시보드도 만들어서 히트율을 계속 추적하세요." 캐시 시스템 구축을 제대로 이해하면 이론을 실전에 적용하는 능력이 생깁니다. 단계별로 차근차근 구축하면 누구나 할 수 있습니다.

실전 팁

💡 - 에러 처리를 반드시 추가하여 Redis 장애 시에도 서비스가 동작하도록 하세요

• 캐시 키에 버전 정보를 포함하여 업데이트 시 유연하게 대응하세요
• 통계를 주기적으로 확인하고 TTL과 캐시 전략을 최적화하세요

---

5. 실습: 응답 속도 개선

시스템을 구축한 김개발 씨는 이제 실제 성능을 측정하고 개선하는 단계에 들어섰습니다. "캐시를 적용했으니 얼마나 빨라졌는지 정확히 측정해봐야겠어." 박시니어 씨가 조언했습니다.

"벤치마크를 돌려보고, 병목 지점을 찾아서 하나씩 최적화하세요."

응답 속도 개선은 벤치마킹과 프로파일링을 통해 시스템의 성능을 측정하고 최적화하는 실습입니다. 마치 자동차 정비사가 엔진을 튜닝하듯이, 각 단계의 소요 시간을 측정하고 개선점을 찾아냅니다.

실무에서 요구되는 성능 기준을 달성하는 방법을 배웁니다.

다음 코드를 살펴봅시다.

import time
from contextlib import contextmanager
from typing import Dict, List
import statistics

class PerformanceMonitor:
    def __init__(self):
        self.metrics: Dict[str, List[float]] = {}

    @contextmanager
    def measure(self, operation: str):
        # 특정 작업의 실행 시간 측정
        start = time.time()
        try:
            yield
        finally:
            elapsed = time.time() - start
            if operation not in self.metrics:
                self.metrics[operation] = []
            self.metrics[operation].append(elapsed)

    def get_stats(self, operation: str) -> dict:
        # 작업별 통계 계산
        if operation not in self.metrics:
            return {}

        times = self.metrics[operation]
        return {
            'count': len(times),
            'mean': statistics.mean(times),
            'median': statistics.median(times),
            'min': min(times),
            'max': max(times),
            'p95': statistics.quantiles(times, n=20)[18] if len(times) > 20 else max(times)
        }

    def print_report(self):
        # 전체 성능 리포트 출력
        print("\n=== Performance Report ===")
        for operation in self.metrics:
            stats = self.get_stats(operation)
            print(f"\n{operation}:")
            print(f"  Count: {stats['count']}")
            print(f"  Mean: {stats['mean']:.3f}s")
            print(f"  Median: {stats['median']:.3f}s")
            print(f"  P95: {stats['p95']:.3f}s")
            print(f"  Min/Max: {stats['min']:.3f}s / {stats['max']:.3f}s")

# 최적화된 RAG 시스템
class OptimizedRAGSystem(RAGCacheSystem):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.monitor = PerformanceMonitor()

    def query_rag_with_monitoring(self, query: str) -> tuple[str, dict]:
        # 각 단계별 성능 측정
        with self.monitor.measure('total_query'):
            # 쿼리 캐시 확인
            with self.monitor.measure('query_cache_lookup'):
                cache_key = self.QUERY_PREFIX + self._hash_text(query)
                cached_response = self.redis_client.get(cache_key)

            if cached_response:
                return cached_response.decode('utf-8'), {'cache': 'hit'}

            # 임베딩 생성
            with self.monitor.measure('embedding_generation'):
                query_embedding = self.get_embedding(query)

            # 벡터 검색
            with self.monitor.measure('vector_search'):
                documents = self.vector_search(query_embedding)

            # 응답 생성
            with self.monitor.measure('llm_generation'):
                response = self.generate_response(query, documents)

            # 캐시 저장
            with self.monitor.measure('cache_write'):
                self.redis_client.setex(cache_key, 1800, response)

            return response, {'cache': 'miss'}

    def benchmark(self, queries: List[str], iterations: int = 3):
        # 벤치마크 실행
        print(f"Running benchmark with {len(queries)} queries, {iterations} iterations each...")

        for iteration in range(iterations):
            for query in queries:
                self.query_rag_with_monitoring(query)

        # 결과 출력
        self.monitor.print_report()

        # 캐시 효율 출력
        cache_stats = self.get_stats()
        print(f"\n=== Cache Efficiency ===")
        print(f"Query Hit Rate: {cache_stats['query_hit_rate']:.1%}")
        print(f"Embedding Hit Rate: {cache_stats['embedding_hit_rate']:.1%}")

# 실제 사용 예제
monitor = PerformanceMonitor()
optimized_rag = OptimizedRAGSystem()

# 테스트 쿼리 (일부러 중복 포함)
test_queries = [
    "배송 조회 방법 알려줘",
    "환불 정책이 어떻게 되나요",
    "배송 조회 방법 알려줘",  # 중복
    "교환은 가능한가요",
    "환불 정책이 어떻게 되나요",  # 중복
]

# 벤치마크 실행
optimized_rag.benchmark(test_queries, iterations=2)

김개발 씨는 캐시 시스템을 프로덕션에 배포했습니다. 사용자들의 반응은 좋았지만, 김개발 씨는 "정말로 얼마나 빨라진 걸까?" 하는 궁금증이 생겼습니다.

감으로는 빨라진 것 같은데, 정확한 숫자가 필요했습니다. 박시니어 씨가 슬랙으로 메시지를 보냈습니다.

"김개발 씨, 성능 측정 안 하셨어요? 최적화는 측정에서 시작됩니다.

무엇을 측정할 수 없으면 개선할 수도 없어요." 김개발 씨는 고개를 끄덕이며 성능 모니터링 코드를 작성하기 시작했습니다. 그렇다면 응답 속도 개선은 어떻게 해야 할까요?

쉽게 비유하자면, 성능 개선은 마치 마라톤 선수가 기록을 단축하는 것과 같습니다. 먼저 현재 기록을 측정하고, 어느 구간이 느린지 분석합니다.

그다음 그 구간을 집중적으로 훈련하여 전체 기록을 향상시킵니다. 측정 없이 최적화를 시도하면 어떻게 될까요?

많은 개발자들이 "이 부분이 느릴 것 같아"라는 직감으로 최적화를 시작합니다. 하지만 실제로 측정해보면 전혀 다른 곳이 병목인 경우가 많습니다.

시간을 낭비하고 잘못된 곳을 최적화하는 것입니다. 더 큰 문제는 개선 효과를 알 수 없다는 점입니다.

최적화 전후를 비교할 데이터가 없으면, 정말로 나아졌는지 확인할 방법이 없습니다. 바로 이런 이유로 성능 모니터링이 필수입니다.

먼저 PerformanceMonitor 클래스를 만들었습니다. 이것은 각 작업의 실행 시간을 추적합니다.

measure 메서드는 컨텍스트 매니저로 구현되어, with 문으로 간편하게 사용할 수 있습니다. get_stats 메서드는 평균, 중앙값, 최소/최대, P95 같은 통계를 계산합니다.

P95는 95번째 백분위수로, 대부분의 요청이 이 시간 안에 완료된다는 의미입니다. 실무에서 SLA를 정할 때 자주 사용하는 지표입니다.

위의 코드를 더 자세히 살펴보겠습니다. OptimizedRAGSystem은 기존 RAGCacheSystem을 상속받아 모니터링 기능을 추가합니다.

query_rag_with_monitoring 메서드는 각 단계를 별도로 측정합니다. 먼저 query_cache_lookup으로 쿼리 캐시 조회 시간을 측정합니다.

이것은 보통 1-2ms로 매우 빠릅니다. 만약 이것이 느리다면 Redis 네트워크 지연이나 연결 풀 설정을 확인해야 합니다.

다음으로 embedding_generation은 임베딩 생성 시간을 측정합니다. 캐시 히트면 1ms 이하, 미스면 API 호출로 100-200ms가 걸립니다.

이 수치를 보면 임베딩 캐싱의 효과를 명확히 알 수 있습니다. vector_search는 벡터 데이터베이스 검색 시간입니다.

인덱스가 잘 구축되어 있으면 10-50ms 정도입니다. 만약 이것이 느리다면 인덱스 최적화나 샤딩을 고려해야 합니다.

llm_generation은 보통 가장 시간이 많이 걸립니다. GPT-4는 2-5초, GPT-3.5-turbo는 0.5-2초 정도입니다.

이것을 줄이려면 스트리밍 응답이나 더 빠른 모델을 사용해야 합니다. benchmark 메서드는 실제 부하 테스트를 수행합니다.

여러 쿼리를 반복 실행하여 캐시 효과를 측정합니다. 중복 쿼리를 의도적으로 포함하여 캐시 히트율을 확인합니다.

실제 현업에서는 어떻게 활용할까요? 예를 들어 서비스 목표가 "95%의 쿼리를 1초 이내에 응답"이라고 가정해봅시다.

벤치마크를 돌려서 현재 P95가 3.2초라는 것을 확인했습니다. 각 단계를 분석한 결과, LLM 생성이 2.8초를 차지했습니다.

해결책은 여러 가지입니다. 더 빠른 모델로 전환하거나, 프롬프트를 짧게 만들거나, 쿼리 캐싱 TTL을 늘려 히트율을 올리는 것입니다.

실제로 한 팀은 TTL을 30분에서 2시간으로 늘려 히트율을 65%에서 85%로 올렸고, P95를 1.2초로 줄였습니다. 하지만 주의할 점도 있습니다.

초보 개발자들이 흔히 하는 실수 중 하나는 평균값만 보는 것입니다. 평균이 0.5초여도 일부 사용자는 10초를 기다릴 수 있습니다.

따라서 P95, P99 같은 백분위수를 반드시 확인해야 합니다. 또 다른 실수는 로컬 환경에서만 테스트하는 것입니다.

실제 프로덕션 환경은 네트워크 지연, 동시 접속, 다양한 쿼리 패턴 등 변수가 많습니다. 스테이징 환경에서 실제와 유사한 조건으로 테스트해야 합니다.

다시 김개발 씨의 이야기로 돌아가 봅시다. 벤치마크 결과를 본 김개발 씨는 놀랐습니다.

첫 번째 실행은 평균 3.1초였는데, 두 번째 실행은 0.8초로 줄었습니다. 캐시 덕분이었습니다.

더 자세히 분석해보니, 쿼리 캐시 히트율은 60%, 임베딩 캐시 히트율은 80%였습니다. "임베딩 캐싱이 더 효과적이네.

같은 단어가 다른 질문에도 자주 쓰이니까 그런가?" 박시니어 씨가 리포트를 보며 칭찬했습니다. "잘했어요.

이제 이 데이터를 바탕으로 더 최적화할 수 있겠죠?" 김개발 씨는 다음 단계를 계획했습니다. 임베딩 배치 처리, 벡터 인덱스 튜닝, 쿼리 캐시 TTL 조정 등 할 일이 많았습니다.

하지만 이제는 무엇을 해야 할지 명확했습니다. 응답 속도 개선을 제대로 이해하면 데이터 기반으로 시스템을 최적화할 수 있습니다.

측정하고, 분석하고, 개선하는 사이클을 반복하면 누구나 고성능 시스템을 만들 수 있습니다.

실전 팁

💡 - 평균뿐만 아니라 P95, P99 같은 백분위수를 반드시 확인하세요

• 프로덕션과 유사한 환경에서 테스트하여 현실적인 결과를 얻으세요
• 각 단계별 소요 시간을 측정하여 진짜 병목을 찾아내세요

---

이상으로 학습을 마칩니다. 위 내용을 직접 코드로 작성해보면서 익혀보세요!