Skip to main content

​
Corrective RAG (CRAG)

Corrective RAG는 검색된 문서의 품질을 평가하고, 품질이 낮으면 웹 검색으로 전환하여 보완하는 아키텍처입니다. 벡터 데이터베이스의 검색 결과가 부정확할 때, 외부 소스로 자동 폴백하여 답변의 정확도를 유지합니다.

​
핵심 아이디어

CRAG의 핵심은 검색 결과에 대한 3가지 판정입니다.
판정조건행동
Correct관련 문서가 충분함검색 결과를 그대로 사용하여 생성
Incorrect관련 문서가 없음웹 검색으로 전환하여 새로운 정보 확보
Ambiguous일부만 관련 있음검색 결과 + 웹 검색 결과를 함께 사용
추가로 Knowledge Refinement 과정을 통해, 검색된 문서를 작은 단위(knowledge strip)로 분해한 뒤 관련 있는 부분만 필터링하여 노이즈를 제거합니다.
원 논문의 CRAG는 검색 결과에 대해 confidence score 기반의 연속적 판정을 수행합니다. 위 3분류(Correct/Incorrect/Ambiguous)는 이해를 돕기 위한 단순화된 표현이며, 실제 논문에서는 confidence 임계값에 따라 더 세밀하게 행동을 결정합니다.
Knowledge Refinement은 검색된 문서를 그대로 사용하지 않고, 더 작은 단위로 분해하여 정제하는 과정입니다.
  1. 분해(Decompose): 검색된 문서를 문장 또는 의미 단위의 knowledge strip으로 분할합니다
  2. 필터링(Filter): 각 strip이 질문과 관련 있는지 평가하여 관련 없는 strip을 제거합니다
  3. 재구성(Recompose): 관련 있는 strip만 모아 정제된 컨텍스트를 구성합니다
이 과정을 통해 원본 문서의 노이즈(질문과 무관한 내용)를 제거하고, LLM이 핵심 정보에 집중할 수 있게 합니다.

​
동작 방식

​
LangGraph 구현

​
상태 정의

from typing import TypedDict, List, Literal
from langchain_core.documents import Document

class GraphState(TypedDict):
    question: str
    documents: List[Document]
    web_results: List[Document]
    grade: str  # "correct", "incorrect", "ambiguous"
    generation: str

​
노드 함수

from langchain.chat_models import init_chat_model
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain_community.tools import TavilySearchResults
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

llm = init_chat_model("gpt-4o-mini", temperature=0)

