ARCH · RAG

Advanced RAG

🔀 Hybrid Search 🪄 HyDE & RAPTOR ✅ Corrective RAG

기본 RAG를 넘어 Hybrid Search, HyDE, RAPTOR, Corrective RAG, Self-RAG 등 프로덕션 수준의 고급 검색-증강 패턴을 구현합니다.

RAG 진화 단계

단계기법핵심 개선적합 케이스
Basic RAGDense Vector Search시맨틱 검색소규모 단일 도메인
Advanced RetrievalHybrid Search, HyDE검색 품질 향상다양한 질의 유형
Adaptive RAGCorrective RAG, Self-RAG검색 결과 검증높은 정확도 요구
Hierarchical RAGRAPTOR다층 요약 인덱스대규모 문서 컬렉션

Hybrid Search — Dense + Sparse 결합

벡터 검색(의미)과 BM25(키워드)를 결합해 각각의 단점을 보완합니다. 정확한 키워드 매칭이 필요한 코드 검색이나 고유명사에서 특히 효과적입니다.

pythonhybrid_search.py — Dense + BM25 앙상블
from langchain_chroma import Chroma
from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever
from langchain_openai import OpenAIEmbeddings
from langchain_core.documents import Document

def build_hybrid_retriever(
    documents: list[Document],
    dense_weight: float = 0.6,
    sparse_weight: float = 0.4,
    k: int = 6
) -> EnsembleRetriever:
    """Dense + Sparse 검색기 결합"""

    # Dense: 시맨틱 벡터 검색
    vectorstore = Chroma.from_documents(
        documents,
        embedding=OpenAIEmbeddings()
    )
    dense_retriever = vectorstore.as_retriever(
        search_type="mmr",  # Maximum Marginal Relevance — 다양성 보장
        search_kwargs={"k": k, "fetch_k": k * 3}
    )

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

    # 앙상블 (RRF: Reciprocal Rank Fusion)
    return EnsembleRetriever(
        retrievers=[dense_retriever, bm25_retriever],
        weights=[dense_weight, sparse_weight]
    )

# 사용
retriever = build_hybrid_retriever(docs)
results = retriever.invoke("LangGraph StateGraph 사용법")

HyDE — Hypothetical Document Embeddings

HyDE는 질문을 직접 임베딩하는 대신, LLM이 가상의 답변 문서를 먼저 생성하고 그것을 임베딩해 검색합니다. 질문-답변 임베딩 공간 불일치 문제를 해결합니다.

pythonhyde.py — 가상 문서 임베딩 검색
from langchain_anthropic import ChatAnthropic
from langchain_openai import OpenAIEmbeddings
import numpy as np

llm = ChatAnthropic(model="claude-haiku-4-5-20251001")
embeddings = OpenAIEmbeddings()

async def hyde_retrieve(
    query: str,
    vectorstore,
    k: int = 5,
    n_hypothetical: int = 3
) -> list:
    """HyDE: 가상 답변 N개 생성 → 임베딩 평균 → 검색"""

    # 가상 답변 N개 생성
    hypothetical_docs = []
    for _ in range(n_hypothetical):
        response = await llm.ainvoke([{
            "role": "user",
            "content": f"""다음 질문에 대한 이상적인 답변 문서를 작성하세요.
실제 데이터가 없어도 전문가가 작성할 법한 내용으로 작성합니다.

질문: {query}
답변 (3~5 문장):"""
        }])
        hypothetical_docs.append(response.content)

    # 가상 문서들의 임베딩 평균
    doc_embeddings = await embeddings.aembed_documents(hypothetical_docs)
    avg_embedding = np.mean(doc_embeddings, axis=0).tolist()

    # 평균 임베딩으로 검색
    results = vectorstore.similarity_search_by_vector(avg_embedding, k=k)
    return results

Corrective RAG — 검색 결과 검증 및 수정

검색 결과의 관련성을 LLM이 평가하고, 관련성이 낮으면 웹 검색으로 보완하는 자기교정 RAG 패턴입니다.

