4장. 검색과 응답을 최적화하는 RAG 고도화 전략

출처: 『RAG 마스터: 랭체인으로 완성하는 LLM 서비스』(브라이스 유·조경아·박수진·김재웅, 프리렉 2025) | 공식: https://www.langchain.com/

기본 RAG는 사용자 질문을 그대로 검색에 넣고, 단순 유사도로 문서를 가져와 모델에 던진다. 이 장은 그 각 단계—문서 분할, 질의 변형, 검색 알고리즘, 후처리, 전체 흐름 제어—를 어떻게 고도화할 수 있는지 기법별로 설명한다. 기본 RAG가 왜 부족한가를 먼저 보고, 각 단계에서 어떤 문제를 어떤 방식으로 푸는지 따라가면 된다.

실습 — 책 공식 노트북: Ch04. Advanced Rag 폴더 — 검색 알고리즘 4종. 코랩 바로 열기: BM25·Dense·Ensemble 하이브리드·HyDE.

학습 목표

이 장을 끝내면 다음을 할 수 있다.

• 부모-자식 분할의 구조와 동작 원리를 설명하고, ParentDocumentRetriever로 구현한다.
• MultiQueryRetriever와 HyDE의 접근 방식을 비교하고, 상황에 맞는 질의 변형 전략을 선택한다.
• BM25·FAISS·EnsembleRetriever를 구분하고, 도메인 특성에 따라 가중치를 설정한다.
• LLM 기반 리랭킹과 크로스 인코더 리랭킹의 장단점을 비교해 실무 상황에 맞는 방식을 선택한다.
• Self-RAG의 3단계 흐름(검색 판단→관련성 평가→품질 검증)을 설명하고, 적합한 적용 시나리오를 구분한다.

전체 흐름도

RAG 고도화는 기본 파이프라인의 각 단계를 개별적으로 강화하거나 조합해 사용한다.

[ 원본 문서 ]
   │
   ▼
[ 청킹 전략 ]
   ├─ 고정 길이 분할 (기본)
   └─ 부모-자식 분할 (ParentDocumentRetriever) — 정밀 검색 + 넓은 컨텍스트
   │
   ▼
[ 질의 변형 ]
   ├─ MultiQueryRetriever — 다중 관점 쿼리로 커버리지 확대
   └─ HyDE — 가상 답변 임베딩으로 의미 거리 단축
   │
   ▼
[ 검색 알고리즘 ]
   ├─ 희소 검색 (BM25) — 키워드 정확 일치
   ├─ 밀집 검색 (FAISS) — 의미 기반 유사도
   └─ 앙상블 (EnsembleRetriever) — 두 방식 가중치 결합
   │
   ▼
[ 리랭킹 (후처리) ]
   ├─ LLM 기반 — 높은 정확도, 높은 비용
   └─ 크로스 인코더 — 속도·비용 절충
   │
   ▼
[ Self-RAG — 흐름 자체 제어 ]
   ├─ 검색 필요 여부 자가 판단
   ├─ 관련성 자가 평가 (ISREL)
   └─ 답변 품질 자가 검증 (ISSUP + ISUSE)
   │
   ▼
[ 최종 답변 ]

0. 사전 필수 용어

• 청킹(Chunking) — 긴 문서를 검색과 생성에 최적화된 작은 단위로 잘라 나누는 과정. 분할 전략에 따라 검색 품질이 크게 달라진다.
• 부모-자식 분할(Parent-Child Chunking) — 원본 문서를 큰 '부모' 단위와 작은 '자식' 단위로 이중 분할하는 기법. 자식으로 정밀 검색, 부모를 컨텍스트로 반환.
• HyDE(Hypothetical Document Embeddings, 가상 문서 임베딩) — 질문에 대한 가상 답변을 LLM으로 만들어, 그 답변 텍스트를 질문 대신 검색에 활용하는 기법.
• BM25 — 단어 빈도(TF)와 희소성(IDF), 문서 길이 정규화를 결합한 키워드 기반 랭킹 알고리즘. 희소 검색의 표준.
• 희소 검색(Sparse Retrieval) — 단어 존재 여부·빈도를 기반으로 희소 벡터를 만들어 문서를 찾는 방식. 키워드 일치 강함, 의미 포착 약함.
• 밀집 검색(Dense Retrieval) — 트랜스포머 기반 임베딩으로 텍스트를 고차원 밀집 벡터로 변환해 유사도를 계산하는 방식. 의미 기반, 계산 비용 높음.
• 앙상블 검색(Ensemble Retrieval) — 희소 검색과 밀집 검색을 가중치로 결합해 두 방식의 장점을 모두 활용하는 하이브리드 방식.
• 리랭킹(Reranking) — 초기 검색 결과를 더 정교한 모델로 재평가·재정렬해 관련성 높은 문서만 남기는 후처리 단계.
• 크로스 인코더(Cross-Encoder) — 질문과 문서를 함께 인코더에 입력해 관련성 점수를 직접 산출하는 방식. 정확도 높음, 속도 느림.
• Self-RAG — LLM이 검색 필요 여부를 스스로 판단하고, 검색 결과의 관련성과 생성 답변의 품질을 자체 평가·개선하는 고도화 방법론.

