RAG 기법 (Naive, Advanced, Modular, Agentic)

LLM의 한계 : 사내 문서를 알지 못함
-> 필요한 지식을 프롬포트에 입력해서 넣어줘야 함
=> RAG = Retrieve(검색), Augment(증강), Generation(생성)

RAG의 진화

1. Naive RAG : 검색->생성
2. Advanced RAG : 품질 기법
3. Modular: 재구성
4. Agentic: 스스로 판단

RAG PipeLine (Naive RAG)

Indexing

= Data Loading -> Chunking -> Embedding -> Storing

Retrieval & Generation

= Query -> Embedding -> Retrieval -> Top-K Chunks -> LLM -> Response

로딩 (Loading)

문서를 읽어 LangChain의 Document(내용 page_content + 출처 metadata)로 만든다.

from langchain_core.documents import Document

# 문서를 읽어 Document로 만든다 (page_content = 내용, metadata = 출처 등)
text = open("spacex_ipo_2026_report.txt", encoding="utf-8").read()
docs = [Document(page_content=text, metadata={"source": "spacex_ipo_2026_report.txt"})]

print(f"문서 {len(docs)}개, 길이 {len(docs[0].page_content)}자")
print("metadata:", docs[0].metadata)

"""
문서 1개, 길이 1871자
metadata: {'source': 'spacex_ipo_2026_report.txt'}
"""

청킹 (Chunking)

긴 문서를 그대로 임베딩하면 의미가 뭉뚱그려지고 검색 단위도 너무 커짐. => 의미 단위의 적당한 크기로 청크를 나눠야함

• chunk_size: 한 청크의 길이 -> 청크가 너무 크면 무관한 내용까지 딸려오고, 너무 작으면 맥락이 끊긴다
• chunk_overlap : 이웃 청크와 겹치는 부분 -> 경계에서 맥락이 잘리지 않게 한다

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(chunk_size=300, chunk_overlap=50)
chunks = splitter.split_documents(docs)

print(f"청크 {len(chunks)}개")
print("--- 첫 청크 ---")
print(chunks[0].page_content)

"""
청크 9개
--- 첫 청크 ---
SpaceX 2026 기업공개(IPO) 종합 리포트

작성 목적: 본 문서는 RAG 실습용 자료로, 2026년 6월 공개 보도를 바탕으로 정리한 SpaceX 상장 리포트다.

1. 개요
스페이스엑스(SpaceX)는 2026년 6월 12일 나스닥(Nasdaq)에 정식 상장했다. 종목 티커는 SPCX이다. 공모가는 2026년 6월 11일에 확정됐으며, 이 상장은 역사상 최대 규모의 기업공개로 기록됐다. 일론 머스크가 이끄는 이 우주 기업은 상장을 통해 약 750억 달러를 조달했다.
"""

=> 이 단계에선 각 Chunk가 Text임

임베딩 (Embedding)

검색을 위해 텍스트의 의미를 숫자 벡터로 바꾸는 것. 의미가 비슷하면 벡터도 가깝다.

코사인 유사도

두 벡터가 이루는 각도로 의미의 유사도를 잰다. 방향이 비슷할수록 1에 가깝다. -> 크기가 아니라 방향을 보므로, 문장 길이에 덜 휘둘린다.

검색

= '질문 벡터와 코사인이 가장 높은 청크' 를 고르는 것

• 같은 의미 -> 코사인 = 0.9

무관  -> 코사인 = 0.1 

임베딩 모델

이 벡터를 만드는 모델

from langchain_openai import OpenAIEmbeddings
import numpy as np

def cosine(a, b):
    a, b = np.array(a), np.array(b)
    return a @ b / (np.linalg.norm(a) * np.linalg.norm(b))

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

v_q = embeddings.embed_query("SpaceX 공모가가 얼마야?")
v_related = embeddings.embed_query("SpaceX는 주당 135달러에 상장했다.")
v_unrelated = embeddings.embed_query("오늘 점심은 김치찌개를 먹었다.")

print("벡터 차원:", len(v_q))
print("관련 문장 유사도:", round(cosine(v_q, v_related), 3))
print("무관 문장 유사도:", round(cosine(v_q, v_unrelated), 3))

"""
벡터 차원: 1536
관련 문장 유사도: 0.582
무관 문장 유사도: 0.04
"""

Vector DB

임베딩된 청크를 벡터DB에 저장한다.

from langchain_community.vectorstores import FAISS

# 청크들을 임베딩해 FAISS 인덱스로 만든다 (지식베이스 구축)
vectorstore = FAISS.from_documents(chunks, embeddings)
print("벡터DB에 저장된 청크 수:", vectorstore.index.ntotal)

