LangGraph (2) - State, Node, Edge

LangGraph State, Node, Edge와 흐름 제어

LangGraph의 Graph API에서는 하나의 실행이 State를 공유하면서 여러 Node를 지나고, 실행 결과에 따라 다음 Node를 선택하는 과정으로 표현됩니다. 코드를 이해하려면 여기서 데이터가 어떻게 바뀌는지와 실행 위치가 어떻게 바뀌는지를 따로 볼 필요가 있습니다.

State는 단순한 전역 딕셔너리가 아닙니다. Node는 전체 State를 새로 만드는 대신 자신이 변경할 값만 Partial<State> 형태로 반환하고, 각 State key에 정의된 규칙에 따라 기존 값과 합쳐집니다. 별도의 reducer가 없는 key는 새로운 값으로 갱신되고, Annotated로 reducer를 지정한 key는 리스트를 누적하는 것처럼 이전 값과 새로운 update를 병합할 수 있습니다. 여러 Node가 병렬로 같은 key를 갱신할 때도 이 reducer가 어떤 결과를 남길지 결정합니다.

Node가 State를 어떻게 바꿀지 결정한다면, Edge는 다음에 무엇을 실행할지 결정합니다. 고정된 Edge는 정해진 순서로 Node를 연결하고, conditional edge는 현재 State를 보고 다음 경로를 선택합니다.

조금 더 동적인 흐름에서는 Command와 Send도 사용할 수 있습니다. Command는 하나의 Node에서 State를 갱신하면서 다음에 이동할 Node까지 함께 결정할 때 사용하고, Send는 특정 Node에 별도의 입력을 보내 여러 작업을 동적으로 만들어 내는 데 사용합니다. 같은 Node를 여러 입력으로 실행한 뒤 결과를 reducer로 모으는 map-reduce 형태가 대표적입니다.

State schema와 reducer가 데이터의 변화를 어떻게 정의하는지부터 살펴보고, Node의 partial update가 실제 State에 반영되는 과정을 따라갑니다. 이어서 고정 Edge와 conditional edge, Command, Send가 각각 어떤 종류의 제어 흐름을 표현하는지 연결해서 봅니다. 이렇게 보면 StateGraph 코드를 읽을 때 State가 어떻게 변하는가와 다음 실행 위치가 어떻게 결정되는가를 분리해서 이해할 수 있습니다.

---

1장. State는 그래프가 공유하는 데이터다

먼저 그래프 전체가 알아야 할 데이터를 하나의 스키마로 정의합니다.

from typing_extensions import TypedDict

class State(TypedDict):
    question: str
    context: str
    answer: str

이 State는 함수 하나의 지역 변수와 다릅니다. 여러 Node가 같은 실행 동안 공유하는 데이터 구조입니다.

retrieve
    question 읽음
    context 갱신

        ↓ 같은 State

generate
    question, context 읽음
    answer 갱신

State에 무엇을 넣을지는 워크플로에 따라 달라집니다.

예	용도
messages	대화와 도구 호출 이력
context	검색한 문서
draft	작성 중인 결과
score	평가 결과
iteration	반복 횟수
approved	사람 승인 여부

중요한 것은 모든 값을 무조건 State에 넣는 것이 아닙니다. 여러 Node가 함께 알아야 하거나 다음 경로를 결정하는 데 필요한 값을 State로 올리는 편이 좋습니다.

---

2장. Node는 State의 일부를 갱신한다

Node는 일반 파이썬 함수입니다. State를 입력으로 받고 필요한 작업을 한 뒤 변경할 값을 반환합니다.

def retrieve(state: State):
    docs = retriever.invoke(state["question"])
    context = "\n\n".join(doc.page_content for doc in docs)
    return {"context": context}

공식 StateGraph 설명은 Node의 형태를 State -> Partial<State>로 표현합니다. 여기서 Partial<State>가 중요합니다.

