MINERVA LangGraph 기초

StateGraph, Node, Edge, Conditional Edge — Chain에서 Graph로 가는 이유

Phase B에서 만든 LCEL 기반 Chain 파이프라인의 한계를 짚고, LangGraph StateGraph가 그 한계를 어떻게 푸는지 정리한다. Node·Edge·State·Conditional Edge의 의미를 MINERVA의 RAG 흐름에 매핑하고, 같은 RAG를 Chain과 StateGraph 두 방식으로 작성해 비교한다. Phase C-2 전환의 이론적 출발점이며, 14편(노드 분해)과 15편(State 설계)의 선행 학습이다.

Agent
저자

Kwangmin Kim

공개

2026년 05월 06일

1 왜 Chain에서 Graph로 가는가

Phase B에서 MINERVA의 RAG 파이프라인을 LCEL(prompt | llm | StrOutputParser)로 묶어 동작시켰다. 단일 호출에서 빠르고 코드도 짧다. 그런데 데이터 흐름 추적 편에서 본 것처럼 실제로는 “검색 → 리랭크 → Parent 매핑 → 컨텍스트 조립 → LLM → 응답 후처리”의 6~7 단계가 _prepare() 함수 안에 순차적으로 묶여 있다.

이 묶음은 다음 네 가지 상황에서 한계를 드러낸다.

  1. 분기 처리 — 검색 실패 시 Web 검색으로 폴백, 인용 없는 답변일 때 재시도, 사용자 의도에 따라 Sub-Agent로 위임 — 이런 분기가 if 사슬로 함수 안에 누적된다.
  2. 순환 처리 — Self-RAG처럼 “답변 평가 → 부족하면 재검색”의 루프, ReAct처럼 “도구 호출 → 관찰 → 다시 추론”의 반복이 LCEL 단일 그래프로는 표현되지 않는다.
  3. 중단·재개 — 긴 작업에서 사람의 승인을 받거나, 실패 지점부터 재실행하려면 중간 상태를 외부에 영속화해야 한다.
  4. 노드별 관찰성 — 어느 단계에서 시간을 잃었는지, 어느 노드가 실패했는지 추적하려면 단계가 함수가 아니라 일급(first-class) 단위여야 한다.

LangGraph는 위 네 가지를 표현할 수 있는 명시적 그래프 실행 엔진이다. Chain이 “함수 합성”이라면 LangGraph는 “상태 머신”이다.

선행 학습
더 깊은 LangGraph API 학습

이 포스트는 MINERVA 적용 관점이다. LangGraph 자체의 API를 처음 보는 독자는 다음 두 편을 함께 읽으면 좋다.

2 LangGraph 핵심 개념 4가지

정의: LangGraph

LangGraph는 LangChain 위에서 동작하는 그래프 기반 실행 프레임워크다. “노드(함수) + 엣지(전이) + 상태(State)” 세 요소로 에이전트의 실행 경로를 명시적으로 정의하며, 순환·분기·중단·재개를 일급(first-class) 기능으로 제공한다.

  • 핵심 단위: StateGraph — 노드와 엣지를 등록한 후 .compile() 하여 실행 가능한 그래프 생성
  • 차별점: LCEL Chain은 DAG(비순환). LangGraph는 순환 가능

2.1 Node — 상태를 받아 상태를 갱신하는 함수

Node는 state -> dict 시그니처의 순수 함수다. 입력으로 현재 State를 받고, 갱신할 필드만 dict로 반환한다.

def retrieve_node(state: AgentState) -> dict:
    docs = retriever.invoke(state["question"])
    return {"docs": docs}        # docs 필드만 갱신

반환 dict의 키는 State에 정의된 필드여야 한다. 반환되지 않은 필드는 그대로 유지된다.

2.2 Edge — 노드 간 전이

Edge는 다음에 어느 Node로 갈지 결정한다. 두 종류가 있다.

종류 정의 사용
정적 Edge 항상 같은 다음 노드로 전이 graph.add_edge("retrieve", "generate")
Conditional Edge 함수가 다음 노드 이름을 반환 graph.add_conditional_edges("grade", route_fn, {"good": "generate", "bad": "rewrite"})

