✍️ 필사 모드: Advanced RAG 파이프라인 완전 가이드 2025: 청킹 전략, 리랭킹, 에이전틱 RAG, 평가

사용자 질문 → [검색(Retrieval)] → [관련 문서] → [LLM + 문서 컨텍스트] → 답변 생성

Naive RAG (2023초)
├── 고정 크기 청킹
├── 단일 임베딩 검색
├── Top-K 결과를 그대로 LLM에 전달
└── 문제: 검색 정확도 낮음, 컨텍스트 노이즈

Advanced RAG (2024)
├── 시맨틱 청킹 + 메타데이터
├── 하이브리드 검색 (벡터 + 키워드)
├── 리랭킹으로 검색 결과 정제
├── 쿼리 변환 (HyDE, Multi-Query)
└── 컨텍스트 압축

Modular RAG (2025)
├── 에이전틱 RAG (동적 라우팅)
├── Self-RAG (자기 반성)
├── CRAG (Corrective RAG)
├── Multi-modal RAG
├── Knowledge Graph + RAG
└── 모듈 조합형 파이프라인

나쁜 청킹: 문장 중간에서 잘림 → 의미 손실 → 검색 실패 → 잘못된 답변
좋은 청킹: 의미 단위로 분리 → 풍부한 컨텍스트 → 정확한 검색 → 정확한 답변

from langchain.text_splitter import CharacterTextSplitter

splitter = CharacterTextSplitter(
    chunk_size=500,      # 청크 크기
    chunk_overlap=50,    # 오버랩 (경계 정보 보존)
    separator="\n\n"     # 분리자
)

chunks = splitter.split_text(document_text)

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", ". ", " ", ""]  # 우선순위
)

chunks = splitter.split_text(document_text)
# 먼저 \n\n으로 분리 시도, 크기 초과 시 \n, 그다음 마침표...

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# 시맨틱 청킹
chunker = SemanticChunker(
    embeddings,
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=95  # 유사도가 95번째 백분위수 이하면 분할
)

chunks = chunker.split_text(document_text)

문장1: "벡터DB는 AI 인프라의 핵심이다." ──→ 임베딩 [0.1, 0.3, ...]
문장2: "HNSW는 가장 빠른 검색 알고리즘이다." ──→ 유사도 0.85 (높음 → 같은 청크)
문장3: "한편, 한국의 날씨는 맑다." ──→ 유사도 0.15 (낮음 → 새 청크 시작!)

from langchain.text_splitter import MarkdownHeaderTextSplitter

headers_to_split = [
    ("#", "Header 1"),
    ("##", "Header 2"),
    ("###", "Header 3"),
]

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

# 각 청크에 제목 메타데이터 자동 포함
for chunk in chunks:
    print(f"메타데이터: {chunk.metadata}")
    # {'Header 1': 'Vector Database', 'Header 2': '인덱싱 알고리즘'}

from openai import OpenAI

client = OpenAI()

