[LangGraph] 단순 조회를 넘어 스스로 생각하는 AI: Agent RAG와 Self-RAG 정리

1. 도입부 (Introduction)

데이터베이스에서 관련 문서를 찾아서 LLM에게 넘겨주는 '표준 RAG(Naive RAG)' 시스템은 구현하기 매우 쉽다는 장점이 있다. 하지만 현업의 실제 비즈니스 도메인에 이를 적용하는 순간, 성능 평가 지표는 곤두박질치기 시작한다.

가장 빈번하게 발생하며 시스템 신뢰도를 무너뜨리는 세 가지 고질적인 실패 지점은 다음과 같다.

1. 부적절한 문서 검색(Low Quality Retrieval): 질문과 전혀 상관없는 문서를 들고 와 엉뚱한 답변을 낸다.
2. 모델의 환각(Hallucination): 검색해 온 참고 문서에 전혀 없는 가상의 수치나 사실을 지어내어 답변한다.
3. 불충분한 답변 품질(Low Helpfulness): 정답이 문서 어딘가에 존재함에도 불구하고 질문자가 원하는 본질적인 궁금증을 시원하게 긁어주지 못한다.

이 세 가지 결함을 소프트웨어 설계적으로 극복하기 위해 등장한 개념이 바로 에이전트 RAG(Agentic RAG)와 자가 교정형 RAG(Self-RAG)다. 이 아키텍처 하에서는 AI가 단순히 데이터를 읽고 쓰는 수동적인 도구가 아니다. 검색된 문서가 쓸모없다면 질문을 재작성하여 다시 찾아오고, 자신이 뱉은 답변이 환각에 오염되었다면 스스로 지우고 다시 답변을 지어내며, 질문에 충분히 대답하지 못했다면 단락을 분석하여 수정하는 '사유하는 주체(Agent)'가 된다.

이 글에서는 LangGraph의 유연한 상태 제어(State Management) 기법을 활용해 이 두 가지 진화된 RAG 아키텍처를 구현하는 과정을 코드와 원리를 통해 완벽히 파헤친다.

 

 

2. 주요 아키텍처 특징 및 핵심 개념 (Core Architecture & Concepts)

두 시스템이 동작하는 원리와 위상의 흐름을 직관적으로 이해하기 위해 구조 차이를 비교해 본다.

2.1 에이전트 RAG (Agent RAG - Query Rewrite Loop)

가장 기초적인 에이전트 RAG 모델은 검색 결과에 대해 피드백을 수동적으로 반영한다. 검색기(Retriever)가 찾아온 내용이 질문에 충분히 부합하는지 '평가 엔진(Document Relevance Grader)'이 심사한다. 만약 관련도가 떨어진다면, 사전에 내장된 표준 용어 사전(거주자 용어 치환 등)과 재작성 프롬프트를 활용해 질문 자체를 더 정교하게 변형(Rewrite)하여 재수색 루프를 타게 만든다.

 

2.2 고성능 Self-RAG (Self-Correction & Triple Critic Loop)

이 구조에서 한 단계 더 나아간 Self-RAG는 생성된 답변 자체에 대해 3단계 검증 시스템(Triple Critic Grader)을 순차적으로 기동시킨다.

1. Document Relevance Grader: 가져온 문맥 데이터(context)가 사용자 질문(query)에 실제로 대답할 수 있는 최소한의 가치를 지녔는가?
2. Hallucination Grader: 생성된 답변(answer)이 오직 제공된 문맥 데이터(context)에 기반하여 작성되었는가? (지어낸 사실이 없는가?)
3. Answer Helpfulness Grader: 생성된 답변(answer)이 사용자가 처음 물어봤던 본래 의도(query)에 명확하게 해답을 주는가?

 

 

3. 상세 가이드 및 심층 코드 구현 (Detailed Guide)

필습이 권장되는 두 가지 랭그래프 RAG 구현 코드를 단계별로 배치하고 핵심 라인별로 깊이 있게 분석해 보겠다.

STEP 1. 패키지 설치 및 벡터 데이터베이스 환경 동기화

우선 이전에 적재해 둔 income_tax_collection Chroma 데이터베이스를 안전하게 메모리로 올리기 위한 기본 선언 및 로더 구성을 세팅한다.

from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings

# 1. 문서 벡터 변환을 책임지는 대용량 임베딩 엔진 정의
embedding_function = OpenAIEmbeddings(model='text-embedding-3-large')

# 2. 로컬 디스크에 기록되어 있던 세금 문서 수집 컬렉션 세팅
vector_store = Chroma(
    embedding_function=embedding_function,
    collection_name='income_tax_collection',
    persist_directory='./income_tax_collection'
)

