Graph API 사용하기

Graph API 사용하기 (Use the graph API)

이 가이드는 LangGraph Graph API의 기초를 보여줘요. 상태 정의·갱신을 다루고, 시퀀스, 브랜치, 루프 같은 흔한 그래프 구조를 조합하는 방법을 살펴봐요. 또 map-reduce 워크플로를 위한 Send API와 상태 업데이트와 노드 간 "이동(hop)"을 결합하는 Command API 같은 제어 기능도 다뤄요.

출처: 문서

본문

설정 (Setup)

langgraph를 설치해요.

pip install -U langgraph

uv를 쓴다면:

uv add langgraph

더 나은 디버깅을 위한 LangSmith 설정 LangSmith에 가입하면 LangGraph 프로젝트의 문제를 빠르게 발견하고 성능을 개선할 수 있어요. LangSmith로 트레이스 데이터를 활용해 LangGraph로 만든 LLM 앱을 디버깅·테스트·모니터링할 수 있어요. 시작 방법은 문서를 참고해요.

상태 정의와 갱신 (Define and update state)

LangGraph에서 상태를 정의하고 갱신하는 방법을 보여줄게요. 다음을 시연해요.

상태로 그래프의 스키마를 정의하는 방법
리듀서로 상태 업데이트가 처리되는 방식을 제어하는 방법

상태 정의 (Define state)

LangGraph의 상태는 TypedDict, Pydantic 모델, 또는 dataclass가 될 수 있어요. 아래에서는 TypedDict를 사용할게요. Pydantic 사용에 대한 자세한 내용은 그래프 상태에 Pydantic 모델 사용하기를 참고해요.

기본적으로 그래프는 같은 입력·출력 스키마를 가지며, 상태가 그 스키마를 결정해요. 별도의 입력·출력 스키마를 정의하는 방법은 입력·출력 스키마 정의를 참고해요.

messages를 사용한 간단한 예시를 살펴봐요. 이것은 많은 LLM 애플리케이션에서 다재다능한 상태 표현이에요. 더 자세한 내용은 개념 페이지를 참고해요.

from langchain.messages import AnyMessage
from typing_extensions import TypedDict

class State(TypedDict):
    messages: list[AnyMessage]
    extra_field: int

이 상태는 message 객체 목록과 추가 정수 필드를 추적해요.

상태 갱신 (Update state)

단일 노드가 있는 예시 그래프를 만들어 볼게요. 우리의 노드는 그래프의 상태를 읽고 갱신하는 일반 Python 함수일 뿐이에요. 이 함수의 첫 번째 인자는 항상 상태예요.

from langchain.messages import AIMessage

def node(state: State):
    messages = state["messages"]
    new_message = AIMessage("Hello!")
    return {"messages": messages + [new_message], "extra_field": 10}

이 노드는 메시지 목록에 단순히 메시지를 추가하고, 추가 필드에 값을 채워요.

노드는 상태를 변형하는 대신 상태에 대한 업데이트를 직접 반환해야 해요.

다음으로 이 노드를 포함한 간단한 그래프를 정의해요. StateGraph를 사용해 이 상태에서 동작하는 그래프를 정의하고, add_node로 그래프를 채워요.

from langgraph.graph import StateGraph

builder = StateGraph(State)
builder.add_node(node)
builder.set_entry_point("node")
graph = builder.compile()

LangGraph는 그래프 시각화를 위한 내장 유틸리티를 제공해요. 그래프를 살펴볼게요. 시각화에 대한 자세한 내용은 그래프 시각화를 참고해요.

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))

이 경우 그래프는 단일 노드만 실행해요. 간단한 호출로 진행할게요.

from langchain.messages import HumanMessage

result = graph.invoke({"messages": [HumanMessage("Hi")]})
result

{'messages': [HumanMessage(content='Hi'), AIMessage(content='Hello!')], 'extra_field': 10}

주의할 점:

상태의 단일 키를 갱신하는 것으로 호출을 시작했어요.
호출 결과로 전체 상태를 받아요.

편의상 message 객체의 내용을 pretty-print로 자주 살펴봐요.

for message in result["messages"]:
    message.pretty_print()

================================ Human Message ================================

Hi
================================== Ai Message ==================================

Hello!

리듀서로 상태 업데이트 처리 (Process state updates with reducers)

상태의 각 키는 노드의 업데이트가 어떻게 적용될지 제어하는 독립적인 리듀서 함수를 가질 수 있어요. 리듀서 함수가 명시적으로 지정되지 않으면 해당 키의 모든 업데이트가 그것을 덮어써야 한다고 가정해요.

TypedDict 상태 스키마의 경우 상태의 해당 필드를 리듀서 함수로 어노테이션해 리듀서를 정의할 수 있어요.

앞의 예시에서 우리 노드는 메시지를 추가해 상태의 "messages" 키를 갱신했어요. 아래에서는 이 키에 리듀서를 추가해 업데이트가 자동으로 추가되게 해요.

from typing_extensions import Annotated

def add(left, right):
    """Can also import `add` from the `operator` built-in."""
    return left + right

class State(TypedDict):
    messages: Annotated[list[AnyMessage], add]  # [!code highlight]
    extra_field: int

이제 우리 노드를 단순화할 수 있어요.

def node(state: State):
    new_message = AIMessage("Hello!")
    return {"messages": [new_message], "extra_field": 10}  # [!code highlight]

from langgraph.graph import START

graph = StateGraph(State).add_node(node).add_edge(START, "node").compile()

result = graph.invoke({"messages": [HumanMessage("Hi")]})

for message in result["messages"]:
    message.pretty_print()

================================ Human Message ================================

Hi
================================== Ai Message ==================================

Hello!

MessagesState

실무에서는 메시지 목록을 갱신할 때 추가 고려사항이 있어요.

상태의 기존 메시지를 갱신하고 싶을 수 있어요.
OpenAI 형식 같은 메시지 형식의 약칭(short-hand)을 받아들이고 싶을 수 있어요.

LangGraph는 이러한 고려사항을 처리하는 내장 리듀서 add_messages를 포함해요.

from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]  # [!code highlight]
    extra_field: int

def node(state: State):
    new_message = AIMessage("Hello!")
    return {"messages": [new_message], "extra_field": 10}

graph = StateGraph(State).add_node(node).set_entry_point("node").compile()

input_message = {"role": "user", "content": "Hi"}  # [!code highlight]

result = graph.invoke({"messages": [input_message]})

for message in result["messages"]:
    message.pretty_print()

================================ Human Message ================================

Hi
================================== Ai Message ==================================

Hello!

이것은 채팅 모델을 수반하는 애플리케이션의 다재다능한 상태 표현이에요. LangGraph는 편의를 위해 prebuilt MessagesState를 포함하므로 다음과 같이 할 수 있어요.

from langgraph.graph import MessagesState

class State(MessagesState):
    extra_field: int

`Overwrite`로 리듀서 우회하기 (Bypass reducers with `Overwrite`)

어떤 경우에는 리듀서를 우회하고 상태 값을 직접 덮어쓰고 싶을 수 있어요. LangGraph는 이를 위해 Overwrite 타입을 제공해요. 노드가 Overwrite로 감싼 값을 반환하면 리듀서가 우회되고 채널이 그 값으로 직접 설정돼요.

누적된 상태를 기존 값과 병합하는 대신 재설정하거나 교체하고 싶을 때 유용해요.

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

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

def add_message(state: State):
    return {"messages": ["first message"]}

def replace_messages(state: State):
    # Bypass the reducer and replace the entire messages list
    return {"messages": Overwrite(["replacement message"])}

builder = StateGraph(State)
builder.add_node("add_message", add_message)
builder.add_node("replace_messages", replace_messages)
builder.add_edge(START, "add_message")
builder.add_edge("add_message", "replace_messages")
builder.add_edge("replace_messages", END)

graph = builder.compile()

result = graph.invoke({"messages": ["initial"]})
print(result["messages"])

['replacement message']

특수 키 "__overwrite__"를 가진 JSON 형식도 사용할 수 있어요.

def replace_messages(state: State):
    return {"messages": {"__overwrite__": ["replacement message"]}}

노드들이 병렬로 실행될 때, 주어진 슈퍼스텝에서 하나의 노드만 같은 상태 키에 Overwrite를 사용할 수 있어요. 여러 노드가 같은 슈퍼스텝에서 같은 키를 덮어쓰려 하면 InvalidUpdateError가 발생해요.

입력·출력 스키마 정의 (Define input and output schemas)

기본적으로 StateGraph는 단일 스키마로 동작하며, 모든 노드가 그 스키마로 통신할 것으로 기대해요. 하지만 그래프에 대해 별도의 입력·출력 스키마를 정의하는 것도 가능해요.

별도 스키마가 지정되면 노드 간 통신에는 여전히 내부 스키마가 사용돼요. 입력 스키마는 제공된 입력이 예상 구조와 일치하도록 보장하고, 출력 스키마는 정의된 출력 스키마에 따라 관련 정보만 반환하도록 내부 데이터를 필터링해요.

아래에서 별도 입력·출력 스키마를 정의하는 방법을 볼게요.

from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict

# Define the schema for the input
class InputState(TypedDict):
    question: str

# Define the schema for the output
class OutputState(TypedDict):
    answer: str

# Define the overall schema, combining both input and output
class OverallState(InputState, OutputState):
    pass

# Define the node that processes the input and generates an answer
def answer_node(state: InputState):
    # Example answer and an extra key
    return {"answer": "bye", "question": state["question"]}

# Build the graph with input and output schemas specified
builder = StateGraph(OverallState, input_schema=InputState, output_schema=OutputState)
builder.add_node(answer_node)  # Add the answer node
builder.add_edge(START, "answer_node")  # Define the starting edge
builder.add_edge("answer_node", END)  # Define the ending edge
graph = builder.compile()  # Compile the graph

# Invoke the graph with an input and print the result
print(graph.invoke({"question": "hi"}))

{'answer': 'bye'}

invoke의 출력에는 출력 스키마만 포함된다는 점을 주목하세요.

노드 간 비공개 상태 전달 (Pass private state between nodes)

