Self-RAG
Self-RAG는 LLM이 검색 여부를 스스로 판단하고, 생성한 답변의 품질을 자체 평가하여 기준에 미달하면 재생성하는 아키텍처입니다. 4가지 Reflection Token을 통해 검색 필요성, 문서 관련성, 답변의 근거 유무, 답변의 유용성을 단계적으로 검증합니다.
핵심 아이디어
Naive RAG는 항상 검색을 수행하고, 결과를 검증하지 않습니다. Self-RAG는 네 가지 Reflection Token을 도입하여 검색부터 생성까지 각 단계에서 자기 평가를 수행합니다.
Reflection Token의 이름과 판정값은 Asai et al. (2023) 원 논문을 기준으로 합니다.
Reflection Token 평가 대상 판정값 Retrieve 검색이 필요한가? yes / no / continueIsRel (Is Relevant)검색된 문서가 질문과 관련 있는가? relevant / irrelevantIsSup (Is Supported)생성된 답변이 문서에 근거하는가? fully supported / partially supported / no supportIsUse (Is Useful)최종 답변이 질문에 유용한가? 5 / 4 / 3 / 2 / 1 (5점 척도)
Retrieve : 생성 도중 추가 검색이 필요한지 판단합니다. yes면 검색을 수행하고, no면 검색 없이 생성을 계속하며, continue는 현재 생성을 이어갑니다. 이 토큰 덕분에 Self-RAG는 항상 검색하지 않고 필요할 때만 검색 합니다.IsRel : 검색된 각 문서가 질문에 관련 있는지 평가합니다. irrelevant 문서는 컨텍스트에서 제외됩니다.IsSup : 생성된 문장이 검색 문서에 의해 뒷받침되는지 평가합니다. 3단계 판정을 통해 fully supported(완전히 근거 있음), partially supported(부분적 근거), no support(근거 없음)로 나눕니다.IsUse : 최종 답변의 전반적인 유용성을 1~5점으로 평가합니다. 높은 점수일수록 질문에 대해 유용한 답변입니다.
동작 방식
네 단계의 자기 평가를 거치며, 각 단계에서 기준을 충족하지 못하면 이전 단계로 돌아가 재시도합니다.
LangGraph 구현
원 논문에서는 Reflection Token을 모델 학습 단계에서 직접 생성하도록 훈련합니다. 아래 구현은 LLM 프롬프팅으로 동일한 판정 로직을 근사하는 방식입니다.
상태 정의
from typing import TypedDict, List, Literal
from langchain_core.documents import Document
class GraphState ( TypedDict ):
question: str
documents: List[Document]
generation: str
retry_count: int
노드 함수 작성
검색 노드
문서 평가 (IsRel)
생성 노드
생성 평가 (IsSup + IsUse)
from langchain.chat_models import init_chat_model
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
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 })
def retrieve ( state : GraphState) -> GraphState:
"""문서를 검색합니다."""
question = state[ "question" ]
documents = retriever.invoke(question)
return { "documents" : documents, "retry_count" : state.get( "retry_count" , 0 )}
def grade_documents ( state : GraphState) -> GraphState:
"""검색된 문서의 관련성을 평가합니다 (IsRel)."""
question = state[ "question" ]
documents = state[ "documents" ]
prompt = ChatPromptTemplate.from_messages([
( "system" , (
"문서가 질문에 관련이 있는지 판단하세요. \n "
"'relevant' 또는 'irrelevant'만 출력하세요."
)),
( "human" , "질문: {question} \n\n 문서: {document} " ),
])
chain = prompt | llm | StrOutputParser()
relevant_docs = []
for doc in documents:
result = chain.invoke({ "question" : question, "document" : doc.page_content})
if "relevant" in result.lower() and "irrelevant" not in result.lower():
relevant_docs.append(doc)
return { "documents" : relevant_docs}
def generate ( state : GraphState) -> GraphState:
"""관련 문서를 기반으로 답변을 생성합니다."""
question = state[ "question" ]
documents = state[ "documents" ]
prompt = ChatPromptTemplate.from_messages([
( "system" , "다음 컨텍스트만을 참고하여 질문에 답변하세요. \n\n {context} " ),
( "human" , " {question} " ),
])
chain = prompt | llm | StrOutputParser()
context = " \n\n " .join(doc.page_content for doc in documents)
generation = chain.invoke({ "context" : context, "question" : question})
return { "generation" : generation}
def grade_generation ( state : GraphState) -> GraphState:
"""생성된 답변의 근거와 유용성을 평가합니다 (IsSup + IsUse)."""
question = state[ "question" ]
documents = state[ "documents" ]
generation = state[ "generation" ]
# IsSup: 근거 확인 (3단계)
supported_prompt = ChatPromptTemplate.from_messages([
( "system" , (
"답변이 주어진 문서에 근거하는지 판단하세요. \n "
"'fully_supported', 'partially_supported', 'no_support' 중 하나만 출력하세요."
)),
( "human" , "문서: {documents} \n\n 답변: {generation} " ),
])
# IsUse: 유용성 확인 (5점 척도)
useful_prompt = ChatPromptTemplate.from_messages([
( "system" , (
"답변이 질문에 얼마나 유용한지 1~5점으로 평가하세요. \n "
"숫자만 출력하세요."
)),
( "human" , "질문: {question} \n\n 답변: {generation} " ),
])
docs_text = " \n " .join(doc.page_content for doc in documents)
supported_chain = supported_prompt | llm | StrOutputParser()
useful_chain = useful_prompt | llm | StrOutputParser()
supported = supported_chain.invoke({ "documents" : docs_text, "generation" : generation})
useful_score = useful_chain.invoke({ "question" : question, "generation" : generation})
state[ "is_supported" ] = "fully_supported" in supported.lower()
try :
state[ "usefulness_score" ] = int (useful_score.strip())
except ValueError :
state[ "usefulness_score" ] = 1
state[ "retry_count" ] = state.get( "retry_count" , 0 ) + 1
return state
조건부 엣지 정의
def decide_to_generate ( state : GraphState) -> Literal[ "generate" , "retrieve" ]:
"""관련 문서가 있으면 생성, 없으면 재검색합니다."""
if state[ "documents" ]:
return "generate"
return "retrieve"
def check_generation ( state : GraphState) -> Literal[ "end" , "retrieve" ]:
"""생성 결과가 기준을 충족하는지 확인합니다."""
if state.get( "retry_count" , 0 ) >= 3 :
return "end" # 최대 재시도 횟수 초과
if state.get( "is_supported" ) and state.get( "usefulness_score" , 0 ) >= 4 :
return "end"
return "retrieve"
그래프 구성 및 실행
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( "generate" , generate)
workflow.add_node( "grade_generation" , grade_generation)
# 엣지 연결
# 참고: 원 논문의 Self-RAG는 Retrieve 토큰으로 "검색 필요 여부"를 먼저 판단합니다 (on-demand retrieval).
# 아래 구현은 단순화를 위해 항상 검색부터 시작하고, 생성 품질 평가 후 재검색하는 루프로 구성합니다.
workflow.add_edge( START , "retrieve" )
workflow.add_edge( "retrieve" , "grade_documents" )
workflow.add_conditional_edges( "grade_documents" , decide_to_generate, {
"generate" : "generate" ,
"retrieve" : "retrieve" ,
})
workflow.add_edge( "generate" , "grade_generation" )
workflow.add_conditional_edges( "grade_generation" , check_generation, {
"end" : END ,
"retrieve" : "retrieve" ,
})
# 컴파일 및 실행
app = workflow.compile()
result = app.invoke({ "question" : "Self-RAG의 동작 원리는?" , "retry_count" : 0 })
print (result[ "generation" ])
Self-RAG는 매 단계마다 LLM 호출이 발생하므로 Naive RAG 대비 토큰 사용량이 증가합니다. 재시도 횟수에 상한을 설정하여 무한 루프를 방지해야 합니다.
Naive RAG와의 차이
항목 Naive RAG Self-RAG 검색 판단 항상 검색 Retrieve 토큰으로 필요 시 검색 문서 관련성 평가 없음 IsRel로 필터링 환각 검증 없음 IsSup로 근거 확인 (3단계) 답변 유용성 평가 없음 IsUse로 유용성 평가 (5점 척도) 재시도 불가 기준 미달 시 재검색/재생성 그래프 구조 선형 루프 포함
참고 논문
논문 학회 링크 Self-RAG: Learning to Retrieve, Generate, and Critique through Self-Reflection (Asai et al., 2023) ICLR 2024 arXiv 2310.11511