LangGraph — 에이전트를 위한 상태 머신(State Machines for Agents)

손으로 작성한 ReAct 루프(ReAct loop)는 결국 while True입니다. 반면 LangGraph로 작성한 ReAct 루프는 체크포인트(checkpoint)를 남기고, 중단(interrupt)하고, 분기(branch)하고, 시간 여행(time-travel)까지 할 수 있는 그래프(graph)입니다. 에이전트(agent) 자체가 바뀐 것이 아니라, 에이전트를 둘러싼 하니스(harness)가 바뀐 것입니다.

유형: Build 언어: Python 선수 지식: Phase 11 · 09(Function Calling), Phase 11 · 14(Model Context Protocol) 예상 시간: 약 75분

문제

함수 호출(function calling) 기반 에이전트를 배포했다고 가정해 봅시다. 처음 세 턴(turn)은 잘 동작합니다. 그러다 무언가 어긋나기 시작합니다. 모델이 호출한 도구(tool)가 500 에러를 반환하거나, 사용자가 작업 도중 마음을 바꾸거나, 에이전트가 사람의 승인(human sign-off) 없이 주문 환불(order refund)을 결정해 버립니다. while True: 루프에는 끼어들 훅(hook)이 없습니다. 잠깐 멈출 수도, 되감을 수도, "모델이 다른 도구를 골랐다면 어땠을까"라는 분기를 만들어 볼 수도 없습니다. 데모를 넘어 실서비스(production)로 나가는 순간, 에이전트는 성공했거나 실패했다는 결과만 남기는 블랙박스(black box)가 됩니다.

한 번 보고 나면 다음 단계는 분명합니다. 에이전트는 이미 상태 머신(state machine)입니다. 시스템 프롬프트(system prompt), 메시지 이력(message history), 대기 중인 도구 호출(pending tool calls), 다음 행동(next action)이 모두 상태(state)에 해당합니다. 이 상태 머신을 명시적으로 드러내면 됩니다. "모델이 사고한다", "도구를 실행한다", "사람이 승인한다" 같은 단계를 노드(node)로 두고, 그 사이의 조건부 전이(conditional transition)를 엣지(edge)로 둡니다. 그래프가 명시적이면 하니스(harness)는 네 가지 기능을 거의 공짜로 얻습니다. 단계 사이의 상태를 저장하는 체크포인팅(checkpointing), 사람을 기다리며 멈추는 중단(interrupts), 토큰과 중간 이벤트(intermediate event)를 흘려보내는 스트리밍(streaming), 이전 상태로 되감아 다른 분기를 시도하는 시간 여행(time-travel)이 그것입니다.

LangGraph는 바로 이 추상화(abstraction)를 제공하는 라이브러리(library)입니다. LangChain 식의 "AgentExecutor를 줄 테니 알아서 해보라"는 에이전트 프레임워크(agent framework)가 아닙니다. LangGraph는 일급(first-class) 상태, 일급 영속성(persistence), 일급 중단을 갖춘 그래프 런타임(graph runtime)입니다. 에이전트 루프(agent loop)는 손으로 작성하는 것이 아니라 그림으로 그리는 대상이 됩니다.

개념

StateGraph에는 세 가지 구성 요소가 있습니다.

1. 상태(State). 그래프를 따라 흐르는 타입 지정 딕셔너리(TypedDict 또는 Pydantic model)입니다. 모든 노드는 전체 상태(full state)를 입력으로 받고 부분 갱신(partial update)을 반환합니다. LangGraph는 필드(field)마다 정해진 리듀서(reducer)를 사용해 이 갱신을 병합합니다. 누적되어야 하는 리스트에는 operator.add를 쓰고, 기본 동작은 덮어쓰기(overwrite)입니다.
2. 노드(Nodes). state -> partial_state 형태의 Python 함수입니다. "모델을 호출한다", "도구를 실행한다", "요약한다"처럼 하나의 독립된 단계(discrete step)를 나타냅니다.
3. 엣지(Edges). 노드 사이의 전이입니다. 정적 엣지(static edge)는 한 곳으로만 향합니다. 조건부 엣지(conditional edge)는 라우터 함수(router function) state -> next_node_name을 받아, 모델의 출력에 따라 그래프가 분기할 수 있게 해 줍니다.

그래프는 컴파일(compile)을 통해 사용할 수 있는 형태가 됩니다. 컴파일 단계에서는 그래프의 위상(topology)을 결합하고, 체크포인터(checkpointer)를 붙인 뒤, 실행 가능한 객체(runnable)를 반환합니다. 실서비스에서는 체크포인터가 선택 사항처럼 보여도 사실상 필수입니다. 컴파일된 그래프는 초기 상태와 thread_id로 호출(invoke)합니다. 실행의 매 단계는 (thread_id, checkpoint_id)를 키로 삼아 체크포인트를 저장합니다.