어떤 경우에는 중간 로직에 중요하지만 그래프의 메인 스키마의 일부일 필요는 없는 정보를 노드들이 교환하고 싶을 수 있어요. 이 비공개 데이터는 그래프의 전체 입력·출력과 관련이 없으며 특정 노드 사이에서만 공유되어야 해요.

아래에서 세 개의 노드(node_1, node_2, node_3)로 이루어진 순차 그래프 예시를 만들 거예요. 첫 두 단계(node_1, node_2) 사이에는 비공개 데이터를 전달하고, 세 번째 단계(node_3)는 공개된 전체 상태에만 접근할 수 있어요.

from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict

# The overall state of the graph (this is the public state shared across nodes)
class OverallState(TypedDict):
    a: str

# Output from node_1 contains private data that is not part of the overall state
class Node1Output(TypedDict):
    private_data: str

# The private data is only shared between node_1 and node_2
def node_1(state: OverallState) -> Node1Output:
    output = {"private_data": "set by node_1"}
    print(f"Entered node `node_1`:\n\tInput: {state}.\n\tReturned: {output}")
    return output

# Node 2 input only requests the private data available after node_1
class Node2Input(TypedDict):
    private_data: str

def node_2(state: Node2Input) -> OverallState:
    output = {"a": "set by node_2"}
    print(f"Entered node `node_2`:\n\tInput: {state}.\n\tReturned: {output}")
    return output

# Node 3 only has access to the overall state (no access to private data from node_1)
def node_3(state: OverallState) -> OverallState:
    output = {"a": "set by node_3"}
    print(f"Entered node `node_3`:\n\tInput: {state}.\n\tReturned: {output}")
    return output

# Connect nodes in a sequence
# node_2 accepts private data from node_1, whereas
# node_3 does not see the private data.
builder = StateGraph(OverallState).add_sequence([node_1, node_2, node_3])
builder.add_edge(START, "node_1")
graph = builder.compile()

# Invoke the graph with the initial state
response = graph.invoke(
    {
        "a": "set at start",
    }
)

print()
print(f"Output of graph invocation: {response}")

Entered node `node_1`:
    Input: {'a': 'set at start'}.
    Returned: {'private_data': 'set by node_1'}
Entered node `node_2`:
    Input: {'private_data': 'set by node_1'}.
    Returned: {'a': 'set by node_2'}
Entered node `node_3`:
    Input: {'a': 'set by node_2'}.
    Returned: {'a': 'set by node_3'}

Output of graph invocation: {'a': 'set by node_3'}

그래프 상태에 Pydantic 모델 사용하기 (Use pydantic models for graph state)

StateGraph는 초기화 시 그래프의 노드들이 접근하고 갱신할 수 있는 상태의 "형태"를 지정하는 state_schema 인자를 받아요.

예시에서는 보통 state_schema에 python 네이티브 TypedDict나 dataclass를 쓰지만, state_schema는 어떤 타입이든 될 수 있어요.

여기서는 Pydantic BaseModel을 state_schema로 사용해 입력에 대한 런타임 검증을 추가하는 방법을 볼게요.

알려진 제한 사항 (Known Limitations)

현재 그래프의 출력은 Pydantic 모델의 인스턴스가 아닐 거예요.

런타임 검증은 그래프의 첫 노드 입력에만 발생하며, 후속 노드나 출력에서는 발생하지 않아요.

Pydantic의 검증 오류 트레이스는 오류가 어느 노드에서 발생했는지 보여주지 않아요.

Pydantic의 재귀 검증은 느릴 수 있어요. 성능이 중요한 애플리케이션이라면 대신 dataclass를 고려해 보세요.

from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
from pydantic import BaseModel

# The overall state of the graph (this is the public state shared across nodes)
class OverallState(BaseModel):
    a: str

def node(state: OverallState):
    return {"a": "goodbye"}

# Build the state graph
builder = StateGraph(OverallState)
builder.add_node(node)  # node_1 is the first node
builder.add_edge(START, "node")  # Start the graph with node_1
builder.add_edge("node", END)  # End the graph after node_1
graph = builder.compile()

# Test the graph with a valid input
graph.invoke({"a": "hello"})

잘못된 입력으로 그래프를 호출해요.

try:
    graph.invoke({"a": 123})  # Should be a string
except Exception as e:
    print("An exception was raised because `a` is an integer rather than a string.")
    print(e)

An exception was raised because `a` is an integer rather than a string.
1 validation error for OverallState
a
  Input should be a valid string [type=string_type, input_value=123, input_type=int]
    For further information visit https://errors.pydantic.dev/2.9/v/string_type

직렬화 동작 (Serialization Behavior)

Pydantic 모델을 상태 스키마로 쓸 때 직렬화가 어떻게 동작하는지 이해하는 게 중요해요. 특히:

Pydantic 객체를 입력으로 전달할 때
그래프에서 출력을 받을 때
중첩된 Pydantic 모델로 작업할 때

이 동작들을 실제로 살펴봐요.

from langgraph.graph import StateGraph, START, END
from pydantic import BaseModel

class NestedModel(BaseModel):
    value: str

class ComplexState(BaseModel):
    text: str
    count: int
    nested: NestedModel

def process_node(state: ComplexState):
    # Node receives a validated Pydantic object
    print(f"Input state type: {type(state)}")
    print(f"Nested type: {type(state.nested)}")
    # Return a dictionary update
    return {"text": state.text + " processed", "count": state.count + 1}

# Build the graph
builder = StateGraph(ComplexState)
builder.add_node("process", process_node)
builder.add_edge(START, "process")
builder.add_edge("process", END)
graph = builder.compile()

# Create a Pydantic instance for input
input_state = ComplexState(text="hello", count=0, nested=NestedModel(value="test"))
print(f"Input object type: {type(input_state)}")

# Invoke graph with a Pydantic instance
result = graph.invoke(input_state)
print(f"Output type: {type(result)}")
print(f"Output content: {result}")

# Convert back to Pydantic model if needed
output_model = ComplexState(**result)
print(f"Converted back to Pydantic: {type(output_model)}")

런타임 타입 강제 변환 (Runtime Type Coercion)

Pydantic은 특정 데이터 타입에 대해 런타임 타입 강제 변환을 수행해요. 이것은 유용할 수 있지만 그걸 인지하지 못하면 예상치 못한 동작을 일으킬 수도 있어요.

from langgraph.graph import StateGraph, START, END
from pydantic import BaseModel

class CoercionExample(BaseModel):
    # Pydantic will coerce string numbers to integers
    number: int
    # Pydantic will parse string booleans to bool
    flag: bool

def inspect_node(state: CoercionExample):
    print(f"number: {state.number} (type: {type(state.number)})")
    print(f"flag: {state.flag} (type: {type(state.flag)})")
    return {}

builder = StateGraph(CoercionExample)
builder.add_node("inspect", inspect_node)
builder.add_edge(START, "inspect")
builder.add_edge("inspect", END)
graph = builder.compile()

# Demonstrate coercion with string inputs that will be converted
result = graph.invoke({"number": "42", "flag": "true"})

# This would fail with a validation error
try:
    graph.invoke({"number": "not-a-number", "flag": "true"})
except Exception as e:
    print(f"\nExpected validation error: {e}")

메시지 모델 작업 (Working with Message Models)

상태 스키마에서 LangChain 메시지 타입으로 작업할 때 직렬화에 대한 중요한 고려사항이 있어요. 메시지 객체를 wire로 통해 사용할 때 올바른 직렬화/역직렬화를 위해 BaseMessage 대신 AnyMessage를 사용해야 해요.

from langgraph.graph import StateGraph, START, END
from pydantic import BaseModel
from langchain.messages import HumanMessage, AIMessage, AnyMessage
from typing import List

class ChatState(BaseModel):
    messages: List[AnyMessage]
    context: str

def add_message(state: ChatState):
    return {"messages": state.messages + [AIMessage(content="Hello there!")]}

builder = StateGraph(ChatState)
builder.add_node("add_message", add_message)
builder.add_edge(START, "add_message")
builder.add_edge("add_message", END)
graph = builder.compile()

# Create input with a message
initial_state = ChatState(
    messages=[HumanMessage(content="Hi")], context="Customer support chat"
)

result = graph.invoke(initial_state)
print(f"Output: {result}")

# Convert back to Pydantic model to see message types
output_model = ChatState(**result)
for i, msg in enumerate(output_model.messages):
    print(f"Message {i}: {type(msg).__name__} - {msg.content}")

런타임 설정 추가 (Add runtime configuration)

때로는 그래프를 호출할 때 구성할 수 있길 원해요. 예를 들어 그래프 상태를 이 파라미터들로 오염시키지 않고 런타임에 어떤 LLM이나 시스템 프롬프트를 사용할지 지정하고 싶을 수 있어요.

런타임 설정을 추가하려면:

설정 스키마 지정
노드나 조건부 엣지의 함수 시그니처에 설정 추가
그래프에 설정 전달

간단한 예시를 확인해요.

from langgraph.graph import END, StateGraph, START
from langgraph.runtime import Runtime
from typing_extensions import TypedDict

# 1. Specify config schema
class ContextSchema(TypedDict):
    my_runtime_value: str

# 2. Define a graph that accesses the config in a node
class State(TypedDict):
    my_state_value: str

def node(state: State, runtime: Runtime[ContextSchema]):  # [!code highlight]
    if runtime.context["my_runtime_value"] == "a":  # [!code highlight]
        return {"my_state_value": 1}
    elif runtime.context["my_runtime_value"] == "b":  # [!code highlight]
        return {"my_state_value": 2}
    else:
        raise ValueError("Unknown values.")

builder = StateGraph(State, context_schema=ContextSchema)  # [!code highlight]
builder.add_node(node)
builder.add_edge(START, "node")
builder.add_edge("node", END)

graph = builder.compile()

# 3. Pass in configuration at runtime:
print(graph.invoke({}, context={"my_runtime_value": "a"}))  # [!code highlight]
print(graph.invoke({}, context={"my_runtime_value": "b"}))  # [!code highlight]

{'my_state_value': 1}
{'my_state_value': 2}

확장 예시: 런타임에 LLM 지정

아래에서 런타임에 어떤 LLM을 사용할지 구성하는 실용적 예시를 보여줘요. OpenAI와 Anthropic 모델을 모두 사용할 거예요.