def agentic_chunk(text, max_chunk_size=1500):
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": """주어진 텍스트를 의미적으로 독립적인 청크로 분할하세요.
각 청크는 자체적으로 이해 가능해야 합니다.
JSON 배열로 청크 텍스트를 반환하세요."""},
            {"role": "user", "content": text}
        ],
        response_format={"type": "json_object"}
    )
    return response.choices[0].message.content

일반 텍스트: 500~1000 토큰 (오버랩 10~20%)
기술 문서: 800~1500 토큰 (섹션 단위)
법률/의료 문서: 300~500 토큰 (정밀도 중시)
코드: 함수/클래스 단위 (구조 기반)
대화/QA: 대화 턴 단위

비용 중시 + 범용        → text-embedding-3-small
최고 성능 + 무료        → BGE-M3 (셀프호스팅)
다국어 + 검색 특화      → Cohere embed-v3.0
코드 검색              → voyage-code-3
긴 문서 (8K+ 토큰)     → Jina-embeddings-v3
프라이빗 데이터 + 온프레미스 → BGE-M3 or Nomic

# ColBERT v2 예시
from ragatouille import RAGPretrainedModel

rag = RAGPretrainedModel.from_pretrained("colbert-ir/colbertv2.0")

# 인덱싱
rag.index(
    collection=[doc1, doc2, doc3],
    index_name="my_index"
)

# 검색 (토큰 수준 매칭)
results = rag.search(query="벡터 데이터베이스 성능 비교", k=5)

원본 쿼리: "RAG 느려요" (모호, 짧음)
→ 변환 후: "RAG 파이프라인 검색 지연시간 최적화 방법" (구체적, 검색 적합)

from openai import OpenAI

client = OpenAI()

def hyde_search(query, collection):
    # 1. LLM으로 가상 답변 생성
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "주어진 질문에 대한 상세한 답변을 작성하세요. 정확하지 않아도 됩니다."},
            {"role": "user", "content": query}
        ]
    )
    hypothetical_doc = response.choices[0].message.content
    
    # 2. 가상 답변을 임베딩하여 검색
    embedding = get_embedding(hypothetical_doc)
    results = collection.search(query_vector=embedding, limit=5)
    
    return results

def multi_query_transform(original_query):
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": """주어진 질문을 3가지 다른 관점에서 재작성하세요.
각 질문은 원래 의도를 유지하되 다른 키워드를 사용하세요.
줄바꿈으로 구분하여 3개의 질문만 반환하세요."""},
            {"role": "user", "content": original_query}
        ]
    )
    queries = response.choices[0].message.content.strip().split("\n")
    return [original_query] + queries

# 사용
queries = multi_query_transform("RAG 파이프라인 성능 최적화")
# → ["RAG 파이프라인 성능 최적화",
#    "검색 증강 생성 시스템의 응답 시간 개선 방법",
#    "RAG 아키텍처에서 검색 정확도를 높이는 전략",
#    "LLM 기반 문서 검색 시스템 최적화 기법"]

# 각 쿼리로 검색 후 결과 합치기 (deduplicate)
all_results = set()
for q in queries:
    results = search(q)
    all_results.update(results)

def step_back_prompt(query):
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": """주어진 질문보다 한 단계 더 일반적이고 넓은 질문을 생성하세요.
이 일반적 질문의 답변이 원래 질문에 답하는 데 도움이 되어야 합니다."""},
            {"role": "user", "content": query}
        ]
    )
    return response.choices[0].message.content

# 예시
original = "Qdrant에서 HNSW M 파라미터를 32로 설정했을 때 메모리 영향은?"
step_back = step_back_prompt(original)
# → "벡터 데이터베이스에서 HNSW 인덱스 파라미터가 성능과 리소스에 미치는 영향은?"

def decompose_query(complex_query):
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": """복잡한 질문을 간단한 하위 질문들로 분해하세요.
각 하위 질문은 독립적으로 답변 가능해야 합니다.
JSON 배열로 반환하세요."""},
            {"role": "user", "content": complex_query}
        ],
        response_format={"type": "json_object"}
    )
    return response.choices[0].message.content

# 예시
complex_q = "Pinecone과 Qdrant의 성능, 비용, 운영 편의성을 비교하고 10M 벡터 규모에서 어떤 것을 추천하나요?"
sub_questions = decompose_query(complex_q)
# → ["Pinecone의 10M 벡터 규모 성능은?",
#    "Qdrant의 10M 벡터 규모 성능은?",
#    "Pinecone의 비용 구조는?",
#    "Qdrant의 비용 구조는?",
#    "Pinecone의 운영 편의성은?",
#    "Qdrant의 운영 편의성은?"]

from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from langchain_community.vectorstores import Qdrant

# BM25 (키워드) 검색기
bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 5

# 벡터 검색기
vector_retriever = qdrant_vectorstore.as_retriever(
    search_kwargs={"k": 5}
)

# 앙상블 (하이브리드)
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6]  # 벡터에 가중치
)

results = ensemble_retriever.invoke("RAG 파이프라인 최적화 방법")

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor

compressor = LLMChainExtractor.from_llm(llm)

compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vector_retriever
)

# 검색 + 압축: 관련 부분만 추출
compressed_docs = compression_retriever.invoke("HNSW 알고리즘의 M 파라미터 역할")
# 원본 문서의 전체가 아닌, 질문과 관련된 문단만 반환

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
from langchain.text_splitter import RecursiveCharacterTextSplitter

# 작은 청크: 검색용
child_splitter = RecursiveCharacterTextSplitter(chunk_size=200)

# 큰 청크: 컨텍스트용
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000)

store = InMemoryStore()

retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=store,
    child_splitter=child_splitter,
    parent_splitter=parent_splitter,
)

# 200토큰 청크로 정밀 검색 → 2000토큰 상위 문서 반환
retriever.add_documents(documents)
results = retriever.invoke("HNSW 파라미터 튜닝")
# 검색은 작은 청크로 정확하게, 반환은 큰 컨텍스트로 풍부하게

from langchain.retrievers.multi_vector import MultiVectorRetriever

# 문서별로 요약 + 가상 질문 생성
summaries = generate_summaries(documents)
hypothetical_questions = generate_questions(documents)

# 요약/질문으로 검색, 원본 문서 반환
retriever = MultiVectorRetriever(
    vectorstore=vectorstore,  # 요약/질문 벡터 저장
    docstore=store,           # 원본 문서 저장
    id_key="doc_id"
)

# 요약의 임베딩으로 검색하되 원본 전체 문서를 반환

1단계 (Bi-encoder): 쿼리 벡터 vs 문서 벡터 → 빠르지만 부정확 → Top 20
2단계 (Cross-encoder): (쿼리, 문서) 쌍을 직접 비교 → 느리지만 정확 → Top 5

from sentence_transformers import CrossEncoder

# Cross-encoder 모델 로드
model = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

# 초기 검색 결과
initial_results = vector_search(query, top_k=20)

# 리랭킹
pairs = [(query, doc.content) for doc in initial_results]
scores = model.predict(pairs)

# 점수 기준 재정렬
reranked = sorted(
    zip(initial_results, scores),
    key=lambda x: x[1],
    reverse=True
)[:5]

import cohere

co = cohere.Client("YOUR_API_KEY")

# 초기 검색 결과 텍스트
documents = [doc.content for doc in initial_results]

# Cohere 리랭킹
response = co.rerank(
    model="rerank-v3.5",
    query="RAG 파이프라인 최적화",
    documents=documents,
    top_n=5
)

for result in response.results:
    print(f"Index: {result.index}, Score: {result.relevance_score:.4f}")
    print(f"Text: {documents[result.index][:100]}...")

from ragatouille import RAGPretrainedModel

# ColBERT 모델 로드
rag = RAGPretrainedModel.from_pretrained("colbert-ir/colbertv2.0")

# 리랭킹
reranked = rag.rerank(
    query="RAG 파이프라인 최적화 방법",
    documents=[doc.content for doc in initial_results],
    k=5
)

def llm_rerank(query, documents, top_n=5):
    scored_docs = []
    
    for doc in documents:
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": "질문과 문서의 관련도를 0~10 점수로 평가하세요. 숫자만 반환하세요."},
                {"role": "user", "content": f"질문: {query}\n\n문서: {doc.content[:500]}"}
            ]
        )
        score = float(response.choices[0].message.content.strip())
        scored_docs.append((doc, score))
    
    return sorted(scored_docs, key=lambda x: x[1], reverse=True)[:top_n]

전통 RAG: 질문 → 검색 → 생성 (고정 파이프라인)
에이전틱 RAG: 질문 → [에이전트가 판단]
              ├── 검색이 필요한가? → 검색 → 결과 충분한가?
              │                         ├── Yes → 생성
              │                         └── No → 다른 소스 검색 / 쿼리 수정
              └── 검색 없이 답변 가능 → 직접 생성

def self_rag(query):
    # 1. 검색 필요 여부 판단
    need_retrieval = judge_retrieval_need(query)
    
    if not need_retrieval:
        return generate_without_context(query)
    
    # 2. 검색 수행
    documents = retrieve(query)
    
    # 3. 각 문서의 관련성 판단
    relevant_docs = []
    for doc in documents:
        if is_relevant(query, doc):
            relevant_docs.append(doc)
    
    if not relevant_docs:
        # 관련 문서 없으면 쿼리 수정 후 재검색
        refined_query = refine_query(query)
        documents = retrieve(refined_query)
        relevant_docs = [d for d in documents if is_relevant(refined_query, d)]
    
    # 4. 답변 생성
    answer = generate_with_context(query, relevant_docs)
    
    # 5. 답변 품질 자체 평가
    if not is_supported(answer, relevant_docs):
        # 답변이 문서에 근거하지 않으면 재생성
        answer = regenerate(query, relevant_docs)
    
    return answer

def corrective_rag(query):
    # 1. 초기 검색
    documents = retrieve(query)
    
    # 2. 검색 결과 품질 평가
    quality = evaluate_retrieval_quality(query, documents)
    
    if quality == "CORRECT":
        # 검색 결과 우수 → 지식 정제 후 생성
        refined_knowledge = refine_knowledge(documents)
        return generate(query, refined_knowledge)
    
    elif quality == "AMBIGUOUS":
        # 검색 결과 애매 → 웹 검색으로 보충
        web_results = web_search(query)
        combined = documents + web_results
        refined = refine_knowledge(combined)
        return generate(query, refined)
    
    elif quality == "INCORRECT":
        # 검색 결과 부적절 → 웹 검색으로 대체
        web_results = web_search(query)
        refined = refine_knowledge(web_results)
        return generate(query, refined)

def adaptive_rag(query):
    # 쿼리 복잡도 분류
    complexity = classify_query(query)
    
    if complexity == "SIMPLE":
        # 간단한 사실 질문 → 직접 검색 + 생성
        docs = simple_retrieve(query, top_k=3)
        return generate(query, docs)
    
    elif complexity == "MODERATE":
        # 보통 → 멀티쿼리 + 리랭킹
        queries = multi_query_transform(query)
        docs = hybrid_search(queries)
        reranked = rerank(query, docs)
        return generate(query, reranked)
    
    elif complexity == "COMPLEX":
        # 복잡 → 쿼리 분해 + 다단계 추론
        sub_queries = decompose_query(query)
        sub_answers = []
        for sq in sub_queries:
            docs = search(sq)
            sub_answers.append(generate(sq, docs))
        return synthesize(query, sub_answers)

def query_router(query):
    # LLM으로 라우팅 결정
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": """쿼리를 분석하여 적절한 데이터 소스를 선택하세요:
- VECTOR_DB: 내부 문서 검색이 필요한 경우
- WEB_SEARCH: 최신 정보나 외부 정보가 필요한 경우
- SQL_DB: 정형 데이터 쿼리가 필요한 경우
- DIRECT: LLM이 직접 답변 가능한 경우
한 단어만 반환하세요."""},
            {"role": "user", "content": query}
        ]
    )
    route = response.choices[0].message.content.strip()
    
    if route == "VECTOR_DB":
        return vector_db_search(query)
    elif route == "WEB_SEARCH":
        return web_search(query)
    elif route == "SQL_DB":
        return sql_query(query)
    else:
        return direct_answer(query)

from langchain_community.document_loaders import UnstructuredPDFLoader

# PDF에서 텍스트 + 이미지 + 표 추출
loader = UnstructuredPDFLoader(
    "document.pdf",
    mode="elements",
    strategy="hi_res"  # 고해상도 이미지/표 추출
)
elements = loader.load()

# 이미지 요소에 대해 GPT-4o로 설명 생성
for element in elements:
    if element.metadata.get("type") == "Image":
        description = describe_image_with_vision(element)
        element.page_content = description  # 이미지 설명을 텍스트로 변환

# 모든 요소 (텍스트 + 이미지 설명)를 벡터 DB에 저장

def process_table(table_element):
    """표를 검색 가능한 형태로 변환"""
    # 1. 표를 마크다운으로 변환
    md_table = table_element.metadata.get("text_as_html", "")
    
    # 2. LLM으로 표 요약 생성
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "다음 표의 내용을 자연어로 요약하세요."},
            {"role": "user", "content": md_table}
        ]
    )
    summary = response.choices[0].message.content
    
    # 3. 요약을 메타데이터와 함께 저장
    return {
        "content": summary,
        "metadata": {
            "type": "table",
            "original_html": md_table,
            "source": table_element.metadata.get("source")
        }
    }

import base64

def describe_image_with_vision(image_path):
    with open(image_path, "rb") as f:
        image_data = base64.b64encode(f.read()).decode()
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": "이 이미지의 내용을 상세히 설명하세요. 다이어그램이면 구조를, 차트면 데이터를 설명하세요."},
                    {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{image_data}"}}
                ]
            }
        ]
    )
    return response.choices[0].message.content

일반 RAG: 문서 청크를 독립적으로 검색
GraphRAG: 엔티티 관계를 활용하여 연결된 정보까지 검색

예시: "HNSW를 사용하는 벡터 데이터베이스는?"
일반 RAG: HNSW에 대한 청크만 반환
GraphRAG: HNSW → (사용됨) → Qdrant, Weaviate, Milvus
         + 각 DB의 HNSW 설정 관련 청크까지 반환

from langchain_community.graphs import Neo4jGraph
from langchain.chains import GraphCypherQAChain

graph = Neo4jGraph(
    url="bolt://localhost:7687",
    username="neo4j",
    password="password"
)

# 지식 그래프 + LLM QA 체인
chain = GraphCypherQAChain.from_llm(
    llm=llm,
    graph=graph,
    verbose=True
)

result = chain.invoke("Qdrant가 지원하는 인덱싱 알고리즘은 무엇인가요?")
# LLM이 Cypher 쿼리를 생성하여 Neo4j에서 관계를 탐색

def hybrid_graph_vector_rag(query):
    # 1. 벡터 검색으로 관련 문서 청크
    vector_results = vector_search(query, top_k=5)
    
    # 2. 청크에서 엔티티 추출
    entities = extract_entities(query)
    
    # 3. 지식 그래프에서 관련 엔티티 탐색
    graph_results = graph_query(entities, depth=2)
    
    # 4. 두 결과를 결합
    combined_context = merge_results(vector_results, graph_results)
    
    # 5. LLM 생성
    return generate(query, combined_context)

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

# 평가 데이터셋 준비
eval_data = {
    "question": ["RAG란 무엇인가?", "HNSW 알고리즘의 원리는?"],
    "answer": ["RAG는 검색 증강 생성으로...", "HNSW는 계층적 그래프..."],
    "contexts": [
        ["RAG(Retrieval-Augmented Generation)은...", "검색 기반 생성 기술..."],
        ["HNSW는 Hierarchical Navigable...", "그래프 기반 ANN 알고리즘..."]
    ],
    "ground_truth": ["RAG는 LLM이 외부 지식을...", "HNSW는 다층 그래프 구조의..."]
}

dataset = Dataset.from_dict(eval_data)

# RAGAS 평가 실행
results = evaluate(
    dataset,
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall]
)

print(results)
# {'faithfulness': 0.89, 'answer_relevancy': 0.92,
#  'context_precision': 0.85, 'context_recall': 0.78}

from trulens.apps.langchain import TruChain
from trulens.core import Feedback, TruSession
from trulens.providers.openai import OpenAI as TruOpenAI

session = TruSession()
provider = TruOpenAI()

# 피드백 함수 정의
f_relevance = Feedback(provider.relevance).on_input_output()
f_groundedness = Feedback(provider.groundedness_measure_with_cot_reasons).on(
    source=context, statement=output
)

# RAG 체인 래핑
tru_chain = TruChain(
    rag_chain,
    app_name="RAG Pipeline v1",
    feedbacks=[f_relevance, f_groundedness]
)

# 평가 실행
with tru_chain as recording:
    response = rag_chain.invoke("RAG 파이프라인 최적화 방법은?")

# 대시보드 확인
session.get_leaderboard()

def llm_judge(question, answer, context, ground_truth):
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": """RAG 시스템의 답변을 평가하세요.
다음 기준으로 1~5점을 매기세요:
1. 정확성: 답변이 사실적으로 정확한가
2. 관련성: 질문에 적합한 답변인가
3. 근거성: 제공된 컨텍스트에 기반한 답변인가
4. 완전성: 질문에 충분히 답변했는가
JSON으로 각 점수와 이유를 반환하세요."""},
            {"role": "user", "content": f"""질문: {question}