Conditional Edge가 분기·순환의 핵심이다. 라우팅 함수는 State를 받아 string을 반환한다.

def route_after_grading(state: AgentState) -> str:
    if state["relevance_score"] >= 0.7:
        return "generate"          # 검색 결과 충분 → 답변 생성
    elif state["retry_count"] < 2:
        return "rewrite_query"     # 부족하면 쿼리 재작성
    else:
        return "fallback"          # 한계 도달 → 폴백

2.3 State — 노드 간 공유 자료 구조

State는 그래프 전체에서 노드 사이를 흘러다니는 “백팩”이다. TypedDict로 정의하며, 각 필드의 누적·덮어쓰기 전략을 Annotated 타입으로 지정할 수 있다.

from typing import Annotated, TypedDict
from operator import add
from langchain_core.documents import Document

class AgentState(TypedDict):
    question: str
    docs: list[Document]
    answer: str
    # 누적되는 필드 — 여러 노드가 이어붙임
    log_messages: Annotated[list[str], add]

Annotated[list[str], add]로 표시한 필드는 각 노드 반환값이 누적된다. 미표시 필드는 가장 최근 반환값으로 덮어쓴다. State 설계의 자세한 패턴은 15편 — State 설계에서 다룬다.

2.4 Conditional Edge — 분기와 순환의 단위

LangGraph의 차별성은 Conditional Edge가 자기 자신 또는 이전 노드로 돌아갈 수 있다는 점이다. 이로써 ReAct, Self-RAG, Plan-and-Execute 같은 반복 패턴이 자연스럽게 표현된다.

[generate] ──→ [grade_answer] ──good──→ END
                     │
                     └──bad──→ [retrieve]   ← 순환

LCEL Chain은 위와 같은 그림을 표현할 수 없다. 함수 합성은 한 방향뿐이다.

3 Chain vs StateGraph 직접 비교

같은 RAG를 두 방식으로 작성해 차이를 본다. MINERVA의 단순화한 흐름 — 질문 → 검색 → 답변 만 다룬다.

3.1 LCEL Chain 버전 (Phase B 현재)

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnableLambda

prompt = PromptTemplate.from_template(
    "다음 문서를 참고해 질문에 답하라.\n\n{context}\n\n질문: {question}"
)

def build_inputs(payload: dict) -> dict:
    docs = retriever.invoke(payload["question"])
    context = "\n\n".join(d.page_content for d in docs)
    return {"context": context, "question": payload["question"]}

chain = RunnableLambda(build_inputs) | prompt | llm | StrOutputParser()
answer = chain.invoke({"question": "데이터 표준화의 핵심 원칙은?"})

특징

  • 한 줄로 묶이고, 각 단계가 함수다.
  • 중간 결과(docs, context)는 함수 내부에서만 보이며, 그래프 차원에서 추적되지 않는다.
  • 분기·순환은 표현 불가. if로 분기하려면 별도의 RunnableBranch 또는 함수 안에서 처리.

3.2 LangGraph StateGraph 버전

from typing import Annotated, TypedDict
from operator import add
from langchain_core.documents import Document
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langgraph.graph import StateGraph, START, END


class AgentState(TypedDict):
    question: str
    docs: list[Document]
    context: str
    answer: str
    log_messages: Annotated[list[str], add]


def retrieve(state: AgentState) -> dict:
    docs = retriever.invoke(state["question"])
    return {"docs": docs, "log_messages": [f"retrieved {len(docs)} docs"]}


def build_context(state: AgentState) -> dict:
    context = "\n\n".join(d.page_content for d in state["docs"])
    return {"context": context, "log_messages": ["built context"]}


def generate(state: AgentState) -> dict:
    prompt = PromptTemplate.from_template(
        "다음 문서를 참고해 질문에 답하라.\n\n{context}\n\n질문: {question}"
    )
    chain = prompt | llm | StrOutputParser()
    answer = chain.invoke({
        "context": state["context"],
        "question": state["question"],
    })
    return {"answer": answer, "log_messages": ["generated answer"]}