from dataclasses import dataclass

from langchain.chat_models import init_chat_model
from langgraph.graph import MessagesState, END, StateGraph, START
from langgraph.runtime import Runtime
from typing_extensions import TypedDict

@dataclass
class ContextSchema:
    model_provider: str = "anthropic"

MODELS = {
    "anthropic": init_chat_model("claude-haiku-4-5-20251001"),
    "openai": init_chat_model("gpt-5.4-mini"),
}

def call_model(state: MessagesState, runtime: Runtime[ContextSchema]):
    model = MODELS[runtime.context.model_provider]
    response = model.invoke(state["messages"])
    return {"messages": [response]}

builder = StateGraph(MessagesState, context_schema=ContextSchema)
builder.add_node("model", call_model)
builder.add_edge(START, "model")
builder.add_edge("model", END)

graph = builder.compile()

# Usage
input_message = {"role": "user", "content": "hi"}
# With no configuration, uses default (Anthropic)
response_1 = graph.invoke({"messages": [input_message]}, context=ContextSchema())["messages"][-1]
# Or, can set OpenAI
response_2 = graph.invoke({"messages": [input_message]}, context={"model_provider": "openai"})["messages"][-1]

print(response_1.response_metadata["model_name"])
print(response_2.response_metadata["model_name"])

claude-haiku-4-5-20251001
gpt-5.4-mini

확장 예시: 런타임에 모델과 시스템 메시지 지정

아래에서 LLM과 시스템 메시지라는 두 파라미터를 런타임에 구성하는 실용적 예시를 보여줘요.

from dataclasses import dataclass
from langchain.chat_models import init_chat_model
from langchain.messages import SystemMessage
from langgraph.graph import END, MessagesState, StateGraph, START
from langgraph.runtime import Runtime
from typing_extensions import TypedDict

@dataclass
class ContextSchema:
    model_provider: str = "anthropic"
    system_message: str | None = None

MODELS = {
    "anthropic": init_chat_model("claude-haiku-4-5-20251001"),
    "openai": init_chat_model("gpt-5.4-mini"),
}

def call_model(state: MessagesState, runtime: Runtime[ContextSchema]):
    model = MODELS[runtime.context.model_provider]
    messages = state["messages"]
    if (system_message := runtime.context.system_message):
        messages = [SystemMessage(system_message)] + messages
    response = model.invoke(messages)
    return {"messages": [response]}

builder = StateGraph(MessagesState, context_schema=ContextSchema)
builder.add_node("model", call_model)
builder.add_edge(START, "model")
builder.add_edge("model", END)

graph = builder.compile()

# Usage
input_message = {"role": "user", "content": "hi"}
response = graph.invoke({"messages": [input_message]}, context={"model_provider": "openai", "system_message": "Respond in Italian."})
for message in response["messages"]:
    message.pretty_print()

================================ Human Message ================================

hi
================================== Ai Message ==================================

Ciao! Come posso aiutarti oggi?

재시도 정책 추가 (Add retry policies)

API를 호출하거나, DB를 조회하거나, LLM을 호출하는 등 노드에 커스텀 재시도 정책을 두고 싶은 사용 사례가 많아요. LangGraph는 노드에 재시도 정책을 추가하게 해줘요.

재시도 정책을 구성하려면 add_node에 retry_policy 파라미터를 전달해요. retry_policy 파라미터는 RetryPolicy named tuple 객체를 받아요. 아래에서 기본 파라미터로 RetryPolicy 객체를 만들고 노드와 연결해요.

from langgraph.types import RetryPolicy

builder.add_node(
    "node_name",
    node_function,
    retry_policy=RetryPolicy(),
)

기본적으로 retry_on 파라미터는 다음을 제외한 모든 예외에서 재시도하는 default_retry_on 함수를 사용해요.

ValueError
TypeError
ArithmeticError
ImportError
LookupError
NameError
SyntaxError
RuntimeError
ReferenceError
StopIteration
StopAsyncIteration
OSError

또한 requests와 httpx 같은 인기 HTTP 요청 라이브러리의 예외에 대해서는 5xx 상태 코드에서만 재시도해요.

확장 예시: 재시도 정책 커스터마이즈

SQL 데이터베이스에서 읽는 예시를 생각해요. 아래에서 노드 두 개에 서로 다른 재시도 정책을 전달해요.

import sqlite3
from typing_extensions import TypedDict
from langchain.chat_models import init_chat_model
from langgraph.graph import END, MessagesState, StateGraph, START
from langgraph.types import RetryPolicy
from langchain.messages import AIMessage

con = sqlite3.connect(":memory:")
model = init_chat_model("claude-haiku-4-5-20251001")

def query_database(state: MessagesState):
    cursor = con.cursor()
    cursor.execute("SELECT * FROM Artist LIMIT 10;")
    query_result = str(cursor.fetchall())
    return {"messages": [AIMessage(content=query_result)]}

def call_model(state: MessagesState):
    response = model.invoke(state["messages"])
    return {"messages": [response]}

# Define a new graph
builder = StateGraph(MessagesState)
builder.add_node(
    "query_database",
    query_database,
    retry_policy=RetryPolicy(retry_on=sqlite3.OperationalError),
)
builder.add_node("model", call_model, retry_policy=RetryPolicy(max_attempts=5))
builder.add_edge(START, "model")
builder.add_edge("model", "query_database")
builder.add_edge("query_database", END)
graph = builder.compile()

노드 타임아웃 설정 (Set node timeouts)

add_node와 함께 timeout 파라미터를 사용해 단일 async 노드 호출이 얼마나 오래 실행될 수 있는지 제한해요. 타임아웃을 초 단위나 datetime.timedelta로 제공해요.

import asyncio
from typing_extensions import TypedDict

from langgraph.errors import NodeTimeoutError
from langgraph.graph import END, START, StateGraph


class State(TypedDict):
    value: str


async def call_model(state: State) -> State:
    await asyncio.sleep(2)
    return {"value": "done"}


builder = StateGraph(State)
builder.add_node("model", call_model, timeout=1.0)
builder.add_edge(START, "model")
builder.add_edge("model", END)
graph = builder.compile()

try:
    await graph.ainvoke({"value": "start"})
except NodeTimeoutError:
    print("Node timed out")

노드 타임아웃은 async 노드에서만 지원돼요. 동기 노드에 timeout을 설정하면 그래프가 컴파일될 때 LangGraph가 오류를 발생시켜요. 동기 Python 실행은 프로세스 내에서 안전하게 취소될 수 없기 때문이에요.

노드가 타임아웃을 초과하면 LangGraph는 Python 내장 TimeoutError를 상속하는 NodeTimeoutError를 발생시켜요. 노드에 TimeoutError나 NodeTimeoutError를 재시도하는 retry_policy가 있다면, 타임아웃된 시도가 재시도돼요. 타임아웃은 각 시도에 독립적으로 적용되므로, 재시도마다 타이머가 리셋돼요.

타임아웃된 시도는 버퍼된 쓰기를 커밋하지 않아요. 이렇게 하면 타임아웃 경계 이후에 상태 업데이트나 하위 태스크 스케줄링이 새어 나가는 걸 방지해요.

노드 타임아웃 구성 (Configure node timeouts)

add_node의 timeout= 파라미터는 단일 async 노드 시도가 얼마나 오래 실행될 수 있는지 제한해요. 숫자(초), timedelta, 또는 실행·유휴 타임아웃을 세밀하게 제어하는 TimeoutPolicy를 전달해요. 한도를 초과하면 LangGraph는 NodeTimeoutError를 발생시키고 재시도 정책이 재시도 여부를 결정하게 해요.

노드별 타임아웃은 langgraph>=1.2가 필요해요.

from langgraph.types import TimeoutPolicy

builder.add_node(
    "call_model",
    call_model,
    timeout=TimeoutPolicy(run_timeout=120, idle_timeout=30),
)

전체 타임아웃 수명 주기, 유휴 타임아웃 갱신 소스, runtime.heartbeat()는 Fault tolerance를 참고해요.

노드 오류 처리 (Handle node errors)

add_node의 error_handler= 파라미터는 노드가 실패하고 모든 재시도를 소진한 후 실행되는 함수를 등록해요. 핸들러는 현재 상태와 실패 컨텍스트를 담은 타입화된 NodeError를 받고, Command로 복구 분기로 라우팅할 수 있어요.

노드 수준 오류 핸들러는 langgraph>=1.2가 필요해요.

from langgraph.errors import NodeError
from langgraph.types import Command, RetryPolicy

def payment_error_handler(state: State, error: NodeError) -> Command:
    return Command(
        update={"status": f"compensated: {error.error}"},
        goto="finalize",
    )

builder.add_node(
    "charge_payment",
    charge_payment,
    retry_policy=RetryPolicy(max_attempts=3, retry_on=ConnectionError),
    error_handler=payment_error_handler,
)

보상 패턴과 Command 라우팅은 Fault tolerance를 참고해요.

그래프 전체 노드 기본값 설정 (Set graph-wide node defaults)

langgraph>=1.2가 필요해요.

set_node_defaults를 사용해 retry_policy, timeout, cache_policy, error_handler를 매 add_node 호출마다 반복하는 대신 그래프의 모든 노드에 한 번에 설정해요. 노드별 값이 항상 우선하며, 기본값은 StateGraph.compile 시점에 적용돼요.

from langgraph.types import RetryPolicy, TimeoutPolicy

graph = (
    StateGraph(State)
    .set_node_defaults(
        retry_policy=RetryPolicy(max_attempts=3),
        timeout=TimeoutPolicy(run_timeout=30),
        error_handler=fallback_handler,
    )
    .add_node("a", node_a)
    .add_node("b", node_b, retry_policy=RetryPolicy(max_attempts=5))  # overrides default
    .add_edge(START, "a")
    .compile()
)

retry_policy와 timeout 기본값은 오류 처리 노드를 포함한 모든 노드에 적용돼요. cache_policy와 error_handler 기본값은 일반 노드에만 적용돼요 — 핸들러가 자기 자신을 catch하지 않고, 핸들러 결과를 캐시하는 건 안전하지 않기 때문이에요. 기본값은 서브그래프에 상속되지 않아요.

전체 우선순위 규칙과 적용 표는 Fault tolerance를 참고해요.