네 가지 슈퍼파워(superpowers)

체크포인팅(Checkpointing). 모든 노드 전이는 새로운 상태를 저장소(store)에 기록합니다. 테스트에서는 인메모리(in-memory) 저장소를, 실서비스에서는 Postgres, Redis, SQLite를 쓸 수 있습니다. 같은 thread_id로 그래프를 다시 호출하면, 그래프는 멈췄던 지점에서 이어서 실행됩니다.

중단(Interrupts). 노드에 interrupt_before=["human_review"]를 표시하면, 해당 노드가 실행되기 직전에 실행이 멈춥니다. 이때 상태는 유지된 채로 남아 있습니다. 여러분의 API는 사용자에게 "승인을 기다리는 중(awaiting approval)"이라는 응답을 돌려줄 수 있습니다. 나중에 같은 thread_id로 Command(resume=...)을 보내면 실행이 재개됩니다.

스트리밍(Streaming). graph.stream(state, mode="updates")는 상태 변화(delta)가 발생할 때마다 그것을 흘려보냅니다. mode="messages"는 모델 노드 내부에서 생성되는 LLM 토큰(token)을 스트리밍합니다. mode="values"는 전체 스냅샷(snapshot)을 흘려보냅니다. UI에 어떤 정보를 표면화할지는 선택할 수 있습니다.

시간 여행(Time-travel). graph.get_state_history(thread_id)는 전체 체크포인트 로그를 반환합니다. 임의의 이전 checkpoint_id를 graph.invoke에 넘기면 그 지점에서 분기됩니다. "모델이 다른 도구를 골랐다면 어떻게 됐을까" 같은 디버깅이나, 실서비스 트레이스(trace)를 재현하는 회귀 테스트(regression test)에 유용합니다.

리듀서(Reducer)가 핵심입니다

상태의 모든 필드에는 리듀서가 붙어 있습니다. 대부분은 기본 동작으로도 충분합니다. 새 값이 이전 값을 덮어쓰는 식입니다. 하지만 메시지 리스트(message list)에는 operator.add가 필요합니다. 그래야 새 메시지가 기존 이력을 대체하지 않고 뒤에 덧붙습니다. 병렬 엣지(parallel edge)는 리듀서를 통해 갱신을 합칩니다. 만약 두 노드가 모두 messages를 갱신했는데 Annotated[list, add_messages]를 빠뜨렸다면, 두 번째 갱신이 조용히 첫 갱신을 덮어쓰고 한 턴의 절반이 사라져 버립니다. 리듀서는 이 라이브러리에서 유일하게 미묘한(subtle) 부분입니다. 여기만 정확히 잡으면 나머지는 자연스럽게 조합됩니다.

네 노드로 구성한 ReAct 그래프

실서비스 수준의 ReAct 에이전트는 네 개의 노드와 두 개의 엣지로 표현됩니다.

1. agent — 현재 메시지 이력으로 LLM을 호출합니다. 어시스턴트 메시지(assistant message)를 반환하고, 이 메시지에는 tool_calls가 포함될 수 있습니다.
2. tools — 마지막 어시스턴트 메시지의 tool_calls를 실행하고, 도구 결과를 도구 메시지(tool message)로 이어 붙입니다.
3. agent에서 출발하는 조건부 엣지. 마지막 메시지에 tool_calls가 있으면 tools로, 없으면 END로 라우팅합니다.
4. tools에서 agent로 돌아가는 정적 엣지.

이게 전부입니다. Thought → Action → Observation → Thought → … 으로 이어지는 전체 ReAct 루프를 체크포인팅, 중단, 스트리밍과 함께 약 40줄의 코드로 얻습니다.

StateGraph와 Send(팬아웃; fanout)

Send(node_name, state)는 하나의 노드가 여러 개의 병렬 서브그래프(parallel subgraph)를 디스패치(dispatch)할 수 있게 해 줍니다. 예를 들어 에이전트가 세 개의 리트리버(retriever)를 동시에 질의(query)하도록 결정할 수 있습니다. 각 Send는 대상 노드의 병렬 실행을 하나씩 띄우고, 각 결과는 상태 리듀서를 통해 합쳐집니다. LangGraph가 스레딩(threading) 기본기 없이도 오케스트레이터-워커(orchestrator-workers) 패턴을 표현하는 방식이 바로 이것입니다.

서브그래프(Subgraphs)

