[LangGraph] 로컬 데이터의 한계 극복: Corrective RAG(CRAG)와 계층형 서브그래프 설계 가이드

1. 도입부 (Introduction)

엔터프라이즈 환경에서 검색 증강 생성(RAG, Retrieval-Augmented Generation)을 상용화할 때, 사내 구축용 문서(Knowledge Base)만을 바라보게 설계된 단순한 구조는 쉽게 무너진다. 사용자가 사내 문서 범주를 벗어난 최신 트렌드를 묻거나, 도메인에 해당하지 않는 질문(예: 세법 상담 챗봇에게 맛집 추천 요구)을 던졌을 때 시스템은 멍청하게 침묵하거나 잘못된 문서를 억지로 끼워 맞춰 환각(Hallucination)을 생성하게 된다.

이러한 한계를 원천 타파하기 위해 등장한 혁신적인 설계 기법이 바로 Corrective RAG(CRAG)와 계층형 서브그래프(Subgraph) 패턴이다.

CRAG는 로컬 벡터 스페이스(Vector Store)에서 검색된 정보의 적합성이 떨어지면, 즉각적으로 웹 검색 도구(Tavily Search)를 트리거하여 누락된 지식을 자가 보완해 내는 능동적 지식 치유 시스템이다. 한편, 서브그래프 설계 패턴은 전체 에이전트의 뇌 역할을 하는 코디네이터(Coordinator) 상위 그래프 하위에 기능별 독립 에이전트 그래프를 단일 노드(Node) 형태로 중첩하는 모듈러 아키텍처다.

이 글에서는 LangGraph의 핵심적인 강점인 '조건부 분기(Conditional Edge)'와 '독립적 상태 캡슐화'를 백십분 활용하여, 두 진화형 아키텍처를 프로덕션 레벨 코드로 구현하는 방법론을 체계적으로 전개한다.

 

 

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

우리가 설계할 두 시스템이 어떠한 위상 구조와 메시지 파이프라인을 지니는지 다이어그램 가이드를 통해 시각적으로 조감해 보자.

2.1 Corrective RAG (CRAG) 아키텍처 원리

로컬 검색이 실패하는 예외 상황을 어떻게 웹 검색으로 극복하는지 그 복구 메커니즘의 뼈대를 살펴본다.

 

2.2 계층형 서브그래프 (Subgraph / Sub-agent) 아키텍처 원리

거대한 오케스트레이션에서 단일 그래프가 수십 개의 노드를 제어하는 Ad-hoc 구조는 디버깅이 불가능하다. 각각의 비즈니스 도메인을 독립된 서브에이전트 그래프로 격리하고 계층적 우선순위로 통제하는 방안이다.

 

 

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

작동 방식의 뼈대를 이해했으니, 실전에 가동될 파이썬 코드를 낱낱이 해부하고 주석을 통해 각 컴포넌트의 유기적 매핑을 학습해 본다.

 

STEP 1. Corrective RAG (CRAG) 구현

아래 코드는 로컬 벡터 DB의 check_doc_relevance 스코어링 결과에 기반하여, 필요시 질문을 변환(rewrite)한 후 실시간 웹 검색 도구(TavilySearch)를 호출하는 에이전트 RAG 파이프라인의 완성형 구조다.

# %%
from dotenv import load_dotenv
load_dotenv()

from typing_extensions import List, TypedDict
from typing import Literal
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_tavily import TavilySearch
from langsmith import Client
from langgraph.graph import StateGraph, START, END

# 1. 랭그래프 공통 전역 상태(State) 설계
class AgentState(TypedDict):
    query: str
    context: list  # 로컬 문서 청크 혹은 웹 검색 결과 리포트가 적재되는 필드
    answer: str
    
graph_builder = StateGraph(AgentState)

