그래프 API 사용하기 (Use the Graph API)

그래프 API 사용하기 (Use the Graph API)

이 가이드는 LangGraph의 그래프 API(Graph API) 기초를 다룹니다. 상태(state)를 정의하고 갱신하는 방법부터, 시퀀스·브랜치·루프 같은 흔한 그래프 구조를 어떻게 조합하는지까지 차근차근 살펴볼게요. 또 LangGraph의 제어 기능, 즉 map-reduce 워크플로를 위한 Send API와, 상태 갱신과 "점프(hop)"를 함께 처리하는 Command API도 함께 다룹니다.

설정 (Setup)

langgraph를 설치합니다:

pip install -U langgraph

uv add langgraph

더 나은 디버깅을 위해 LangSmith 설정하기

LangSmith에 가입하면 LangGraph 프로젝트의 문제를 빠르게 찾고 성능을 개선할 수 있어요. LangSmith는 LangGraph로 만든 LLM 앱을 트레이스 데이터를 활용해 디버깅하고, 테스트하고, 모니터링하게 해 줍니다. 시작하는 방법은 문서에서 더 자세히 볼 수 있어요.

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

여기서는 LangGraph에서 상태(state)를 정의하고 갱신하는 법을 보여드릴게요. 크게 두 가지를 다룹니다:

1. 상태를 사용해 그래프의 스키마(schema)를 정의하는 법
2. 리듀서(reducers)를 사용해 상태 갱신이 처리되는 방식을 제어하는 법

상태 정의 (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

이 상태는 메시지 객체 리스트와, 추가 정수 필드 하나를 추적합니다.

상태 갱신 (Update state)

단일 노드를 가진 예시 그래프를 만들어 볼게요. 우리의 노드(node)는 그저 그래프의 상태를 읽고 갱신하는 파이썬 함수일 뿐이에요. 이 함수의 첫 번째 인자는 항상 상태입니다:

from langchain.messages import AIMessage

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

이 노드는 메시지 리스트에 메시지 하나를 추가하고, extra 필드 하나를 채워 넣습니다.

경고

노드는 상태를 직접 변경(mutate)하기보다, 상태에 대한 갱신을 바로 반환해야 합니다.

이제 이 노드를 포함하는 간단한 그래프를 정의해 볼게요. 이 상태에서 동작하는 그래프를 정의하려면 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}

이때 몇 가지 짚고 갈 점이 있어요:

• 상태의 키 하나만 갱신하면서 호출을 시작했어요.
• 호출 결과로 상태 전체를 받아요.

편의상 메시지 객체의 내용을 pretty-print로 자주 확인합니다:

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

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

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

Hello!

리듀서로 상태 갱신 처리하기 (Process state updates with reducers)

상태의 각 키는 자신만의 독립적인 리듀서(reducer) 함수를 가질 수 있는데, 이 함수가 노드의 갱신이 어떻게 적용되는지를 제어합니다. 리듀서 함수를 명시적으로 지정하지 않으면, 해당 키에 대한 모든 갱신이 기존 값을 덮어쓴다고 가정합니다.

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 형식 같은 메시지 형식의 축약형을 받아들이고 싶을 수 있어요.

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!

이것은 챗 모델(chat models)을 활용하는 애플리케이션에 아주 유연한 상태 표현이에요. LangGraph는 편의를 위해 사전 구성된 MessagesState도 제공해서, 다음과 같이 쓸 수 있습니다:

from langgraph.graph import MessagesState

class State(MessagesState):
    extra_field: int

리듀서 우회하기: Overwrite

어떤 경우에는 리듀서를 우회하고 상태 값을 직접 덮어써야 할 수도 있어요. LangGraph는 이런 용도를 위해 Overwrite 타입을 제공합니다. 노드가 Overwrite로 감싼 값을 반환하면, 리듀서는 우회되고 채널(channel)이 그 값으로 직접 설정됩니다.

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

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"]}}

경고

노드가 병렬로 실행될 때, 같은 super-step 안에서 같은 상태 키에 Overwrite를 사용할 수 있는 노드는 하나뿐입니다. 같은 super-step에서 여러 노드가 같은 키를 덮어쓰려 하면 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 인자를 받는데, 이 인자가 그래프 안의 노드들이 접근하고 갱신할 수 있는 상태의 "형태"를 지정합니다.