1. 청킹 전략

기본 분할의 한계

가장 단순한 방법인 고정 길이 문자 분할은 구현이 쉽지만 문장·단락의 의미 구조를 무시해 중요한 맥락이 잘린다. 재귀적 문자 분할(RecursiveCharacterTextSplitter)은 구조를 고려하지만 계층적 관계까지 보존하진 못한다.

1.1 부모-자식 분할 — ParentDocumentRetriever

핵심 아이디어: "정밀하게 찾되, 넓게 반환한다."

• 자식 문서 (작은 청크, 예: 200자) → 벡터 DB에 임베딩 저장 → 질문과의 유사도 검색에 사용
• 부모 문서 (큰 청크, 예: 1000자) → 별도 문서 저장소(InMemoryStore)에 원본 보관 → 최종 반환

검색 흐름: 1. 사용자 쿼리 → 자식 문서(벡터 DB) 유사도 검색 2. 매칭된 자식 문서의 메타데이터에서 부모 문서 식별자 확인 3. 부모 문서 저장소에서 부모 문서 반환 → LLM 컨텍스트로 전달

언제 쓰는가: 긴 문서에서 특정 정보를 정밀하게 찾으면서도 충분한 맥락이 필요할 때. 법률 문서, 기술 문서처럼 맥락이 중요한 도메인에 적합.

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 부모 문서 분할기 (크게)
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=1000)
# 자식 문서 분할기 (작게 — 검색용)
child_splitter = RecursiveCharacterTextSplitter(chunk_size=200)

# 자식 문서용 벡터 저장소 (임베딩 검색)
vectorstore = Chroma(
    collection_name="split-parents",
    embedding_function=OpenAIEmbeddings()
)
# 부모 문서용 키-값 저장소 (원본 보관)
store = InMemoryStore()

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

# 문서 추가 — 자동으로 부모/자식으로 분할해 각 저장소에 저장
retriever.add_documents(docs)
print(f"부모 문서 수: {len(list(store.yield_keys()))}")  # 예: 219

# 검색 — 자식으로 찾고 부모를 반환
retrieved_docs = retriever.invoke("What are the types of investments?")
print(retrieved_docs[0].page_content)  # 부모 문서 (넓은 맥락 포함)

# 자식 문서 직접 확인
sub_docs = vectorstore.similarity_search("What are the types of investments?")
print(sub_docs[0].page_content)  # 자식 문서 (정확한 키워드 매칭)

2. 질의 변형

기본 RAG의 한계: 사용자 질문이 모호하거나 검색에 최적화되지 않은 표현이면 관련 문서를 제대로 못 찾는다. 질의 변형은 원래 질문을 재구성해 검색 품질을 높이는 과정이다.

2.1 다중 질의 생성 — MultiQueryRetriever

핵심 아이디어: 하나의 질문을 여러 관점으로 변형해 병렬 검색, 중복 제거 후 통합.

흐름: 1. LLM이 원본 질문을 다양한 관점의 쿼리 N개로 변형 2. 각 쿼리로 독립 검색 3. 결과 통합 + 중복 제거 → 더 넓은 관련 문서 커버리지

언제 쓰는가: 질문이 모호하거나 다양한 측면을 포함할 때. 단일 쿼리로 놓치는 관련 문서를 포착할 때.

import logging
from langchain.retrievers import MultiQueryRetriever
from langchain_community.vectorstores import Chroma
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 생성된 쿼리 로깅 설정
logging.basicConfig()
logging.getLogger("langchain.retrievers.multi_query").setLevel(logging.INFO)

# 벡터 저장소 구성 (문서 로드 → 분할 → 임베딩은 기존 패턴과 동일)
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(documents=split_docs, embedding=embeddings)