Node는 전체 State를 다시 만들 필요가 없습니다.

# 필요한 부분만 반환
def retrieve(state: State):
    ...
    return {"context": context}

이 방식은 어느 Node가 어떤 값을 바꾸는지 코드에 드러나게 합니다. 다른 Node가 관리하는 값을 실수로 덮어쓸 가능성도 줄어듭니다.

Node 안에서 모델이나 LCEL 체인을 그대로 사용할 수도 있습니다.

def summarize(state: State):
    summary = summary_chain.invoke({"text": state["context"]})
    return {"answer": summary}

즉 LangGraph가 모델 호출 방식을 새로 만드는 것은 아닙니다. 기존 실행 단위를 Node 안에 넣고 그 사이의 상태 흐름을 관리합니다.

---

3장. 기본 동작은 덮어쓰기다

State key에 별도 병합 규칙이 없으면 새 값이 이전 값을 대체합니다.

class State(TypedDict):
    answer: str

어떤 Node가 다음처럼 반환하면:

return {"answer": "새 답변"}

answer는 새 값으로 바뀝니다. 현재 단계나 최종 답변처럼 마지막 값 하나만 필요할 때는 이 동작이 자연스럽습니다.

문제는 값이 누적되어야 할 때입니다.

def node_a(state):
    return {"logs": ["A 완료"]}


def node_b(state):
    return {"logs": ["B 완료"]}

logs를 단순한 list[str]로 정의하면 각 업데이트를 어떻게 합칠지 별도 규칙이 없습니다. 특히 병렬 Node가 같은 key를 갱신하는 경우에는 병합 방법을 명시해야 합니다.

이때 reducer를 사용합니다.

---

4장. reducer는 값을 합치는 규칙이다

reducer는 기존 값과 새 값을 받아 다음 State 값을 만드는 함수입니다.

기존 값 + 새 값
      ↓ reducer
    다음 값

리스트를 이어 붙이고 싶다면 다음처럼 정의할 수 있습니다.

import operator
from typing import Annotated
from typing_extensions import TypedDict

class State(TypedDict):
    logs: Annotated[list[str], operator.add]

이제 여러 업데이트가 들어오면 operator.add로 합칩니다.

["A 완료"] + ["B 완료"]
          ↓
["A 완료", "B 완료"]

reducer의 형태는 개념적으로 다음과 같습니다.

def reducer(current, new):
    return merged

어떤 key에 reducer가 필요한지는 값의 의미를 보면 됩니다.

값의 성격	일반적인 처리
최종 답변, 현재 단계	새 값으로 교체
로그, 이벤트 목록	누적
숫자 합계	합산
메시지 이력	메시지 전용 병합 규칙

reducer는 단순한 리스트 append 기능이 아닙니다. 한 State key에 여러 업데이트가 들어왔을 때 그 의미를 어떻게 보존할지 정의하는 규칙입니다.

---

5장. 메시지는 add_messages로 다룬다

대화 이력은 단순히 리스트 두 개를 이어 붙이는 것보다 조금 더 복잡합니다. 기존 메시지와 같은 ID의 새 메시지가 들어오면 해당 메시지를 갱신해야 할 수 있습니다.

LangGraph는 메시지를 위한 add_messages reducer를 제공합니다.

from typing import Annotated
from langchain_core.messages import AnyMessage
from langgraph.graph import add_messages
from typing_extensions import TypedDict

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

add_messages는 기본적으로 새 메시지를 추가하지만, 같은 ID의 메시지가 들어오면 기존 메시지를 새 내용으로 갱신합니다.

이 패턴이 매우 흔해서 MessagesState도 제공됩니다.

from langgraph.graph import MessagesState

class AgentState(MessagesState):
    iteration: int
    approved: bool

이전에는 메시지만 상태로 쓰는 MessageGraph도 많이 사용됐지만 현재는 deprecated입니다. 메시지 외의 값을 함께 다루기 쉬운 StateGraph와 MessagesState 조합이 더 일반적입니다.