우리 예시들에서는 보통 파이썬 네이티브 TypedDict나 dataclass를 state_schema로 쓰지만, state_schema는 어떤 type이든 될 수 있어요.

여기서는 Pydantic BaseModel을 state_schema로 사용해서 입력에 런타임 검증(run-time validation)을 추가하는 방법을 볼게요.

알려진 제한 사항

• 현재 그래프의 출력은 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

Pydantic 모델 상태의 추가 기능들은 아래에서 볼 수 있어요.

직렬화 동작 (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 메시지 타입으로 작업할 때는 직렬화에 중요한 고려 사항이 있어요. 메시지 객체를 네트워크를 통해 전달할 때는 올바른 직렬화/역직렬화를 위해 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이나 시스템 프롬프트를 사용할지 지정하고 싶을 수 있죠.

런타임 구성을 추가하려면:

1. 구성에 대한 스키마를 지정합니다.
2. 노드 또는 조건부 엣지의 함수 시그니처에 구성을 추가합니다.
3. 구성을 그래프에 전달합니다.

간단한 예시를 볼게요:

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를 호출하거나, 데이터베이스를 조회하거나, LLM을 호출할 때 그렇죠. LangGraph는 노드에 재시도 정책을 추가할 수 있게 해 줍니다.

재시도 정책을 구성하려면 retry_policy 매개변수를 add_node에 전달합니다. 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 노드에서만 지원돼요. sync 노드에 timeout을 설정하면, sync 파이썬 실행은 프로세스 내에서 안전하게 취소할 수 없기 때문에 그래프를 컴파일할 때 LangGraph가 오류를 발생시킵니다.

노드가 타임아웃을 초과하면 LangGraph는 NodeTimeoutError를 발생시키는데, 이것은 파이썬 내장 TimeoutError의 하위 클래스예요. 노드에 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,
)

보상(compensation) 패턴과 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 기본값은 일반 노드에만 적용됩니다—핸들러는 결코 스스로를 잡지 않으며, 핸들러 결과를 캐싱하는 것은 안전하지 않기 때문이에요. 기본값은 서브그래프에는 상속되지 않습니다.

전체 우선순위 규칙과 적용 가능성 표는 Fault tolerance를 참고하세요.

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

runtime.execution_info를 통해 실행 ID와 재시도 정보에 접근할 수 있어요. 이는 config에서 직접 읽지 않고도 스레드, 런, 체크포인트 식별자와 재시도 상태를 드러냅니다.

Attribute	Type	Description
thread_id	str | None	현재 실행의 Thread ID. 체크포인터가 없으면 None입니다.
run_id	str | None	현재 실행의 Run ID. config에 제공되지 않으면 None입니다.
checkpoint_id	str	현재 실행의 Checkpoint ID.
checkpoint_ns	str	현재 실행의 Checkpoint namespace.
task_id	str	현재 실행의 Task ID.
node_attempt	int	현재 실행 시도 번호(1부터 시작). 첫 시도는 1, 첫 재시도는 2 등.
node_first_attempt_time	float | None	첫 시도가 시작된 Unix 타임스탬프(초). 재시도해도 동일하게 유지됩니다.

스레드 및 런 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 키에서 직접 읽지 않고도 어시스턴트 ID, 그래프 ID, 인증된 사용자를 드러냅니다.

Attribute	Type	Description
assistant_id	str	현재 배포의 어시스턴트 ID.
graph_id	str	현재 배포의 그래프 ID.
user	BaseUser | None	custom auth가 구성된 경우 인증된 사용자.

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)이 필요합니다.

노드 내부에서 drain 상태 접근 (Access drain state inside a node)

정상 종료(graceful shutdown)가 요청되면 runtime.drain_requested는 True가 됩니다. 노드 내부에서 이를 읽어 다음 superstep 경계 전에 비싼 작업을 건너뛸 수 있어요:

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	이 런에 RunControl.request_drain()이 호출되었으면 True.
drain_reason	str | None	request_drain()에 전달된 이유 문자열. drain이 요청되지 않았으면 None.

langgraph>=1.2가 필요합니다. 전체 RunControl API는 Graceful shutdown을 참고하세요.

노드 캐싱 추가 (Add node caching)

노드 캐싱은 비싼(시간이나 비용 측면에서) 작업을 반복하지 않으려 할 때 유용해요. LangGraph는 그래프의 노드에 개별화된 캐싱 정책을 추가할 수 있게 해 줍니다.