# MultiQueryRetriever: 기본 검색기 + 쿼리 변형용 LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0.2)
retriever = MultiQueryRetriever.from_llm(
    retriever=vectorstore.as_retriever(),
    llm=llm,
)

question = "주식 투자를 처음 시작하려면 어떻게 해야 하나요?"
unique_docs = retriever.invoke(question)
# 로그 출력: ['주식 투자 초보자가 알아야 할 기본 단계는?', '... 준비 사항은?', '...조언이 있나요?']
print(f"검색된 문서 수: {len(unique_docs)}")  # 예: 6개

2.2 가상 문서 임베딩 — HyDE

핵심 아이디어: "질문보다 답변이 문서와 더 유사하다."

질문을 그대로 검색에 쓰면 키워드가 광범위하거나 의미적으로 모호할 수 있다. 반면 LLM이 만든 가상 답변은 실제 문서에서 쓰이는 전문 용어·표현을 포함한다.

흐름: 질문 → LLM으로 가상 답변 생성 → 가상 답변 임베딩으로 벡터 DB 검색 → 검색 문서 + 원본 질문 → 최종 답변

장점: 구현이 간단하고 기존 RAG와 쉽게 통합 가능. 논문에서 11개 쿼리 세트에서 기존 방식 대비 우수한 성능 입증. 단점: LLM 가상 답변 생성에 추가 비용·지연 발생.

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnableLambda

# 1. 가상 문서 생성 체인
def create_virtual_doc_chain():
    system = "당신은 고도로 숙련된 AI입니다."
    user = "주어진 질문 '{query}'에 대해 직접적으로 답변하는 가상의 문서를 생성하세요. 크기는 {chunk_size}자 언저리로."
    prompt = ChatPromptTemplate.from_messages([("system", system), ("human", user)])
    llm = ChatOpenAI(model="gpt-4o", temperature=0.2)
    return prompt | llm | StrOutputParser()

# 2. 문서 검색 체인 (가상 문서로 벡터 DB 검색)
def create_retrieval_chain(retriever):
    return RunnableLambda(lambda x: retriever.invoke(x["virtual_doc"]))

# 유틸리티: 문서 목록 → 단일 문자열
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

# 3. 최종 응답 체인
def create_final_response_chain():
    final_prompt = ChatPromptTemplate.from_template(
        "다음 정보와 질문을 바탕으로 답변해주세요:\n컨텍스트: {context}\n질문: {question}"
    )
    return final_prompt | ChatOpenAI(model="gpt-4o", temperature=0.2) | StrOutputParser()

# 전체 파이프라인 연결
virtual_doc_chain = create_virtual_doc_chain()
retrieval_chain = create_retrieval_chain(retriever)
final_chain = create_final_response_chain()

def hyde_pipeline(question):
    virtual_doc = virtual_doc_chain.invoke({"query": question, "chunk_size": 200})
    retrieved = retrieval_chain.invoke({"virtual_doc": virtual_doc})
    context = format_docs(retrieved)
    return final_chain.invoke({"context": context, "question": question})

response = hyde_pipeline("주식 시장의 변동성이 높을 때 투자 전략은 무엇인가요?")
print(response)

3. 검색 알고리즘

희소 검색 vs. 밀집 검색 vs. 앙상블 — 비교

3.1 희소 검색 — BM25

TF-IDF의 한계: 문서 길이 편향 + 단어 빈도 선형 증가 문제. BM25의 개선: 문서 길이 정규화 + 단어 빈도 포화 처리(빈도가 높아도 점수 증가율 감소).

BM25 점수 계산식:

Score(D, Q) = Σ IDF(qi) × [f(qi, D) × (k1 + 1)] / [f(qi, D) + k1 × (1 - b + b × |D|/avgdl)]

• f(qi, D): 문서 D에서 쿼리 단어 qi의 빈도
• |D|/avgdl: 문서 길이를 평균 대비 정규화
• k1(1.2~2.0): 단어 빈도 포화 계수 / b(0.75): 길이 정규화 강도

한국어는 교착어 특성상 형태소 분석 전처리가 BM25 성능을 크게 높인다.

from langchain_community.retrievers import BM25Retriever
from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from kiwipiepy import Kiwi  # 한국어 형태소 분석기

# 문서 로드 및 분할
loader = PyPDFLoader("투자설명서.pdf")
doc_splitter = RecursiveCharacterTextSplitter(chunk_size=2000, chunk_overlap=200)
docs = loader.load_and_split(doc_splitter)

