RAG 파이프라인 프로덕션 구축 가이드: 벡터 DB 선택부터 청킹·리랭킹·평가까지

[사용자 질의]
    │
    ▼
[Query Processing] ← 쿼리 분해, HyDE, 쿼리 확장
    │
    ▼
[Retrieval] ← BM25 + 벡터 검색 (하이브리드)
    │
    ▼
[Reranking] ← Cross-encoder, Cohere Rerank
    │
    ▼
[Context Assembly] ← 청크 병합, 중복 제거, 토큰 제한
    │
    ▼
[Generation] ← LLM에 컨텍스트와 질의 전달
    │
    ▼
[Post-processing] ← 출처 표시, 신뢰도 점수, 환각 검증

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# 1. 문서 청킹
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", ".", " "]
)
chunks = text_splitter.split_documents(documents)

# 2. 임베딩 및 벡터 저장소 생성
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    collection_name="production_docs"
)

# 3. Retriever 설정
retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 5, "fetch_k": 20}
)

# 4. RAG 체인 구성
template = """다음 컨텍스트를 기반으로 질문에 답변하세요.
컨텍스트에서 답을 찾을 수 없다면 "정보가 부족합니다"라고 답하세요.

컨텍스트:
{context}

질문: {question}
답변:"""

prompt = ChatPromptTemplate.from_template(template)
llm = ChatOpenAI(model="gpt-4o", temperature=0)

rag_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

response = rag_chain.invoke("RAG 파이프라인의 핵심 구성요소는?")

from sentence_transformers import SentenceTransformer
import numpy as np
from typing import List

class EmbeddingBenchmark:
    """임베딩 모델 비교 벤치마크"""

    def __init__(self, models: List[str]):
        self.models = {}
        for model_name in models:
            self.models[model_name] = SentenceTransformer(model_name)

    def evaluate_retrieval_quality(
        self,
        queries: List[str],
        relevant_docs: List[List[str]],
        corpus: List[str]
    ) -> dict:
        results = {}
        for name, model in self.models.items():
            corpus_embeddings = model.encode(corpus, normalize_embeddings=True)
            query_embeddings = model.encode(queries, normalize_embeddings=True)

            # 코사인 유사도 계산
            scores = np.dot(query_embeddings, corpus_embeddings.T)

            # Recall@k 계산
            recall_at_5 = self._compute_recall(scores, relevant_docs, k=5)
            recall_at_10 = self._compute_recall(scores, relevant_docs, k=10)

            results[name] = {
                "recall@5": recall_at_5,
                "recall@10": recall_at_10,
                "embedding_dim": model.get_sentence_embedding_dimension(),
                "encoding_speed": self._measure_speed(model, queries)
            }
        return results

    def _compute_recall(self, scores, relevant_docs, k):
        recalls = []
        for i, rel_docs in enumerate(relevant_docs):
            top_k_indices = np.argsort(scores[i])[-k:]
            retrieved = set(top_k_indices)
            relevant = set(rel_docs)
            recalls.append(len(retrieved & relevant) / len(relevant))
        return np.mean(recalls)

    def _measure_speed(self, model, texts, n_runs=3):
        import time
        times = []
        for _ in range(n_runs):
            start = time.time()
            model.encode(texts)
            times.append(time.time() - start)
        return np.mean(times)

from qdrant_client import QdrantClient
from qdrant_client.models import (
    Distance, VectorParams, PointStruct,
    Filter, FieldCondition, MatchValue,
    SearchParams, HnswConfigDiff
)
import uuid