# 인덱스를 디스크에 저장하고 다시 불러오기 (매번 재구축하지 않아도 됨)
vectorstore.save_local("faiss_spacex")
loaded = FAISS.load_local("faiss_spacex", embeddings,
                          allow_dangerous_deserialization=True)  # 직접 만든 인덱스라 허용
print("불러온 청크 수:", loaded.index.ntotal)

검색기 (Retriever)

질문을 임베딩해, 벡터 DB에서 가장 가까운 top-k 청크를 돌려준다.

• k = 가져올 청크 수, 너무 적으면 누락&너무 많으면 노이즈
• 점수(score)나 메타데이터로 결과를 더 거를 수 있다

(중요!) Vector DB가 검색기 역할도 한다.

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

def format_docs(docs):
    return "\n\n".join(d.page_content for d in docs)

prompt = ChatPromptTemplate.from_messages([
    ("system", "너는 문서 기반 비서다. 아래 문서 내용만 근거로 답하고, 문서에 없으면 '문서에 없습니다'라고 말해라.\n\n[문서]\n{context}"),
    ("human", "{question}"),
])

# 검색 → 근거 주입 → 생성 (LCEL 한 줄기)
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt | model | StrOutputParser()
)
print("RAG 체인 준비 완료")

# 아까 RAG 없이 못 맞힌 '같은 질문' — 이번엔 RAG로
print(rag_chain.invoke("SpaceX 2026 상장의 공모가(주당 가격)와 첫날 종가, 티커를 알려줘."))

"""
SpaceX 2026 상장의 공모가는 주당 135달러였으며, 첫날 종가는 160.95달러였습니다. 티커는 SPCX입니다.
"""

Advanced RAG

Multi-Query Retriever

LLM이 질문을 N개로 변형

; 원래 질문 "휴가 며칠?" 한 개로만 검색하면, 표현이 다른 자료 ("연차", "베케이션") 누락 가능. LLM 에게 질문을 N 개로 다시 써달라고 하고, 각각 검색 해 결과를 합친다

원래 질문: "휴가 며칠?"
   ↓ LLM 이 변형
1. "정규직 연차는 며칠인가요?"
2. "휴가·연차 일수가 어떻게 되죠?"
3. "베케이션 정책이 어떻게 됩니까?"
   ↓ 각각 의미 검색
   ↓ 결과 합치고 중복 제거→ 더 풍부한 후

MultiQueryRetriever

q = '한국은행 기준금리 인하 영향'

base_retriever = vectorstore.as_retriever(search_kwargs={'k': 3})

• 객체 정의

# LLM 으로 질문을 N 개로 변형해 각각 검색
from langchain_classic.retrievers.multi_query import MultiQueryRetriever

mq_retriever = MultiQueryRetriever.from_llm(
    retriever=base_retriever,
    llm=model,
)
print(type(mq_retriever).__name__)

"""
MultiQueryRetriever
"""

• 호출

results_mq = mq_retriever.invoke(q)

# 비교 — 기본 vs Multi-Query 의 *반환 청크 페이지* 차이
base_pages = set(r.metadata.get('page') for r in results_base)
mq_pages   = set(r.metadata.get('page') for r in results_mq)
print(f'기본       페이지: {sorted(base_pages)}')
print(f'Multi-Query 페이지: {sorted(mq_pages)}')
print(f'  추가로 잡힌    : {sorted(mq_pages - base_pages)}')

"""
기본       페이지: [31, 36, 81]
Multi-Query 페이지: [21, 31, 36, 77, 79, 81, 86]
  추가로 잡힌    : [21, 77, 79, 86]
"""

BM25

전통적 키워드 검색 점수 알고리즘. 글자가 정확히 같은 청크를 우선 시

• 의미 검색, 즉 벡터 기반 검색은 비슷한 의미에 강하지만, 제품번호•법령조항 같은 정확 키워드엔 약함
  -> 이거 보완 !

BM25Retriever

# BM25 — 임베딩 X. 청크 리스트만 있으면 즉시 retriever 가능
from langchain_community.retrievers import BM25Retriever

bm25 = BM25Retriever.from_documents(chunks)
bm25.k = 3
print(type(bm25).__name__)