graph = StateGraph(AgentState)
graph.add_node("retrieve", retrieve)
graph.add_node("build_context", build_context)
graph.add_node("generate", generate)

graph.add_edge(START, "retrieve")
graph.add_edge("retrieve", "build_context")
graph.add_edge("build_context", "generate")
graph.add_edge("generate", END)

app = graph.compile()
final_state = app.invoke({"question": "데이터 표준화의 핵심 원칙은?"})
print(final_state["answer"])
print(final_state["log_messages"])  # 각 노드의 로그가 누적되어 있음

무엇이 달라졌는가

측면 Chain StateGraph
단계 표현 함수 합성 (파이프 연산자) 명시적 add_node/add_edge
중간 상태 가시성 함수 내부에 매몰 State 필드로 외부 관찰 가능
노드별 단위 테스트 함수 단위로 가능하나 결합 강함 State 입출력만 mocking → 단순
분기 RunnableBranch 또는 함수 내 if add_conditional_edges 일급 지원
순환 불가 Conditional Edge로 자연 표현
중단·재개 별도 영속화 필요 Checkpointer 내장
코드량 짧음 길지만 명시적

코드는 길어진다. 그러나 분기·순환·관찰성·재개라는 네 가지 능력이 생긴다. 1.5명이 PoC를 만들 때는 Chain이 빠르지만, 50명이 같은 그래프를 협업으로 다듬으려면 명시성이 비용을 정당화한다.

4 Conditional Edge로 자기 평가 루프 만들기

위 그래프에 “답변이 인용을 충분히 포함하지 않으면 재검색”이라는 평가 루프를 더한다. Self-RAG의 단순화 버전이다.

루프를 위해 AgentState에 두 필드를 더했다고 가정한다 — citation_count: int(평가 결과 점수)와 retry_count: int(재시도 누적). 이렇게 노드가 늘어날 때마다 State가 함께 자라는 것이 일반적이며, 자세한 설계 패턴은 15편 — State 설계에서 다룬다.

def grade_answer(state: AgentState) -> dict:
    """답변에 [N] 마커가 2개 이상 있어야 통과."""
    citation_count = state["answer"].count("[")
    return {
        "citation_count": citation_count,
        "log_messages": [f"graded answer: {citation_count} citations"],
    }


def route_after_grading(state: AgentState) -> str:
    if state.get("citation_count", 0) >= 2:
        return "ok"
    if state.get("retry_count", 0) >= 1:
        return "ok"          # 한 번 재시도 후에는 그대로 종료
    return "retry"


def increment_retry(state: AgentState) -> dict:
    return {
        "retry_count": state.get("retry_count", 0) + 1,
        "log_messages": ["retry"],
    }


graph = StateGraph(AgentState)
graph.add_node("retrieve", retrieve)
graph.add_node("build_context", build_context)
graph.add_node("generate", generate)
graph.add_node("grade", grade_answer)
graph.add_node("retry", increment_retry)

graph.add_edge(START, "retrieve")
graph.add_edge("retrieve", "build_context")
graph.add_edge("build_context", "generate")
graph.add_edge("generate", "grade")
graph.add_conditional_edges(
    "grade",
    route_after_grading,
    {"ok": END, "retry": "retry"},
)
graph.add_edge("retry", "retrieve")        # 순환!

app = graph.compile()

그래프는 다음 모양이 된다.

START → retrieve → build_context → generate → grade ──ok──→ END
           ▲                                    │
           └──────────── retry ←────────────────┘ (citations < 2)

LCEL로는 이 그림을 직접 그릴 수 없다. Chain의 합성 연산자(|)는 단방향이라 grade -> retrieve 같은 역방향 엣지를 표현할 수단이 없다. 함수 안에 while 루프를 넣을 수는 있지만 그 순간 “그래프”가 아니라 “스크립트”가 된다 — 외부 도구가 그래프를 그릴 수도, 중간에 끊을 수도 없다.

5 MINERVA 현재 구조의 LangGraph 매핑

데이터 흐름 추적 편에서 본 _prepare() 흐름을 노드로 분해하면 다음과 같다.

