9장: 고급 RAG 패턴 - Agentic RAG와 Self-Correcting RAG

1. Retrieve Token: 검색이 필요한가?
   - [검색 필요] --> 검색 수행
   - [검색 불필요] --> 직접 응답
 
2. Relevance Token: 검색된 문서가 관련 있는가?
   - [관련 있음] --> 생성에 활용
   - [관련 없음] --> 폐기 후 재검색
 
3. Critique Token: 생성된 답변이 충실한가?
   - [충실함] --> 최종 응답으로 출력
   - [충실하지 않음] --> 재생성

from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Literal
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
 
class SelfRAGState(TypedDict):
    question: str
    retrieved_docs: List[str]
    generation: str
    relevance_check: str
    faithfulness_check: str
    retry_count: int
 
llm = ChatOpenAI(model="gpt-4o", temperature=0)
 
def decide_retrieval(state: SelfRAGState) -> SelfRAGState:
    """검색이 필요한지 판단"""
    prompt = ChatPromptTemplate.from_messages([
        ("system", """질문에 답변하기 위해 외부 문서 검색이 필요한지 판단하세요.
일반 상식으로 답변 가능하면 "불필요", 특정 정보가 필요하면 "필요"로만 답하세요."""),
        ("human", "{question}")
    ])
    result = (prompt | llm).invoke({"question": state["question"]})
    state["relevance_check"] = result.content.strip()
    return state
 
def retrieve_documents(state: SelfRAGState) -> SelfRAGState:
    """문서 검색 수행"""
    # 실제 구현에서는 벡터 검색 + 하이브리드 검색 + 리랭킹
    results = retriever.invoke(state["question"])
    state["retrieved_docs"] = [doc.page_content for doc in results]
    return state
 
def check_relevance(state: SelfRAGState) -> SelfRAGState:
    """검색 결과의 관련성 평가"""
    prompt = ChatPromptTemplate.from_messages([
        ("system", """검색된 문서가 질문에 답하기에 충분한 정보를 포함하는지 평가하세요.
"관련" 또는 "비관련"으로만 답하세요."""),
        ("human", """질문: {question}
 
검색 결과:
{documents}""")
    ])
    docs_text = "\n\n".join(state["retrieved_docs"])
    result = (prompt | llm).invoke({
        "question": state["question"],
        "documents": docs_text
    })
    state["relevance_check"] = result.content.strip()
    return state
 
def generate_answer(state: SelfRAGState) -> SelfRAGState:
    """답변 생성"""
    prompt = ChatPromptTemplate.from_messages([
        ("system", "주어진 컨텍스트만을 사용하여 질문에 답변하세요."),
        ("human", """컨텍스트:
{context}
 
질문: {question}""")
    ])
    context = "\n\n".join(state["retrieved_docs"])
    result = (prompt | llm).invoke({
        "question": state["question"],
        "context": context
    })
    state["generation"] = result.content
    return state
 
def check_faithfulness(state: SelfRAGState) -> SelfRAGState:
    """생성된 답변의 충실도 검증"""
    prompt = ChatPromptTemplate.from_messages([
        ("system", """답변이 컨텍스트에 있는 정보만을 사용했는지 검증하세요.
환각이 없으면 "충실", 환각이 있으면 "불충실"로만 답하세요."""),
        ("human", """컨텍스트:
{context}
 
답변: {answer}""")
    ])
    context = "\n\n".join(state["retrieved_docs"])
    result = (prompt | llm).invoke({
        "context": context,
        "answer": state["generation"]
    })
    state["faithfulness_check"] = result.content.strip()
    return state
 
# 라우팅 함수
def route_retrieval(state: SelfRAGState) -> Literal["retrieve", "generate_direct"]:
    if "필요" in state.get("relevance_check", "필요"):
        return "retrieve"
    return "generate_direct"
 
def route_relevance(state: SelfRAGState) -> Literal["generate", "retry"]:
    if "관련" in state.get("relevance_check", ""):
        return "generate"
    return "retry"
 
def route_faithfulness(state: SelfRAGState) -> Literal["output", "retry"]:
    if "충실" in state.get("faithfulness_check", ""):
        return "output"
    if state.get("retry_count", 0) >= 2:
        return "output"  # 최대 재시도 후 현재 결과 반환
    return "retry"
 
def increment_retry(state: SelfRAGState) -> SelfRAGState:
    state["retry_count"] = state.get("retry_count", 0) + 1
    return state
 
# 그래프 구성
workflow = StateGraph(SelfRAGState)
 
workflow.add_node("decide_retrieval", decide_retrieval)
workflow.add_node("retrieve", retrieve_documents)
workflow.add_node("check_relevance", check_relevance)
workflow.add_node("generate", generate_answer)
workflow.add_node("check_faithfulness", check_faithfulness)
workflow.add_node("increment_retry", increment_retry)
 