---

6장. Edge는 다음 실행 위치를 정한다

Node를 만들었으면 어떤 순서로 실행할지 연결합니다.

고정 순서는 일반 Edge로 표현합니다.

from langgraph.graph import START, END

builder.add_edge(START, "retrieve")
builder.add_edge("retrieve", "generate")
builder.add_edge("generate", END)

조건에 따라 다음 Node가 달라지면 조건부 Edge를 사용합니다.

def route(state: State):
    if state["score"] >= 0.8:
        return "finish"
    return "retry"

builder.add_conditional_edges(
    "evaluate",
    route,
    {
        "retry": "generate",
        "finish": END,
    },
)

조건부 Edge의 판단 함수도 State를 읽습니다. 그래서 score, iteration, approved처럼 다음 경로를 정하는 값은 State에 있어야 합니다.

순환도 별도 특별한 문법이 있는 것이 아닙니다. 조건부 Edge가 앞의 Node를 다시 가리키면 됩니다.

generate → evaluate
   ▲          │
   └── retry ─┘

---

7장. 병렬 실행에서는 reducer가 더 중요하다

LangGraph는 한 super-step에서 여러 Node를 병렬로 실행할 수 있습니다. 두 Node가 서로 다른 key를 갱신한다면 특별한 문제가 없습니다.

search_web → web_docs
search_db  → db_docs

하지만 두 Node가 같은 key를 동시에 갱신한다면 그 값을 어떻게 합칠지 정의해야 합니다.

node A → logs = ["A"]
node B → logs = ["B"]
             ↓
        reducer 필요

그래서 reducer는 순차 흐름보다 병렬 그래프에서 더 분명하게 중요해집니다. 병렬 실행을 설계할 때는 같은 State key를 여러 Node가 쓸 수 있는지 먼저 확인하는 습관이 좋습니다.

---

8장. Command와 Send는 동적인 흐름을 표현한다

조건부 Edge는 다음 위치만 결정할 때 가장 단순합니다. 그런데 한 Node가 State를 갱신하면서 동시에 다음 Node까지 결정해야 하는 경우에는 Command가 더 자연스럽습니다.

from typing import Literal
from langgraph.types import Command


def review(state: State) -> Command[Literal["approve", "revise"]]:
    if state["score"] >= 0.8:
        return Command(
            update={"approved": True},
            goto="approve",
        )

    return Command(
        update={"approved": False},
        goto="revise",
    )

Command(update=..., goto=...)는 State 업데이트와 라우팅을 한 함수에서 함께 표현합니다. 반대로 State를 바꿀 필요 없이 다음 위치만 고른다면 조건부 Edge가 더 읽기 쉽습니다. 같은 Node에서 정적 add_edge()와 Command(goto=...)를 동시에 사용하면 두 경로가 모두 실행될 수 있으므로 한 가지 라우팅 방식으로 정리하는 편이 좋습니다.

Send는 실행 시점에 몇 개의 병렬 작업이 생길지 미리 알 수 없을 때 사용합니다. 대표적인 예가 map-reduce입니다.

from langgraph.types import Send


def fan_out(state: State):
    return [
        Send("summarize_one", {"doc": doc})
        for doc in state["documents"]
    ]

문서가 세 개면 같은 Node가 세 번, 열 개면 열 번 실행됩니다. 각 결과가 다시 같은 State key로 합쳐진다면 앞에서 정의한 reducer가 필요합니다.

그래서 흐름 제어를 다음처럼 구분하면 이해하기 쉽습니다.

add_edge              고정된 다음 단계
conditional edge      State를 보고 다음 단계 선택
Command               State 업데이트 + 다음 단계 선택
Send                  실행 중 동적으로 병렬 작업 생성

---

9장. compile하면 실행 가능한 그래프가 된다