class ProductionVectorStore:
    """프로덕션 벡터 저장소 관리자"""

    def __init__(self, url: str, api_key: str, collection_name: str):
        self.client = QdrantClient(url=url, api_key=api_key)
        self.collection_name = collection_name

    def create_collection(self, vector_size: int = 1536):
        """HNSW 인덱스 최적화된 컬렉션 생성"""
        self.client.create_collection(
            collection_name=self.collection_name,
            vectors_config=VectorParams(
                size=vector_size,
                distance=Distance.COSINE,
                on_disk=True  # 대규모 데이터셋을 위한 디스크 기반 저장
            ),
            hnsw_config=HnswConfigDiff(
                m=16,              # 그래프 연결 수 (기본: 16)
                ef_construct=128,  # 인덱스 빌드 시 탐색 범위
                full_scan_threshold=10000  # 이 수 이하면 전체 스캔
            ),
            optimizers_config={
                "indexing_threshold": 20000,
                "memmap_threshold": 50000
            }
        )

    def upsert_documents(self, documents: list, embeddings: list, metadata: list):
        """배치 업서트"""
        points = [
            PointStruct(
                id=str(uuid.uuid4()),
                vector=embedding,
                payload={**meta, "text": doc}
            )
            for doc, embedding, meta in zip(documents, embeddings, metadata)
        ]

        # 배치 크기 100으로 분할 업로드
        batch_size = 100
        for i in range(0, len(points), batch_size):
            batch = points[i:i + batch_size]
            self.client.upsert(
                collection_name=self.collection_name,
                points=batch,
                wait=True
            )

    def hybrid_search(self, query_vector: list, query_text: str,
                      filters: dict = None, top_k: int = 10):
        """하이브리드 검색 (벡터 + 메타데이터 필터)"""
        search_filter = None
        if filters:
            conditions = [
                FieldCondition(key=k, match=MatchValue(value=v))
                for k, v in filters.items()
            ]
            search_filter = Filter(must=conditions)

        results = self.client.search(
            collection_name=self.collection_name,
            query_vector=query_vector,
            query_filter=search_filter,
            limit=top_k,
            search_params=SearchParams(
                hnsw_ef=128,  # 검색 시 탐색 범위 (높을수록 정확, 느림)
                exact=False   # 근사 검색 사용
            )
        )
        return results

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings
import numpy as np

class AdvancedChunker:
    """다양한 청킹 전략을 지원하는 고급 청커"""

    def __init__(self, embedding_model: str = "text-embedding-3-small"):
        self.embeddings = OpenAIEmbeddings(model=embedding_model)

    def semantic_chunk(self, text: str, breakpoint_threshold: float = 0.5):
        """시맨틱 청킹 - 의미 변화 지점에서 분할"""
        chunker = SemanticChunker(
            self.embeddings,
            breakpoint_threshold_type="percentile",
            breakpoint_threshold_amount=breakpoint_threshold * 100
        )
        return chunker.split_text(text)

    def parent_child_chunk(self, text: str,
                            parent_size: int = 2000,
                            child_size: int = 400,
                            child_overlap: int = 50):
        """부모-자식 청킹 - 검색은 자식, 컨텍스트는 부모"""
        from langchain.text_splitter import RecursiveCharacterTextSplitter

        # 부모 청크 생성
        parent_splitter = RecursiveCharacterTextSplitter(
            chunk_size=parent_size,
            chunk_overlap=0
        )
        parent_chunks = parent_splitter.split_text(text)

        # 각 부모에서 자식 청크 생성
        child_splitter = RecursiveCharacterTextSplitter(
            chunk_size=child_size,
            chunk_overlap=child_overlap
        )

        result = []
        for i, parent in enumerate(parent_chunks):
            children = child_splitter.split_text(parent)
            for child in children:
                result.append({
                    "child_text": child,
                    "parent_text": parent,
                    "parent_id": i
                })
        return result

    def markdown_structure_chunk(self, markdown_text: str):
        """마크다운 문서 구조 기반 청킹"""
        from langchain.text_splitter import MarkdownHeaderTextSplitter

        headers_to_split_on = [
            ("#", "h1"),
            ("##", "h2"),
            ("###", "h3"),
        ]

        splitter = MarkdownHeaderTextSplitter(
            headers_to_split_on=headers_to_split_on
        )
        chunks = splitter.split_text(markdown_text)

        # 각 청크에 계층 구조 메타데이터 추가
        enriched_chunks = []
        for chunk in chunks:
            enriched_chunks.append({
                "text": chunk.page_content,
                "metadata": chunk.metadata,
                "hierarchy": " > ".join(
                    chunk.metadata.get(h, "")
                    for h in ["h1", "h2", "h3"]
                    if chunk.metadata.get(h)
                )
            })
        return enriched_chunks