workflow.set_entry_point("decide_retrieval")
workflow.add_conditional_edges("decide_retrieval", route_retrieval)
workflow.add_edge("retrieve", "check_relevance")
workflow.add_conditional_edges("check_relevance", route_relevance)
workflow.add_edge("generate", "check_faithfulness")
workflow.add_conditional_edges("check_faithfulness", route_faithfulness)
workflow.add_edge("increment_retry", "retrieve")
workflow.add_edge("generate_direct", END)
 
graph = workflow.compile()

검색 결과 평가:
  |
  +--> [정확] --> 검색 결과를 그대로 사용하여 생성
  |
  +--> [부정확] --> 웹 검색 등 외부 소스에서 추가 검색
  |
  +--> [모호] --> 검색 결과 유지 + 외부 소스 보완 검색

from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Literal
from langchain_openai import ChatOpenAI
from langchain_community.tools import TavilySearchResults
 
class CRAGState(TypedDict):
    question: str
    retrieved_docs: List[str]
    web_results: List[str]
    quality_assessment: str
    final_context: List[str]
    generation: str
 
llm = ChatOpenAI(model="gpt-4o", temperature=0)
web_search = TavilySearchResults(max_results=3)
 
def retrieve(state: CRAGState) -> CRAGState:
    """내부 지식 베이스에서 검색"""
    results = retriever.invoke(state["question"])
    state["retrieved_docs"] = [doc.page_content for doc in results]
    return state
 
def assess_quality(state: CRAGState) -> CRAGState:
    """검색 결과 품질 평가"""
    prompt = ChatPromptTemplate.from_messages([
        ("system", """검색된 문서가 질문에 정확히 답할 수 있는지 평가하세요.
- "정확": 질문에 대한 명확한 답이 문서에 있음
- "부정확": 질문과 관련 없는 문서가 대부분임
- "모호": 관련 정보가 일부 있으나 완전하지 않음
위 세 가지 중 하나로만 답하세요."""),
        ("human", """질문: {question}
 
검색 결과:
{documents}""")
    ])
    docs_text = "\n\n".join(state["retrieved_docs"][:3])
    result = (prompt | llm).invoke({
        "question": state["question"],
        "documents": docs_text
    })
    state["quality_assessment"] = result.content.strip()
    return state
 
def route_by_quality(state: CRAGState) -> Literal["correct", "incorrect", "ambiguous"]:
    assessment = state.get("quality_assessment", "")
    if "정확" in assessment:
        return "correct"
    elif "부정확" in assessment:
        return "incorrect"
    return "ambiguous"
 
def use_retrieved(state: CRAGState) -> CRAGState:
    """검색 결과를 그대로 사용"""
    state["final_context"] = state["retrieved_docs"]
    return state
 
def search_web(state: CRAGState) -> CRAGState:
    """웹 검색으로 대체"""
    results = web_search.invoke(state["question"])
    state["web_results"] = [r["content"] for r in results]
    state["final_context"] = state["web_results"]
    return state
 
def combine_sources(state: CRAGState) -> CRAGState:
    """내부 검색 + 웹 검색 결합"""
    results = web_search.invoke(state["question"])
    state["web_results"] = [r["content"] for r in results]
    state["final_context"] = state["retrieved_docs"] + state["web_results"]
    return state
 
def generate(state: CRAGState) -> CRAGState:
    """최종 답변 생성"""
    prompt = ChatPromptTemplate.from_messages([
        ("system", "주어진 컨텍스트를 기반으로 질문에 답변하세요."),
        ("human", """컨텍스트:
{context}
 
질문: {question}""")
    ])
    context = "\n\n".join(state["final_context"])
    result = (prompt | llm).invoke({
        "question": state["question"],
        "context": context
    })
    state["generation"] = result.content
    return state
 
# CRAG 그래프 구성
crag = StateGraph(CRAGState)
crag.add_node("retrieve", retrieve)
crag.add_node("assess", assess_quality)
crag.add_node("use_retrieved", use_retrieved)
crag.add_node("search_web", search_web)
crag.add_node("combine", combine_sources)
crag.add_node("generate", generate)
 
crag.set_entry_point("retrieve")
crag.add_edge("retrieve", "assess")
crag.add_conditional_edges("assess", route_by_quality, {
    "correct": "use_retrieved",
    "incorrect": "search_web",
    "ambiguous": "combine"
})
crag.add_edge("use_retrieved", "generate")
crag.add_edge("search_web", "generate")
crag.add_edge("combine", "generate")
crag.add_edge("generate", END)
 
crag_graph = crag.compile()

from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Literal
 
class AdaptiveRAGState(TypedDict):
    question: str
    complexity: str
    retrieved_docs: List[str]
    sub_questions: List[str]
    intermediate_answers: List[str]
    final_answer: str
 