컴파일된 그래프는 다른 그래프의 노드가 될 수 있습니다. 바깥 그래프(outer graph)에서는 단일 노드처럼 보이지만, 안쪽 그래프(inner graph)는 자신만의 상태와 체크포인트를 가집니다. 슈퍼바이저-워커(supervisor-worker) 형태의 에이전트를 구성할 때 이 방식을 씁니다. 슈퍼바이저 그래프가 사용자 의도(user intent)를 도메인별 워커 서브그래프(per-domain worker subgraph)로 라우팅하는 식입니다.

직접 만들기

Step 1: 상태와 노드 정의하기

from typing import Annotated, TypedDict
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

def agent_node(state: State) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def should_continue(state: State) -> str:
    last = state["messages"][-1]
    return "tools" if getattr(last, "tool_calls", None) else END

tool_node = ToolNode(tools=[search_web, read_file])

graph = StateGraph(State)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)
graph.set_entry_point("agent")
graph.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
graph.add_edge("tools", "agent")

app = graph.compile(checkpointer=MemorySaver())

add_messages는 메시지 리스트가 덮어쓰기되지 않고 누적되게 만드는 리듀서입니다. 이를 빠뜨리는 것이 LangGraph에서 가장 흔히 발생하는 버그입니다.

Step 2: 스레드(thread)로 실행하기

config = {"configurable": {"thread_id": "user-42"}}
for event in app.stream(
    {"messages": [HumanMessage("find the Anthropic headquarters address")]},
    config,
    stream_mode="updates",
):
    print(event)

모든 업데이트는 {node_name: state_delta} 형태의 딕셔너리입니다. 프런트엔드(frontend)는 이를 UI로 흘려보내, 사용자에게 "에이전트가 생각하는 중… search_web 호출 중… 결과 수신… 답변 작성 중" 같은 진행 상태를 보여줄 수 있습니다.

Step 3: 사람이 개입하는(human-in-the-loop) 중단 추가하기

특정 노드를 표시해 두면, 실행이 그 노드 앞에서 잠시 멈추게 만들 수 있습니다.

app = graph.compile(
    checkpointer=MemorySaver(),
    interrupt_before=["tools"],  # 모든 tool 호출 직전에 멈춥니다
)

state = app.invoke({"messages": [HumanMessage("delete the production database")]}, config)
# state["__interrupt__"] 가 설정됩니다. 제안된 tool_calls를 확인합니다.
# 승인한 경우:
from langgraph.types import Command
app.invoke(Command(resume=True), config)
# 거절한 경우: 거절 메시지를 기록하고 재개합니다.
app.update_state(config, {"messages": [AIMessage("Blocked by human reviewer.")]})

상태, 체크포인트, 스레드는 중단 지점을 넘어서도 모두 보존됩니다. 실행 중이 아닐 때 메모리에만 위태롭게 떠 있는 상태가 아닙니다.

Step 4: 디버깅을 위한 시간 여행

history = list(app.get_state_history(config))
for snapshot in history:
    print(snapshot.values["messages"][-1].content[:80], snapshot.config)

# 이전 checkpoint에서 분기합니다
target = history[3].config  # 세 단계 이전
for event in app.stream(None, target, stream_mode="values"):
    pass  # 그 지점부터 앞으로 재생합니다

입력으로 None을 넘기면 지정한 체크포인트에서 그대로 재생(replay)합니다. 값을 넘기면 재개하기 전에 그 값을 해당 체크포인트 상태에 갱신으로 덧붙입니다. 전체 대화를 처음부터 다시 돌리지 않고도 잘못된 에이전트 실행을 그대로 재현하는 방법입니다.

Step 5: 실서비스용 체크포인터로 교체하기

from langgraph.checkpoint.postgres import PostgresSaver

with PostgresSaver.from_conn_string("postgresql://...") as checkpointer:
    checkpointer.setup()
    app = graph.compile(checkpointer=checkpointer)

SQLite, Redis, Postgres 기반의 세이버(saver)가 모두 제공됩니다. MemorySaver는 테스트용입니다. 재시작을 넘어 영속성이 필요한 경우라면 실제 저장소를 사용해야 합니다.

스킬(The Skill)

에이전트는 while True 루프가 아니라 그래프로 만듭니다.

LangGraph를 곧바로 집어 들기 전에, 60초 정도 다음과 같이 설계해 봅니다.