# 2. 로컬 벡터 스토어 바인딩 및 retriever 선언
embedding_function = OpenAIEmbeddings(model='text-embedding-3-large')
vector_store = Chroma(
    embedding_function=embedding_function,
    collection_name='income_tax_collection',
    persist_directory='./income_tax_collection'
)
retriever = vector_store.as_retriever(search_kwargs={'k': 3})

# 3. 로컬 DB 수색 노드 구현
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
)

def generate(state: AgentState):
    context = state['context']
    query = state['query']
    rag_chain = generate_prompt | llm | StrOutputParser()
    response = rag_chain.invoke({'question': query, 'context': context})
    return {'answer': response}

# 5. 로컬 문서의 유용성을 판별하는 평가 관문 (Relevance 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['relevant', 'irrelevant']:
    query = state['query']
    context = state['context']
    
    doc_relevance_chain = doc_relevance_prompt | llm
    response = doc_relevance_chain.invoke({'question': query, 'documents': context})
    
    # 판정 결과 출력 및 에지 라우팅 결정
    if response.get('Score') == 1 or response.get('score') == 1:
        print("[CRAG] 로컬 지식 문서가 적합한 것으로 판정되었습니다. 생성 단계로 진입합니다.")
        return 'relevant'
    
    print("[CRAG] 적절한 로컬 문서를 수집하지 못했습니다. 웹 수색 트랙으로 전이합니다.")
    return 'irrelevant'