def classify_complexity(state: AdaptiveRAGState) -> AdaptiveRAGState:
    """질문 복잡도 분류"""
    prompt = ChatPromptTemplate.from_messages([
        ("system", """질문의 복잡도를 분류하세요.
- "단순": 하나의 문서에서 답을 찾을 수 있는 사실 질문
- "비교": 여러 항목을 비교해야 하는 질문
- "분석": 여러 문서를 종합하여 추론이 필요한 질문
위 세 가지 중 하나로만 답하세요."""),
        ("human", "{question}")
    ])
    result = (prompt | llm).invoke({"question": state["question"]})
    state["complexity"] = result.content.strip()
    return state
 
def route_by_complexity(state: AdaptiveRAGState) -> Literal["simple", "compare", "analyze"]:
    complexity = state.get("complexity", "")
    if "단순" in complexity:
        return "simple"
    elif "비교" in complexity:
        return "compare"
    return "analyze"
 
def simple_retrieval(state: AdaptiveRAGState) -> AdaptiveRAGState:
    """단순 검색 - 단일 쿼리"""
    results = retriever.invoke(state["question"])
    state["retrieved_docs"] = [doc.page_content for doc in results[:5]]
    return state
 
def decompose_question(state: AdaptiveRAGState) -> AdaptiveRAGState:
    """복잡한 질문을 하위 질문으로 분해"""
    prompt = ChatPromptTemplate.from_messages([
        ("system", """주어진 질문을 답변하기 위해 필요한 하위 질문들을 생성하세요.
각 줄에 하나의 하위 질문을 작성하세요. 최대 3개."""),
        ("human", "{question}")
    ])
    result = (prompt | llm).invoke({"question": state["question"]})
    state["sub_questions"] = [
        q.strip() for q in result.content.strip().split("\n")
        if q.strip()
    ]
    return state
 
def multi_step_retrieval(state: AdaptiveRAGState) -> AdaptiveRAGState:
    """멀티스텝 검색 - 하위 질문별 검색 후 통합"""
    all_docs = []
    intermediate = []
 
    for sub_q in state["sub_questions"]:
        results = retriever.invoke(sub_q)
        docs = [doc.page_content for doc in results[:3]]
        all_docs.extend(docs)
 
        # 하위 질문에 대한 중간 답변 생성
        context = "\n\n".join(docs)
        answer = (
            ChatPromptTemplate.from_messages([
                ("human", "컨텍스트:\n{context}\n\n질문: {question}")
            ]) | llm
        ).invoke({"context": context, "question": sub_q})
        intermediate.append(answer.content)
 
    state["retrieved_docs"] = all_docs
    state["intermediate_answers"] = intermediate
    return state

일반 RAG:
  "프로젝트 A의 일정은?" --> 해당 청크 검색 --> 답변
 
Graph RAG:
  "프로젝트 간의 의존 관계는?" --> 지식 그래프 탐색 --> 관계 기반 답변

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
import json
import networkx as nx
 
def extract_entities_and_relations(text, llm):
    """텍스트에서 엔티티와 관계를 추출"""
    prompt = ChatPromptTemplate.from_messages([
        ("system", """텍스트에서 주요 엔티티(사람, 조직, 개념, 기술 등)와
그들 사이의 관계를 추출하세요.
 
JSON 형식으로 응답하세요:
[
  {"source": "엔티티A", "relation": "관계", "target": "엔티티B"}
]"""),
        ("human", "{text}")
    ])
 
    result = (prompt | llm).invoke({"text": text})
    try:
        relations = json.loads(result.content)
        return relations
    except json.JSONDecodeError:
        return []
 
def build_knowledge_graph(documents, llm):
    """문서들로부터 지식 그래프 구축"""
    graph = nx.DiGraph()
 
    for doc in documents:
        relations = extract_entities_and_relations(doc.page_content, llm)
        for rel in relations:
            graph.add_edge(
                rel["source"],
                rel["target"],
                relation=rel["relation"]
            )
 
    print(f"노드 수: {graph.number_of_nodes()}")
    print(f"엣지 수: {graph.number_of_edges()}")
    return graph
 
# 그래프 기반 검색
def graph_search(graph, query_entities, max_hops=2):
    """지식 그래프에서 관련 서브그래프 추출"""
    relevant_nodes = set()
    for entity in query_entities:
        if entity in graph:
            # BFS로 인접 노드 탐색
            for node in nx.bfs_tree(graph, entity, depth_limit=max_hops):
                relevant_nodes.add(node)
 
    subgraph = graph.subgraph(relevant_nodes)
    # 서브그래프의 관계를 텍스트로 변환
    context_parts = []
    for source, target, data in subgraph.edges(data=True):
        context_parts.append(
            f"{source} --[{data['relation']}]--> {target}"
        )
    return "\n".join(context_parts)