질문 입력 벡터 검색 K개 문서 검색 관련성 평가 LLM이 각 청크를 Relevant/Irrelevant 관련 ✅ 문서 정제 비관련 🌐 웹 검색 보완 검색 📝 생성 정제된 컨텍스트 관련성 판단 기준: 질문-청크 간 의미적 충분성 그림 1. Corrective RAG 흐름 — 검색 결과 검증 후 부족하면 웹 검색으로 보완
pythoncorrective_rag.py — LangGraph Corrective RAG
from langgraph.graph import StateGraph, END
from langchain_anthropic import ChatAnthropic
from langchain_community.tools.tavily_search import TavilySearchResults
from typing import TypedDict, Literal

class RAGState(TypedDict):
    question: str
    documents: list
    generation: str
    web_search_needed: bool

llm = ChatAnthropic(model="claude-haiku-4-5-20251001")
web_search = TavilySearchResults(max_results=3)

async def retrieve_node(state: RAGState) -> dict:
    docs = retriever.invoke(state["question"])
    return {"documents": docs}

async def grade_documents_node(state: RAGState) -> dict:
    """각 문서의 관련성 평가"""
    question = state["question"]
    filtered_docs = []
    web_needed = False

    for doc in state["documents"]:
        response = await llm.ainvoke([{
            "role": "user",
            "content": f"""문서가 질문에 관련이 있으면 'yes', 없으면 'no'만 답하세요.

질문: {question}
문서: {doc.page_content[:500]}"""
        }])

        if "yes" in response.content.lower():
            filtered_docs.append(doc)
        else:
            web_needed = True

    return {"documents": filtered_docs, "web_search_needed": web_needed}

async def web_search_node(state: RAGState) -> dict:
    """웹 검색으로 부족한 컨텍스트 보완"""
    results = web_search.invoke(state["question"])
    from langchain_core.documents import Document
    web_docs = [Document(page_content=r["content"]) for r in results]
    return {"documents": state["documents"] + web_docs}

async def generate_node(state: RAGState) -> dict:
    context = "\n\n".join(d.page_content for d in state["documents"])
    response = await ChatAnthropic(model="claude-opus-4-7").ainvoke([{
        "role": "user",
        "content": f"컨텍스트:\n{context}\n\n질문: {state['question']}"
    }])
    return {"generation": response.content}

def route_after_grading(state: RAGState) -> str:
    if state["web_search_needed"] and not state["documents"]:
        return "web_search"
    return "generate"

builder = StateGraph(RAGState)
builder.add_node("retrieve",        retrieve_node)
builder.add_node("grade_documents", grade_documents_node)
builder.add_node("web_search",      web_search_node)
builder.add_node("generate",        generate_node)
builder.set_entry_point("retrieve")
builder.add_edge("retrieve", "grade_documents")
builder.add_conditional_edges("grade_documents", route_after_grading, {
    "web_search": "web_search",
    "generate":   "generate"
})
builder.add_edge("web_search", "generate")
builder.add_edge("generate",   END)
crag_graph = builder.compile()

청크 전략 비교

전략방법장점단점
Fixed SizeN토큰 고정 분할단순, 균일문맥 단절 가능성
Recursive Character문단→문장→단어 순 분할자연스러운 경계청크 크기 불균일
Semantic Chunking임베딩 유사도 기반의미 보존비용 높음
Parent-Child큰 청크 검색 + 작은 청크 임베딩정밀도 + 컨텍스트구현 복잡
Sliding Window오버랩 포함 분할경계 손실 감소중복 저장
💡 RAG 파이프라인 선택 기준
  • 정확한 키워드 검색 중요 → Hybrid Search (BM25 + Dense)
  • 질문이 단답형/짧음 → HyDE로 임베딩 공간 매칭 개선
  • 검색 정확도 최우선 → Corrective RAG + 크로스인코더 리랭킹
  • 수천 페이지 이상 문서 → RAPTOR (계층적 요약 트리)
↑ 목차