노드 안에서 실행 정보 접근 (Access execution info inside a node)

runtime.execution_info로 실행 식별자와 재시도 정보에 접근할 수 있어요. config에서 직접 읽지 않고도 스레드, 실행, 체크포인트 식별자와 재시도 상태를 노출해요.

Attribute	Type	Description
`thread_id`	`str \| None`	Thread ID for the current execution. `None` without a checkpointer.
`run_id`	`str \| None`	Run ID for the current execution. `None` when not provided in config.
`checkpoint_id`	`str`	Checkpoint ID for the current execution.
`checkpoint_ns`	`str`	Checkpoint namespace for the current execution.
`task_id`	`str`	Task ID for the current execution.
`node_attempt`	`int`	Current execution attempt number (1-indexed). `1` on the first try, `2` on the first retry, etc.
`node_first_attempt_time`	`float \| None`	Unix timestamp (seconds) of when the first attempt started. Stays the same across retries.

스레드·실행 ID 접근 (Access thread and run IDs)

execution_info를 사용해 노드 안에서 스레드 ID, 실행 ID, 기타 식별 필드에 접근해요.

from langgraph.graph import StateGraph, START, END
from langgraph.runtime import Runtime
from typing_extensions import TypedDict

class State(TypedDict):
    result: str

def my_node(state: State, runtime: Runtime):
    info = runtime.execution_info
    print(f"Thread: {info.thread_id}, Run: {info.run_id}")  # [!code highlight]
    return {"result": "done"}

builder = StateGraph(State)
builder.add_node("my_node", my_node)
builder.add_edge(START, "my_node")
builder.add_edge("my_node", END)
graph = builder.compile()

재시도 상태에 따라 동작 조정 (Adjust behavior based on retry state)

노드에 재시도 정책이 있다면 execution_info를 사용해 현재 시도 번호를 살펴보고 첫 시도 실패 후 폴백으로 전환해요.

from langgraph.graph import StateGraph, START, END
from langgraph.runtime import Runtime
from langgraph.types import RetryPolicy
from typing_extensions import TypedDict

class State(TypedDict):
    result: str

def my_node(state: State, runtime: Runtime):
    info = runtime.execution_info
    if info.node_attempt > 1:  # [!code highlight]
        # use a fallback on retries
        return {"result": call_fallback_api()}
    return {"result": call_primary_api()}

builder = StateGraph(State)
builder.add_node("my_node", my_node, retry_policy=RetryPolicy(max_attempts=3))
builder.add_edge(START, "my_node")
builder.add_edge("my_node", END)
graph = builder.compile()

execution_info는 재시도 정책이 없어도 Runtime 객체에서 사용할 수 있어요. node_attempt는 1을 기본값으로 하고, node_first_attempt_time은 노드가 실행을 시작한 시간으로 설정돼요.

노드 안에서 서버 정보 접근 (Access server info inside a node)

그래프가 LangGraph Server에서 실행될 때 runtime.server_info로 서버별 메타데이터에 접근할 수 있어요. config 메타데이터나 configurable 키에서 직접 읽지 않고도 assistant ID, graph ID, 인증된 사용자를 노출해요.

Attribute	Type	Description
`assistant_id`	`str`	The assistant ID for the current deployment.
`graph_id`	`str`	The graph ID for the current deployment.
`user`	`BaseUser \| None`	The authenticated user, if custom auth is configured.

from langgraph.graph import StateGraph, START, END
from langgraph.runtime import Runtime
from typing_extensions import TypedDict

class State(TypedDict):
    result: str

def my_node(state: State, runtime: Runtime):
    server = runtime.server_info
    if server is not None:
        print(f"Assistant: {server.assistant_id}, Graph: {server.graph_id}")  # [!code highlight]
        if server.user is not None:
            print(f"User: {server.user.identity}")
    return {"result": "done"}

builder = StateGraph(State)
builder.add_node("my_node", my_node)
builder.add_edge(START, "my_node")
builder.add_edge("my_node", END)
graph = builder.compile()

server_info는 그래프가 LangGraph Server에서 실행되지 않을 때(예: 로컬 개발·테스트 중) None이에요.

runtime.execution_info와 runtime.server_info에는 deepagents>=0.5.0(또는 langgraph>=1.1.5)가 필요해요.

노드 안에서 드레인 상태 접근 (Access drain state inside a node)

정상 종료(graceful shutdown)가 요청되면 runtime.drain_requested는 True가 돼요. 노드 안에서 이를 읽어 다음 슈퍼스텝 경계 전에 비싼 작업을 건너뛰어요.

from langgraph.runtime import Runtime

def my_node(state: State, runtime: Runtime) -> State:
    if runtime.drain_requested:  # [!code highlight]
        return {"status": "skipped", "reason": runtime.drain_reason}
    return {"status": do_work()}

Property	Type	Description
`drain_requested`	`bool`	`True` if `RunControl.request_drain()` has been called for this run.
`drain_reason`	`str \| None`	The reason string passed to `request_drain()`, or `None` if drain was not requested.

langgraph>=1.2가 필요해요. 전체 RunControl API는 Graceful shutdown을 참고해요.

노드 캐싱 추가 (Add node caching)

노드 캐싱은 비용이 큰 작업(시간이나 비용 측면에서)을 할 때 반복 작업을 피하고 싶은 경우 유용해요. LangGraph는 그래프의 노드에 개별화된 캐싱 정책을 추가하게 해줘요.

캐시 정책을 구성하려면 add_node 함수에 cache_policy 파라미터를 전달해요. 다음 예시에서는 120초의 time-to-live와 기본 key_func 생성기로 CachePolicy 객체를 만들고, 노드와 연결해요.

from langgraph.types import CachePolicy

builder.add_node(
    "node_name",
    node_function,
    cache_policy=CachePolicy(ttl=120),
)

그런 다음 그래프의 노드 수준 캐싱을 활성화하려면 그래프를 컴파일할 때 cache 인자를 설정해요. 아래 예시는 InMemoryCache를 사용해 인메모리 캐시로 그래프를 구성하지만, SqliteCache도 사용할 수 있어요.

from langgraph.cache.memory import InMemoryCache

graph = builder.compile(cache=InMemoryCache())

단계 시퀀스 만들기 (Create a sequence of steps)

사전 준비 (Prerequisites) 이 가이드는 위의 상태 섹션에 대한 이해를 가정해요.

간단한 단계 시퀀스를 구성하는 방법을 보여줄게요. 다음을 보여줄 거예요.

순차 그래프를 만드는 방법
유사한 그래프를 구성하는 내장 약칭(short-hand)

노드 시퀀스를 추가하려면 그래프의 add_node와 add_edge 메서드를 사용해요.

from langgraph.graph import START, StateGraph

builder = StateGraph(State)

# Add nodes
builder.add_node(step_1)
builder.add_node(step_2)
builder.add_node(step_3)

# Add edges
builder.add_edge(START, "step_1")
builder.add_edge("step_1", "step_2")
builder.add_edge("step_2", "step_3")

내장 약칭 .add_sequence도 사용할 수 있어요.

builder = StateGraph(State).add_sequence([step_1, step_2, step_3])
builder.add_edge(START, "step_1")

왜 애플리케이션 단계를 LangGraph로 시퀀스로 나누나요?

LangGraph는 애플리케이션에 기본 영속성 레이어를 쉽게 추가하게 해줘요. 이를 통해 노드 실행 사이에 상태가 체크포인트되므로 LangGraph 노드들이 다음을 관장해요.

상태 업데이트가 어떻게 체크포인트되는지
human-in-the-loop 워크플로에서 인터럽션이 어떻게 재개되는지
LangGraph의 타임 트래블 기능으로 실행을 어떻게 "되감고" 분기하는지

또한 실행 단계가 어떻게 스트리밍되는지, Studio로 애플리케이션이 어떻게 시각화·디버깅되는지도 결정해요.

end-to-end 예시를 시연해 볼게요. 세 단계 시퀀스를 만들 거예요.

상태의 키에 값을 채우기
같은 값을 갱신
다른 값을 채우기

먼저 상태를 정의해요. 이것이 그래프의 스키마를 관장하고, 업데이트를 어떻게 적용할지도 지정할 수 있어요. 자세한 내용은 리듀서로 상태 업데이트 처리를 참고해요.

이 경우 두 값만 추적할 거예요.

from typing_extensions import TypedDict

class State(TypedDict):
    value_1: str
    value_2: int

우리의 노드는 그래프의 상태를 읽고 갱신하는 일반 Python 함수일 뿐이에요. 이 함수의 첫 번째 인자는 항상 상태예요.

def step_1(state: State):
    return {"value_1": "a"}

def step_2(state: State):
    current_value_1 = state["value_1"]
    return {"value_1": f"{current_value_1} b"}

def step_3(state: State):
    return {"value_2": 10}

상태에 업데이트를 발행할 때 각 노드는 갱신하려는 키의 값만 지정하면 된다는 점을 주목하세요.

기본적으로 이것은 해당 키의 값을 덮어써요. 리듀서를 사용해 업데이트가 처리되는 방식을 제어할 수도 있어요 — 예를 들어 연속된 업데이트를 키에 추가(append)할 수 있어요. 자세한 내용은 리듀서로 상태 업데이트 처리를 참고해요.

마지막으로 그래프를 정의해요. StateGraph를 사용해 이 상태에서 동작하는 그래프를 정의해요.

그런 다음 add_node와 add_edge를 사용해 그래프를 채우고 그 제어 흐름을 정의해요.

from langgraph.graph import START, StateGraph

builder = StateGraph(State)

# Add nodes
builder.add_node(step_1)
builder.add_node(step_2)
builder.add_node(step_3)

# Add edges
builder.add_edge(START, "step_1")
builder.add_edge("step_1", "step_2")
builder.add_edge("step_2", "step_3")

커스텀 이름 지정 (Specifying custom names) add_node를 사용해 노드에 커스텀 이름을 지정할 수 있어요.
builder.add_node("my_node", step_1)

주목할 점:

add_edge는 노드 이름을 받으며, 함수의 경우 기본적으로 node.__name__이 돼요.
그래프의 진입점(entry point)을 지정해야 해요. 이를 위해 START 노드로 엣지를 추가해요.
실행할 노드가 더 없으면 그래프가 정지해요.