# 한국어 형태소 분석 함수 (BM25 전처리용)
kiwi_tokenizer = Kiwi()
def kiwi_tokenize(text):
    return [token.form for token in kiwi_tokenizer.tokenize(text)]

# BM25 리트리버 생성 — 형태소 분석 전처리 적용
bm25_retriever = BM25Retriever.from_documents(docs, preprocess_func=kiwi_tokenize)
bm25_retriever.k = 2  # 상위 2개 반환

# QA 체인과 연결
from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI

qa_chain = RetrievalQA.from_chain_type(
    llm=ChatOpenAI(temperature=0.2, model="gpt-4o"),
    chain_type="stuff",
    retriever=bm25_retriever,
    return_source_documents=True
)
result = qa_chain.invoke("이 회사가 발행한 주식의 총 발행량이 어느 정도야?")
print(result["result"])  # 13,602,977주

3.2 밀집 검색 — FAISS

밀집 벡터는 텍스트의 의미를 수치로 표현한 것이다. "테슬라 주가 급등"과 "전기차 수요 증가로 주식 상승"은 키워드가 다르지만 밀집 벡터 공간에서 가깝다.

FAISS: 대규모 벡터 유사도 검색 라이브러리. IVF(Inverted File) 인덱스로 클러스터링 후 근접 클러스터만 탐색 → 검색 속도 대폭 향상.

from langchain_openai.embeddings import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI

# 임베딩 모델 설정
embedding = OpenAIEmbeddings(model="text-embedding-3-large")

# FAISS DB 생성 및 저장
faiss_store = FAISS.from_documents(docs, embedding)
faiss_store.save_local("/content/DB")

# 저장된 DB 로드
vectordb = FAISS.load_local(
    "/content/DB",
    embeddings=embedding,
    allow_dangerous_deserialization=True
)

# 리트리버 생성 (상위 2개 반환)
faiss_retriever = vectordb.as_retriever(search_kwargs={"k": 2})

qa_chain = RetrievalQA.from_chain_type(
    llm=ChatOpenAI(temperature=0.2, model="gpt-4o"),
    chain_type="stuff",
    retriever=faiss_retriever,
    return_source_documents=True
)
result = qa_chain.invoke("이 회사가 발행한 주식의 총 발행량이 어느 정도야?")
print(result["result"])  # 보통주 13,602,977주

3.3 앙상블 검색 — EnsembleRetriever

언제 쓰는가: 키워드 정확도(희소)와 의미 기반 검색(밀집) 모두 필요할 때. 가중치로 도메인 특성에 맞게 비율 조정 가능.

from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain.chains import RetrievalQA

# BM25 리트리버 (희소)
bm25_retriever = BM25Retriever.from_documents(docs, preprocess_func=kiwi_tokenize)
bm25_retriever.k = 4

# FAISS 리트리버 (밀집)
embedding = OpenAIEmbeddings(model="text-embedding-3-large")
faiss_store = FAISS.from_documents(docs, embedding)
faiss_retriever = faiss_store.as_retriever(search_kwargs={"k": 4})

# 앙상블 — 가중치 0.5:0.5 (기술 문서라면 BM25 높이기, 일반 QA라면 FAISS 높이기)
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, faiss_retriever],
    weights=[0.5, 0.5]
)

qa_chain = RetrievalQA.from_chain_type(
    llm=ChatOpenAI(temperature=0.2, model="gpt-4o"),
    chain_type="stuff",
    retriever=ensemble_retriever,
    return_source_documents=True
)
result = qa_chain.invoke("이 회사가 발행한 주식의 총 발행량이 어느 정도야?")
print(result["result"])

4. 문서 후처리 — 리랭킹

초기 검색(BM25, FAISS)은 빠르게 후보를 좁히는 데 탁월하지만 관련성을 완벽하게 평가하지 못한다. 리랭킹은 초기 검색 결과를 더 정교한 방법으로 재평가·재정렬해 최종 컨텍스트 품질을 높인다.

4.1 고성능 LLM 기반 리랭킹 vs. 4.2 크로스 인코더 기반 리랭킹 — 비교

Bi-Encoder vs. Cross-Encoder 추가 비교: - Bi-Encoder: 각 문장 독립 인코딩 → 빠름 (밀집 검색에 사용) - Cross-Encoder: 쌍을 함께 인코딩 → 더 정확하지만 조합 수 폭발 (리랭킹에 사용) - 실전 패턴: Bi-Encoder로 후보 추리고(빠름) → Cross-Encoder로 재평가(정확) → 균형 달성