from rank_bm25 import BM25Okapi
from langchain_openai import OpenAIEmbeddings
import numpy as np
from typing import List, Tuple

class HybridRetriever:
    """BM25 + 벡터 검색을 결합한 하이브리드 검색기"""

    def __init__(self, documents: List[str], embeddings_model: str = "text-embedding-3-large"):
        self.documents = documents
        self.embeddings = OpenAIEmbeddings(model=embeddings_model)

        # BM25 인덱스 구축
        tokenized_docs = [doc.lower().split() for doc in documents]
        self.bm25 = BM25Okapi(tokenized_docs)

        # 벡터 인덱스 구축
        self.doc_embeddings = np.array(
            self.embeddings.embed_documents(documents)
        )

    def search(self, query: str, top_k: int = 5,
               alpha: float = 0.5) -> List[Tuple[str, float]]:
        """
        하이브리드 검색 수행
        alpha: 벡터 검색 가중치 (0=BM25만, 1=벡터만)
        """
        # BM25 점수
        bm25_scores = self.bm25.get_scores(query.lower().split())
        bm25_scores = self._normalize_scores(bm25_scores)

        # 벡터 유사도 점수
        query_embedding = np.array(self.embeddings.embed_query(query))
        vector_scores = np.dot(self.doc_embeddings, query_embedding)
        vector_scores = self._normalize_scores(vector_scores)

        # 가중 결합 (Reciprocal Rank Fusion 대안)
        combined_scores = alpha * vector_scores + (1 - alpha) * bm25_scores

        # 상위 k개 반환
        top_indices = np.argsort(combined_scores)[-top_k:][::-1]
        return [
            (self.documents[i], combined_scores[i])
            for i in top_indices
        ]

    def _normalize_scores(self, scores: np.ndarray) -> np.ndarray:
        """Min-Max 정규화"""
        min_s, max_s = scores.min(), scores.max()
        if max_s == min_s:
            return np.zeros_like(scores)
        return (scores - min_s) / (max_s - min_s)

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate

class HyDERetriever:
    """HyDE 기반 검색기"""

    def __init__(self, vectorstore):
        self.vectorstore = vectorstore
        self.llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-large")

        self.hyde_prompt = ChatPromptTemplate.from_template(
            "다음 질문에 대해 가상의 답변을 작성해주세요. "
            "실제 정확성보다는 관련 키워드와 개념을 포함하는 것이 중요합니다.\n\n"
            "질문: {question}\n\n"
            "가상 답변:"
        )

    def retrieve(self, query: str, top_k: int = 5):
        """HyDE 검색 수행"""
        # 1. 가상 답변 생성
        chain = self.hyde_prompt | self.llm
        hypothetical_answer = chain.invoke({"question": query}).content

        # 2. 가상 답변의 임베딩 생성
        hyde_embedding = self.embeddings.embed_query(hypothetical_answer)

        # 3. 가상 답변 임베딩으로 검색
        results = self.vectorstore.similarity_search_by_vector(
            hyde_embedding, k=top_k
        )
        return results

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
import json

class QueryDecomposer:
    """복잡한 쿼리를 하위 쿼리로 분해"""

    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4o", temperature=0)
        self.decompose_prompt = ChatPromptTemplate.from_template(
            "다음 질문을 검색에 최적화된 2-4개의 하위 질문으로 분해하세요.\n"
            "JSON 배열 형태로 반환하세요.\n\n"
            "원래 질문: {question}\n\n"
            "하위 질문들:"
        )

    def decompose(self, question: str) -> list:
        """쿼리 분해"""
        chain = self.decompose_prompt | self.llm
        result = chain.invoke({"question": question}).content

        try:
            sub_queries = json.loads(result)
        except json.JSONDecodeError:
            sub_queries = [question]

        return sub_queries

    def retrieve_and_merge(self, question: str, retriever, top_k: int = 3):
        """분해된 쿼리로 검색 후 결과 병합"""
        sub_queries = self.decompose(question)

        all_docs = []
        seen_contents = set()

        for sub_query in sub_queries:
            docs = retriever.invoke(sub_query)[:top_k]
            for doc in docs:
                if doc.page_content not in seen_contents:
                    seen_contents.add(doc.page_content)
                    all_docs.append(doc)

        return all_docs