이제 그래프를 컴파일해요. 이 단계에서 그래프 구조에 대한 몇 가지 기본 검사(예: 고아 노드 식별)를 제공해요. checkpointer로 애플리케이션에 영속성을 추가한다면 여기서도 전달할 거예요.

graph = builder.compile()

LangGraph는 그래프 시각화를 위한 내장 유틸리티를 제공해요. 우리 시퀀스를 살펴볼게요. 시각화에 대한 자세한 내용은 그래프 시각화를 참고해요.

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))

간단한 호출로 진행할게요.

graph.invoke({"value_1": "c"})

{'value_1': 'a b', 'value_2': 10}

주목할 점:

단일 상태 키에 대한 값을 제공하는 것으로 호출을 시작했어요. 항상 적어도 하나의 키에 대한 값을 제공해야 해요.
우리가 전달한 값은 첫 노드에서 덮어써졌어요.
두 번째 노드는 값을 갱신했어요.
세 번째 노드는 다른 값을 채웠어요.

내장 약칭 (Built-in shorthand) langgraph>=0.2.46은 노드 시퀀스를 추가하는 내장 약칭 add_sequence를 포함해요. 같은 그래프를 다음과 같이 컴파일할 수 있어요.
builder = StateGraph(State).add_sequence([step_1, step_2, step_3])  # [!code highlight]
builder.add_edge(START, "step_1")

graph = builder.compile()

graph.invoke({"value_1": "c"})

브랜치 만들기 (Create branches)

노드의 병렬 실행은 전체 그래프 연산 속도를 높이는 데 필수적이에요. LangGraph는 노드의 병렬 실행을 네이티브로 지원해, 그래프 기반 워크플로의 성능을 크게 향상시킬 수 있어요. 이 병렬화는 표준 엣지와 conditional_edges를 모두 활용한 fan-out과 fan-in 메커니즘으로 이루어져요. 아래에 브랜칭 데이터 흐름을 추가하는 몇 가지 예시가 있어요.

그래프 노드를 병렬로 실행 (Run graph nodes in parallel)

이 예시에서는 Node A에서 B and C로 fan-out한 다음 D로 fan-in해요. 우리 상태로 리듀서 추가 연산을 지정해요. 이렇게 하면 State의 특정 키에 대해 값을 단순히 덮어쓰는 대신 결합하거나 누적해요. 리스트의 경우 새 리스트를 기존 리스트와 연결(concatenate)하는 것을 의미해요. 리듀서로 상태를 갱신하는 자세한 내용은 위의 상태 리듀서 섹션을 참고해요.

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

class State(TypedDict):
    # The operator.add reducer fn makes this append-only
    aggregate: Annotated[list, operator.add]

def a(state: State):
    print(f'Adding "A" to {state["aggregate"]}')
    return {"aggregate": ["A"]}

def b(state: State):
    print(f'Adding "B" to {state["aggregate"]}')
    return {"aggregate": ["B"]}

def c(state: State):
    print(f'Adding "C" to {state["aggregate"]}')
    return {"aggregate": ["C"]}

def d(state: State):
    print(f'Adding "D" to {state["aggregate"]}')
    return {"aggregate": ["D"]}

builder = StateGraph(State)
builder.add_node(a)
builder.add_node(b)
builder.add_node(c)
builder.add_node(d)
builder.add_edge(START, "a")
builder.add_edge("a", "b")
builder.add_edge("a", "c")
builder.add_edge("b", "d")
builder.add_edge("c", "d")
builder.add_edge("d", END)
graph = builder.compile()

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))

리듀서 덕에 각 노드에서 추가된 값이 누적되는 것을 볼 수 있어요.

graph.invoke({"aggregate": []}, {"configurable": {"thread_id": "foo"}})

Adding "A" to []
Adding "B" to ['A']
Adding "C" to ['A']
Adding "D" to ['A', 'B', 'C']

위 예시에서 노드 "b"와 "c"는 같은 슈퍼스텝(superstep)에서 동시에 실행돼요. 같은 단계에 있으므로 노드 "d"는 "b"와 "c"가 모두 끝난 후에 실행돼요.

중요한 점: 병렬 슈퍼스텝의 업데이트는 일관되게 정렬되지 않을 수 있어요. 병렬 슈퍼스텝의 업데이트에 대해 일관되고 미리 정해진 정렬이 필요하다면, 출력을 정렬할 값과 함께 상태의 별도 필드에 써야 해요.

예외 처리 (Exception handling)?

LangGraph는 노드를 슈퍼스텝 안에서 실행해요. 즉 병렬 브랜치가 병렬로 실행되는 동안 전체 슈퍼스텝은 트랜잭션적이에요. 브랜치 중 하나라도 예외를 발생시키면, 업데이트 중 아무것도 상태에 적용되지 않아요(전체 슈퍼스텝이 오류로 처리돼요).

중요한 점: checkpointer를 쓸 때 슈퍼스텝 내 성공한 노드의 결과는 저장되며, 재개할 때 반복되지 않아요.

오류가 발생하기 쉬운 경우(불안정한 API 호출을 처리하고 싶을 때), LangGraph는 두 가지 해결 방법을 제공해요.

노드 안에서 일반 Python 코드를 작성해 예외를 catch하고 처리할 수 있어요.
특정 타입의 예외를 발생시키는 노드를 그래프가 재시도하도록 지시하는 **RetryPolicy**를 설정할 수 있어요. 실패한 브랜치만 재시도되므로 중복 작업을 걱정할 필요가 없어요.

이 둘을 함께 쓰면 병렬 실행을 수행하고 예외 처리를 완전히 제어할 수 있어요.

최대 동시성 설정 (Set max concurrency) 그래프를 호출할 때 configuration에서 max_concurrency를 설정해 최대 동시 태스크 수를 제어할 수 있어요.
graph.invoke({"value_1": "c"}, {"configurable": {"max_concurrency": 10}})

노드 실행 지연 (Defer node execution)

노드 실행 지연은 보류 중인 다른 태스크가 모두 완료될 때까지 노드의 실행을 미루고 싶을 때 유용해요. 브랜치들의 길이가 다를 때 특히 관련이 있는데, map-reduce 흐름 같은 워크플로에서 흔해요.

위의 예시는 각 경로가 한 단계일 때 fan-out과 fan-in을 보여줬어요. 하지만 브랜치 하나가 두 단계 이상이라면 어떨까요? "b" 브랜치에 "b_2" 노드를 추가해 볼게요.

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

class State(TypedDict):
    # The operator.add reducer fn makes this append-only
    aggregate: Annotated[list, operator.add]

def a(state: State):
    print(f'Adding "A" to {state["aggregate"]}')
    return {"aggregate": ["A"]}

def b(state: State):
    print(f'Adding "B" to {state["aggregate"]}')
    return {"aggregate": ["B"]}

def b_2(state: State):
    print(f'Adding "B_2" to {state["aggregate"]}')
    return {"aggregate": ["B_2"]}

def c(state: State):
    print(f'Adding "C" to {state["aggregate"]}')
    return {"aggregate": ["C"]}

def d(state: State):
    print(f'Adding "D" to {state["aggregate"]}')
    return {"aggregate": ["D"]}

builder = StateGraph(State)
builder.add_node(a)
builder.add_node(b)
builder.add_node(b_2)
builder.add_node(c)
builder.add_node(d, defer=True)  # [!code highlight]
builder.add_edge(START, "a")
builder.add_edge("a", "b")
builder.add_edge("a", "c")
builder.add_edge("b", "b_2")
builder.add_edge("b_2", "d")
builder.add_edge("c", "d")
builder.add_edge("d", END)
graph = builder.compile()

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))

graph.invoke({"aggregate": []})

Adding "A" to []
Adding "B" to ['A']
Adding "C" to ['A']
Adding "B_2" to ['A', 'B', 'C']
Adding "D" to ['A', 'B', 'C', 'B_2']

위 예시에서 노드 "b"와 "c"는 같은 슈퍼스텝에서 동시에 실행돼요. 노드 d에 defer=True를 설정해 보류 중인 모든 태스크가 끝날 때까지 실행되지 않게 했어요. 이 경우 "d"는 "b" 브랜치 전체가 끝날 때까지 실행을 기다려요.

모든 브랜치가 항상 실행된다면 defer=True 대신 목록 형태의 엣지로 기다릴 수 있어요. add_edge는 시작 노드 목록도 받아요. 이것은 별도의 add_edge 호출들의 약칭이 아니에요. 두 형태는 다르게 동작해요.

builder.add_edge(["b_2", "c"], "d")  # d runs once, after both b_2 and c complete

시작 노드 목록은 나열된 노드가 모두 완료된 후 d를 한 번 실행해요. 그중 하나가 실행되지 않으면(예: 조건부 엣지가 그 브랜치를 선택하지 않으면) d는 실행되지 않고 오류도 발생하지 않아요. 완료된 브랜치의 상태 업데이트는 그래프 상태에 남지만 d는 그것을 소비하지 않아요.
별도 엣지는 들어오는 브랜치가 완료된 각 슈퍼스텝마다 d를 한 번 실행해요. 길이가 같은 브랜치면 한 번 실행되고, 길이가 다른 브랜치면 d가 두 번 이상 실행돼요.
defer=True가 있는 별도 엣지(위 예시의 패턴)는 fan-out이 모든 브랜치를 선택했든 일부만 선택했든, 선택된 모든 브랜치가 완료된 후 d를 한 번 실행해요.

defer=True는 노드를 그래프 어디에든 보류 중인 태스크가 없을 때까지 연기해요. 그 노드를 공급하는 브랜치뿐 아니라요. 그 노드로 주소가 지정된 Send는 여전히 별도로 호출해요.

조건부 브랜칭 (Conditional branching)

fan-out이 런타임에 상태에 따라 달라져야 한다면 add_conditional_edges로 그래프 상태를 사용해 하나 이상의 경로를 선택할 수 있어요. 아래 예시를 보세요. 노드 a가 다음 노드를 결정하는 상태 업데이트를 생성해요.

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

class State(TypedDict):
    aggregate: Annotated[list, operator.add]
    # Add a key to the state. We will set this key to determine
    # how we branch.
    which: str