4.1 LLM 기반 리랭킹 코드

from pydantic import BaseModel, Field
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from langchain_openai import ChatOpenAI
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from typing import List
from langchain_core.documents import Document

# 관련성 점수 스키마
class RelevanceScore(BaseModel):
    relevance_score: float = Field(description="문서가 쿼리와 얼마나 관련 있는지 나타내는 점수.")

def reranking_documents(query: str, docs: List[Document], top_n: int = 2) -> List[Document]:
    parser = JsonOutputParser(pydantic_object=RelevanceScore)
    prompt = PromptTemplate(
        template="""1~10점으로 다음 문서가 질문과 얼마나 관련 있는지 평가해주세요.
단순 키워드 일치가 아닌 쿼리의 맥락과 의도를 고려하세요.
{format_instructions}
question: {query}
document: {doc}
relevance_score:""",
        input_variables=["query", "doc"],
        partial_variables={"format_instructions": parser.get_format_instructions()}
    )
    llm = ChatOpenAI(temperature=0, model="gpt-4o", max_tokens=3000)
    chain = prompt | llm | parser

    scored_docs = []
    for doc in docs:
        try:
            score = float(chain.invoke({"query": query, "doc": doc.page_content})["relevance_score"])
        except Exception:
            score = 5.0  # 오류 시 기본값
        scored_docs.append((doc, score))

    reranked = sorted(scored_docs, key=lambda x: x[1], reverse=True)
    return [doc for doc, _ in reranked[:top_n]]

# 사용 예
embedding = OpenAIEmbeddings(model="text-embedding-3-large")
vectordb = FAISS.load_local("/content/DB", embeddings=embedding, allow_dangerous_deserialization=True)
query = "이 회사의 2022년 영업손실이 정확히 얼마야?"
initial_docs = vectordb.similarity_search(query, k=4)
reranked_docs = reranking_documents(query, initial_docs)
print(f"리랭킹 후 {len(reranked_docs)}개 문서 선택")

4.2 크로스 인코더 기반 리랭킹 코드

from sentence_transformers import CrossEncoder
from langchain_core.retrievers import BaseRetriever
from langchain_core.documents import Document
from pydantic import BaseModel, Field
from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI
from typing import Any, List

# cross-encoder/ms-marco-MiniLM-L-12-v2: MS MARCO 데이터셋으로 학습된 경량 크로스 인코더
crossencoder = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-12-v2")

class RetrieverWithCrossEncoder(BaseRetriever):
    vectorstore: Any = Field(description="초기 검색용 벡터 저장소")
    crossencoder: Any = Field(description="재순위화용 크로스 인코더 모델")
    k: int = Field(default=5, description="초기 검색 문서 수")
    rerank_top_k: int = Field(default=2, description="리랭킹 후 최종 반환 문서 수")

    class Config:
        arbitrary_types_allowed = True

    def _get_relevant_documents(self, query: str) -> List[Document]:
        # 1. 초기 밀집 검색으로 후보 추리기
        initial_docs = self.vectorstore.similarity_search(query, k=self.k)
        # 2. 크로스 인코더용 (질문, 문서) 쌍 준비
        pairs = [[query, doc.page_content] for doc in initial_docs]
        # 3. 크로스 인코더로 관련성 점수 산출
        scores = self.crossencoder.predict(pairs)
        # 4. 점수 내림차순 정렬 후 상위 반환
        scored_docs = sorted(zip(initial_docs, scores), key=lambda x: x[1], reverse=True)
        return [doc for doc, _ in scored_docs[:self.rerank_top_k]]

cross_encoder_retriever = RetrieverWithCrossEncoder(
    vectorstore=vectordb,
    crossencoder=crossencoder,
    k=4,
    rerank_top_k=2
)

qa_chain = RetrievalQA.from_chain_type(
    llm=ChatOpenAI(temperature=0.2, model="gpt-4o"),
    chain_type="stuff",
    retriever=cross_encoder_retriever,
    return_source_documents=True
)
result = qa_chain.invoke("이 회사의 2022년 영업손실이 정확히 얼마야?")
print(result["result"])  # 149억원 (또는 149.1억원)

5. 확장된 RAG 방법론 — Self-RAG

5.1 Self-RAG 개요