답변: {answer}
컨텍스트: {context}
정답: {ground_truth}"""}
        ],
        response_format={"type": "json_object"}
    )
    return response.choices[0].message.content

import hashlib
import json
from functools import lru_cache

class RAGCache:
    def __init__(self, redis_client):
        self.redis = redis_client
        self.ttl = 3600  # 1시간
    
    def _hash_query(self, query):
        return hashlib.md5(query.encode()).hexdigest()
    
    def get_cached_response(self, query):
        key = self._hash_query(query)
        cached = self.redis.get(f"rag:{key}")
        if cached:
            return json.loads(cached)
        return None
    
    def cache_response(self, query, response):
        key = self._hash_query(query)
        self.redis.setex(
            f"rag:{key}",
            self.ttl,
            json.dumps(response)
        )

# 시맨틱 캐싱: 유사한 쿼리도 캐시 히트
class SemanticCache:
    def __init__(self, vectorstore, threshold=0.95):
        self.store = vectorstore
        self.threshold = threshold
    
    def get(self, query):
        results = self.store.similarity_search_with_score(query, k=1)
        if results and results[0][1] >= self.threshold:
            return results[0][0].metadata["response"]
        return None

from openai import OpenAI

client = OpenAI()

def stream_rag_response(query, context):
    stream = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": f"컨텍스트:\n{context}"},
            {"role": "user", "content": query}
        ],
        stream=True
    )
    
    for chunk in stream:
        if chunk.choices[0].delta.content:
            yield chunk.choices[0].delta.content

# FastAPI 스트리밍 엔드포인트
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.get("/rag/stream")
async def rag_stream(query: str):
    context = retrieve_context(query)
    return StreamingResponse(
        stream_rag_response(query, context),
        media_type="text/event-stream"
    )

def rag_with_fallback(query):
    try:
        # 1차: 벡터 DB 검색
        docs = vector_search(query)
        
        if not docs or max_relevance_score(docs) < 0.5:
            # 2차: 웹 검색 폴백
            docs = web_search_fallback(query)
        
        if not docs:
            # 3차: LLM 직접 답변 (검색 없이)
            return direct_llm_answer(query)
        
        return generate_with_context(query, docs)
    
    except Exception as e:
        # 에러 폴백
        return {
            "answer": "죄송합니다. 일시적인 오류가 발생했습니다.",
            "error": str(e),
            "fallback": True
        }

# 핵심 RAG 메트릭
monitoring = {
    "검색 지연 시간": "목표 p99 200ms",
    "생성 지연 시간": "목표 p99 3s (스트리밍 first-token 500ms)",
    "검색 적합도 (Relevance)": "자동 평가 0.85+",
    "답변 근거성 (Groundedness)": "자동 평가 0.90+",
    "캐시 히트율": "목표 30%+",
    "사용자 피드백 (좋아요/싫어요)": "긍정 80%+",
    "토큰 사용량": "비용 추적",
    "에러율": "목표 1% 이하",
}

def debug_rag_pipeline(query):
    print(f"=== RAG 디버깅: {query} ===\n")
    
    # 1. 검색 결과 확인
    results = vector_search(query, top_k=10)
    print("--- 검색 결과 ---")
    for i, r in enumerate(results):
        print(f"#{i+1} Score: {r.score:.4f} | {r.content[:100]}...")
    
    # 2. 리랭킹 결과 확인
    reranked = rerank(query, results)
    print("\n--- 리랭킹 후 ---")
    for i, (r, score) in enumerate(reranked):
        print(f"#{i+1} Score: {score:.4f} | {r.content[:100]}...")
    
    # 3. 최종 컨텍스트 확인
    context = build_context(reranked[:5])
    print(f"\n--- 컨텍스트 길이: {len(context)} chars ---")
    
    # 4. 생성 결과 확인
    answer = generate(query, context)
    print(f"\n--- 답변 ---\n{answer}")
    
    return {"results": results, "reranked": reranked, "answer": answer}

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Qdrant
from langchain.retrievers import EnsembleRetriever, ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CohereRerank

class AdvancedRAGPipeline:
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4o")
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
        self.vectorstore = Qdrant.from_existing_collection(
            embedding=self.embeddings,
            collection_name="knowledge_base"
        )
        self.cache = SemanticCache(self.vectorstore)
    
    def query(self, question):
        # 1. 캐시 확인
        cached = self.cache.get(question)
        if cached:
            return cached
        
        # 2. 쿼리 변환
        queries = self.multi_query_transform(question)
        
        # 3. 하이브리드 검색
        all_docs = []
        for q in queries:
            docs = self.hybrid_search(q)
            all_docs.extend(docs)
        
        # 4. 중복 제거
        unique_docs = self.deduplicate(all_docs)
        
        # 5. 리랭킹
        reranked = self.rerank(question, unique_docs)
        
        # 6. 컨텍스트 압축
        compressed = self.compress_context(question, reranked)
        
        # 7. 답변 생성
        answer = self.generate(question, compressed)
        
        # 8. 답변 검증
        if not self.verify_groundedness(answer, compressed):
            answer = self.regenerate_with_instruction(question, compressed)
        
        # 9. 캐시 저장
        self.cache.set(question, answer)
        
        return answer