[현재 — 단일 함수]
_prepare(query):
    _ensure_initialized()
    _retrieve_docs(query.text, document_id, source_filter)
    _build_chain_inputs(query, docs)          # context + chat_history + question
    chain = self._prompt | llm | StrOutputParser()
    return docs, chain, chain_inputs, model_used

run(query):
    docs, chain, chain_inputs, model_used = _prepare(query)
    answer = chain.invoke(chain_inputs)
    return _build_response(answer, docs, query)
[LangGraph 분해 — 5개 노드]
START
  ▼
[ensure_init_node]      lazy init (prompt·llm 로드)
  ▼
[retrieve_node]         _retrieve_docs → docs
  ▼
[build_inputs_node]     _build_chain_inputs → chain_inputs
  ▼
[generate_node]         chain.invoke(chain_inputs) → answer
  ▼
[build_response_node]   _build_response → response
  ▼
END

각 단계가 일급 노드가 되면 다음이 가능해진다.

  1. 단계별 timing 자동 수집 — [timing] prepare: 로그를 직접 찍지 않아도, LangGraph의 stream_events가 노드별 시작/종료 이벤트를 토해낸다.
  2. 검색 실패 시 폴백 노드 추가 — retrieve_node가 docs=[]를 반환하면 web_search_node로 우회.
  3. Reranker A/B를 노드 교체로 처리 — rerank_cosine vs rerank_flashrank를 두 노드로 두고, A/B 실험 arm에 따라 그래프를 두 가지로 컴파일.
  4. HITL — 위험한 답변 전 사람 승인 — generate_node 직후 interrupt를 걸어 검토 후 재개. 자세한 흐름은 16편 — Checkpointing과 HITL에서 다룬다.

이 분해의 실제 코드는 14편 — RAG Chain 분해에서 작성한다.

Data Standardizer는 이미 sub_agent supervisor

QnA Chatbot이 단일 LCEL 체인이라 위 분해가 깔끔하지만, Data Standardizer는 이미 agent.py가 RagRecommender·DomainAuditor·post_processing/{tables,code} 4개 sub_agent를 직접 호출하는 정적 파이프라인이다(02·08·10편 참조). LangGraph로 옮길 때 두 가지 선택이 있다:

  1. 각 sub_agent를 일급 노드로 평탄화 — recommend_node → post_process_node → audit_node로 펼친다. 단순하지만 sub_agent 내부 분기(코드 모드 vs 데이터 모드)가 다시 노드 폭증을 일으킨다.
  2. 각 sub_agent를 sub-graph로 캡슐화 — recommend_subgraph·audit_subgraph를 별도 StateGraph로 만들고 부모 그래프가 호출. 노드 수는 적게 유지되고, sub_agent 단위 재사용성이 보장된다.

상위 노드 수가 4개 이하로 유지되면 (1)이 가독성이 좋고, sub_agent가 5개 이상이거나 ReAct 루프가 들어가면 (2)가 합리적이다. MINERVA의 경우 Phase C-3에서 ToolNode를 도입하면 sub_agent 일부가 도구로 흡수되므로 (1)에 가까운 형태로 수렴할 가능성이 크다.

6 컴파일과 실행 — 그래프의 생애 주기

StateGraph는 정의 단계와 실행 단계가 분리된다.

# 정의 단계
graph = StateGraph(AgentState)
graph.add_node(...)
graph.add_edge(...)

# 컴파일 — 검증 + 최적화
app = graph.compile()        # checkpointer / interrupt_before 옵션 지정 가능

# 실행 — 세 가지 모드
result = app.invoke(initial_state)             # 동기 — 최종 State 반환
for chunk in app.stream(initial_state):        # 노드별 출력 스트리밍
    print(chunk)
async for event in app.astream_events(...):    # 토큰·이벤트 단위 비동기 스트리밍
    ...

compile() 시점에 다음이 검증된다.

  • 모든 노드가 등록되어 있는가
  • 도달 불가능한 노드가 있는가 (시작점에서 닿지 않는 고립 노드)
  • Conditional Edge의 라우팅 함수 반환값이 등록된 노드 이름인가

이 검증은 LCEL Chain에 없던 안전망이다. Chain은 런타임에 함수가 호출되어야 비로소 형 미스매치가 드러난다.