전통 RAG의 구조적 한계: - 검색이 필요 없는 질문에도 무조건 문서를 가져와 프롬프트에 넣는다 - 관련 없는 문서가 들어와도 그대로 사용한다 - 생성된 답변이 검색 내용과 얼마나 일치하는지 검증하지 않는다

Self-RAG는 LLM이 각 단계에 직접 개입해 특수 토큰(Retrieve, ISREL, ISSUP, ISUSE)으로 흐름을 제어한다.

Self-RAG 3단계 흐름:

질문 입력
   ↓
[1단계] 검색 필요 여부 판단
  Retrieve=Yes → 외부 문서 검색
  Retrieve=No  → 검색 없이 바로 생성
   ↓ (Retrieve=Yes인 경우)
[2단계] 관련성 평가 (ISREL)
  ISREL=Relevant   → 해당 문서로 답변 생성
  ISREL=Irrelevant → 해당 문서 제외
   ↓
[3단계] 답변 평가 (ISSUP + ISUSE)
  ISSUP: Fully supported / Partially supported / No support
  ISUSE: 1~5점 유용성 점수
   ↓
최고 답변 선택 (Fully supported + 유용성 최고점 우선)

5.2 Self-RAG 구현

실제 Self-RAG는 특수 토큰으로 학습된 오픈소스 모델이 필요하지만, 여기서는 GPT-4o로 각 단계의 판단 로직을 시뮬레이션한다.

from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from langchain_core.prompts import PromptTemplate
from typing import Literal

llm = ChatOpenAI(model="gpt-4o", max_tokens=2000, temperature=0.2)

# --- 단계별 출력 스키마 ---
class RetrievalResponse(BaseModel):
    Reasoning: str = Field(description="검색 필요 여부 추론 (2~3문장)")
    Retrieve: Literal["Yes", "No"] = Field(description="검색 필요 여부")

class RelevanceResponse(BaseModel):
    Reasoning: str = Field(description="관련성 평가 추론 (2~3문장)")
    ISREL: Literal["Relevant", "Irrelevant"] = Field(description="관련성 평가 결과")

class GenerationResponse(BaseModel):
    response: str = Field(description="생성된 답변")

class SupportResponse(BaseModel):
    Reasoning: str = Field(description="지원 평가 추론 (2~3문장)")
    ISSUP: Literal["Fully supported", "Partially supported", "No support"]

class UtilityResponse(BaseModel):
    Reasoning: str = Field(description="유용성 평가 추론")
    ISUSE: Literal[1, 2, 3, 4, 5] = Field(description="유용성 점수 1~5")

# --- 각 단계 체인 ---
retrieval_prompt = PromptTemplate(
    input_variables=["query"],
    template="주어진 질문에 외부 문서 검색이 도움이 되는지 판단하세요.\n질문: {query}"
)
retrieval_chain = retrieval_prompt | llm.with_structured_output(RetrievalResponse)

relevance_prompt = PromptTemplate(
    input_variables=["query", "context"],
    template="연관 문서가 질문에 관련 있는지 평가하세요.\n질문: {query}\n연관 문서: {context}"
)
relevance_chain = relevance_prompt | llm.with_structured_output(RelevanceResponse)

generation_prompt = PromptTemplate(
    input_variables=["query", "context"],
    template="질문 '{query}'와 연관 문서 '{context}'를 기반으로 답변을 만들어주세요."
)
generation_chain = generation_prompt | llm.with_structured_output(GenerationResponse)

# support_chain, utility_chain도 동일 패턴으로 구성 (생략)