import cohere
from typing import List, Dict

class CohereReranker:
    """Cohere Rerank API를 활용한 리랭킹"""

    def __init__(self, api_key: str, model: str = "rerank-v3.5"):
        self.client = cohere.Client(api_key)
        self.model = model

    def rerank(self, query: str, documents: List[str],
               top_n: int = 5) -> List[Dict]:
        """문서 리랭킹"""
        response = self.client.rerank(
            model=self.model,
            query=query,
            documents=documents,
            top_n=top_n,
            return_documents=True
        )

        results = []
        for hit in response.results:
            results.append({
                "text": hit.document.text,
                "relevance_score": hit.relevance_score,
                "index": hit.index
            })
        return results


class CrossEncoderReranker:
    """로컬 Cross-Encoder 모델 기반 리랭킹"""

    def __init__(self, model_name: str = "cross-encoder/ms-marco-MiniLM-L-12-v2"):
        from sentence_transformers import CrossEncoder
        self.model = CrossEncoder(model_name)

    def rerank(self, query: str, documents: List[str],
               top_n: int = 5) -> List[Dict]:
        """Cross-encoder로 리랭킹"""
        pairs = [[query, doc] for doc in documents]
        scores = self.model.predict(pairs)

        # 점수 기준 정렬
        scored_docs = list(zip(documents, scores, range(len(documents))))
        scored_docs.sort(key=lambda x: x[1], reverse=True)

        results = []
        for text, score, idx in scored_docs[:top_n]:
            results.append({
                "text": text,
                "relevance_score": float(score),
                "index": idx
            })
        return results

class RAGPipelineWithReranking:
    """리랭킹이 통합된 RAG 파이프라인"""

    def __init__(self, retriever, reranker, llm):
        self.retriever = retriever
        self.reranker = reranker
        self.llm = llm

    def query(self, question: str, top_k_retrieve: int = 20,
              top_k_rerank: int = 5) -> str:
        """검색 -> 리랭킹 -> 생성"""
        # 1단계: 초기 검색 (넓게)
        initial_docs = self.retriever.invoke(question)[:top_k_retrieve]
        doc_texts = [doc.page_content for doc in initial_docs]

        # 2단계: 리랭킹 (정밀)
        reranked = self.reranker.rerank(
            query=question,
            documents=doc_texts,
            top_n=top_k_rerank
        )

        # 3단계: 컨텍스트 조립
        context = "\n\n---\n\n".join(
            r["text"] for r in reranked
        )

        # 4단계: LLM 생성
        prompt = f"""다음 컨텍스트를 기반으로 질문에 답변하세요.
각 정보의 출처를 표시하세요.

컨텍스트:
{context}

질문: {question}
답변:"""

        response = self.llm.invoke(prompt)
        return response.content

from ragas import evaluate
from ragas.metrics import (
    context_precision,
    context_recall,
    faithfulness,
    answer_relevancy,
    answer_correctness
)
from datasets import Dataset