7 응용 분야 — MINERVA 외 사례

분야 LangGraph 적용 시나리오
QnA Chatbot Self-RAG 자기 평가 루프 + 인용 부족 시 재검색
Data Standardizer 후보 생성 → 사용자 검토(HITL) → 승인 후 영속화
Plan-and-Execute Agent 계획 노드 → 단계별 실행 노드 → 자기 수정 루프
Multi-Agent Supervisor Supervisor 노드가 라우팅 함수로 하위 에이전트 선택
코드 리뷰 에이전트 Static analyzer → LLM 리뷰 → 신뢰도 낮으면 재분석
데이터 파이프라인 추출 → 검증 → 실패 시 재시도 → 적재

위 사례들은 모두 “분기 또는 순환”을 본질적으로 갖는다. Chain으로는 표현할 수 없거나, 표현하려면 함수 안에 로직이 매몰된다.

8 자주 발생하는 오류 패턴

WRONG:

def retrieve_node(state: AgentState) -> dict:
    state["docs"] = retriever.invoke(state["question"])  # State 직접 변경
    return state

CORRECT:

def retrieve_node(state: AgentState) -> dict:
    docs = retriever.invoke(state["question"])
    return {"docs": docs}        # 갱신할 필드만 dict로 반환

LangGraph는 State 직접 변경이 아닌 “갱신 dict 반환” 모델이다. 직접 변경하면 누적 필드(Annotated)의 reducer가 작동하지 않는다.

WRONG:

graph.add_conditional_edges("grade", route_fn, ["good", "bad"])  # 리스트

CORRECT:

graph.add_conditional_edges(
    "grade",
    route_fn,
    {"good": "generate", "bad": "rewrite"},   # 라우팅 함수 반환값 → 노드 이름 dict
)

Conditional Edge는 dict로 매핑한다. 리스트는 라우팅 함수가 노드 이름을 직접 반환하는 경우에만 허용된다.

WRONG:

graph.add_edge("generate", "grade")
graph.add_edge("grade", "retrieve")     # END로 가는 경로 없음 → 무한 루프

CORRECT:

graph.add_conditional_edges(
    "grade",
    route_fn,
    {"ok": END, "retry": "retrieve"},   # 종료 조건 명시
)

순환 그래프는 반드시 END로 가는 분기를 포함해야 한다. 컴파일 시점에 도달 불가능 검사는 통과해도, 실행 중 루프 카운터(recursion_limit)에 걸려 예외가 발생한다.

9 다음 편 예고 — RAG Chain의 실제 분해

이번 편은 LangGraph가 무엇이고 왜 필요한지를 다뤘다. 다음 편(14 — RAG Chain 분해)에서는 MINERVA의 _prepare() 함수를 5개 노드로 분해해 실제 코드를 작성한다.

분해 후 다음을 확인한다.

  • 노드별 단위 테스트가 어떻게 단순해지는가 (mock 표면적 감소)
  • A/B 실험 arm을 노드 교체로 표현하는 방법
  • 검색 실패 시 폴백 노드를 추가하는 conditional edge

10 정리

항목 핵심
Chain의 한계 분기·순환·중단·노드별 관찰성 표현 불가
StateGraph 4 요소 Node(함수), Edge(전이), State(공유 자료), Conditional Edge(분기/순환)
같은 RAG 비교 Chain은 짧고, Graph는 길지만 명시적 — 50명 협업 단계에서 비용 정당화
MINERVA 매핑 _prepare의 6단계 → 5개 일급 노드로 분해 가능
다음 단계 14편에서 실제 분해 코드 작성, 15편에서 State 설계 패턴

Chain은 “기능이 동작하면 끝”인 단계에 적합하다. LangGraph는 “기능에 분기·순환·관찰성·거버넌스를 더해야 하는 단계”에서 비용을 회수한다. MINERVA가 1.5명 PoC에서 50명 플랫폼으로 진화하는 길목에서 LangGraph 전환은 선택이 아니라 전제다.

11 관련 주제

선행 학습

후속 주제

다른 카테고리 연결

Subscribe

Enjoy this blog? Get notified of new posts by email: