10장: 프로덕션 RAG 파이프라인 구축

프로덕션 RAG 아키텍처:
 
사용자 --> API Gateway --> 캐시 확인 --> RAG 파이프라인
                              |
                     [캐시 히트] --> 캐시된 응답 반환
                     [캐시 미스] --> 검색 --> 리랭킹 --> LLM 생성
                                                          |
                                                    응답 + 로깅
                                                          |
                                                    모니터링 대시보드

import hashlib
import json
import time
import numpy as np
from langchain_openai import OpenAIEmbeddings
 
class SemanticCache:
    """시맨틱 유사도 기반 캐시"""
 
    def __init__(self, embeddings_model, similarity_threshold=0.95, ttl=3600):
        self.embeddings = embeddings_model
        self.threshold = similarity_threshold
        self.ttl = ttl  # 캐시 유효 시간 (초)
        self.cache = {}  # {key: {embedding, response, timestamp}}
 
    def _get_embedding(self, text):
        return self.embeddings.embed_query(text)
 
    def get(self, query):
        """캐시에서 유사한 질문의 답변을 검색"""
        query_embedding = self._get_embedding(query)
        current_time = time.time()
 
        best_match = None
        best_similarity = 0
 
        for key, entry in list(self.cache.items()):
            # TTL 확인
            if current_time - entry["timestamp"] > self.ttl:
                del self.cache[key]
                continue
 
            similarity = np.dot(query_embedding, entry["embedding"])
            if similarity > best_similarity:
                best_similarity = similarity
                best_match = entry
 
        if best_match and best_similarity >= self.threshold:
            return best_match["response"]
        return None
 
    def set(self, query, response):
        """캐시에 질문-답변 쌍 저장"""
        embedding = self._get_embedding(query)
        key = hashlib.md5(query.encode()).hexdigest()
        self.cache[key] = {
            "embedding": embedding,
            "response": response,
            "timestamp": time.time()
        }
 
# 캐시 적용 RAG 파이프라인
class CachedRAGPipeline:
    def __init__(self, rag_pipeline, cache):
        self.rag = rag_pipeline
        self.cache = cache
 
    def query(self, question):
        # 1. 캐시 확인
        cached = self.cache.get(question)
        if cached:
            return {"answer": cached, "source": "cache"}
 
        # 2. 캐시 미스 시 RAG 실행
        answer = self.rag.query(question)
 
        # 3. 결과 캐싱
        self.cache.set(question, answer)
        return {"answer": answer, "source": "rag"}

# LangSmith 트레이싱 설정
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-api-key"
os.environ["LANGCHAIN_PROJECT"] = "rag-production"
 
# 이후 LangChain 코드는 자동으로 트레이싱됨
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o")
 
# 모든 LLM 호출, 검색, 리랭킹 단계가 LangSmith에 기록됨

import time
import logging
from dataclasses import dataclass, field
 
@dataclass
class RAGMetrics:
    """RAG 파이프라인 메트릭 수집"""
    query_count: int = 0
    cache_hits: int = 0
    avg_latency_ms: float = 0
    avg_retrieval_ms: float = 0
    avg_reranking_ms: float = 0
    avg_generation_ms: float = 0
    avg_docs_retrieved: float = 0
    error_count: int = 0
    _latencies: list = field(default_factory=list)
 
    def record_query(self, latency_ms, retrieval_ms, reranking_ms,
                     generation_ms, docs_count, cached=False):
        self.query_count += 1
        if cached:
            self.cache_hits += 1
        self._latencies.append(latency_ms)
        self.avg_latency_ms = sum(self._latencies) / len(self._latencies)
        self.avg_retrieval_ms = (
            self.avg_retrieval_ms * (self.query_count - 1) + retrieval_ms
        ) / self.query_count
        self.avg_reranking_ms = (
            self.avg_reranking_ms * (self.query_count - 1) + reranking_ms
        ) / self.query_count
        self.avg_generation_ms = (
            self.avg_generation_ms * (self.query_count - 1) + generation_ms
        ) / self.query_count
        self.avg_docs_retrieved = (
            self.avg_docs_retrieved * (self.query_count - 1) + docs_count
        ) / self.query_count
 
    def record_error(self):
        self.error_count += 1
 
    def get_cache_hit_rate(self):
        if self.query_count == 0:
            return 0
        return self.cache_hits / self.query_count
 
    def summary(self):
        return {
            "total_queries": self.query_count,
            "cache_hit_rate": f"{self.get_cache_hit_rate():.1%}",
            "avg_latency": f"{self.avg_latency_ms:.0f}ms",
            "avg_retrieval": f"{self.avg_retrieval_ms:.0f}ms",
            "avg_reranking": f"{self.avg_reranking_ms:.0f}ms",
            "avg_generation": f"{self.avg_generation_ms:.0f}ms",
            "error_rate": f"{self.error_count/max(self.query_count,1):.1%}",
        }