def a(state: State):
    print(f'Adding "A" to {state["aggregate"]}')
    return {"aggregate": ["A"], "which": "c"}  # [!code highlight]

def b(state: State):
    print(f'Adding "B" to {state["aggregate"]}')
    return {"aggregate": ["B"]}

def c(state: State):
    print(f'Adding "C" to {state["aggregate"]}')
    return {"aggregate": ["C"]}

builder = StateGraph(State)
builder.add_node(a)
builder.add_node(b)
builder.add_node(c)
builder.add_edge(START, "a")
builder.add_edge("b", END)
builder.add_edge("c", END)

def conditional_edge(state: State) -> Literal["b", "c"]:
    # Fill in arbitrary logic here that uses the state
    # to determine the next node
    return state["which"]

builder.add_conditional_edges("a", conditional_edge)  # [!code highlight]

graph = builder.compile()

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))

result = graph.invoke({"aggregate": []})
print(result)

Adding "A" to []
Adding "C" to ['A']
{'aggregate': ['A', 'C'], 'which': 'c'}

조건부 엣지는 여러 목적지 노드로 라우팅할 수 있어요. 예를 들어:
def route_bc_or_cd(state: State) -> Sequence[str]:
    if state["which"] == "cd":
        return ["c", "d"]
    return ["b", "c"]

맵-리듀스와 Send API (Map-Reduce and the send API)

LangGraph는 Send API를 사용해 map-reduce와 다른 고급 브랜칭 패턴을 지원해요. 사용법 예시는 다음과 같아요.

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

class OverallState(TypedDict):
    topic: str
    subjects: list[str]
    jokes: Annotated[list[str], operator.add]
    best_selected_joke: str

def generate_topics(state: OverallState):
    return {"subjects": ["lions", "elephants", "penguins"]}

def generate_joke(state: OverallState):
    joke_map = {
        "lions": "Why don't lions like fast food? Because they can't catch it!",
        "elephants": "Why don't elephants use computers? They're afraid of the mouse!",
        "penguins": "Why don't penguins like talking to strangers at parties? Because they find it hard to break the ice."
    }
    return {"jokes": [joke_map[state["subject"]]]}

def continue_to_jokes(state: OverallState):
    return [Send("generate_joke", {"subject": s}) for s in state["subjects"]]

def best_joke(state: OverallState):
    return {"best_selected_joke": "penguins"}

builder = StateGraph(OverallState)
builder.add_node("generate_topics", generate_topics)
builder.add_node("generate_joke", generate_joke)
builder.add_node("best_joke", best_joke)
builder.add_edge(START, "generate_topics")
builder.add_conditional_edges("generate_topics", continue_to_jokes, ["generate_joke"])
builder.add_edge("generate_joke", "best_joke")
builder.add_edge("best_joke", END)
graph = builder.compile()

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))

# Call the graph: here we call it to generate a list of jokes
stream = graph.stream_events({"topic": "animals"}, version="v3")
for message in stream.messages:
    for token in message.text:
        print(token, end="", flush=True)

{'generate_topics': {'subjects': ['lions', 'elephants', 'penguins']}}
{'generate_joke': {'jokes': ["Why don't lions like fast food? Because they can't catch it!"]}}
{'generate_joke': {'jokes': ["Why don't elephants use computers? They're afraid of the mouse!"]}}
{'generate_joke': {'jokes': ['Why don't penguins like talking to strangers at parties? Because they find it hard to break the ice.']}}
{'best_joke': {'best_selected_joke': 'penguins'}}

루프 만들기와 제어 (Create and control loops)

루프가 있는 그래프를 만들 때는 실행을 종료하는 메커니즘이 필요해요. 가장 흔한 방법은 어떤 종료 조건에 도달하면 END 노드로 라우팅하는 조건부 엣지를 추가하는 거예요.

그래프를 호출하거나 스트리밍할 때 재귀 한도(recursion limit)를 설정할 수도 있어요. 재귀 한도는 그래프가 오류를 발생시키기 전에 실행할 수 있는 슈퍼스텝의 수를 설정해요. 재귀 한도 개념에 대해 더 읽어보세요.

이 메커니즘들이 어떻게 동작하는지 이해하기 위해 루프가 있는 간단한 그래프를 살펴볼게요.

재귀 한도 오류 대신 상태의 마지막 값을 반환하려면 다음 섹션을 참고해요.

루프를 만들 때 종료 조건을 지정하는 조건부 엣지를 포함할 수 있어요.

builder = StateGraph(State)
builder.add_node(a)
builder.add_node(b)

def route(state: State) -> Literal["b", END]:
    if termination_condition(state):
        return END
    else:
        return "b"

builder.add_edge(START, "a")
builder.add_conditional_edges("a", route)
builder.add_edge("b", "a")
graph = builder.compile()

재귀 한도를 제어하려면 config에서 "recursion_limit"을 지정해요. 그러면 catch해서 처리할 수 있는 GraphRecursionError가 발생해요.

from langgraph.errors import GraphRecursionError

try:
    graph.invoke(inputs, {"recursion_limit": 3})
except GraphRecursionError:
    print("Recursion Error")

간단한 루프가 있는 그래프를 정의해 볼게요. 종료 조건을 구현하는 데 조건부 엣지를 사용한다는 점을 주목하세요.

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

class State(TypedDict):
    # The operator.add reducer fn makes this append-only
    aggregate: Annotated[list, operator.add]

def a(state: State):
    print(f'Node A sees {state["aggregate"]}')
    return {"aggregate": ["A"]}

def b(state: State):
    print(f'Node B sees {state["aggregate"]}')
    return {"aggregate": ["B"]}

# Define nodes
builder = StateGraph(State)
builder.add_node(a)
builder.add_node(b)

# Define edges
def route(state: State) -> Literal["b", END]:
    if len(state["aggregate"]) < 7:
        return "b"
    else:
        return END

builder.add_edge(START, "a")
builder.add_conditional_edges("a", route)
builder.add_edge("b", "a")
graph = builder.compile()

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))

이 구조는 ReAct 에이전트와 비슷해요. 노드 "a"가 도구 호출 모델이고, 노드 "b"가 도구를 나타내요.

route 조건부 엣지에서 상태의 "aggregate" 목록이 임계 길이를 넘으면 종료하도록 지정했어요.

그래프를 호출하면 종료 조건에 도달할 때까지 노드 "a"와 "b"를 오가며 교대하는 걸 볼 수 있어요.

graph.invoke({"aggregate": []})

Node A sees []
Node B sees ['A']
Node A sees ['A', 'B']
Node B sees ['A', 'B', 'A']
Node A sees ['A', 'B', 'A', 'B']
Node B sees ['A', 'B', 'A', 'B', 'A']
Node A sees ['A', 'B', 'A', 'B', 'A', 'B']

재귀 한도 부과 (Impose a recursion limit)

어떤 애플리케이션에서는 주어진 종료 조건에 도달할 것이라는 보장이 없을 수 있어요. 이런 경우 그래프의 재귀 한도를 설정할 수 있어요. 그러면 주어진 수의 슈퍼스텝 이후 GraphRecursionError가 발생해요. 이 예외를 catch해 처리할 수 있어요.

from langgraph.errors import GraphRecursionError

try:
    graph.invoke({"aggregate": []}, {"recursion_limit": 4})
except GraphRecursionError:
    print("Recursion Error")

Node A sees []
Node B sees ['A']
Node C sees ['A', 'B']
Node D sees ['A', 'B']
Node A sees ['A', 'B', 'C', 'D']
Recursion Error

확장 예시: 재귀 한도 도달 시 상태 반환

GraphRecursionError를 발생시키는 대신, 재귀 한도까지 남은 단계 수를 추적하는 새 키를 상태에 도입할 수 있어요. 그런 다음 이 키로 실행을 종료할지 결정할 수 있어요.

LangGraph는 특수 RemainingSteps 어노테이션을 구현해요. 내부적으로 ManagedValue 채널을 만들어요. 그 채널은 그래프 실행이 지속되는 동안만 존재하는 상태 채널이에요.

import operator
from typing import Annotated, Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.managed.is_last_step import RemainingSteps

class State(TypedDict):
    aggregate: Annotated[list, operator.add]
    remaining_steps: RemainingSteps

def a(state: State):
    print(f'Node A sees {state["aggregate"]}')
    return {"aggregate": ["A"]}

def b(state: State):
    print(f'Node B sees {state["aggregate"]}')
    return {"aggregate": ["B"]}

# Define nodes
builder = StateGraph(State)
builder.add_node(a)
builder.add_node(b)

# Define edges
def route(state: State) -> Literal["b", END]:
    if state["remaining_steps"] <= 2:
        return END
    else:
        return "b"

builder.add_edge(START, "a")
builder.add_conditional_edges("a", route)
builder.add_edge("b", "a")
graph = builder.compile()

# Test it out
result = graph.invoke({"aggregate": []}, {"recursion_limit": 4})
print(result)

Node A sees []
Node B sees ['A']
Node A sees ['A', 'B']
{'aggregate': ['A', 'B', 'A']}

확장 예시: 브랜치가 있는 루프

재귀 한도가 어떻게 동작하는지 더 잘 이해하기 위해 더 복잡한 예시를 살펴볼게요. 아래에서는 루프를 구현하되, 한 단계가 두 노드로 fan-out돼요.

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

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

def a(state: State):
    print(f'Node A sees {state["aggregate"]}')
    return {"aggregate": ["A"]}

def b(state: State):
    print(f'Node B sees {state["aggregate"]}')
    return {"aggregate": ["B"]}

def c(state: State):
    print(f'Node C sees {state["aggregate"]}')
    return {"aggregate": ["C"]}

def d(state: State):
    print(f'Node D sees {state["aggregate"]}')
    return {"aggregate": ["D"]}

# Define nodes
builder = StateGraph(State)
builder.add_node(a)
builder.add_node(b)
builder.add_node(c)
builder.add_node(d)

# Define edges
def route(state: State) -> Literal["b", END]:
    if len(state["aggregate"]) < 7:
        return "b"
    else:
        return END

builder.add_edge(START, "a")
builder.add_conditional_edges("a", route)
builder.add_edge("b", "c")
builder.add_edge("b", "d")
builder.add_edge(["c", "d"], "a")
graph = builder.compile()

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))