# 3. 검색 정밀도 극대화를 위해 유사도 상위 3개 단락을 수집하는 검색기 추출
retriever = vector_store.as_retriever(search_kwargs={'k': 3})

 

STEP 2. [아키텍처 1] 에이전트 RAG 구현 (Query Rewrite Loop)

이 파트에서는 검색 효율성이 떨어졌을 때 질문을 스스로 고쳐서 재검색을 처리하는 비교적 가벼운 루프형 RAG 에이전트를 조립한다.

from typing_extensions import List, TypedDict
from langchain_core.documents import Document
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from langsmith import Client
from typing import Literal
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 1. 상태 객체 선언 (에이전트 전체 노드에서 수시로 읽고 쓸 핵심 데이터 공유 룸)
class AgentState(TypedDict):
    query: str
    context: List[Document]
    answer: str

# 2. 그래프 빌더 수립
graph_builder = StateGraph(AgentState)

# 3. 검색 노드 정의
def retrieve(state: AgentState):
    query = state['query']
    docs = retriever.invoke(query)
    return {'context': docs}

# 4. LLM 및 표준 프롬프트 원격 풀링
llm = ChatOpenAI(model='gpt-4o', temperature=0)

client = Client()
generate_prompt = client.pull_prompt(
    "rlm/rag-prompt",
    dangerously_pull_public_prompt=True
)

# 5. 생성 노드 정의
def generate(state: AgentState):
    context = state['context']
    query = state['query']
    rag_chain = generate_prompt | llm
    response = rag_chain.invoke({'question': query, 'context': context})
    # 최종 답변을 가공하여 전달
    return {'answer': response.content}

# 6. 문서 적합성을 심사하는 조건부 라우팅 함수 (Grader)
doc_relevance_prompt = client.pull_prompt(
    "langchain-ai/rag-document-relevance:0626f225",
    dangerously_pull_public_prompt=True
)

def check_doc_relevance(state: AgentState) -> Literal['generate', 'rewrite']:
    query = state['query']
    context = state['context']
    
    # Grader 체인 기동
    doc_relevance_chain = doc_relevance_prompt | llm
    response = doc_relevance_chain.invoke({'question': query, 'documents': context})
    
    # LangSmith 허브 프롬프트는 판단 결과를 스코어로 반환함
    if response.get('Score') == 1 or response.get('score') == 1:
        print("[INFO] 검색된 문서가 질문과 일치합니다. 답변 생성 노드로 이동합니다.")
        return 'generate'
    
    print("[WARN] 질문과 일치하는 정보를 문서에서 찾지 못했습니다. 쿼리를 고치러 이동합니다.")
    return 'rewrite'

# 7. 질문 교정 노드 정의 (사전 매핑 기반 용어 치환 수색법)
dictionary = ['사뢈과 관련된 표현 -> 거주자']
rewrite_prompt = PromptTemplate.from_template(f"""
    사용자의 질문을 보고, 아래에 제공하는 표준 사전을 적극 참고하여 검색엔진에 최적화된 질문으로 고쳐줘.
    사전: {dictionary}
    기존 질문: {{query}}
    출력 결과는 오직 정교하게 번역된 질문 한 줄만 반환해야 한다.
""")

def rewrite(state: AgentState):
    query = state['query']
    rewrite_chain = rewrite_prompt | llm | StrOutputParser()
    response = rewrite_chain.invoke({'query': query})
    print(f"[REWRITE] 원본 질문이 다음으로 정규화되었습니다: {response}")
    return {'query': response}

# 8. 그래프의 정교한 매핑 빌드업
graph_builder.add_node('retrieve', retrieve)
graph_builder.add_node('generate', generate)
graph_builder.add_node('rewrite', rewrite)

graph_builder.add_edge(START, 'retrieve')
# 조건부 라우팅 에지 장착: 검색 문서 품질을 보고 generate로 갈지 rewrite로 갈지 판단
graph_builder.add_conditional_edges('retrieve', check_doc_relevance)
graph_builder.add_edge('rewrite', 'retrieve') # 질문을 다시 쓰고 나면 retrieve 노드로 백헤드
graph_builder.add_edge('generate', END)

# 9. 컴파일
simple_agent_graph = graph_builder.compile()

💡 심층 분석: check_doc_relevance의 조건부 에지 바인딩 원리

이 단순 에이전트 RAG 구조는 retrieve 노드가 동작을 마치고 반환한 상태값을 보고 스스로 분기(Conditional Edge)를 튼다. Grader 체인이 Score: 1을 출력하면 문서들이 유의미하므로 즉시 generate로 직진하지만, 0을 주면 사용자가 세법 전문 용어 대신 일상 구어체를 남발한 것으로 보고 질문 자체를 정규화하여 쿼리를 변환한 뒤 다시 retrieve로 루프를 돌리는 기법이다. 이 덕분에 잘못 수색된 결과물로 인해 엉뚱한 답변을 뱉는 병폐를 초기 단계에서 완전히 차단할 수 있다.

 