# --- SelfRAG 클래스 ---
class SelfRAG:
    def __init__(self, vectorstore, retrieval_chain, relevance_chain,
                 generation_chain, support_chain, utility_chain, top_k=4):
        self.vectorstore = vectorstore
        self.retrieval_chain = retrieval_chain
        self.relevance_chain = relevance_chain
        self.generation_chain = generation_chain
        self.support_chain = support_chain
        self.utility_chain = utility_chain
        self.top_k = top_k

    def process_query(self, query):
        # 1단계: 검색 필요 여부
        decision = self.retrieval_chain.invoke({"query": query}).Retrieve
        if decision != "Yes":
            return self.generation_chain.invoke(
                {"query": query, "context": "관련 컨텍스트 없음"}
            ).response

        # 2단계: 관련 문서 검색
        docs = self.vectorstore.similarity_search(query, k=self.top_k)
        contexts = [doc.page_content for doc in docs]

        # 3단계: 관련성 평가 — Relevant 문서만 남기기
        relevant_contexts = [
            ctx for ctx in contexts
            if self.relevance_chain.invoke({"query": query, "context": ctx}).ISREL == "Relevant"
        ]
        if not relevant_contexts:
            return self.generation_chain.invoke(
                {"query": query, "context": "관련 컨텍스트 없음"}
            ).response

        # 4단계: 답변 생성
        responses = [
            self.generation_chain.invoke({"query": query, "context": ctx}).response
            for ctx in relevant_contexts
        ]

        # 5~6단계: 지원·유용성 평가
        assessed = []
        for resp, ctx in zip(responses, relevant_contexts):
            sup = self.support_chain.invoke({"query": query, "response": resp, "context": ctx}).ISSUP
            use = int(self.utility_chain.invoke({"query": query, "response": resp}).ISUSE)
            assessed.append((resp, sup, use))

        # 최고 답변 선택: Fully supported → Partially supported → 유용성 점수 순
        for support_level in ["Fully supported", "Partially supported"]:
            candidates = [r for r in assessed if r[1] == support_level]
            if candidates:
                return max(candidates, key=lambda x: x[2])[0]
        return max(assessed, key=lambda x: x[2])[0]

# 사용 예
self_rag = SelfRAG(
    vectorstore=vectordb,
    retrieval_chain=retrieval_chain,
    relevance_chain=relevance_chain,
    generation_chain=generation_chain,
    support_chain=support_chain,
    utility_chain=utility_chain,
    top_k=4
)
response = self_rag.process_query("이 회사의 바이오 의약품 라이센스 아웃 수익을 알려줘")
print(response)

Self-RAG 장단점 요약: - 장점: 불필요한 검색 없음, 관련 없는 문서 자동 제거, 생성 품질 자체 검증 - 단점: 여러 단계에 LLM 반복 호출 → 처리 시간 및 비용 증가 - 언제: 정확성·신뢰성이 최우선인 도메인(의료·법률·금융)에 적합. 빠른 응답이 필요한 서비스에는 비적합.

핵심 개념 정리

한 문장 요약 — 기본 RAG의 고정된 파이프라인을 분할 전략(부모-자식), 질의 변형(MultiQuery·HyDE), 검색 알고리즘(BM25·FAISS·앙상블), 후처리(리랭킹), 흐름 제어(Self-RAG) 다섯 축에서 개선하면, 관련성과 신뢰성을 크게 높인 RAG 시스템을 만들 수 있다.

실무 체크리스트

• [ ] 긴 문서에서 맥락이 중요하다면 ParentDocumentRetriever로 부모-자식 계층 구성.
• [ ] 질문이 모호하거나 다각도 검색이 필요하면 MultiQueryRetriever 적용.
• [ ] 질문과 문서 표현이 크게 다른 도메인이라면 HyDE 고려 (단, API 비용·지연 감수).
• [ ] 한국어 BM25는 반드시 형태소 분석 전처리(kiwipiepy, konlpy 등) 추가.
• [ ] 정확한 용어 매칭 + 의미 기반 검색 모두 필요하면 EnsembleRetriever, weights 도메인별 조정.
• [ ] 1차 검색 후 리랭킹 추가 시: 속도·비용 여유 있으면 LLM 리랭킹, 제약 있으면 Cross-Encoder.
• [ ] Bi-Encoder(FAISS)로 후보 추리고 Cross-Encoder로 재평가하는 2단계 패턴이 실전 표준.
• [ ] Self-RAG는 정확성이 최우선인 도메인에만 적용 — 처리 시간·비용 증가를 고려.
• [ ] RAG 고도화 기법은 조합 사용 가능: 부모-자식 + 앙상블 + 크로스 인코더 리랭킹 등.

연습문제

1. 비교. MultiQueryRetriever와 HyDE는 둘 다 질의 변형 기법이다. 각각의 접근 방식과 적합한 상황을 비교하라.
2. 개념. 부모-자식 분할에서 "자식으로 검색하고 부모를 반환"하는 이유를 정확도와 컨텍스트 측면에서 설명하라.
3. 비교. 희소 검색(BM25)과 밀집 검색(FAISS)의 차이를 벡터 표현 방식, 의미 포착 능력, 계산 비용 세 가지 측면에서 비교하라.
4. 설계. 법률 문서 RAG 시스템을 구축한다. 앙상블 검색의 weights를 어떻게 설정하겠는가? 근거를 들어라.
5. 분석. LLM 기반 리랭킹과 Cross-Encoder 기반 리랭킹 중 실시간 고객 상담 챗봇에 어느 것이 더 적합한가? Self-RAG의 적용 가능성도 함께 논하라.