class QualityMonitor:
    """실시간 응답 품질 모니터링"""
 
    def __init__(self, alert_threshold=0.5):
        self.threshold = alert_threshold
        self.quality_scores = []
 
    def check_response_quality(self, question, context, answer):
        """경량 품질 검사"""
        issues = []
 
        # 1. 빈 응답 검사
        if not answer or len(answer.strip()) < 10:
            issues.append("빈 응답 또는 너무 짧은 응답")
 
        # 2. 컨텍스트 미사용 검사
        context_words = set(context.lower().split())
        answer_words = set(answer.lower().split())
        overlap = len(context_words & answer_words) / max(len(answer_words), 1)
        if overlap < 0.1:
            issues.append("컨텍스트와 답변 간 어휘 중복이 매우 낮음")
 
        # 3. 불확실성 표현 검사
        uncertainty_phrases = [
            "잘 모르겠", "확실하지 않", "정보를 찾을 수 없",
            "알 수 없", "판단하기 어려"
        ]
        has_uncertainty = any(
            phrase in answer for phrase in uncertainty_phrases
        )
 
        # 4. 거부 응답 검사 (검색 실패 시 정상적인 거부인지)
        if has_uncertainty and len(context.strip()) > 100:
            issues.append("컨텍스트가 있는데 거부 응답을 생성함")
 
        quality_score = 1.0 - (len(issues) * 0.25)
        self.quality_scores.append(quality_score)
 
        if quality_score < self.threshold:
            logging.warning(
                f"품질 경고 - 점수: {quality_score}, "
                f"문제: {issues}, 질문: {question[:50]}"
            )
 
        return {"score": quality_score, "issues": issues}

import re
 
class PromptInjectionDetector:
    """프롬프트 인젝션 탐지"""
 
    SUSPICIOUS_PATTERNS = [
        r"ignore\s+(previous|above|all)\s+instructions",
        r"you\s+are\s+now",
        r"forget\s+(everything|all|your)",
        r"system\s*:\s*you",
        r"new\s+instructions?:",
        r"override\s+your",
        r"disregard\s+(previous|above)",
    ]
 
    def check(self, text):
        """텍스트에서 프롬프트 인젝션 패턴 검사"""
        text_lower = text.lower()
        for pattern in self.SUSPICIOUS_PATTERNS:
            if re.search(pattern, text_lower):
                return True
        return False
 
    def sanitize_context(self, documents):
        """검색된 문서에서 위험한 내용 필터링"""
        safe_docs = []
        for doc in documents:
            if not self.check(doc.page_content):
                safe_docs.append(doc)
            else:
                logging.warning(
                    f"프롬프트 인젝션 의심 문서 필터링: "
                    + doc.metadata.get("source", "unknown")
                )
        return safe_docs
 
detector = PromptInjectionDetector()

def secure_retrieval(query, user_id, user_permissions):
    """접근 제어가 적용된 검색"""
    # 메타데이터 필터로 접근 가능한 문서만 검색
    allowed_departments = user_permissions.get("departments", [])
 
    results = vectorstore.similarity_search(
        query,
        k=20,
        filter={
            "department": {"$in": allowed_departments}
        }
    )
 
    # 추가 보안 검증
    verified_results = []
    for doc in results:
        doc_acl = doc.metadata.get("access_level", "public")
        user_level = user_permissions.get("access_level", "basic")
 
        if can_access(user_level, doc_acl):
            verified_results.append(doc)
 
    return verified_results

import re
 
class PIIFilter:
    """개인식별정보 필터링"""
 
    PATTERNS = {
        "주민등록번호": r"\d{6}-[1-4]\d{6}",
        "전화번호": r"01[016789]-?\d{3,4}-?\d{4}",
        "이메일": r"[\w.+-]+@[\w-]+\.[\w.]+",
        "계좌번호": r"\d{3,4}-\d{2,6}-\d{2,6}",
    }
 
    def filter_response(self, text):
        """응답에서 PII를 마스킹"""
        filtered = text
        for pii_type, pattern in self.PATTERNS.items():
            filtered = re.sub(
                pattern,
                f"[{pii_type} 마스킹됨]",
                filtered
            )
        return filtered
 
pii_filter = PIIFilter()

import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
 
async def async_rag_query(question, retriever, llm):
    """비동기 RAG 쿼리"""
    # 검색은 동기적으로 (벡터 DB 클라이언트에 따라 다름)
    docs = await asyncio.to_thread(
        retriever.invoke, question
    )
 
    context = "\n\n".join([doc.page_content for doc in docs])
 
    # LLM 생성은 비동기 스트리밍
    prompt = ChatPromptTemplate.from_messages([
        ("system", "컨텍스트를 기반으로 답변하세요."),
        ("human", "컨텍스트:\n{context}\n\n질문: {question}")
    ])
 
    chain = prompt | llm
    response = await chain.ainvoke({
        "context": context,
        "question": question
    })
 
    return response.content
 
# FastAPI 엔드포인트
from fastapi import FastAPI
from pydantic import BaseModel
 
app = FastAPI()
 
class QueryRequest(BaseModel):
    question: str
 
class QueryResponse(BaseModel):
    answer: str
    latency_ms: float
 