이 그래프는 복잡해 보이지만 슈퍼스텝의 루프로 개념화할 수 있어요.

Node A
Node B
Nodes C and D
Node A
...

네 개의 슈퍼스텝 루프가 있고, 노드 C와 D는 동시에 실행돼요. 목록 형태의 엣지는 "a"로 돌아가기 전에 "c"와 "d"를 모두 기다려요. 목록 형태 엣지가 같은 노드로 들어가는 별도 엣지와 어떻게 다른지는 노드 실행 지연을 참고해요.

전처럼 그래프를 호출하면 종료 조건에 도달하기 전에 두 번의 완전한 "바퀴(lap)"를 완주하는 걸 볼 수 있어요.

result = graph.invoke({"aggregate": []})

Node A sees []
Node B sees ['A']
Node D sees ['A', 'B']
Node C sees ['A', 'B']
Node A sees ['A', 'B', 'C', 'D']
Node B sees ['A', 'B', 'C', 'D', 'A']
Node D sees ['A', 'B', 'C', 'D', 'A', 'B']
Node C sees ['A', 'B', 'C', 'D', 'A', 'B']
Node A sees ['A', 'B', 'C', 'D', 'A', 'B', 'C', 'D']

하지만 재귀 한도를 4로 설정하면 각 바퀴가 네 개의 슈퍼스텝이므로 한 바퀴만 완주해요.

from langgraph.errors import GraphRecursionError

try:
    result = graph.invoke({"aggregate": []}, {"recursion_limit": 4})
except GraphRecursionError:
    print("Recursion Error")

Node A sees []
Node B sees ['A']
Node C sees ['A', 'B']
Node D sees ['A', 'B']
Node A sees ['A', 'B', 'C', 'D']
Recursion Error

비동기 (Async)

비동기 프로그래밍 패러다임을 사용하면 IO 바운드 코드를 동시에 실행할 때(예: 채팅 모델 프로바이더에 동시 API 요청) 상당한 성능 향상을 얻을 수 있어요.

그래프의 sync 구현을 async 구현으로 변환하려면 다음을 해야 해요.

nodes를 def 대신 async def로 갱신.
내부 코드를 await를 적절히 사용하도록 갱신.
원하는 대로 .ainvoke 또는 .astream으로 그래프를 호출.

많은 LangChain 객체가 모든 sync 메서드의 async 변형을 가진 Runnable Protocol을 구현하므로, 보통 sync 그래프를 async 그래프로 업그레이드하는 건 꽤 빠르다.

아래 예시를 보세요. 기본 LLM의 async 호출을 시연하기 위해 채팅 모델을 포함할게요.

OpenAI

👉 OpenAI 채팅 모델 통합 문서를 읽어보세요.

pip install -U "langchain[openai]"

uv를 쓴다면:

uv add "langchain[openai]"

import os
from langchain.chat_models import init_chat_model

os.environ["OPENAI_API_KEY"] = "sk-..."

model = init_chat_model("gpt-5.5")

또는 모델 클래스 직접:

import os
from langchain_openai import ChatOpenAI

os.environ["OPENAI_API_KEY"] = "sk-..."

model = ChatOpenAI(model="gpt-5.5")

Anthropic

👉 Anthropic 채팅 모델 통합 문서를 읽어보세요.

pip install -U "langchain[anthropic]"

uv를 쓴다면:

uv add "langchain[anthropic]"

import os
from langchain.chat_models import init_chat_model

os.environ["ANTHROPIC_API_KEY"] = "sk-..."

model = init_chat_model("claude-sonnet-4-6")

또는 모델 클래스 직접:

import os
from langchain_anthropic import ChatAnthropic

os.environ["ANTHROPIC_API_KEY"] = "sk-..."

model = ChatAnthropic(model="claude-sonnet-4-6")

Azure

👉 Azure 채팅 모델 통합 문서를 읽어보세요.

pip install -U "langchain[openai]"

uv를 쓴다면:

uv add "langchain[openai]"

import os
from langchain.chat_models import init_chat_model

os.environ["AZURE_OPENAI_API_KEY"] = "..."
os.environ["AZURE_OPENAI_ENDPOINT"] = "..."
os.environ["OPENAI_API_VERSION"] = "2025-03-01-preview"

model = init_chat_model(
    "azure_openai:gpt-5.5",
    azure_deployment=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"],
)

또는 모델 클래스 직접:

import os
from langchain_openai import AzureChatOpenAI

os.environ["AZURE_OPENAI_API_KEY"] = "..."
os.environ["AZURE_OPENAI_ENDPOINT"] = "..."
os.environ["OPENAI_API_VERSION"] = "2025-03-01-preview"

model = AzureChatOpenAI(
    model="gpt-5.5",
    azure_deployment=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"]
)

Google Gemini

👉 Google GenAI 채팅 모델 통합 문서를 읽어보세요.

pip install -U "langchain[google-genai]"

uv를 쓴다면:

uv add "langchain[google-genai]"

import os
from langchain.chat_models import init_chat_model

os.environ["GOOGLE_API_KEY"] = "..."

model = init_chat_model("google_genai:gemini-3.7-flash")

또는 모델 클래스 직접:

import os
from langchain_google_genai import ChatGoogleGenerativeAI

os.environ["GOOGLE_API_KEY"] = "..."

model = ChatGoogleGenerativeAI(model="gemini-3.7-flash")

AWS Bedrock

👉 AWS Bedrock 채팅 모델 통합 문서를 읽어보세요.

pip install -U "langchain[aws]"

uv를 쓴다면:

uv add "langchain[aws]"

from langchain.chat_models import init_chat_model

# Follow the steps here to configure your credentials:
# https://docs.aws.amazon.com/bedrock/latest/userguide/getting-started.html

model = init_chat_model(
    "us.anthropic.claude-sonnet-4-6",
    model_provider="bedrock_converse",
)

또는 모델 클래스 직접:

from langchain_aws import ChatBedrock

model = ChatBedrock(model="us.anthropic.claude-sonnet-4-6")

HuggingFace

👉 HuggingFace 채팅 모델 통합 문서를 읽어보세요.

pip install -U "langchain[huggingface]"

uv를 쓴다면:

uv add "langchain[huggingface]"

import os
from langchain.chat_models import init_chat_model

os.environ["HUGGINGFACEHUB_API_TOKEN"] = "hf_..."

model = init_chat_model(
    "microsoft/Phi-3-mini-4k-instruct",
    model_provider="huggingface",
    temperature=0.7,
    max_tokens=1024,
)

또는 모델 클래스 직접:

import os
from langchain_huggingface import ChatHuggingFace, HuggingFaceEndpoint

os.environ["HUGGINGFACEHUB_API_TOKEN"] = "hf_..."

llm = HuggingFaceEndpoint(
    repo_id="microsoft/Phi-3-mini-4k-instruct",
    temperature=0.7,
    max_length=1024,
)
model = ChatHuggingFace(llm=llm)

OpenRouter

👉 OpenRouter 채팅 모델 통합 문서를 읽어보세요.

pip install -U "langchain-openrouter"

uv를 쓴다면:

uv add "langchain-openrouter"

import os
from langchain.chat_models import init_chat_model

os.environ["OPENROUTER_API_KEY"] = "sk-..."

model = init_chat_model(
    "auto",
    model_provider="openrouter",
)

또는 모델 클래스 직접:

import os
from langchain_openrouter import ChatOpenRouter

os.environ["OPENROUTER_API_KEY"] = "sk-..."

model = ChatOpenRouter(model="auto")

from langchain.chat_models import init_chat_model
from langgraph.graph import MessagesState, StateGraph

async def node(state: MessagesState):  # [!code highlight]
    new_message = await llm.ainvoke(state["messages"])  # [!code highlight]
    return {"messages": [new_message]}

builder = StateGraph(MessagesState).add_node(node).set_entry_point("node")
graph = builder.compile()

input_message = {"role": "user", "content": "Hello"}
result = await graph.ainvoke({"messages": [input_message]})  # [!code highlight]

비동기 스트리밍 (Async streaming) 비동기와 함께 스트리밍하는 예시는 스트리밍 가이드를 참고해요.

`Command`로 제어 흐름과 상태 업데이트 결합하기 (Combine control flow and state updates with Command)

제어 흐름(엣지)과 상태 업데이트(노드)를 결합하는 것이 유용할 수 있어요. 예를 들어 같은 노드에서 상태 업데이트를 수행하고 다음에 갈 노드를 결정하고 싶을 수 있어요. LangGraph는 노드 함수에서 Command 객체를 반환해 이렇게 할 수 있게 해줘요.

def my_node(state: State) -> Command[Literal["my_other_node"]]:
    return Command(
        # state update
        update={"foo": "bar"},
        # control flow
        goto="my_other_node"
    )

아래에서 end-to-end 예시를 보여줄게요. A, B, C 세 노드로 이루어진 간단한 그래프를 만들 거예요. 먼저 노드 A를 실행한 다음, 노드 A의 출력에 따라 다음에 노드 B로 갈지 C로 갈지 결정해요.

import random
from typing_extensions import TypedDict, Literal
from langgraph.graph import StateGraph, START
from langgraph.types import Command

# Define graph state
class State(TypedDict):
    foo: str

# Define the nodes

def node_a(state: State) -> Command[Literal["node_b", "node_c"]]:
    print("Called A")
    value = random.choice(["b", "c"])
    # this is a replacement for a conditional edge function
    if value == "b":
        goto = "node_b"
    else:
        goto = "node_c"

    # note how Command allows you to BOTH update the graph state AND route to the next node
    return Command(
        # this is the state update
        update={"foo": value},
        # this is a replacement for an edge
        goto=goto,
    )

def node_b(state: State):
    print("Called B")
    return {"foo": state["foo"] + "b"}

def node_c(state: State):
    print("Called C")
    return {"foo": state["foo"] + "c"}

이제 위의 노드들로 StateGraph를 만들 수 있어요. 라우팅을 위한 조건부 엣지가 없다는 점을 주목하세요! 이것은 제어 흐름이 node_a 안의 Command로 정의되기 때문이에요.