# 6. 구체적인 검색 키워드 최적화 변환(Rewrite) 노드 구현
rewrite_prompt = PromptTemplate.from_template("""
    사용자의 원본 질문을 면밀히 분석한 후, 대형 웹 포털이나 검색 엔진에서 최적의 결론을 이끌어낼 수 있는 정교한 검색 쿼리용 키워드로 다듬어줘.
    원본 질문: {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}

# 7. Tavily 실시간 고속 웹 검색 노드 구현
tavily_search_tool = TavilySearch(
    max_results=3,
    include_answer=True,
    include_raw_content=True,
    search_depth="advanced"
)

def web_search(state: AgentState):
    query = state['query']
    results = tavily_search_tool.invoke(query)
    print(f"[WEB SEARCH] Tavily 엔진이 다음 데이터를 수집했습니다: {results}")
    return {'context': results}

# 8. 그래프 노드 바인딩 및 조건부 라우팅 설정
graph_builder.add_node('retrieve', retrieve)
graph_builder.add_node('generate', generate)
graph_builder.add_node('rewrite', rewrite)
graph_builder.add_node('web_search', web_search)

graph_builder.add_edge(START, 'retrieve')

# 조건부 분기 설정
graph_builder.add_conditional_edges(
    'retrieve',
    check_doc_relevance,
    {
        'relevant': 'generate',     # 문서 적합 시 곧바로 답변 생성
        'irrelevant': 'rewrite'     # 부적합 시 질문을 다시 쓰고 웹 수색 가동
    }
)
graph_builder.add_edge('rewrite', 'web_search')
graph_builder.add_edge('web_search', 'generate')
graph_builder.add_edge('generate', END)

# 9. CRAG 컴파일
crag_graph = graph_builder.compile()

 

STEP 2. 계층형 서브그래프 및 구조적 출력 라우터 구현

단순한 일차원적 구조를 탈피하여, 랭그래프의 정수인 '컴파일된 타 그래프의 단일 노드화(Subgraphing)' 패턴을 적용한 엔터프라이즈 코디네이터 아키텍처다.

상위 게이트웨이에서는 Route 객체를 pydantic 모델로 구조화하여 수집 도메인을 가르며, 복잡한 세금 관련 연산은 이미 컴파일된 하위 그래프인 income_tax_subgraph에 상태를 통째로 이관해 수행하게 만든다.

# %%
from typing_extensions import TypedDict
from typing import Literal
from pydantic import BaseModel, Field
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langgraph.graph import StateGraph, START, END

# 1. 상위 게이트웨이 그래프에서 공유할 코디네이터 상태(State) 정의
class CoordinatorState(TypedDict):
    query: str
    context: list
    answer: str

parent_builder = StateGraph(CoordinatorState)

# 2. Pydantic을 활용한 라우팅 의사결정 모델 설계
class Route(BaseModel):
    target: Literal['vector_store', 'llm', 'web_search'] = Field(
        description = "사용자 질문을 효율적으로 해결하기 위해 매핑할 최적의 대상 스토어 및 도구"
    )

# 3. 고정 조건 기반의 지능형 시스템 라우터 설계
router_system_prompt = """
You are an expert at routing a user's question to 'vector_store', 'llm', or 'web_search'.
- 거주자, 소득세법, 종합소득세, 누진공제, 과세표준 구간 등 세무/세법 지식을 요구하는 경우, 반드시 'vector_store'로 라우팅해라.
- 맛집 추천, 날씨 정보, 최신 연예 소식 등 사내 세법 데이터를 벗어난 실시간 외부 지식을 묻는 경우 'web_search'를 선택해라.
- 단순 일상 인사, 아주 단순한 개념 정의 등은 'llm'을 선택해라.
"""

router_prompt = ChatPromptTemplate.from_messages([
    ('system', router_system_prompt),
    ('user', '{query}')
])

# 라우팅은 정확하고 빠른 토큰 처리가 생명인 영역이므로 gpt-4o-mini를 적용한다.
router_llm = ChatOpenAI(model="gpt-4o-mini")
structured_router_llm = router_llm.with_structured_output(Route)

def router(state: CoordinatorState):
    query = state['query']
    router_chain = router_prompt | structured_router_llm
    route = router_chain.invoke({'query': query})
    print(f"[GATEWAY ROUTER] 판별된 실행 목적지: {route.target}")
    return route.target

# 4. 일반 일상 대화를 다루는 basic_generate 노드 구현
basic_llm = ChatOpenAI(model="gpt-4o-mini")

def basic_generate(state: CoordinatorState):
    query = state['query']
    # ⚠️ 중요 리팩터링 노트: 기존 코드의 typo인 'basic_ll_chain'을 'basic_llm_chain'으로 안정 정교화 완료
    basic_llm_chain = basic_llm | StrOutputParser()
    llm_response = basic_llm_chain.invoke(query)
    return {'answer': llm_response}

# 5. 일반 외부 검색 연산용 Tavily 및 제네릭 생성 노드 구현
tavily_search_tool = TavilySearch(max_results=3, search_depth="advanced")

def parent_web_search(state: CoordinatorState):
    query = state['query']
    results = tavily_search_tool.invoke(query)
    return {'context': results}

generate_llm = ChatOpenAI(model="gpt-4o")

def web_generate(state: CoordinatorState):
    context = state['context']
    query = state['query']
    rag_chain = generate_prompt | generate_llm | StrOutputParser()
    response = rag_chain.invoke({'question': query, 'context': context})
    return {'answer': response}

# 6. 하위 세법 Q&A 그래프(서브그래프) 바인딩
# 사전에 정교하게 컴파일해 둔 하위 세법 전용 RAG 그래프 인스턴스를 가져온다.
# 이 서브그래프는 relevance checking, hallucination grader, helpfulness grader를 자체 내장하고 있다.
from income_tax_graph import graph as income_tax_subgraph

# 7. 노드 장착 및 계층적 워크플로우 구성
parent_builder.add_node('income_tax_agent', income_tax_subgraph) # 컴파일된 그래프 객체를 노드로 직접 적재
parent_builder.add_node('basic_generate', basic_generate)
parent_builder.add_node('web_search', parent_web_search)
parent_builder.add_node('web_generate', web_generate)

# 조건부 분기 게이트 설정 (START에서 바로 라우터 기동)
parent_builder.add_conditional_edges(
    START,
    router,
    {
        'vector_store': 'income_tax_agent', # 세법 서브그래프 실행
        'llm': 'basic_generate',           # 단순 LLM Q&A
        'web_search': 'web_search'          # 일반 인터넷 수색 트랙
    }
)

parent_builder.add_edge('web_search', 'web_generate')
parent_builder.add_edge('web_generate', END)
parent_builder.add_edge('basic_generate', END)
parent_builder.add_edge('income_tax_agent', END) # 세법 서브에이전트가 종료되면 전체 종료 처리

# 8. 상위 오케스트레이터 컴파일 및 실행
orchestrator_agent = parent_builder.compile()

 

 

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

엔터프라이즈 레벨의 다중 루프 서브그래프를 설계하고 클라우드 인프라 위에서 운용할 때, 장애 및 비용 예방을 위한 핵심 팁을 엄선하여 제안한다.

1) 서브그래프와 상위 그래프 간 상태 스키마 정렬 (State Mapping)

하위 그래프를 상위 그래프의 노드로 끼워 넣을 때, 가장 자주 터지는 에러는 '상태 키의 불일치'다.

• 설계 원칙: 상위 그래프가 공유하는 CoordinatorState 필드셋과 서브그래프가 다루는 AgentState 필드가 명밀히 호환되어야 한다. LangGraph는 하위 노드로 진입할 때 상태 필드명이 동일하다면 값을 자동으로 상속(Inherit)하여 분배하고, 작업 완료 시 병합(Merge)해 준다.
• 해결 전략: 만약 하위 노드와 상위 노드의 TypedDict 스펙을 다르게 설계해야 한다면, 아래 예시처럼 서브그래프를 호출하는 시점에 명시적으로 상태 매핑 래퍼(Wrapper) 함수를 선언해 상위 게이트웨이의 변수를 알맞게 주입 및 수집해야 한다.

# 상태 변수 스키마 매핑 래퍼 예시
def income_tax_subgraph_wrapper(state: CoordinatorState):
    # 상위 그래프의 query를 하부 서브그래프가 이해할 수 있는 입력 스펙으로 정규화
    subgraph_input = {"query": state["query"]}
    # 하위 서브그래프 구동 및 결과 확보
    subgraph_result = income_tax_subgraph.invoke(subgraph_input)
    # 수집한 결론을 상위 그래프의 스키마 구조에 맞춰 가공 및 전파
    return {
        "answer": subgraph_result["answer"],
        "context": subgraph_result.get("context", [])
    }

 

2) Tavily Search 호출 튜닝 및 가용성 확보

TavilySearch 엔진은 매우 강력하지만 매 호출마다 고정된 리서치 지연이 발생하며 API 호출 비용이 청구된다.

• 해결책: 실무에서는 TavilySearch 인스턴스화 시 search_depth 매개변수를 질문 유형에 따라 동적으로 가르거나, 기본 basic 모드로 유지하다가 타임아웃 발생 시 혹은 재탐색 시에만 advanced 모드로 승격하여 기동하는 것이 응답 속도 사수와 예산 통제라는 두 마리 토끼를 다 잡는 유일한 지름길이다.

 

 

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

로컬 벡터 DB의 탐색 한계를 웹 검색 엔진Fallback으로 보완하는 Corrective RAG(CRAG)와 거대 에이전트를 모듈 단위로 결리하고 협업하도록 돕는 서브그래프(Subgraph) 구조는 현대적인 지능형 비즈니스 플랫폼 개발의 표준(Standard)으로 자리 잡았다.

단방향 체인 설계에서는 상상할 수 없었던 고차원적인 피드백 제어가 가능해짐에 따라, 우리는 비즈니스 영역의 변경 사항이나 정책을 다른 에이전트의 오염 우려 없이 독립된 서브그래프 내부에만 반영할 수 있게 되었다.

오늘 정립한 세법 분기 에이전트 및 웹 검색 동적 폴백 설계를 바탕으로, 실무 환경에서 복잡한 사용자의 요구사항을 안전하고 예측 가능한 코드로 철저하게 제어해 보길 권장한다.

 

 

 

 

#본 이미지들은 AI를 이용하여 생성하였습니다.