@app.post("/query", response_model=QueryResponse)
async def query_endpoint(request: QueryRequest):
    start = time.time()
 
    # 캐시 확인
    cached = cache.get(request.question)
    if cached:
        latency = (time.time() - start) * 1000
        return QueryResponse(answer=cached, latency_ms=latency)
 
    # RAG 실행
    answer = await async_rag_query(
        request.question, retriever, llm
    )
 
    # 캐시 저장
    cache.set(request.question, answer)
 
    # PII 필터링
    answer = pii_filter.filter_response(answer)
 
    latency = (time.time() - start) * 1000
    metrics.record_query(latency_ms=latency, ...)
 
    return QueryResponse(answer=answer, latency_ms=latency)

from fastapi.responses import StreamingResponse
 
@app.post("/query/stream")
async def stream_query(request: QueryRequest):
    docs = await asyncio.to_thread(retriever.invoke, request.question)
    context = "\n\n".join([doc.page_content for doc in docs])
 
    prompt = ChatPromptTemplate.from_messages([
        ("system", "컨텍스트를 기반으로 답변하세요."),
        ("human", "컨텍스트:\n{context}\n\n질문: {question}")
    ])
 
    chain = prompt | llm
 
    async def generate():
        async for chunk in chain.astream({
            "context": context,
            "question": request.question
        }):
            if hasattr(chunk, "content"):
                yield chunk.content
 
    return StreamingResponse(generate(), media_type="text/plain")

from datetime import datetime
 
class FeedbackCollector:
    """사용자 피드백 수집 및 분석"""
 
    def __init__(self, storage):
        self.storage = storage
 
    def record_feedback(self, query_id, question, answer, rating, comment=None):
        """피드백 저장"""
        self.storage.save({
            "query_id": query_id,
            "question": question,
            "answer": answer,
            "rating": rating,  # 1-5 또는 thumbs up/down
            "comment": comment,
            "timestamp": datetime.now().isoformat()
        })
 
    def get_low_rated_queries(self, threshold=3):
        """낮은 평가를 받은 질문들을 분석용으로 추출"""
        all_feedback = self.storage.get_all()
        low_rated = [
            fb for fb in all_feedback if fb["rating"] < threshold
        ]
        return low_rated
 
    def generate_improvement_report(self):
        """개선 포인트 보고서 생성"""
        low_rated = self.get_low_rated_queries()
 
        if not low_rated:
            return "낮은 평가의 질문이 없습니다."
 
        # 실패 패턴 분류
        categories = {}
        for fb in low_rated:
            # LLM으로 실패 원인 분류
            category = classify_failure(fb["question"], fb["answer"])
            categories.setdefault(category, []).append(fb)
 
        report = "RAG 시스템 개선 보고서\n\n"
        for category, items in sorted(
            categories.items(),
            key=lambda x: len(x[1]),
            reverse=True
        ):
            report += f"## {category}: {len(items)}건\n"
            for item in items[:3]:
                report += f"- 질문: {item['question']}\n"
            report += "\n"
 
        return report

인프라:
  [ ] 벡터 데이터베이스 고가용성 구성 (복제, 백업)
  [ ] API 서버 오토스케일링 설정
  [ ] LLM API 레이트 리밋 및 폴백 설정
  [ ] 캐시 레이어 (Redis 등) 구성
 
품질:
  [ ] 평가 데이터셋 구축 (최소 100개 질문-답변 쌍)
  [ ] 기준선(Baseline) 메트릭 측정
  [ ] 실패 케이스 분석 및 대응 전략 수립
  [ ] A/B 테스트 인프라 구축
 
보안:
  [ ] 프롬프트 인젝션 방어 적용
  [ ] PII 필터링 적용
  [ ] 접근 제어 구현 (멀티테넌트 시)
  [ ] 입력 검증 및 레이트 리미팅
 
모니터링:
  [ ] 응답 지연 시간 추적
  [ ] LLM 토큰 사용량 추적
  [ ] 검색 품질 메트릭 대시보드
  [ ] 에러율 알림 설정
  [ ] 사용자 피드백 수집 파이프라인
 
운영:
  [ ] 증분 인덱싱 자동화 (새 문서 추가)
  [ ] 인덱스 백업 및 복구 절차
  [ ] 모델 업데이트 시 마이그레이션 계획
  [ ] 비용 모니터링 및 최적화

비용 구성 요소:
  1. 임베딩 생성:  인덱싱 시 1회, 쿼리마다 1회
  2. 벡터 DB:     저장 용량 + 쿼리 수
  3. LLM 생성:    입력 토큰 (컨텍스트) + 출력 토큰 (답변)
  4. 리랭킹:      검색당 1회 API 호출
 
최적화 방법:
  - 시맨틱 캐싱으로 반복 질문 처리 (비용 30-50% 절감)
  - Matryoshka 임베딩으로 벡터 차원 축소 (저장 비용 절감)
  - 리랭킹으로 top-K를 줄여 LLM 입력 토큰 절감
  - GPT-4o-mini 같은 경량 모델 활용 (단순 질문)
  - 배치 임베딩으로 API 호출 최적화