캐시 정책을 구성하려면 cache_policy 매개변수를 add_node 함수에 전달합니다. 다음 예시에서 CachePolicy 객체는 120초의 TTL과 기본 key_func 생성기로 인스턴스화되고, 노드에 연결됩니다:

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)

전제 조건

이 가이드는 위의 상태(state) 섹션에 익숙하다고 가정합니다.

여기서는 간단한 단계 시퀀스를 만드는 방법을 보여드릴게요. 다음을 다룹니다:

1. 순차 그래프(sequential graph)를 만드는 법
2. 비슷한 그래프를 만드는 내장 축약형(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는 애플리케이션에 기본 영속 계층(persistence layer)을 쉽게 추가할 수 있게 해 줘요. 이를 통해 노드 실행 사이에 상태를 체크포인트할 수 있고, 그래서 LangGraph 노드들은 다음을 담당하게 됩니다:

• 상태 갱신이 어떻게 체크포인트되는지
• human-in-the-loop 워크플로에서 인터럽션이 어떻게 재개되는지
• LangGraph의 time travel 기능으로 실행을 어떻게 "되감고" 분기할 수 있는지

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

엔드투엔드 예시를 보여드릴게요. 세 단계로 이루어진 시퀀스를 만들 거예요:

1. 상태의 한 키에 값을 채웁니다.
2. 같은 값을 갱신합니다.
3. 다른 값을 채웁니다.

먼저 우리의 상태(state)를 정의해 볼게요. 이는 그래프의 스키마를 결정하고, 갱신이 어떻게 적용될지도 지정할 수 있어요. 더 자세한 내용은 "리듀서로 상태 갱신 처리하기"를 참고하세요.

이 경우 두 개의 값만 추적할 거예요:

from typing_extensions import TypedDict

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

우리의 노드들은 그저 그래프의 상태를 읽고 갱신하는 파이썬 함수들이에요. 이 함수의 첫 번째 인자는 항상 상태입니다:

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}

상태에 갱신을 보낼 때 각 노드는 갱신하고자 하는 키의 값만 지정하면 됩니다.

기본적으로 이는 해당 키의 값을 덮어씁니다. 리듀서(reducers)를 사용해 갱신이 처리되는 방식을 제어할 수도 있어요—예를 들어 키에 연속된 갱신을 덧붙이는 방식으로요. 자세한 내용은 "리듀서로 상태 갱신 처리하기"를 참고하세요.

마지막으로 그래프를 정의합니다. 이 상태에서 동작하는 그래프를 정의하려면 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")

커스텀 이름 지정하기

add_node를 사용해 노드에 커스텀 이름을 지정할 수 있어요:

builder.add_node("my_node", step_1)

여기서 몇 가지 집고 갈 점이 있어요:

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

다음으로 그래프를 컴파일(compile)합니다. 이는 그래프 구조에 대한 몇 가지 기본 검사(예: 고아 노드 식별)를 수행해요. 애플리케이션에 체크포인터를 통해 영속성을 추가한다면, 그것도 여기에 전달할 거예요.

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}

여기서 짚고 갈 점:

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

내장 축약형

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와 C로 fan-out하고, 다시 D로 fan-in합니다. 우리 상태에서는 리듀서 add 연산을 지정해요. 이는 State의 특정 키에 대한 값을 단순히 덮어쓰는 대신 결합하거나 누적합니다. 리스트의 경우 새 리스트를 기존 리스트와 연결(concatenating)한다는 뜻이에요. 리듀서로 상태를 갱신하는 자세한 내용은 위의 상태 리듀서 섹션을 참고하세요.

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"가 모두 끝난 후에 실행됩니다.

중요하게도, 병렬 superstep의 갱신 순서는 일관되지 않을 수 있어요. 병렬 superstep의 갱신에 일관되고 미리 결정된 순서가 필요하다면, 출력을 상태의 별도 필드에 정렬 값을 함께 작성해야 합니다.

예외 처리에 대해서는요?

LangGraph는 노드들을 superstep 안에서 실행하는데, 병렬 브랜치들이 병렬로 실행되는 동안에도 전체 superstep은 트랜잭션적입니다. 어떤 브랜치라도 예외를 일으키면, 상태에는 아무 갱신도 적용되지 않습니다(전체 superstep이 오류가 됩니다).