1. 노드 이름 붙이기. 모든 독립적인 결정이나 부작용(side effect)을 일으키는 행동은 노드입니다. "에이전트가 생각한다", "도구가 실행된다", "리뷰어가 승인한다", "응답이 스트리밍된다"처럼 적어 봅니다. 이걸 나열하지 못하겠다면, 아직 에이전트로 풀 만한 작업이 아닙니다.
2. 상태 선언하기. 모든 리스트 필드에 리듀서를 지정한 최소한의 TypedDict를 작성합니다. 모든 정보를 messages에 욱여넣지 말고, 작업별 필드(working plan, budget 카운터, retrieved_docs 리스트 등)는 최상위로 끌어올립니다.
3. 엣지 그리기. 다음 단계가 모델 출력에 의존하지 않는다면 정적 엣지로 둡니다. 조건부 엣지마다 명명된 분기(named branch)를 반환하는 라우터 함수를 둡니다.
4. 체크포인터를 먼저 고르기. 테스트에는 MemorySaver, 그 외에는 Postgres, Redis, SQLite를 씁니다. 체크포인터 없이 배포하지 않습니다. 체크포인터가 없으면 재개도, 중단도, 시간 여행도 할 수 없습니다.
5. 부작용 이전에 중단을 둘지 결정하기. 승인은 부작용을 일으키는 노드로 들어가는 엣지에 둡니다. 그래야 피해가 발생하기 전에 취소할 수 있습니다. 검증은 모델에서 나가는 엣지에 둡니다. 그래야 잘못된 호출을 값싸게 거절할 수 있습니다.
6. 기본은 스트리밍. UI에는 mode="updates", 모델 노드 내부의 토큰 단위 스트리밍에는 mode="messages", 평가(eval) 도중의 전체 스냅샷에는 mode="values"를 씁니다.

체크포인터가 없는 LangGraph 에이전트는 배포하지 않습니다. 부작용이 일어난 뒤에 중단하는 에이전트도 배포하지 않습니다. messages 필드에 add_messages 리듀서가 없는 에이전트도 배포하지 않습니다.

연습문제

1. 쉬움. 위에서 본 네 노드 구조의 ReAct 그래프를 계산기(calculator) 도구와 웹 검색(web-search) 도구로 구현합니다. 두 턴짜리 대화에서 list(app.get_state_history(config))가 최소 네 개의 체크포인트를 반환하는지 확인합니다.
2. 중간. agent 앞에서 먼저 실행되는 planner 노드를 추가하고, 구조화된 plan: list[str]을 상태에 기록합니다. agent가 각 plan 단계를 완료(done)로 표시하도록 만듭니다. 체크포인트에서 재개할 때 plan이 사라지면(잘못된 리듀서를 사용했다면) 테스트가 실패하도록 합니다.
3. 어려움. Send를 사용해 세 개의 서브그래프(researcher, writer, reviewer) 사이를 라우팅하는 슈퍼바이저 그래프를 만듭니다. 각 서브그래프는 자신만의 상태와 체크포인터를 가집니다. 바깥 그래프에 interrupt_before=["writer"]를 추가해, 사람이 리서치 브리프(research brief)를 승인하도록 합니다. 이전 체크포인트에서 시간 여행을 했을 때, 분기된 가지(branch)만 다시 실행되는지 확인합니다.

핵심 용어

더 읽을거리

• LangGraph documentation — StateGraph, 리듀서, 체크포인터, 중단을 다루는 표준 레퍼런스입니다.
• LangGraph concepts: state, reducers, checkpointers — 이 lesson이 사용하는 멘탈 모델(mental model)을 공식 문서에서 그대로 확인할 수 있습니다.
• LangGraph Persistence and Checkpoints — Postgres/SQLite/Redis 저장소, 체크포인트 네임스페이스(namespace), 스레드 ID를 자세히 설명합니다.
• LangGraph Human-in-the-loop — interrupt_before, interrupt_after, Command(resume=...), 그리고 상태 편집 패턴(edit-state pattern)을 다룹니다.
• Yao et al., "ReAct: Synergizing Reasoning and Acting in Language Models" (ICLR 2023) — 모든 LangGraph 에이전트가 구현하는 패턴입니다. 추론 트레이스(reasoning trace)의 근거를 함께 이해할 수 있습니다.
• Anthropic — Building effective agents (Dec 2024) — 체인(chain), 라우터(router), 오케스트레이터-워커(orchestrator-workers), 평가자-최적화자(evaluator-optimizer) 같은 그래프 모양을 언제 골라야 하는지 설명합니다.
• Phase 11 · 09(Function Calling) — 모든 LangGraph 에이전트 노드가 재사용하는 도구 호출(tool-call) 기본기입니다.
• Phase 11 · 14(Model Context Protocol) — MCP 어댑터(adapter)를 통해 LangGraph ToolNode로 연결되는 외부 도구 탐색(external tool discovery)을 다룹니다.
• Phase 11 · 17(Agent framework tradeoffs) — LangGraph를 CrewAI, AutoGen, Agno 대신 언제 선택해야 하는지 비교합니다.