STEP 3. [아키텍처 2] 고성능 Self-RAG 구현 (Triple Critic Grader Loop)

이번 파트에서는 본문에서 다룬 3가지 비평가 관문(적합성 검사, 환각 감지, 도움말 채점)을 완벽하게 통합하여 답변의 무결성을 원천 보장하는 완성형 Self-RAG 파이프라인을 구축한다.

from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langsmith import Client
from typing import Literal

# 그래프 재생성용 인스턴스 초기화
self_rag_builder = StateGraph(AgentState)

# 1. 환각 평가 엔진 (Hallucination Grader) 프롬프트 및 판단 함수 설계
hallucination_prompt = PromptTemplate.from_template("""
You are a teacher tasked with evaluating whether a student's answer is based on documents or not,
Given documents, which are excerpts from income tax law, and a student's answer;
If the student's answer is based on documents, respond with "not hallucinated",
If the student's answer is not based on documents, respond with "hallucinated".

documents: {documents}
student_answer: {student_answer}
""")

# 환각 판독은 무결성이 극대화되어야 하므로 일관성 추론을 위해 온도를 0으로 규정한다.
hallucination_llm = ChatOpenAI(model='gpt-4o', temperature=0)

def check_hallucination(state: AgentState) -> Literal['hallucinated', 'not hallucinated']:
    answer = state['answer']
    context = state['context']
    
    # document 조각들을 스트링 리스트로 단순 정규화
    documents_content = [doc.page_content for doc in context]
    hallucination_chain = hallucination_prompt | hallucination_llm | StrOutputParser()
    
    response = hallucination_chain.invoke({
        'student_answer': answer, 
        'documents': documents_content
    })
    
    # 텍스트 데이터의 줄바꿈이나 공백을 완전히 털어낸 후 비교 검증
    verdict = response.strip()
    print(f"[CRITIC 2 - 환각 판정]: 결과는 '{verdict}' 입니다.")
    return verdict

# 2. 답변 유용성 평가 엔진 (Helpfulness Grader) 풀링 및 가동 함수 설계
client = Client()
helpfulness_prompt = client.pull_prompt(
    "langchain-ai/rag-answer-helpfulness",
    dangerously_pull_public_prompt=True
)

def check_helpfulness_grader(state: AgentState) -> Literal['helpful', 'unhelpful']:
    query = state['query']
    answer = state['answer']
    
    helpfulness_chain = helpfulness_prompt | llm
    response = helpfulness_chain.invoke({
        'question': query, 
        'student_answer': answer
    })
    
    # 응답 사전에 담겨 오는 Score 속성 평가 (1일 시 유용성 합격)
    score = response.get('Score') or response.get('score')
    if score == 1:
        print("[CRITIC 3 - 도움 판정]: 최종 답변이 질문자의 핵심 본질을 시원하게 해결했습니다.")
        return 'helpful'
    
    print("[CRITIC 3 - 도움 판정]: 답변이 너무 모호하거나 빗나갔습니다. 쿼리 개선 후 재수색 단계로 이동합니다.")
    return 'unhelpful'

# 상태 변환에 영향 없이 통과하는 pass-through 노드 정의 (중간 상태 점검용)
def check_helpfulness_passthrough_node(state: AgentState):
    return state

# 3. 노드 등록 프로세스 진행
self_rag_builder.add_node('retrieve', retrieve)
self_rag_builder.add_node('generate', generate)
self_rag_builder.add_node('rewrite', rewrite)
self_rag_builder.add_node('check_helpfulness', check_helpfulness_passthrough_node)

# 4. 고차원적 에지 바인딩 및 3단계 교정 루프 설계
self_rag_builder.add_edge(START, 'retrieve')

# 관문 1: 문서가 유용하지 않으면 즉시 END로 분기하여 탐색 중단
self_rag_builder.add_conditional_edges(
    'retrieve',
    check_doc_relevance,
    {
        'generate': 'generate',
        'rewrite': END # 소득세 관련 사실이 아니면 더 복잡하게 돌지 않고 차단함
    }
)

# 관문 2: 답변 생성 이후 환각 체크 진행
self_rag_builder.add_conditional_edges(
    'generate',
    check_hallucination,
    {
        'not hallucinated': 'check_helpfulness', # 환각이 없으면 다음 도움말 심사로 이동
        'hallucinated': 'generate'              # 환각 발견 시 즉시 자가 교정 및 재합성
    }
)