embeddings = OpenAIEmbeddings()
vectorstore = Chroma(collection_name="documents", embedding_function=embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

web_search = TavilySearchResults(max_results=3)

def retrieve(state: GraphState) -> GraphState:
    """벡터 데이터베이스에서 문서를 검색합니다."""
    question = state["question"]
    documents = retriever.invoke(question)
    return {"documents": documents}

def grade_documents(state: GraphState) -> GraphState:
    """검색된 문서의 전체적인 품질을 평가합니다."""
    question = state["question"]
    documents = state["documents"]

    prompt = ChatPromptTemplate.from_messages([
        ("system", (
            "질문과 검색 결과를 비교하여 검색 품질을 판정하세요.\n"
            "- 대부분 관련 있으면: correct\n"
            "- 대부분 관련 없으면: incorrect\n"
            "- 일부만 관련 있으면: ambiguous\n"
            "'correct', 'incorrect', 'ambiguous' 중 하나만 출력하세요."
        )),
        ("human", "질문: {question}\n\n검색 결과:\n{documents}"),
    ])

    chain = prompt | llm | StrOutputParser()
    docs_text = "\n---\n".join(doc.page_content for doc in documents)
    grade = chain.invoke({"question": question, "documents": docs_text})
    return {"grade": grade.strip().lower()}

def web_search_node(state: GraphState) -> GraphState:
    """웹 검색으로 추가 정보를 확보합니다."""
    question = state["question"]
    results = web_search.invoke(question)
    web_docs = [
        Document(page_content=r["content"], metadata={"source": r["url"]})
        for r in results
    ]
    return {"web_results": web_docs}

def knowledge_refinement(state: GraphState) -> GraphState:
    """검색된 문서에서 관련 있는 부분만 추출합니다."""
    question = state["question"]
    documents = state["documents"]

    prompt = ChatPromptTemplate.from_messages([
        ("system", (
            "문서에서 질문과 관련된 핵심 정보만 추출하세요.\n"
            "관련 없는 내용은 제거하세요."
        )),
        ("human", "질문: {question}\n\n문서: {document}"),
    ])

    chain = prompt | llm | StrOutputParser()

    refined_docs = []
    for doc in documents:
        refined = chain.invoke({"question": question, "document": doc.page_content})
        if refined.strip():
            refined_docs.append(Document(page_content=refined, metadata=doc.metadata))

    return {"documents": refined_docs}

def generate(state: GraphState) -> GraphState:
    """수집된 정보를 기반으로 답변을 생성합니다."""
    question = state["question"]

    all_docs = state.get("documents", []) + state.get("web_results", [])

    prompt = ChatPromptTemplate.from_messages([
        ("system", "다음 컨텍스트를 참고하여 질문에 답변하세요.\n\n{context}"),
        ("human", "{question}"),
    ])

    chain = prompt | llm | StrOutputParser()
    context = "\n\n".join(doc.page_content for doc in all_docs)
    generation = chain.invoke({"context": context, "question": question})
    return {"generation": generation}

​
조건부 엣지

def route_by_grade(state: GraphState) -> Literal["correct", "incorrect", "ambiguous"]:
    """검색 품질 판정에 따라 다음 단계를 결정합니다."""
    grade = state.get("grade", "incorrect")
    if grade == "correct":
        return "correct"
    elif grade == "ambiguous":
        return "ambiguous"
    else:
        return "incorrect"

​
그래프 구성

from langgraph.graph import StateGraph, START, END

workflow = StateGraph(GraphState)

# 노드 추가
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade_documents", grade_documents)
workflow.add_node("knowledge_refinement", knowledge_refinement)
workflow.add_node("web_search", web_search_node)
workflow.add_node("generate", generate)

# 엣지 연결
workflow.add_edge(START, "retrieve")
workflow.add_edge("retrieve", "grade_documents")

workflow.add_conditional_edges("grade_documents", route_by_grade, {
    "correct": "knowledge_refinement",
    "incorrect": "web_search",
    "ambiguous": "knowledge_refinement",
})

workflow.add_edge("knowledge_refinement", "generate")
workflow.add_edge("web_search", "generate")
workflow.add_edge("generate", END)

# 컴파일
app = workflow.compile()
Ambiguous 판정의 경우, knowledge_refinement 후 web_search도 병렬로 실행하는 것이 이상적입니다. 위 예제에서는 단순화를 위해 knowledge_refinement만 수행하지만, 실제 구현에서는 두 경로의 결과를 병합할 수 있습니다.

​
실행

result = app.invoke({"question": "2024년 RAG 최신 동향은?"})
print(result["generation"])

​
Self-RAG와의 차이

항목Self-RAGCRAG
평가 대상생성된 답변검색된 문서
보정 방식재생성 (루프)웹 검색 폴백
외부 소스사용하지 않음웹 검색 활용
Knowledge Refinement없음문서를 strip으로 분해 후 필터링
그래프 구조루프조건부 분기
웹 검색은 외부 API 호출이 필요합니다. 위 예제에서는 Tavily를 사용하며, TAVILY_API_KEY 환경 변수 설정이 필요합니다.

​
참고 논문

논문학회링크
Corrective Retrieval Augmented Generation (Yan et al., 2024)-arXiv 2401.15884