class RAGEvaluator:
    """RAGAS 기반 RAG 평가기"""

    def __init__(self):
        self.metrics = [
            context_precision,
            context_recall,
            faithfulness,
            answer_relevancy,
            answer_correctness
        ]

    def create_eval_dataset(self, eval_samples: list) -> Dataset:
        """평가 데이터셋 생성"""
        data = {
            "question": [],
            "answer": [],
            "contexts": [],
            "ground_truth": []
        }

        for sample in eval_samples:
            data["question"].append(sample["question"])
            data["answer"].append(sample["generated_answer"])
            data["contexts"].append(sample["retrieved_contexts"])
            data["ground_truth"].append(sample["expected_answer"])

        return Dataset.from_dict(data)

    def evaluate(self, eval_samples: list) -> dict:
        """RAG 파이프라인 평가 수행"""
        dataset = self.create_eval_dataset(eval_samples)

        result = evaluate(
            dataset=dataset,
            metrics=self.metrics
        )

        return {
            "context_precision": result["context_precision"],
            "context_recall": result["context_recall"],
            "faithfulness": result["faithfulness"],
            "answer_relevancy": result["answer_relevancy"],
            "answer_correctness": result["answer_correctness"],
            "overall_score": sum(result.values()) / len(result)
        }

    def generate_report(self, results: dict) -> str:
        """평가 리포트 생성"""
        report = "=== RAG Pipeline Evaluation Report ===\n\n"

        for metric, score in results.items():
            status = "PASS" if score >= 0.7 else "WARN" if score >= 0.5 else "FAIL"
            report += f"  {metric}: {score:.4f} [{status}]\n"

        report += f"\n  Overall: {results.get('overall_score', 0):.4f}\n"
        return report

import asyncio
from datetime import datetime
from typing import Optional
import hashlib

class AsyncIndexingPipeline:
    """비동기 문서 인덱싱 파이프라인"""

    def __init__(self, embeddings, vector_store, chunker):
        self.embeddings = embeddings
        self.vector_store = vector_store
        self.chunker = chunker
        self.index_queue = asyncio.Queue()
        self._processed_hashes = set()

    async def enqueue_document(self, doc_id: str, content: str,
                                metadata: Optional[dict] = None):
        """문서를 인덱싱 큐에 추가"""
        doc_hash = hashlib.md5(content.encode()).hexdigest()

        if doc_hash in self._processed_hashes:
            return {"status": "skipped", "reason": "duplicate"}

        await self.index_queue.put({
            "doc_id": doc_id,
            "content": content,
            "metadata": metadata or {},
            "hash": doc_hash,
            "enqueued_at": datetime.utcnow().isoformat()
        })
        return {"status": "enqueued", "queue_size": self.index_queue.qsize()}

    async def process_queue(self, batch_size: int = 10):
        """큐에서 배치 단위로 문서 처리"""
        while True:
            batch = []
            try:
                for _ in range(batch_size):
                    item = self.index_queue.get_nowait()
                    batch.append(item)
            except asyncio.QueueEmpty:
                pass

            if batch:
                await self._process_batch(batch)
            else:
                await asyncio.sleep(1)

    async def _process_batch(self, batch: list):
        """배치 처리: 청킹 -> 임베딩 -> 업서트"""
        all_chunks = []
        all_metadata = []

        for item in batch:
            chunks = self.chunker.semantic_chunk(item["content"])
            for chunk in chunks:
                all_chunks.append(chunk)
                all_metadata.append({
                    "doc_id": item["doc_id"],
                    "hash": item["hash"],
                    **item["metadata"]
                })

        # 배치 임베딩 생성
        chunk_embeddings = self.embeddings.embed_documents(all_chunks)

        # 벡터 저장소에 업서트
        self.vector_store.upsert_documents(
            documents=all_chunks,
            embeddings=chunk_embeddings,
            metadata=all_metadata
        )

        for item in batch:
            self._processed_hashes.add(item["hash"])

import hashlib
import json
from datetime import datetime, timedelta
from typing import Optional
import redis