StateGraph는 빌더이므로 Node와 Edge를 등록한 뒤 compile해야 합니다.

from langgraph.graph import StateGraph

builder = StateGraph(State)
builder.add_node("retrieve", retrieve)
builder.add_node("generate", generate)
builder.add_edge(START, "retrieve")
builder.add_edge("retrieve", "generate")
builder.add_edge("generate", END)

graph = builder.compile()

컴파일된 그래프는 실행할 수 있습니다.

result = graph.invoke({
    "question": "LangGraph가 무엇인가요?",
    "context": "",
    "answer": "",
})

스트리밍과 비동기 실행도 같은 실행체에서 제공합니다.

graph.stream(inputs)
await graph.ainvoke(inputs)

이 지점에서 LangChain의 Runnable과 LangGraph가 연결됩니다. Node 안에서 Runnable을 사용하고, 컴파일된 그래프 자체도 invoke와 stream 같은 공통 실행 방식을 제공합니다.

---

10장. 한 번에 연결해 보기

생성하고 평가한 뒤 점수가 낮으면 다시 생성하는 간단한 그래프를 만들 수 있습니다.

from typing import Annotated
import operator
from typing_extensions import TypedDict
from langgraph.graph import START, END, StateGraph

class State(TypedDict):
    topic: str
    draft: str
    score: float
    iteration: int
    logs: Annotated[list[str], operator.add]


def generate(state: State):
    draft = writer.invoke(state["topic"])
    return {
        "draft": draft,
        "iteration": state["iteration"] + 1,
        "logs": ["초안 생성"],
    }


def evaluate(state: State):
    score = judge.invoke(state["draft"])
    return {
        "score": score,
        "logs": [f"평가 {score}"],
    }


def route(state: State):
    if state["score"] >= 0.8 or state["iteration"] >= 3:
        return "finish"
    return "retry"

builder = StateGraph(State)
builder.add_node("generate", generate)
builder.add_node("evaluate", evaluate)
builder.add_edge(START, "generate")
builder.add_edge("generate", "evaluate")
builder.add_conditional_edges(
    "evaluate",
    route,
    {"retry": "generate", "finish": END},
)

graph = builder.compile()

이 예제에는 State, Node, reducer, 고정 Edge, 조건부 Edge, 순환, 종료 조건이 모두 들어 있습니다.

여기서 가장 중요한 설계 습관은 Node를 크게 만드는 것이 아닙니다. State에서 무엇을 공유할지, 각 Node가 무엇만 갱신할지, 같은 key의 여러 업데이트를 어떻게 합칠지를 먼저 정하는 것입니다.

---

11장. 정리

State, Node, Edge는 따로 외울 개념이 아닙니다. 하나의 흐름으로 연결해서 이해하는 편이 좋습니다.

그래프가 실행되면 Node가 현재 State를 읽습니다. Node는 작업을 수행한 뒤 바꿀 값만 반환합니다. LangGraph는 각 key의 reducer 규칙에 따라 업데이트를 State에 반영하고, Edge가 다음 Node를 정합니다.

State
  ↓
Node 실행
  ↓
Partial<State> 반환
  ↓
reducer로 병합
  ↓
Edge가 다음 Node 결정
  ↓
다음 State

다음 편에서는 이 State와 실행 위치를 호출 사이에서도 이어 가는 persistence를 다룹니다. checkpointer가 thread 상태를 어떻게 저장하는지, interrupt()와 Command(resume=...)가 어떻게 재개를 만드는지, 그리고 thread 밖의 장기 기억을 왜 Store로 분리하는지 살펴봅니다.

---

참고자료

• LangGraph 공식 문서, Graph API
• LangGraph 공식 문서, Graph API 사용법 (Command, Send)
• LangGraph Reference, Command
• LangGraph Reference, Send
• LangGraph Reference, StateGraph
• LangGraph Reference, add_messages
• LangGraph Reference, graph module
• LangGraph PyPI