"""
BM25 Top 1: page 46, 33 
물가 
2-9. 소비자물가 상승률은 1월중 석유류가격 상승률이 큰 폭 낮아지고 농축수산
물가격 오름세...
BM25 Top 2: page 11, < 요약 6/8 > 
6  
  
전망의 리스크 
 
 향후 성장 전망경로에는 반도체 경기, 글로벌 통상환...
BM25 Top 3: page 47, 34 
2-11. 금년중 소비자물가 상승률은 일부 품목전자기기·보험료 등의 비용상승 압력 등으로 11월
전망...
"""
```python
# 둘 비교 — 페이지 set
bm_pages  = set(r.metadata.get('page') for r in results_bm)
vec_pages = set(r.metadata.get('page') for r in results_vec)
print('BM25   :', sorted(bm_pages))
print('Vector :', sorted(vec_pages))
print('BM25 만:', sorted(bm_pages - vec_pages))
print('Vec 만 :', sorted(vec_pages - bm_pages))

"""
BM25   : [11, 46, 47]
Vector : [60, 63, 91]
BM25 만: [11, 46, 47]
Vec 만 : [60, 63, 91]
"""

=> 임베딩 호출 없음 <-> vectorstore 필요 없음 => 비용•지연 X

• 오프라인 인덱스 : 청크 리스트만 있으면 바로 사용 가능

Hybrid Search

• 의미 검색 (Vector) — 표현 변형에 강함
• 키워드 검색 (BM25) — 정확 키워드에 강함

=> 둘의 결과를 weighted 결합 → 두 약점 보완

EnsembleRetriever

# Vector retriever (의미) + BM25 (키워드) 를 가중치로 결합
from langchain_classic.retrievers import EnsembleRetriever

vector_r = vectorstore.as_retriever(search_kwargs={'k': 5})
bm25_r = BM25Retriever.from_documents(chunks); bm25_r.k = 5

hybrid = EnsembleRetriever(
    retrievers=[vector_r, bm25_r],
    weights=[0.7, 0.3],   # 의미 0.7, 키워드 0.3 - 데이터·도메인에 따라 조정 (보통 의미 0.5~0.7).
)
print(type(hybrid).__name__)

Re-ranking

후보를 정밀 점수로 다시 줄 세움

아이디어

검색 결과 Top-K 가 늘 순위대로 정확한 건 아님. 살짝 비슷한 청크가 위에 올 때도 있음. → 일단 후보를 넉넉히 가져온 뒤 (예: 20 개), 별도 모델로 정밀 점수를 매겨 재정렬

Cross-Encoder Reranker

# 크로스인코더 리랭커 다운로드 (다국어 — 한국어 지원). 처음 1회 모델을 받는다.
from langchain_community.cross_encoders import HuggingFaceCrossEncoder

reranker = HuggingFaceCrossEncoder(model_name="BAAI/bge-reranker-v2-m3")
print('리랭커 준비')

# 1) Hybrid 로 후보 *넉넉히* 가져옴 — 예: 10 개
vector_r10 = vectorstore.as_retriever(search_kwargs={'k': 10})
bm25_r10 = BM25Retriever.from_documents(chunks); bm25_r10.k = 10
hybrid10 = EnsembleRetriever(retrievers=[vector_r10, bm25_r10], weights=[0.7, 0.3])

q = '한국은행 기준금리 인하 영향'
candidates = hybrid10.invoke(q)
print(f'1차 후보: {len(candidates)} 개')
for i, r in enumerate(candidates[:5], 1):
    print(f'  Top {i}: page {r.metadata.get("page")}, {r.page_content[:60]}...')

# 2) Cross-Encoder Rerank — (질문, 후보문) 쌍마다 점수 매겨 재정렬
scores = reranker.score([(q, c.page_content) for c in candidates])          # 각 쌍의 관련도 점수
ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)   # 점수 내림차순 정렬

print('Rerank Top-3:')
for d, s in ranked[:3]:
    print(f'  score {s:.4f} page {d.metadata.get("page")}: {d.page_content[:80]}...')

# 비교 — 1차 (Hybrid) vs Rerank 후 Top-3 페이지
hybrid_top3_pages = [c.metadata.get('page') for c in candidates[:3]]
rerank_top3_pages = [d.metadata.get('page') for d, s in ranked[:3]]
print(f'Hybrid 만 Top-3 page : {hybrid_top3_pages}')
print(f'Rerank 후 Top-3 page : {rerank_top3_pages}')

Modular RAG

: 기술이 따로 있는게 아니라, 모듈 형식의 디자인 패턴을 의미

1. 3요소인 Retrieve(검색), Augment(증강), Generation(생성)을
2. 모듈 형태로 다 쪼개서
3. LLM based Application 적용 할 때,
4. 선형 (R->A->G)이 아닌 분기•반복 가능한 유연한 그래프 형태로 만드는 것 (자유롭게 재구성)
5. 필요 시 분기, 반복, 라우팅, 융합, 메모리, 재정렬 모듈도 추가

=> 이 전체 형태 자체를 '모듈러 하다~'

• 필요 기술

1. RAG
2. LangGraph
3. LangChain

1) Naive RAG

retrieve → generate

from typing import TypedDict, List
from langchain_core.documents import Document
from langgraph.graph import StateGraph, START, END
from IPython.display import Image

# State — 이 구현체가 공유할 데이터
class NaiveState(TypedDict):
    question: str
    documents: List[Document]
    generation: str

retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# 노드 2개: 검색 → 생성
def retrieve(state: NaiveState):
    return {"documents": retriever.invoke(state["question"])} # retriever.invoke() -> return chunk = 객체 'Document'

def generate(state: NaiveState):
    context = "\n\n".join(d.page_content for d in state["documents"])
    ans = llm.invoke(f"아래 컨텍스트만 근거로 답해.\n\n[컨텍스트]\n{context}\n\n[질문] {state['question']}").content
    return {"generation": ans}

# 그래프: 한 방향 (Linear)
builder = StateGraph(NaiveState)
builder.add_node("retrieve", retrieve)
builder.add_node("generate", generate)
builder.add_edge(START, "retrieve")
builder.add_edge("retrieve", "generate")
builder.add_edge("generate", END)
graph = builder.compile()

Image(graph.get_graph().draw_mermaid_png())   # 구조 확인

2) Advanced RAG (리랭크)

retrieve(k=5) → rerank(top 3) → generate

# 크로스인코더 리랭커 다운로드 (다국어 — 한국어 지원). 처음 1회 모델을 받는다.
from langchain_community.cross_encoders import HuggingFaceCrossEncoder
reranker = HuggingFaceCrossEncoder(model_name="BAAI/bge-reranker-v2-m3")

class AdvState(TypedDict):
    question: str
    documents: List[Document]
    generation: str

retriever = vectorstore.as_retriever(search_kwargs={"k": 5})   # 일부러 많이 뽑고

def retrieve(state: AdvState):
    return {"documents": retriever.invoke(state["question"])}

def rerank(state: AdvState):
    q, docs = state["question"], state["documents"]
    scores = reranker.score([(q, d.page_content) for d in docs])   # 질문-문서 쌍 점수
    ranked = [d for d, s in sorted(zip(docs, scores), key=lambda x: x[1], reverse=True)]
    print("  [리랭크 상위3]", [d.page_content[:18] for d in ranked[:3]])
    return {"documents": ranked[:3]}                                # 상위 3개만 남김

def generate(state: AdvState):
    context = "\n\n".join(d.page_content for d in state["documents"])
    return {"generation": llm.invoke(f"아래 컨텍스트만 근거로 답해.\n\n[컨텍스트]\n{context}\n\n[질문] {state['question']}").content}

# 그래프: retrieve → rerank → generate
builder = StateGraph(AdvState)
builder.add_node("retrieve", retrieve)
builder.add_node("rerank", rerank)
builder.add_node("generate", generate)
builder.add_edge(START, "retrieve")
builder.add_edge("retrieve", "rerank")
builder.add_edge("rerank", "generate")
builder.add_edge("generate", END)
graph = builder.compile()

Image(graph.get_graph().draw_mermaid_png())   # 구조 확인

3) Adaptive RAG (질의 라우팅)

입구에서 질문 유형을 먼저 판단: 문서 검색이 필요한 지식 질문이면 retrieve → generate, 일반 상식·잡담이면 검색 없이 direct로 바로 답한다. 들어오자마자 길이 갈리는 Conditional 패턴.

라우터 → (vectorstore: retrieve → generate) | (direct: 바로 답)

from pydantic import BaseModel
from typing import Literal

class AdaptState(TypedDict):
    question: str
    documents: List[Document]
    generation: str

retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# 입구 라우터 — LLM이 질문 유형을 둘 중 하나로 분류 (분기 함수)
# ★ 라우터는 '벡터스토어에 무엇이 들었는지'를 알아야 제대로 라우팅한다
class Route(BaseModel):
    datasource: Literal["vectorstore", "direct"]
router = llm.with_structured_output(Route)

def route_question(state: AdaptState):
    r = router.invoke(
        "벡터스토어에는 RAG·LangGraph 관련 문서가 들어 있다. "
        "질문이 그 주제(RAG·CRAG·Modular RAG·LangGraph 등)에 관한 것이면 vectorstore, "
        f"그 외 일반 상식·인사·잡담이면 direct로 분류: {state['question']}"
    )
    print("  (라우팅:", r.datasource, ")")
    return r.datasource

def retrieve(state: AdaptState):
    return {"documents": retriever.invoke(state["question"])}

def generate(state: AdaptState):
    context = "\n\n".join(d.page_content for d in state["documents"])
    return {"generation": llm.invoke(f"아래 컨텍스트만 근거로 답해.\n\n[컨텍스트]\n{context}\n\n[질문] {state['question']}").content}

def direct(state: AdaptState):
    return {"generation": llm.invoke(state["question"]).content}   # 검색 없이 바로 답

# 그래프: START에서 라우터로 분기 (한쪽만 검색)
builder = StateGraph(AdaptState)
builder.add_node("retrieve", retrieve)
builder.add_node("generate", generate)
builder.add_node("direct", direct)
builder.add_conditional_edges(START, route_question, {"vectorstore": "retrieve", "direct": "direct"})
builder.add_edge("retrieve", "generate")
builder.add_edge("generate", END)
builder.add_edge("direct", END)
graph = builder.compile()

Image(graph.get_graph().draw_mermaid_png())   # 구조 확인

4) CRAG (Corrective RAG)

검색한 문서를 평가해서, 관련 없으면 질의를 재작성해 다시 검색한다(교정 루프). 평가·재작성·생성을 노드로 두고, 조건부 엣지로 경로를 바꾸는 Conditional + Loop 패턴.

retrieve → grade → (관련↓ ⟳ transform_query → retrieve) → generate

• 모듈 정의

from pydantic import BaseModel, Field
from typing import Literal

# 이 데모에서 쓸 retriever — 위 vectorstore에서 직접 정의 (멀리 있는 변수 재활용 X)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# ── 모듈 1 · retrieve(검색 노드) ──────────────────────────────
# 역할: 현재 질의로 벡터스토어에서 관련 문서를 가져온다.
def retrieve(state: RAGState):
    docs = retriever.invoke(state["question"])   # 질의 → 유사도 상위 k개 Document
    return {"documents": docs}                    # State.documents 를 검색 결과로 갱신

# ── 모듈 2 · grade_documents(관련성 평가/필터 노드) ───────────
# 역할: 검색된 문서가 '정말' 질문에 관련 있는지 LLM이 yes/no로 채점해 무관 문서를 버린다.
#       (Naive RAG에는 없는 품질 게이트 — 헛문서로 생성되는 걸 막음)
class Grade(BaseModel):
    relevant: Literal["yes", "no"] = Field(description="문서가 질문에 관련 있으면 yes")

grader = llm.with_structured_output(Grade)        # 출력을 yes/no로 강제한 LLM

def grade_documents(state: RAGState):
    kept = []                                      # 관련 있다고 판정된 문서만 모음
    for d in state["documents"]:
        verdict = grader.invoke(
            f"질문: {state['question']}\n문서: {d.page_content}\n이 문서가 질문에 관련 있나?"
        )
        print(f"  [평가] {d.page_content[:24]}... -> {verdict.relevant}")
        if verdict.relevant == "yes":
            kept.append(d)
    return {"documents": kept}                     # 걸러낸 문서로 State 갱신

# ── 모듈 3 · transform_query(질의 재작성 노드) ────────────────
# 역할: 관련 문서가 0개일 때, 검색이 더 잘 되도록 질문을 다시 써서 재검색을 노린다.
def transform_query(state: RAGState):
    better = llm.invoke(
        f"검색이 더 잘 되도록 다음 질문을 다시 써줘. 질문만 출력:\n{state['question']}"
    ).content
    print(f"  [재작성] {state['question']} -> {better}")
    return {"question": better, "tries": state["tries"] + 1}

# ── 모듈 4 · generate(생성 노드) ──────────────────────────────
# 역할: 최종적으로 남은 문서(context)만 근거로 답을 만든다.
def generate(state: RAGState):
    context = "\n\n".join(d.page_content for d in state["documents"]) or "(관련 문서 없음)"
    ans = llm.invoke(
        f"아래 컨텍스트만 근거로 답해. 근거 없으면 모른다고 해.\n\n[컨텍스트]\n{context}\n\n[질문] {state['question']}"
    ).content
    return {"generation": ans}

• 그래프 조립 : 모듈간 동적 라우팅 + 교정 루프

from langgraph.graph import StateGraph, START, END
from IPython.display import Image

# ── 라우터(분기 함수) · decide ────────────────────────────────
# grade_documents 다음에 '어느 노드로 갈지'를 결정한다.
#   · 관련 문서가 하나라도 남았으면 → "generate" (바로 답 생성)
#   · 없으면 → "transform_query" (질의 재작성 후 재검색)
#   · 단, tries 가 2 이상이면 → "generate" (재시도 포기, 가진 것으로 답/모른다)
# 분기 함수는 '다음 노드 이름(문자열)'을 반환하는 게 규칙.
def decide(state: RAGState):
    if state["documents"]:
        return "generate"
    if state["tries"] >= 2:
        return "generate"
    return "transform_query"

# 그래프 = State 위에 노드(모듈)들을 엣지로 연결한 것
builder = StateGraph(RAGState)
builder.add_node("retrieve", retrieve)             # add_node("이름", 함수) — 노드 등록
builder.add_node("grade_documents", grade_documents)
builder.add_node("transform_query", transform_query)
builder.add_node("generate", generate)

builder.add_edge(START, "retrieve")                # 시작 → 검색
builder.add_edge("retrieve", "grade_documents")    # 검색 → 평가
builder.add_conditional_edges(                     # 평가 → decide 결과에 따라 분기
    "grade_documents", decide,
    {"generate": "generate", "transform_query": "transform_query"},  # 반환값 → 실제 노드
)
builder.add_edge("transform_query", "retrieve")    # 재작성 → 다시 검색 (교정 루프 형성)
builder.add_edge("generate", END)                  # 생성 → 끝

graph = builder.compile()                          # 정의를 실행 가능한 그래프로 변환

# 코드가 곧 구조 — 모듈러 RAG 흐름을 그림으로 확인
Image(graph.get_graph().draw_mermaid_png())

5) RAG as Tool

Agent = ReACT = LLM + Tool + Loop

ㄴ 여기에 '검색'을 '도구(@tool)'로 만들어 에이전트한테 주면, 언제•무엇으로 검색할지 LLM이 스스로 결정

from langchain.tools import tool
from langchain.agents import create_agent
from IPython.display import Image

# 이 데모용 retriever — vectorstore에서 직접 정의 (재활용 X)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# 1. RAG(검색)를 '도구'로 감싼다 — docstring이 '이 도구를 언제 쓸지'의 단서가 된다
@tool
def search_knowledge_base(query: str) -> str:
    """업로드한 문서에서 query와 관련된 내용을 검색해 돌려준다."""
    docs = retriever.invoke(query)
    return "\n\n".join(d.page_content for d in docs)

# 2. 에이전트에게 도구를 쥐여준다 — 언제/무엇으로 검색할지는 LLM이 결정 (고정 흐름 X)
rag_agent = create_agent(
    llm,
    tools=[search_knowledge_base],
    system_prompt=(
        "문서 내용에 관한 질문은 반드시 search_knowledge_base 도구로 검색해 "
        "그 결과에 근거해서만 답해라(사전지식으로 추측 금지). 단순 인사·잡담만 도구 없이 답해라."
    ),
)

# 3. 에이전트 내부 그래프 — agent ↔ tools 루프(ReAct)를 create_agent가 자동으로 만든다
Image(rag_agent.get_graph().draw_mermaid_png())

• 실행

# (1) 문서 지식 질문 → 에이전트가 스스로 search_knowledge_base 호출
out = rag_agent.invoke({"messages": [{"role": "user", "content": "Modular RAG가 뭐야?"}]})
print("[지식 질문]", out["messages"][-1].content)

print("\n" + "=" * 50)

# (2) 검색이 필요 없는 질문 → 도구를 안 부르고 바로 답
out2 = rag_agent.invoke({"messages": [{"role": "user", "content": "안녕! 넌 뭘 할 수 있어?"}]})
print("[잡담]", out2["messages"][-1].content)

"""
[지식 질문] Modular RAG는 검색, 평가, 재작성, 생성을 모듈로 분리하고, 조건에 따라 모듈 간 경로를 동적으로 변경하는 RAG(검색-생성) 패러다임입니다. 

기본적으로 Naive RAG는 검색(retrieve) 후 생성(generate)으로만 흐르는 가장 단순한 형태입니다. 반면, Corrective RAG(CRAG)는 검색된 문서의 관련성을 평가하여, 부족할 경우 질의를 재작성하거나 외부 검색으로 보강하는 방식입니다.

==================================================
[잡담] 안녕하세요! 저는 문서에서 정보를 검색하고 질문에 답변할 수 있습니다. 필요한 정보가 있으면 말씀해 주세요!
"""

6) Agentic RAG (멀티툴 자율선택)

에이전트에게 문서 검색 도구 + 웹 검색 도구를 함께 주면, 질문을 보고 무엇을 쓸지 — 문서 검색? 웹 검색? 그냥 답? 를 스스로 판단한다.

이 자율 오케스트레이션이 Agentic RAG의 핵심.

# 이 데모용 retriever — vectorstore에서 직접 정의 (재활용 X)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

from langchain_community.tools import DuckDuckGoSearchRun
ddg = DuckDuckGoSearchRun()   # 인터넷 검색 (키 불필요)

# 도구 2개 — 에이전트가 무엇을 쓸지 스스로 판단한다
@tool
def search_documents(query: str) -> str:
    """업로드한 문서에서 query와 관련된 내용을 검색한다."""
    return "\n\n".join(d.page_content for d in retriever.invoke(query))

@tool
def web_search(query: str) -> str:
    """문서에 없는 최신·외부 정보가 필요할 때 인터넷(DuckDuckGo)에서 검색한다."""
    return ddg.invoke(query)

# 에이전트 — 문서검색 / 웹검색 / 직접답변 중 스스로 선택
agentic_rag = create_agent(
    llm,
    tools=[search_documents, web_search],
    system_prompt=(
        "개념·지식 질문은 반드시 search_documents로 문서를 검색해 그 근거로만 답하고(추측 금지), "
        "문서에 없는 최신·외부 정보는 web_search로 인터넷을 검색해라. 단순 인사만 도구 없이 답해라."
    ),
)

• 실행

# (1) 문서 지식 질문 → search_documents 호출
out = agentic_rag.invoke({"messages": [{"role": "user", "content": "CRAG가 뭐야?"}]})
print("[문서]", out["messages"][-1].content)

print("\n" + "=" * 50)

# (2) 문서에 없는 최신 정보 질문 → web_search 호출
out2 = agentic_rag.invoke({"messages": [{"role": "user", "content": "요즘 화제인 AI 뉴스 알려줘"}]})
print("[웹]", out2["messages"][-1].content)

*응용 : 모듈 총동원 (하이브리드 + 리랭크 + 다단계 LLM + 웹 보강)

지금까지의 모듈을 한 그래프에 모은 실무형 파이프라인이다. 로컬 문서로 답이 부족하면 인터넷 검색으로 보강 한다 (CRAG의 '외부 검색' 아이디어).

구조: expand_query(LLM①) → hybrid_retrieve(벡터+BM25) → rerank(크로스인코더) → generate(LLM②) → verify(LLM③) → 답이 충분하면 END, 부족하면(모른다) web_search(인터넷) → generate → verify 한 번 더.

pre-retrieval — 질문을 검색에 더 잘 맞게 재작성 (LLM①) retrieval — 의미(벡터) + 키워드(BM25) 하이브리드 post-retrieval — 크로스인코더로 정밀 재정렬 generation → 자기검증(LLM③) — 답이 부족하면 DuckDuckGo로 웹 검색해 보강 LLM 노드 3개(재작성·생성·검증) + 하이브리드·리랭크 2단 검색 + 웹 보강 분기까지, 가장 복잡한 형태.

• 모듈(Node) 정의

# 7번은 BM25(키워드)·EnsembleRetriever(하이브리드)·DuckDuckGo(웹검색)가 추가로 필요하다
!pip install -qU rank_bm25 langchain-classic ddgs duckduckgo-search

from typing import TypedDict, List, Literal
from langchain_core.documents import Document
from langchain_community.retrievers import BM25Retriever
from langchain_classic.retrievers import EnsembleRetriever
from langchain_community.tools import DuckDuckGoSearchRun
from pydantic import BaseModel, Field

# State — 이 파이프라인이 공유할 데이터
class ProState(TypedDict):
    question: str
    search_query: str        # expand_query가 만든 검색 질의
    documents: List[Document]
    generation: str
    answered: str            # verify 결과 "yes"/"no"
    web_used: bool           # 웹검색을 이미 썼는지 (무한 루프 방지)

# 하이브리드 검색기 = 의미(벡터) + 키워드(BM25)
vector_r = vectorstore.as_retriever(search_kwargs={"k": 5})
bm25_r = BM25Retriever.from_documents(docs); bm25_r.k = 5          # setup의 docs 사용
hybrid = EnsembleRetriever(retrievers=[vector_r, bm25_r], weights=[0.6, 0.4])

# 인터넷 검색 도구 (키 불필요)
web = DuckDuckGoSearchRun()

# ── LLM① expand_query: 검색용으로 질문 재작성 ──
def expand_query(state: ProState):
    better = llm.invoke(f"다음 질문을 문서 검색에 더 잘 맞는 검색어로 다시 써줘. 검색어만 출력:\n{state['question']}").content
    print("  [재작성]", better)
    return {"search_query": better}

# ── hybrid_retrieve: 벡터+BM25 하이브리드 검색 ──
def hybrid_retrieve(state: ProState):
    return {"documents": hybrid.invoke(state["search_query"])}

# ── rerank: 크로스인코더로 (질문, 문서) 재채점 → 상위 3개 ──
def rerank(state: ProState):
    cand = state["documents"]
    scores = reranker.score([(state["question"], d.page_content) for d in cand])
    ranked = [d for d, s in sorted(zip(cand, scores), key=lambda x: x[1], reverse=True)]
    return {"documents": ranked[:3]}

# ── LLM② generate: 컨텍스트로 답 생성 (없으면 '모른다') ──
def generate(state: ProState):
    context = "\n\n".join(d.page_content for d in state["documents"])
    ans = llm.invoke(f"아래 컨텍스트만 근거로 답해. 컨텍스트에 답이 없으면 '모른다'고만 말해.\n\n[컨텍스트]\n{context}\n\n[질문] {state['question']}").content
    return {"generation": ans}

# ── LLM③ verify: 답이 질문에 실제로 답했는지 ('모른다'면 no) ──
class Check(BaseModel):
    answered: Literal["yes", "no"] = Field(description="질문에 실제로 답하면 yes; '모른다·정보 없음'이면 no")
checker = llm.with_structured_output(Check)

def verify(state: ProState):
    v = checker.invoke(
        "다음 답변이 질문에 실제로 답했는지 보라. 구체적으로 답했으면 yes, "
        "'모른다'·'정보가 없다'는 식이면 no.\n\n"
        f"[질문] {state['question']}\n[답변] {state['generation']}"
    )
    print("  [검증] answered =", v.answered)
    return {"answered": v.answered}

# ── web_search: 로컬로 부족하면 인터넷에서 보강 ──
def web_search(state: ProState):
    print("  [웹검색] 로컬 문서로 부족 → 인터넷 검색")
    try:
        text = web.invoke(state["question"])
    except Exception as e:
        text = f"(웹검색 실패: {e})"
    return {"documents": [Document(page_content=text)], "web_used": True}

# ── 분기: 답 충분하면 끝 / 부족하고 웹 안 썼으면 웹으로 / 이미 썼으면 끝 ──
def decide_after_verify(state: ProState):
    if state["answered"] == "yes":
        return "end"
    if not state.get("web_used", False):
        return "web"
    return "end"

• 그래프 생성

from langgraph.graph import StateGraph, START, END
from IPython.display import Image

builder = StateGraph(ProState)
builder.add_node("expand_query", expand_query)
builder.add_node("hybrid_retrieve", hybrid_retrieve)
builder.add_node("rerank", rerank)
builder.add_node("generate", generate)
builder.add_node("verify", verify)
builder.add_node("web_search", web_search)

builder.add_edge(START, "expand_query")
builder.add_edge("expand_query", "hybrid_retrieve")   # 재작성 → 하이브리드 검색
builder.add_edge("hybrid_retrieve", "rerank")         # 검색 → 리랭크
builder.add_edge("rerank", "generate")                # 리랭크 → 생성
builder.add_edge("generate", "verify")                # 생성 → 자기검증
builder.add_conditional_edges(                        # 부족하면 웹으로, 충분하면 끝
    "verify", decide_after_verify, {"end": END, "web": "web_search"}
)
builder.add_edge("web_search", "generate")            # 웹 결과로 다시 생성
graph = builder.compile()

Image(graph.get_graph().draw_mermaid_png())   # 구조 확인

• 그래프 실행

# (1) 코퍼스에 있는 질문 → 로컬로 충분 (웹 안 씀)
print(graph.invoke({"question": "CRAG가 뭐야?", "web_used": False})["generation"])

print("\n" + "=" * 50)

# (2) 코퍼스에 없는 질문 → 로컬 부족 → 인터넷 검색으로 보강
print(graph.invoke({"question": "에펠탑 높이는 몇 미터야?", "web_used": False})["generation"])

"""
  [재작성] CRAG의 정의와 의미
  [검증] answered = yes
CRAG는 검색된 문서의 관련성을 평가해, 부족하면 질의를 재작성하거나 외부 검색으로 보강하는 방식이다.

==================================================
  [재작성] 에펠탑 높이
  [검증] answered = no
  [웹검색] 로컬 문서로 부족 → 인터넷 검색
  [검증] answered = yes
에펠탑의 높이는 약 300미터이며, 안테나를 포함하면 약 330미터에 달합니다.

"""