class RAGCache:
    """RAG 검색 결과 캐시"""

    def __init__(self, redis_url: str = "redis://localhost:6379",
                 ttl_seconds: int = 3600):
        self.redis = redis.from_url(redis_url)
        self.ttl = ttl_seconds

    def _cache_key(self, query: str, filters: Optional[dict] = None) -> str:
        """쿼리와 필터 기반 캐시 키 생성"""
        key_data = {"query": query.lower().strip(), "filters": filters or {}}
        key_hash = hashlib.sha256(
            json.dumps(key_data, sort_keys=True).encode()
        ).hexdigest()
        return f"rag:cache:{key_hash}"

    def get(self, query: str, filters: Optional[dict] = None):
        """캐시에서 검색 결과 조회"""
        key = self._cache_key(query, filters)
        cached = self.redis.get(key)

        if cached:
            data = json.loads(cached)
            data["cache_hit"] = True
            return data
        return None

    def set(self, query: str, results: list,
            filters: Optional[dict] = None):
        """검색 결과를 캐시에 저장"""
        key = self._cache_key(query, filters)
        data = {
            "results": results,
            "cached_at": datetime.utcnow().isoformat(),
            "query": query
        }
        self.redis.setex(key, self.ttl, json.dumps(data))

    def invalidate_by_pattern(self, pattern: str):
        """패턴 매칭으로 캐시 무효화"""
        keys = self.redis.keys(f"rag:cache:*{pattern}*")
        if keys:
            self.redis.delete(*keys)

import numpy as np
from datetime import datetime
from typing import List

class EmbeddingDriftMonitor:
    """임베딩 드리프트 감지 및 모니터링"""

    def __init__(self, reference_embeddings: np.ndarray):
        self.reference_mean = np.mean(reference_embeddings, axis=0)
        self.reference_std = np.std(reference_embeddings, axis=0)
        self.drift_history = []

    def check_drift(self, new_embeddings: np.ndarray,
                     threshold: float = 0.1) -> dict:
        """새 임베딩의 드리프트 검사"""
        new_mean = np.mean(new_embeddings, axis=0)

        # 코사인 거리로 드리프트 측정
        cosine_sim = np.dot(self.reference_mean, new_mean) / (
            np.linalg.norm(self.reference_mean) * np.linalg.norm(new_mean)
        )
        drift_score = 1 - cosine_sim

        # 분포 비교 (KL Divergence 근사)
        distribution_shift = np.mean(
            np.abs(np.mean(new_embeddings, axis=0) - self.reference_mean)
            / (self.reference_std + 1e-8)
        )

        result = {
            "drift_score": float(drift_score),
            "distribution_shift": float(distribution_shift),
            "is_drifted": drift_score > threshold,
            "timestamp": datetime.utcnow().isoformat(),
            "recommendation": (
                "재인덱싱 필요" if drift_score > threshold
                else "정상 범위"
            )
        }

        self.drift_history.append(result)
        return result

class RAGWithFallback:
    """폴백 전략이 포함된 RAG 파이프라인"""

    def __init__(self, primary_retriever, fallback_retriever,
                 reranker, llm):
        self.primary = primary_retriever
        self.fallback = fallback_retriever
        self.reranker = reranker
        self.llm = llm

    def query(self, question: str) -> dict:
        """폴백을 포함한 질의 처리"""
        # 1차: 기본 검색
        primary_docs = self.primary.invoke(question)

        if not primary_docs:
            # 폴백 1: 필터 완화
            primary_docs = self.fallback.invoke(question)

        if not primary_docs:
            return {
                "answer": "관련 정보를 찾을 수 없습니다.",
                "confidence": 0.0,
                "source": "no_results"
            }

        # 리랭킹
        reranked = self.reranker.rerank(
            query=question,
            documents=[d.page_content for d in primary_docs],
            top_n=5
        )

        # 최소 관련도 검증
        if reranked[0]["relevance_score"] < 0.3:
            return {
                "answer": "높은 신뢰도의 답변을 생성할 수 없습니다.",
                "confidence": reranked[0]["relevance_score"],
                "source": "low_confidence"
            }

        # LLM 생성
        context = "\n\n".join(r["text"] for r in reranked)
        answer = self.llm.invoke(
            f"컨텍스트:\n{context}\n\n질문: {question}\n답변:"
        ).content

        return {
            "answer": answer,
            "confidence": reranked[0]["relevance_score"],
            "source": "primary",
            "num_sources": len(reranked)
        }