# 관문 3: 최종 유용성 체크 진행
self_rag_builder.add_conditional_edges(
    'check_helpfulness',
    check_helpfulness_grader,
    {
        'helpful': END,              # 합격 시 성공적인 탈출
        'unhelpful': 'rewrite'       # 불만족 시 질문을 재규격화(Rewrite)하여 수색기 재가동
    }
)

# 질문 재작성 이후 다시 검색으로 연결
self_rag_builder.add_edge('rewrite', 'retrieve')

# 5. 최종 컴파일
self_rag_graph = self_rag_builder.compile()

 

STEP 4. 세법 RAG 최종 구동 및 응답 수집 검증

이제 구축된 자가 교정형 고성능 Self-RAG 그래프를 활용해, 복잡한 종합소득세 세율 계산식을 정확히 추론해 내는지 직접 구동하여 평가를 진행해 본다.

# 검증 질문: 정확한 세법 명칭과 계산 방법을 유구하는 날카로운 질의
initial_state = {
    'query': '소득세법 제55조 세율표에서 종합소득 과세표준 5천만원 구간의 산출세액 계산식은 무엇인가요??'
}

# 랭그래프 런타임 가동
final_state = self_rag_graph.invoke(initial_state)

print("\n=== [Self-RAG 최종 출력 답변] ===")
print(final_state['answer'])

 

 

4. 실전 팁 및 예외 상황 대응 (Practical Tips & Troubleshooting)

자가 검증 루프가 복잡하게 얽혀 있는 Self-RAG를 프로덕션 환경에 배포하여 운영할 때 반드시 대응해야 할 핵심 엔지니어링 규칙들을 기술한다.

1) 무한 순환 데드락(Infinite Loop)과 Recursion Limit 제약

그레이더 평가 통과율이 나쁘거나 질문이 과도하게 모호할 경우, 그래프가 generate -> hallucinated -> generate 혹은 unhelpful -> rewrite -> retrieve 구간을 무한히 도는 현상(무한 순환 데드락)이 종종 발생할 수 있다. 이 경우 시스템 리소스를 낭비하고 토큰 비용 폭탄을 초래하게 된다.

• 해결책: 랭그래프는 무한 순환 방지를 위한 빌트인 방어 체계를 제공한다. 실행 시 설정 객체(config) 내부에 recursion_limit 값을 제한하여 시스템 안정성을 구축해야 한다.

# 무한 수색 루프 차단용 세션 튜닝 코드
config = {"recursion_limit": 15} # 최대 노드 이동 횟수 도달 시 루프 즉시 제지 및 에러 차단
final_state = self_rag_graph.invoke(initial_state, config=config)

2) LangSmith 비평가 프롬프트 버전 관리와 Local Caching

매 연산 단계마다 client.pull_prompt()를 호출해 원격에서 프롬프트를 풀링해 오면, 원격 네트워크 상황에 따라 미세한 레이턴시 지연을 겪거나 간헐적인 통신 타임아웃 예외가 생길 수 있다.

• 해결책: 개발 완료 이후 스테이징 및 라이브 배포 시점에는 수동 풀링 코드를 제거하고, 해당 프롬프트를 시스템 로컬 텍스트 변수나 로컬 환경 설정 파일로 캐싱하여 탑재하는 편이 성능 면에서 극적으로 유리하다.

 

 

5. 결론 및 마무리하며 (Conclusion)

LangGraph를 활용한 AI Agent 개발를 수강하면서 기존 랭체인이 보여줬던 한계인 '순차적 체이닝'을 완벽하게 해소해 주는 랭그래프(LangGraph)의 상태 전이 기법을 확인해 보았다.

단순히 LLM을 믿고 답을 기다리던 시대를 넘어, 이제 AI 에이전트 시스템은 스스로 수집된 문서를 비평하고, 뱉은 문장의 참과 거짓을 수학적으로 가늠하며, 질문자가 본질적으로 원하는 결론에 도달할 때까지 끊임없이 행동을 정규화해 가는 '자가 평가 구조'로 고도화되고 있다.

이 설계 패턴을 완수하고 나면, 데이터 전처리 단계에서 비전 모델로 표를 정확하게 적재한 뒤, 랭그래프 위에서 그 수치를 교차 비평하는 세상에서 가장 안전하고 똑똑한 소득세법 AI 어드바이저를 탄생시킬 수 있다. 데이터의 오염이 일어날 틈을 주기 싫다면, 오늘 당장 단순 RAG 코드를 걷어내고 Self-RAG의 3중 그레이더 검문소를 설치해 보길 강력히 권장한다.