중요하게도, 체크포인터를 사용할 때 superstep 안에서 성공한 노드의 결과는 저장되며, 재개될 때 반복되지 않아요.

오류가 발생하기 쉬운(아마도 불안정한 API 호출을 처리하고 싶은) 상황이라면 LangGraph는 두 가지 방법을 제공합니다:

1. 노드 안에 일반 파이썬 코드를 작성해 예외를 잡아 처리할 수 있어요.
2. 특정 타입의 예외를 일으키는 노드를 재시도하도록 지시하는 **RetryPolicy**를 설정할 수 있어요. 실패한 브랜치만 재시도되므로, 중복 작업을 수행할 걱정은 없어요.

이 둘을 함께 쓰면 병렬 실행을 하면서도 예외 처리를 완전히 제어할 수 있습니다.

최대 동시성 설정

그래프를 호출할 때 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"는 같은 superstep에서 동시에 실행됩니다. 노드 "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가 그것들을 소비하지는 않습니다.
• 별도의 엣지들은 들어오는 브랜치가 완료되는 각 superstep마다 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)를 설정할 수도 있어요. 재귀 한도는 오류를 발생시키기 전에 그래프가 실행할 수 있는 super-step의 수를 설정합니다. 재귀 한도 개념에 대해 더 읽어보세요.

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

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

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

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"을 지정합니다. 이는 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)

어떤 애플리케이션에서는 주어진 종료 조건에 도달할 것이라는 보장이 없을 수 있어요. 그런 경우 그래프의 재귀 한도를 설정할 수 있습니다. 이는 일정 수의 superstep 후에 GraphRecursionError를 발생시킵니다. 그런 다음 이 예외를 잡아 처리할 수 있어요:

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()))

이 그래프는 복잡해 보이지만, superstep의 루프로 개념화할 수 있어요:

1. Node A
2. Node B
3. Node C와 D
4. Node A
5. ...

네 개의 superstep으로 이루어진 루프가 있고, 노드 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로 설정하면, 각 랩이 네 개의 superstep이므로 랩 한 바퀴만 완료합니다:

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-bound 코드(예: 챗 모델 제공자에 동시 API 요청 보내기)를 동시에 실행할 때 상당한 성능 향상을 얻을 수 있어요.

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

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

많은 LangChain 객체가 모든 sync 메서드의 async 변형을 가진 Runnable Protocol을 구현하기 때문에, sync 그래프를 async 그래프로 업그레이드하는 것은 대개 꽤 빠릅니다.

아래 예시를 볼게요. 기본 LLM의 비동기 호출을 보여드리기 위해 챗 모델을 포함할게요:

OpenAI

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

pip install -U "langchain[openai]"

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 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 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 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 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 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 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와 스트리밍 예시는 스트리밍 가이드를 참고하세요.

제어 흐름과 상태 갱신 결합하기: 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"
    )

아래에서 엔드투엔드 예시를 보여드릴게요. A, B, C 세 개의 노드로 이루어진 간단한 그래프를 만들 거예요. 먼저 노드 A를 실행하고, 그런 다음 노드 A의 출력에 따라 다음으로 Node B로 갈지 Node 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"]]. 이것은 그래프 렌더링에 필요하며, node_a가 node_b와 node_c로 이동할 수 있음을 LangGraph에 알려 줍니다.

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)

서브그래프(subgraphs)를 사용한다면, 서브그래프 안의 노드에서 다른 서브그래프(즉 부모 그래프의 다른 노드)로 이동하고 싶을 수 있어요. 그렇게 하려면 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로 상태 갱신하기

서브그래프 노드에서 부모 그래프 노드로, 부모와 서브그래프 상태 스키마가 모두 공유하는 키에 대한 갱신을 보낼 때는, 부모 그래프 상태에서 그 키에 대한 리듀서를 반드시 정의해야 합니다. 아래 예시를 참고하세요.

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 객체를 반환하는 도구들을 자동으로 처리하고 그래프 상태로 전파하는 사전 구성된 ToolNode를 사용하는 것을 권장해요. 도구를 호출하는 커스텀 노드를 직접 작성한다면, 도구들이 반환한 Command 객체를 노드의 갱신으로 수동 전파해야 합니다.

그래프 시각화하기 (Visualize your graph)

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

StateGraph를 포함해 임의의 Graph를 시각화할 수 있어요.

재미 삼아 프랙탈(fractal)을 그려 보며 시작해 볼게요 :).

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"
    )