builder = StateGraph(State)
builder.add_edge(START, "node_a")
builder.add_node(node_a)
builder.add_node(node_b)
builder.add_node(node_c)
# NOTE: there are no edges between nodes A, B and C!

graph = builder.compile()

Command를 반환 타입 어노테이션으로 쓴 것을 눈치챘을 거예요. 예: Command[Literal["node_b", "node_c"]]. 이것은 그래프 렌더링에 필요하며, LangGraph에 node_a가 node_b와 node_c로 이동할 수 있음을 알려줘요.

from IPython.display import display, Image

display(Image(graph.get_graph().draw_mermaid_png()))

그래프를 여러 번 실행하면 노드 A의 무작위 선택에 따라 서로 다른 경로(A -> B 또는 A -> C)를 취하는 걸 볼 수 있어요.

graph.invoke({"foo": ""})

Called A
Called C

부모 그래프의 노드로 이동 (Navigate to a node in a parent graph)

서브그래프를 사용한다면 서브그래프 안의 노드에서 다른 서브그래프(즉 부모 그래프의 다른 노드)로 이동하고 싶을 수 있어요. 그러려면 Command에서 graph=Command.PARENT를 지정할 수 있어요.

def my_node(state: State) -> Command[Literal["other_subgraph"]]:
    return Command(
        update={"foo": "bar"},
        goto="other_subgraph",  # where `other_subgraph` is a node in the parent graph
        graph=Command.PARENT
    )

위의 예시를 사용해 시연해 볼게요. 예시의 nodeA를 단일 노드 그래프로 바꾸고, 부모 그래프에 서브그래프로 추가할 거예요.

Command.PARENT로 상태 업데이트 (State updates with Command.PARENT) 서브그래프 노드에서 부모 그래프 노드로, 부모와 서브그래프의 상태 스키마가 공유하는 키에 대한 업데이트를 보낼 때는, 부모 그래프 상태에서 갱신하는 키에 대해 리듀서를 반드시 정의해야 해요. 아래 예시를 보세요.

import operator
from typing_extensions import Annotated

class State(TypedDict):
    # NOTE: we define a reducer here
    foo: Annotated[str, operator.add]  # [!code highlight]

def node_a(state: State):
    print("Called A")
    value = random.choice(["a", "b"])
    # this is a replacement for a conditional edge function
    if value == "a":
        goto = "node_b"
    else:
        goto = "node_c"

    # note how Command allows you to BOTH update the graph state AND route to the next node
    return Command(
        update={"foo": value},
        goto=goto,
        # this tells LangGraph to navigate to node_b or node_c in the parent graph
        # NOTE: this will navigate to the closest parent graph relative to the subgraph
        graph=Command.PARENT,  # [!code highlight]
    )

subgraph = StateGraph(State).add_node(node_a).add_edge(START, "node_a").compile()

def node_b(state: State):
    print("Called B")
    # NOTE: since we've defined a reducer, we don't need to manually append
    # new characters to existing 'foo' value. instead, reducer will append these
    # automatically (via operator.add)
    return {"foo": "b"}  # [!code highlight]

def node_c(state: State):
    print("Called C")
    return {"foo": "c"}  # [!code highlight]

builder = StateGraph(State)
builder.add_edge(START, "subgraph")
builder.add_node("subgraph", subgraph)
builder.add_node(node_b)
builder.add_node(node_c)

graph = builder.compile()

graph.invoke({"foo": ""})

Called A
Called C

도구 안에서 사용 (Use inside tools)

도구 안에서 그래프 상태를 갱신하는 것은 흔한 사용 사례예요. 예를 들어 고객 지원 애플리케이션에서 대화 초반에 계정 번호나 ID로 고객 정보를 조회하고 싶을 수 있어요. 도구에서 그래프 상태를 갱신하려면 도구에서 Command(update={"my_custom_key": "foo", "messages": [...]})를 반환하면 돼요.

from langchain.tools import ToolRuntime

@tool
def lookup_user_info(runtime: ToolRuntime):
    """Use this to look up user information to better assist them with their questions."""
    user_info = get_user_info(runtime.server_info.user.identity)  # [!code highlight]
    return Command(
        update={
            # update the state keys
            "user_info": user_info,
            # update the message history
            "messages": [ToolMessage("Successfully looked up user information", tool_call_id=runtime.tool_call_id)]
        }
    )

도구에서 Command를 반환할 때는 Command.update에 messages(또는 메시지 이력에 사용되는 아무 상태 키나)를 반드시 포함하고, messages의 메시지 목록은 ToolMessage를 반드시 포함해야 해요. 이것은 결과 메시지 이력이 유효하도록 하기 위해 필요해요(LLM 프로바이더는 도구 호출이 있는 AI 메시지 뒤에 도구 결과 메시지가 따르도록 요구해요).

Command로 상태를 갱신하는 도구를 쓰고 있다면, Command 객체를 반환하는 도구를 자동으로 처리하고 그래프 상태로 전파하는 prebuilt ToolNode를 사용하는 걸 권장해요. 도구를 호출하는 커스텀 노드를 직접 작성한다면, 도구가 반환한 Command 객체를 노드의 업데이트로 수동 전파해야 해요.

그래프 시각화 (Visualize your graph)

여기서 만든 그래프를 시각화하는 방법을 보여줄게요.

StateGraph를 포함한 어떤 Graph든 시각화할 수 있어요.

프랙탈을 그려서 재미를 좀 볼게요. :)

import random
from typing import Annotated, Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

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

class MyNode:
    def __init__(self, name: str):
        self.name = name
    def __call__(self, state: State):
        return {"messages": [("assistant", f"Called node {self.name}")]}

def route(state) -> Literal["entry_node", END]:
    if len(state["messages"]) > 10:
        return END
    return "entry_node"

def add_fractal_nodes(builder, current_node, level, max_level):
    if level > max_level:
        return
    # Number of nodes to create at this level
    num_nodes = random.randint(1, 3)  # Adjust randomness as needed
    for i in range(num_nodes):
        nm = ["A", "B", "C"][i]
        node_name = f"node_{current_node}_{nm}"
        builder.add_node(node_name, MyNode(node_name))
        builder.add_edge(current_node, node_name)
        # Recursively add more nodes
        r = random.random()
        if r > 0.2 and level + 1 < max_level:
            add_fractal_nodes(builder, node_name, level + 1, max_level)
        elif r > 0.05:
            builder.add_conditional_edges(node_name, route, node_name)
        else:
            # End
            builder.add_edge(node_name, END)

def build_fractal_graph(max_level: int):
    builder = StateGraph(State)
    entry_point = "entry_node"
    builder.add_node(entry_point, MyNode(entry_point))
    builder.add_edge(START, entry_point)
    add_fractal_nodes(builder, entry_point, 1, max_level)
    # Optional: set a finish point if required
    builder.add_edge(entry_point, END)  # or any specific node
    return builder.compile()

app = build_fractal_graph(3)

Mermaid

그래프 클래스를 Mermaid 문법으로 변환할 수도 있어요.

print(app.get_graph().draw_mermaid())

%%{init: {'flowchart': {'curve': 'linear'}}}%%
graph TD;
    tart__([<p>__start__</p>]):::first
    ry_node(entry_node)
    e_entry_node_A(node_entry_node_A)
    e_entry_node_B(node_entry_node_B)
    e_node_entry_node_B_A(node_node_entry_node_B_A)
    e_node_entry_node_B_B(node_node_entry_node_B_B)
    e_node_entry_node_B_C(node_node_entry_node_B_C)
    nd__([<p>__end__</p>]):::last
    tart__ --> entry_node;
    ry_node --> __end__;
    ry_node --> node_entry_node_A;
    ry_node --> node_entry_node_B;
    e_entry_node_B --> node_node_entry_node_B_A;
    e_entry_node_B --> node_node_entry_node_B_B;
    e_entry_node_B --> node_node_entry_node_B_C;
    e_entry_node_A -.-> entry_node;
    e_entry_node_A -.-> __end__;
    e_node_entry_node_B_A -.-> entry_node;
    e_node_entry_node_B_A -.-> __end__;
    e_node_entry_node_B_B -.-> entry_node;
    e_node_entry_node_B_B -.-> __end__;
    e_node_entry_node_B_C -.-> entry_node;
    e_node_entry_node_B_C -.-> __end__;
    ssDef default fill:#f2f0ff,line-height:1.2
    ssDef first fill-opacity:0
    ssDef last fill:#bfb6fc

PNG

원한다면 그래프를 .png로 렌더링할 수 있어요. 여기서 세 가지 옵션을 사용할 수 있어요.

Mermaid.ink API 사용 (추가 패키지 불필요)
Mermaid + Pyppeteer 사용 (pip install pyppeteer 필요)
graphviz 사용 (pip install graphviz 필요)

Mermaid.Ink 사용

기본적으로 draw_mermaid_png()는 Mermaid.Ink의 API를 사용해 다이어그램을 생성해요.

from IPython.display import Image, display
from langchain_core.runnables.graph import CurveStyle, MermaidDrawMethod, NodeStyles

display(Image(app.get_graph().draw_mermaid_png()))

Mermaid + Pyppeteer 사용

import nest_asyncio

nest_asyncio.apply()  # Required for Jupyter Notebook to run async functions

display(
    Image(
        app.get_graph().draw_mermaid_png(
            curve_style=CurveStyle.LINEAR,
            node_colors=NodeStyles(first="#ffdfba", last="#baffc9", default="#fad7de"),
            wrap_label_n_words=9,
            output_file_path=None,
            draw_method=MermaidDrawMethod.PYPPETEER,
            background_color="white",
            padding=10,
        )
    )
)

Graphviz 사용

try:
    display(Image(app.get_graph().draw_png()))
except ImportError:
    print(
        "You likely need to install dependencies for pygraphviz, see more here https://github.com/pygraphviz/pygraphviz/blob/main/INSTALL.txt"
    )

더 알아보기 (Learn more)

상태: 상태 정의·리듀서·스키마의 개념을 다뤄요.
서브그래프: 서브그래프 사용과 Command.PARENT를 다뤄요.
스트리밍: v2 스트리밍 형식과 스트림 모드를 다뤄요.
Fault tolerance: 재시도 정책·타임아웃·오류 처리·그래프 기본값을 다뤄요.

Graph API 사용하기