최신 동향 (2026-05 기준)

최신 동향 (검증 2026-05-21) — 리랭킹 분야에서는 경량 크로스 인코더와 LLM 기반 리랭커의 정확도 격차가 좁혀지고 있으며, 하이브리드 검색(BM25+밀집)은 실질적인 프로덕션 표준으로 자리잡고 있다. Self-RAG 계열은 Corrective RAG(CRAG)·Adaptive RAG 등으로 확장되며 흐름 제어 정교화가 이어지고 있다. 최신 버전·지원 리트리버 목록은 공식 문서(python.langchain.com)에서 확인하라. 책의 개념·흐름은 그대로 유효하다.

부록 A. 용어 사전

부록 B. 핵심 비교표

청킹 전략 비교

검색 알고리즘 비교 (BM25 / FAISS / 앙상블)

부록 C. 추천 참고 자료

검증된 외부 자료 (Tier 1 공식, 생존 확인 2026-05-21)

부록 D. 연습문제 풀이

1. (MultiQueryRetriever: 쿼리 다변화 / HyDE: 가상 답변 임베딩) MultiQueryRetriever는 원본 질문을 LLM으로 N개의 다양한 관점의 쿼리로 변형한 뒤 병렬 검색해 커버리지를 넓힌다. HyDE는 질문에 대한 가상 답변을 LLM으로 생성하고 그 답변 텍스트의 임베딩으로 벡터 DB를 검색한다. MultiQueryRetriever는 질문이 모호하거나 다각도 검색이 필요할 때 적합하고, HyDE는 질문과 문서의 표현 방식이 달라 의미 거리가 먼 도메인(전문 용어 집약적 문서 등)에서 유리하다. 다만 HyDE는 가상 답변 생성에 LLM 호출 비용과 지연이 추가된다.

2. (정밀 검색 + 넓은 컨텍스트 동시 확보) 자식 문서(작은 청크, 예: 200자)는 벡터 DB에 저장돼 질문과의 유사도 검색에 사용된다. 작은 단위일수록 임베딩이 특정 키워드·개념에 집중되어 매칭 정확도가 높다. 그러나 자식 문서 자체는 맥락이 부족하므로, 매칭된 자식의 부모 문서(큰 청크, 예: 1000자)를 별도 저장소에서 꺼내 LLM 컨텍스트로 전달함으로써 충분한 배경 정보를 함께 제공한다. 즉, 정밀도와 컨텍스트 풍부함을 동시에 얻는 구조다.

3. (벡터 표현·의미 포착·계산 비용 3축 비교)

BM25는 정확한 키워드 일치에 강하지만 동의어나 문맥적 유사어를 찾지 못하고, FAISS는 의미 기반 검색에 강하지만 신규 도메인에서 파인튜닝이 필요할 수 있다.

1. (법률 문서: BM25 가중치 높게) 법률 문서는 조문 번호, 고유 법률 용어, 특정 명칭처럼 정확한 키워드 일치가 결과의 신뢰성을 좌우하는 도메인이다. 따라서 EnsembleRetriever의 weights를 BM25(희소) 쪽에 높게 설정하는 것이 적합하다(예: weights=[0.7, 0.3]). 다만 유사 개념이나 다른 표현으로 서술된 조항도 포착해야 할 경우 FAISS 비중을 0.3 이상으로 유지해 의미 기반 검색을 보완 역할로 남긴다.

2. (실시간 챗봇: Cross-Encoder 우선, Self-RAG는 비적합) 실시간 고객 상담 챗봇은 응답 지연에 민감하므로 문서마다 GPT-4o를 호출하는 LLM 기반 리랭킹은 비용·속도 모두 불리하다. Cross-Encoder는 BERT 계열 경량 로컬 모델(예: cross-encoder/ms-marco-MiniLM-L-12-v2)로 구동 가능해 LLM 리랭킹보다 빠르고 저렴하면서 Bi-Encoder 단독보다 정확하다. Self-RAG는 검색 필요 여부 판단·관련성 평가·품질 검증까지 여러 단계에 LLM을 반복 호출하므로 처리 시간과 비용이 크게 증가한다. 본문에서도 "빠른 응답이 필요한 서비스에는 비적합"으로 명시하고 있어, 실시간 챗봇에는 Cross-Encoder 리랭킹이 현실적인 선택이다.