그래프 API 개요 (Graph API Overview)

그래프(Graphs)

LangGraph의 핵심은 에이전트 워크플로를 **그래프(graph)**로 모델링하는 것입니다. 에이전트의 동작은 다음 세 가지 핵심 요소로 정의합니다.

State(상태): 애플리케이션의 현재 스냅샷을 나타내는 공유 데이터 구조입니다. 어떤 데이터 타입이든 될 수 있지만, 보통 공유 상태 스키마(shared state schema)로 정의합니다.
Nodes(노드): 에이전트의 로직을 담은 함수입니다. 현재 상태를 입력으로 받아 어떤 연산이나 부수 효과(side-effect)를 수행하고, 갱신된 상태를 반환합니다.
Edges(엣지): 현재 상태를 바탕으로 다음에 어떤 Node를 실행할지 결정하는 함수입니다. 조건부 분기(conditional branch)일 수도 있고 고정 전환(fixed transition)일 수도 있습니다.

Nodes와 Edges를 조합하면 상태가 시간에 따라 진화하는 복잡한 루프형 워크플로를 만들 수 있습니다. 하지만 진짜 강력함은 LangGraph가 그 상태를 어떻게 관리하는가에서 나옵니다.

강조하자면, Nodes와 Edges는 결국 함수에 불과합니다. 함수 안에는 LLM이 들어갈 수도 있고, 그냥 평범한 코드가 들어갈 수도 있습니다.

한 줄로 요약하면 이렇습니다. 노드는 일을 하고, 엣지는 다음에 무엇을 할지 알려줍니다.

LangGraph의 기반 그래프 알고리즘은 **메시지 전달(message passing)**을 사용해 일반적인 프로그램을 정의합니다. 어떤 Node가 동작을 마치면 하나 이상의 엣지를 따라 다른 노드(들)로 메시지를 보냅니다. 메시지를 받은 노드는 자신의 함수를 실행하고, 그 결과 메시지를 다음 노드 집합으로 넘기며 이런 과정이 이어집니다. Google의 Pregel 시스템에서 영감을 받아, 프로그램은 이산적인 "슈퍼 스텝(super-step)" 단위로 진행됩니다.

슈퍼 스텝은 그래프 노드에 대한 한 번의 반복(iteration)으로 볼 수 있습니다. 병렬로 실행되는 노드들은 같은 슈퍼 스텝에 속하고, 순차적으로 실행되는 노드들은 서로 다른 슈퍼 스텝에 속합니다. 그래프 실행이 시작될 때 모든 노드는 inactive(비활성) 상태에서 시작합니다. 어떤 노드는 자신의 수신 엣지(또는 "채널(channel)") 중 하나에서 새 메시지(상태)를 받으면 active(활성)가 됩니다. 활성화된 노드는 함수를 실행하고 그 결과를 갱신(update)으로 응답합니다. 각 슈퍼 스텝이 끝날 때, 수신 메시지가 없는 노드들은 스스로를 inactive로 표시해 halt(중지)에 투표합니다. 모든 노드가 inactive 상태이고 전달 중인 메시지가 없을 때 그래프 실행은 종료됩니다.

StateGraph

StateGraph 클래스는 사용할 주요 그래프 클래스입니다. 이 클래스는 사용자가 정의한 State 객체로 파라미터화됩니다.

그래프 컴파일(Compiling your graph)

그래프를 만들려면 먼저 상태를 정의하고, 노드와 엣지를 추가한 다음, **컴파일(compile)**해야 합니다. 그래프를 컴파일한다는 게 정확히 무엇이고 왜 필요한 걸까요?

컴파일은 꽤 단순한 단계입니다. 그래프 구조에 대한 몇 가지 기본 검사(예: 고아 노드[orphaned node]가 없는지 등)를 수행합니다. 또한 여기서 **체크포인터(checkpointers)**나 브레이크포인트(breakpoints) 같은 런타임 인자를 지정할 수 있습니다. 컴파일은 .compile 메서드를 호출하기만 하면 됩니다.

graph = graph_builder.compile(...)

그래프를 사용하려면 반드시 컴파일해야 합니다.

상태(State)

그래프를 정의할 때 가장 먼저 하는 일은 그래프의 **State(상태)**를 정의하는 것입니다. State는 그래프의 스키마(schema)와, 상태에 대한 갱신을 어떻게 적용할지 지정하는 **리듀서 함수(reducer functions)**로 구성됩니다.

State의 스키마는 그래프의 모든 Node와 Edge의 입력 스키마가 되며, TypedDict 또는 Pydantic 모델이 될 수 있습니다. 모든 노드는 State에 대한 갱신을 내보내고(emit), 이 갱신은 지정된 리듀서 함수를 사용해 적용됩니다.

스키마(Schema)

그래프의 스키마를 지정하는 공식적인 방법은 **TypedDict**를 사용하는 것입니다. 상태에 기본값을 제공하고 싶다면 **dataclass**를 사용하세요. 재귀적 데이터 검증(recursive data validation)이 필요하다면 Pydantic BaseModel을 그래프 상태로 사용하는 것도 지원합니다. 다만 TypedDict나 dataclass보다 Pydantic이 성능이 더 떨어진다는 점을 알아두세요.

기본적으로 그래프는 동일한 입력·출력 스키마를 갖습니다. 이것을 변경하고 싶다면 입력·출력 스키마를 명시적으로 지정할 수도 있습니다. 키가 많고 일부는 입력 전용, 일부는 출력 전용일 때 유용합니다. 자세한 내용은 가이드를 참고하세요.

참고: langchain의 상위 레벨 팩토리인 create_agent는 Pydantic 상태 스키마를 지원하지 않습니다.

여러 스키마 (Multiple schemas)

일반적으로 모든 그래프 노드는 단일 스키마로 통신합니다. 즉 모든 노드가 동일한 상태 채널을 읽고 씁니다. 하지만 더 세밀하게 제어하고 싶은 경우가 있습니다.

내부 노드는 그래프의 입력/출력에 필요하지 않은 정보도 전달할 수 있습니다.
그래프에 서로 다른 입력/출력 스키마를 쓰고 싶을 수도 있습니다. 예를 들어 출력에는 관련 있는 출력 키 하나만 포함시키고 싶을 수 있습니다.
노드가 내부 노드 통신용으로 그래프 안에 **프라이빗 상태 채널(private state channel)**에 쓰는 것도 가능합니다. 이 경우 프라이빗 스키마 PrivateState를 간단히 정의하면 됩니다.
그래프에 명시적 입력·출력 스키마를 정의할 수도 있습니다. 이런 경우 그래프 연산에 관련된 모든 키를 담은 "내부(internal)" 스키마를 정의하되, 그래프의 입출력을 제약하기 위해 내부 스키마의 부분 집합인 입력·출력 스키마도 함께 정의합니다. "입력·출력 스키마 정의(Define input and output schemas)"를 참고하세요.

예시를 하나 볼게요.

from typing import TypedDict

from langgraph.graph import END, START, StateGraph


class InputState(TypedDict):
    user_input: str


class OutputState(TypedDict):
    graph_output: str


class OverallState(TypedDict):
    foo: str
    user_input: str
    graph_output: str


class PrivateState(TypedDict):
    bar: str


def node_1(state: InputState) -> OverallState:
    # Write to OverallState
    return {"foo": state["user_input"] + " name"}


def node_2(state: OverallState) -> PrivateState:
    # Read from OverallState, write to PrivateState
    return {"bar": state["foo"] + " is"}


def node_3(state: PrivateState) -> OutputState:
    # Read from PrivateState, write to OutputState
    return {"graph_output": state["bar"] + " Lance"}


builder = StateGraph(OverallState, input_schema=InputState, output_schema=OutputState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
builder.add_edge("node_2", "node_3")
builder.add_edge("node_3", END)

graph = builder.compile()
graph.invoke({"user_input": "My"})
# {'graph_output': 'My name is Lance'}

여기서 미묘하지만 중요한 점이 두 가지 있습니다.

우리는 node_1에 입력 스키마로 state: InputState를 전달합니다. 하지만 foo(즉 OverallState의 채널)에 출력합니다. 입력 스키마에 포함되지 않은 상태 채널에 어떻게 쓸 수 있을까요? 노드는 그래프 상태의 아무 상태 채널에나 쓸 수 있기 때문입니다. 그래프 상태는 초기화 때 정의된 상태 채널들의 합집합(union)이며, 여기에는 OverallState와 필터인 InputState, OutputState가 포함됩니다.
그래프를 다음과 같이 초기화합니다.
```
StateGraph(
    OverallState,
    input_schema=InputState,
    output_schema=OutputState
)
```
그렇다면 node_2에서 어떻게 PrivateState에 쓸 수 있을까요? StateGraph 초기화에 전달되지 않았는데 그래프가 이 스키마에 어떻게 접근할까요? 이는 상태 스키마 정의만 존재한다면 노드가 추가적인 상태 채널을 선언할 수 있기 때문입니다. 이 경우 PrivateState 스키마가 정의되어 있으므로, bar를 그래프의 새 상태 채널로 추가해 쓸 수 있습니다.

한 가지 더 중요한 점이 있습니다. 프라이빗 채널은 스트리밍할 때 제외되지 않습니다. 입력·출력·프라이빗 스키마는 각 노드가 읽는 것(입력 스키마)과 invoke가 반환하는 것(출력 스키마)을 제약합니다. 하지만 스트리밍에서 채널을 숨기지는 않습니다.

stream_mode="values"로 스트리밍하면 그래프는 기본적으로 모든 상태 채널을 방출합니다. 프라이빗 채널도 포함해서요. values 스트리밍이 출력 스키마가 아니라 전체 상태 채널 집합을 기본값으로 하기 때문입니다. 그래서 프라이빗 채널인 bar는 invoke에서는 숨겨지지만 스트리밍에서는 보입니다.

stream = graph.stream_events({"user_input": "My"}, version="v3")
for snapshot in stream.values:
    print(snapshot)
# {'user_input': 'My'}
# {'foo': 'My name', 'user_input': 'My'}
# {'foo': 'My name', 'user_input': 'My', 'bar': 'My name is'}        # <-- private channel
# {'foo': 'My name', 'user_input': 'My', 'graph_output': 'My name is Lance', 'bar': 'My name is'}

스트리밍된 값을 특정 채널 집합(예: 출력 스키마만)으로 제한하려면 output_keys를 전달하세요.

stream = graph.stream_events(
    {"user_input": "My"},
    version="v3",
    output_keys=["graph_output"],
)
for snapshot in stream.values:
    print(snapshot)
# {'graph_output': 'My name is Lance'}

각 단계마다 노드가 실제로 생산한 채널만 필요하다면(누적된 전체 상태가 아니라) 대신 stream_mode="updates"를 사용하세요.

리듀서(Reducers)

리듀서는 노드의 갱신이 State에 어떻게 적용되는지를 이해하는 핵심입니다. State의 각 키는 각자 독립적인 리듀서 함수를 갖습니다. 리듀서 함수를 명시적으로 지정하지 않으면 해당 키에 대한 모든 갱신이 그 키를 **덮어쓴다(override)**고 간주합니다. 몇 가지 타입의 리듀서가 있는데, 기본 타입부터 시작할게요.

리듀서 인자 (Reducer arguments)

모든 리듀서는 두 개의 위치 인자를 받는 이항 함수(binary function)입니다.

왼쪽 인자(left): 그 키에 대해 상태에 이미 저장되어 있는 현재 값.
오른쪽 인자(right): 노드가 반환한 그 키에 대한 갱신.

노드가 부분 갱신(partial update)을 반환하면, LangGraph는 갱신된 각 키에 대해 리듀서를 호출하고 그 반환값을 새 상태 값으로 저장합니다.

new_value = reducer(left=current_state[key], right=node_update[key])

left 인자는 항상 누적된 상태(accumulated state)에서 옵니다. right 인자는 항상 가장 최신 노드 갱신에서 옵니다. 다음 예시는 두 인자를 모두 명시적으로 이름 붙인 경우입니다.

from typing import Annotated

from typing_extensions import TypedDict


def append_strings(left: list[str], right: list[str]) -> list[str]:
    """Combine the existing state value (left) with a node update (right)."""
    return left + right


class State(TypedDict):
    tags: Annotated[list[str], append_strings]

상태가 {"tags": ["draft"]}이고 어떤 노드가 {"tags": ["review"]}를 반환한다고 가정해 봅시다. LangGraph는 다음을 호출합니다.

append_strings(left=["draft"], right=["review"])  # returns ["draft", "review"]

tags의 새 상태 값은 ["draft", "review"]가 됩니다. 커스텀 리듀서는 left와 right 인자를 결합하고, 기본 리듀서는 left 인자를 버리고 right만 유지합니다.

기본 리듀서 (Default reducer)

기본 리듀서는 left 인자를 무시하고 상태 값을 right 인자로 교체합니다. 기본 리듀서를 사용하는 예시입니다.

from typing_extensions import TypedDict


class State(TypedDict):
    foo: int
    bar: list[str]

이 예시에서는 어떤 키에도 리듀서 함수를 지정하지 않았습니다. 그래프 입력이 {"foo": 1, "bar": ["hi"]}라고 가정해 봅시다. 첫 번째 노드가 {"foo": 2}를 반환하면 이는 상태에 대한 갱신으로 처리됩니다. 노드가 전체 State 스키마를 반환할 필요는 없고, 갱신만 반환하면 된다는 점을 눈여겨보세요. 이 갱신을 적용한 후 State는 {"foo": 2, "bar": ["hi"]}가 됩니다. 두 번째 노드가 {"bar": ["bye"]}를 반환하면 State는 {"foo": 2, "bar": ["bye"]}가 됩니다.

커스텀 리듀서 (Custom reducers)

커스텀 리듀서는 상태 값을 교체하는 대신 left와 right 인자를 결합합니다. 리스트에 갱신을 누적(append)하는 것처럼 값을 쌓을 때 유용합니다. 커스텀 리듀서를 지정하는 예시입니다.

from operator import add
from typing import Annotated

from typing_extensions import TypedDict


class State(TypedDict):
    foo: int
    bar: Annotated[list[str], add]

이 예시에서는 두 번째 키 bar에 리듀서 함수(operator.add)를 지정하기 위해 Annotated 타입을 사용했습니다. 첫 번째 키는 그대로 두었다는 점을 눈여겨보세요. 그래프 입력이 {"foo": 1, "bar": ["hi"]}라고 가정해 봅시다. 첫 번째 노드가 {"foo": 2}를 반환하면 이는 상태에 대한 갱신으로 처리됩니다(노드는 전체 State 스키마가 아니라 갱신만 반환하면 됩니다). 적용 후 State는 {"foo": 2, "bar": ["hi"]}가 됩니다. 두 번째 노드가 {"bar": ["bye"]}를 반환하면 State는 {"foo": 2, "bar": ["hi", "bye"]}가 됩니다. 여기서 bar 키는 두 리스트를 더해 갱신됩니다.

Overwrite

어떤 경우에는 리듀서를 우회하고 상태 값을 직접 덮어쓰고 싶을 수 있습니다. LangGraph는 이를 위해 Overwrite 타입을 제공합니다. Overwrite 사용법은 여기에서 배울 수 있습니다.

리듀서 필드 리셋 (Resetting a reducer field)

리듀서와 관련해 자주 헷갈리는 부분이 하나 있습니다. 병합(merging) 리듀서를 쓰는 경우 빈 값을 반환한다고 해서 필드가 비워지지 않습니다. 리듀서는 right 인자를 left 인자에 병합하므로, 빈 갱신도 병합되어 이전에 누적된 값은 유지됩니다.

이 패턴은 재시도 사이에 반드시 비워야 하는 오류 버퍼나 재시도 카운터에서 중요합니다.

from operator import add
from typing import Annotated

from typing_extensions import TypedDict


class State(TypedDict):
    errors: Annotated[list[str], add]


# node A returns {"errors": ["bad sql"]}
# node B returns {"errors": []}
# state["errors"] is still ["bad sql"]; the empty list is merged in, not cleared

병합 리듀서를 유지하면서 필드를 비우려면 갱신을 Overwrite로 감싸세요.

from operator import add
from typing import Annotated

from langgraph.types import Overwrite
from typing_extensions import TypedDict


class State(TypedDict):
    errors: Annotated[list[str], add]


def clear_errors(state: State):
    # Bypass the merging reducer and clear the field
    return {"errors": Overwrite([])}

자세한 내용은 "Overwrite로 리듀서 우회하기(Bypass reducers with Overwrite)"를 참고하세요.

그래프 상태에서 메시지 다루기 (Working with messages in graph state)

왜 메시지를 쓸까? (Why use messages?)

대부분의 최신 LLM 공급자는 입력으로 메시지 리스트를 받아들이는 채팅 모델 인터페이스를 갖고 있습니다. 특히 LangChain의 채팅 모델 인터페이스는 메시지 객체 리스트를 입력으로 받아들입니다. 이런 메시지는 HumanMessage(사용자 입력)나 AIMessage(LLM 응답)처럼 다양한 형태로 제공됩니다.

메시지 객체가 무엇인지 더 알고 싶다면 Messages 개념 가이드를 참고하세요.

그래프에서 메시지 사용하기 (Using messages in your graph)

많은 경우 과거 대화 기록을 그래프 상태의 메시지 리스트로 저장하는 것이 유용합니다. 그러려면 그래프 상태에 Message 객체 리스트를 저장할 키(채널)를 추가하고, 리듀서 함수로 주석을 답니다(아래 예시의 messages 키를 보세요). 리듀서 함수는 각 상태 갱신(예: 노드가 갱신을 보낼 때)마다 상태의 Message 객체 리스트를 어떻게 갱신할지 그래프에 알려주는 데 필수적입니다.

리듀서를 지정하지 않으면 모든 상태 갱신이 가장 최근에 제공된 값으로 메시지 리스트를 덮어씁니다. 단순히 기존 리스트에 메시지를 이어 붙이려면 리듀서로 operator.add를 사용할 수 있습니다.

그런데 그래프 상태의 메시지를 직접 수동으로 갱신하고 싶을 수도 있습니다(예: human-in-the-loop). operator.add를 사용하면 그래프에 보내는 수동 상태 갱신이 기존 메시지를 갱신하는 대신 기존 메시지 리스트에 이어 붙여집니다. 이 문제를 피하려면 메시지 ID를 추적하고 갱신된 기존 메시지를 덮어쓰는 리듀서가 필요합니다. 그렇게 하려면 사전 구축된 add_messages 함수를 사용하세요. 완전히 새로운 메시지라면 기존 리스트에 그냥 이어 붙이지만, 기존 메시지의 갱신도 올바르게 처리해 줍니다.

직렬화(Serialization): 메시지 ID를 추적하는 것 외에도, add_messages 함수는 messages 채널에 상태 갱신이 들어올 때마다 메시지를 LangChain Message 객체로 역직렬화(deserialize)하려 시도합니다. 자세한 내용은 LangChain 직렬화/역직렬화를 참고하세요. 이를 통해 다음과 같은 형식으로 그래프 입력/상태 갱신을 보낼 수 있습니다.

# this is supported
{"messages": [HumanMessage(content="message")]}

# and this is also supported
{"messages": [{"type": "human", "content": "message"}]}

add_messages를 사용하면 상태 갱신이 항상 LangChain Message로 역직렬화되므로, state["messages"][-1].content처럼 **점 표기법(dot notation)**으로 메시지 속성에 접근해야 합니다.

add_messages를 리듀서 함수로 사용하는 그래프 예시입니다.

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

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

MessagesState

상태에 메시지 리스트를 두는 일이 너무 흔해서, 메시지를 쉽게 쓸 수 있도록 **사전 구축된 상태인 MessagesState**가 있습니다. MessagesState는 AnyMessage 객체 리스트인 단일 messages 키로 정의되며, add_messages 리듀서를 사용합니다. 보통은 메시지만 추적하는 게 아니라 추적할 상태가 더 많으므로, 이 상태를 서브클래싱하여 필드를 추가하는 식으로 씁니다.

from langgraph.graph import MessagesState

class State(MessagesState):
    documents: list[str]

노드(Nodes)

LangGraph에서 노드는 다음 인자를 받는 Python 함수입니다(동기/비동기 모두 가능).

state — 그래프의 상태
config — thread_id 같은 구성 정보와 tags 같은 트레이싱 정보를 담은 RunnableConfig 객체
runtime — 런타임 context와 store, stream_writer, execution_info, server_info, heartbeat(유휴 타임아웃 갱신용), control(graceful shutdown용) 같은 정보를 담은 Runtime 객체

NetworkX와 비슷하게, 이런 노드는 add_node 메서드로 그래프에 추가합니다.

from dataclasses import dataclass
from typing_extensions import TypedDict

from langgraph.graph import StateGraph
from langgraph.runtime import Runtime

class State(TypedDict):
    input: str
    results: str

@dataclass
class Context:
    user_id: str

builder = StateGraph(State)

def plain_node(state: State):
    return state

def node_with_runtime(state: State, runtime: Runtime[Context]):
    print("In node: ", runtime.context.user_id)
    return {"results": f"Hello, {state['input']}!"}

def node_with_execution_info(state: State, runtime: Runtime):
    print("In node with thread_id: ", runtime.execution_info.thread_id)
    return {"results": f"Hello, {state['input']}!"}


builder.add_node("plain_node", plain_node)
builder.add_node("node_with_runtime", node_with_runtime)
builder.add_node("node_with_execution_info", node_with_execution_info)
...

내부적으로 함수는 RunnableLambda로 변환되며, 여기에 배치(batch)·비동기(async) 지원과 네이티브 트레이싱·디버깅이 더해집니다.

이름을 지정하지 않고 그래프에 노드를 추가하면 함수 이름과 같은 기본 이름이 주어집니다.

builder.add_node(my_node)
# You can then create edges to/from this node by referencing it as `"my_node"`

재실행과 멱등성 (Re-execution and idempotency)

체크포인터로 컴파일하면 LangGraph는 노드 내부의 함수 중간이 아니라 슈퍼 스텝 경계에서 체크포인트를 저장합니다. 실행이 중단됐다가 나중에 재개되면(예: 인터럽트나 재시도 후), 영향을 받은 노드는 함수의 처음부터 다시 실행됩니다. 멈춘 지점 이전의 코드와 부수 효과도 다시 실행됩니다.

멱등성(Idempotency): 재실행이 상태를 손상시키지 않도록 노드 로직을 설계하세요. 노드가 데이터베이스 행을 삽입한다면, 의도적이지 않은 한 두 번 실행해도 중복 행이 생기지 않아야 합니다. 멱등성 키(idempotency keys), upsert, 또는 read-before-write 검사를 사용하세요. interrupt() 주변의 부수 효과에 대해서는 "인터럽트 전에 호출되는 부수 효과는 멱등해야 한다"를 참고하세요.
그래프 변경: 코드 변경에 대한 결정성(determinism) 규칙은 그래프 구조에는 적용되지 않습니다. 기존 스레드의 재개를 깨지 않고도 노드·엣지를 추가/제거할 수 있습니다. 재개된 실행은 저장된 상태를 사용해서 지금 컴파일하는 어떤 그래프든 실행합니다.
노드 내 태스크와 인터럽트: 노드가 태스크(task)나 interrupt를 호출하면 재개 시 더 엄격한 결정성 규칙이 적용됩니다. LangGraph는 완료된 task 결과를 체크포인터에서 복원하지만, 재개 지점 이전에 코드에서 task 또는 interrupt 순서를 바꾸면 캐시된 값과 불일치가 생길 수 있습니다. Functional API entrypoint는 이러한 방식으로 항목 메서드 전체를 실행하는 단일 노드로 컴파일됩니다. 자세한 내용은 Determinism, Idempotency, "노드에서 태스크 사용하기"를 참고하세요.

노드에서 태스크 사용하기 (Using tasks in nodes)

노드에 여러 연산이 들어 있다면, 로직을 여러 노드로 쪼개는 대신 각 연산을 태스크(task)로 구현하는 것이 더 쉬울 수 있습니다. 그래프가 체크포인터를 사용하면 task 결과가 체크포인트 되므로, 스레드를 재개할 때 노드 안에서 완료된 task 작업을 건너뛸 수 있습니다.

[Original]

from typing import NotRequired

import requests
from langchain_core.utils.uuid import uuid7
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from typing_extensions import TypedDict


class State(TypedDict):
    url: str
    result: NotRequired[str]


def call_api(state: State):
    """Example node that makes an API request."""
    result = requests.get(state["url"]).text[:100]
    return {"result": result}


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

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

thread_id = str(uuid7())
config = {"configurable": {"thread_id": thread_id}}

graph.invoke({"url": "https://www.example.com"}, config)

START 노드

START Node는 사용자 입력을 그래프로 보내는 노드를 나타내는 특수 노드입니다. 이 노드를 참조하는 주 목적은 어떤 노드를 먼저 호출할지 결정하기 위해서입니다.

from langgraph.graph import START

graph.add_edge(START, "node_a")

END 노드

END Node는 터미널(종료) 노드를 나타내는 특수 노드입니다. 어느 엣지가 실행 후 더 이상 동작이 없음을 나타내고 싶을 때 이 노드를 참조합니다.

from langgraph.graph import END

graph.add_edge("node_a", END)

노드 캐싱 (Node caching)

LangGraph는 노드 입력을 기반으로 한 태스크/노드 캐싱을 지원합니다. 캐싱을 사용하려면:

그래프를 컴파일할 때(또는 entrypoint를 지정할 때) 캐시를 지정합니다.
노드에 **캐시 정책(cache policy)**을 지정합니다. 각 캐시 정책은 다음을 지원합니다.
- key_func: 노드 입력을 기반으로 캐시 키를 생성하는 데 사용되며, 기본값은 pickle로 입력을 해시하는 것입니다.
- ttl: 캐시의 수명(초). 지정하지 않으면 캐시는 만료되지 않습니다.

예를 들어:

import time
from typing_extensions import TypedDict
from langgraph.graph import StateGraph
from langgraph.cache.memory import InMemoryCache
from langgraph.types import CachePolicy


class State(TypedDict):
    x: int
    result: int


builder = StateGraph(State)


def expensive_node(state: State) -> dict[str, int]:
    # expensive computation
    time.sleep(2)
    return {"result": state["x"] * 2}


builder.add_node("expensive_node", expensive_node, cache_policy=CachePolicy(ttl=3))
builder.set_entry_point("expensive_node")
builder.set_finish_point("expensive_node")

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

print(graph.invoke({"x": 5}, stream_mode='updates'))
# [{'expensive_node': {'result': 10}}]
print(graph.invoke({"x": 5}, stream_mode='updates'))
# [{'expensive_node': {'result': 10}, '__metadata__': {'cached': True}}]

set_entry_point(node)는 그래프가 처음 실행할 노드를 정의합니다. builder.add_edge(START, node)와 동일합니다.
set_finish_point(node)는 그래프의 마지막 노드를 정의합니다. builder.add_edge(node, END)와 동일합니다.
두 메서드 모두 유효하지만, add_edge(START, ...), add_edge(..., END)가 권장되는 현대적 문법입니다.
첫 번째 실행은 (가짜로 비용이 큰 연산 때문에) 실행에 2초가 걸립니다.
두 번째 실행은 캐시를 활용해 빠르게 반환됩니다.

엣지(Edges)

엣지는 로직이 어떻게 라우팅되는지, 그리고 그래프가 언제 멈출지를 정의합니다. 에이전트가 어떻게 동작하고 노드들이 서로 어떻게 통신하는지의 큰 부분을 차지합니다. 주요 엣지 타입은 몇 가지입니다.

일반 엣지(Normal Edges): 한 노드에서 다음 노드로 곧바로 갑니다.
조건부 엣지(Conditional Edges): 다음에 갈 노드(들)를 결정하기 위해 함수를 호출합니다.
진입점(Entry Point): 사용자 입력이 도착했을 때 처음 호출할 노드.
조건부 진입점(Conditional Entry Point): 사용자 입력이 도착했을 때 처음 호출할 노드(들)를 결정하기 위해 함수를 호출합니다.

한 노드는 여러 개의 나가는(outgoing) 엣지를 가질 수 있습니다. 나가는 엣지가 여러 개라면, 그 모든 목적지 노드들은 다음 슈퍼스텝의 일부로 병렬 실행됩니다.

경고: 노드마다 라우팅 메커니즘을 하나만 선택하세요. 정적 라우팅에는 일반 엣지를, 동적 라우팅에는 조건부 엣지/Command를 사용하세요. 같은 노드에서 일반 엣지와 동적 라우팅을 섞지 마세요. 두 경로 모두 실행될 수 있어 그래프 동작을 추론하기 어려워집니다.

일반 엣지 (Normal edges)

항상 노드 A에서 노드 B로 가고 싶다면, add_edge 메서드를 직접 사용하면 됩니다.

graph.add_edge("node_a", "node_b")

조건부 엣지 (Conditional edges)

하나 이상의 엣지로 선택적으로 라우팅하고 싶거나(또는 선택적으로 종료하고 싶다면), add_conditional_edges 메서드를 사용합니다. 이 메서드는 노드 이름과, 그 노드가 실행된 후 호출할 "라우팅 함수(routing function)"를 받습니다.

graph.add_conditional_edges("node_a", routing_function)

노드와 비슷하게, routing_function은 그래프의 현재 state를 받고 값을 반환합니다.

기본적으로 routing_function의 반환값은 상태를 다음으로 보낼 노드(또는 노드 리스트)의 이름으로 사용됩니다. 그 노드들은 모두 다음 슈퍼스텝의 일부로 병렬 실행됩니다.

선택적으로 routing_function의 출력을 다음 노드의 이름에 매핑하는 **사전(dictionary)**을 제공할 수 있습니다.

graph.add_conditional_edges("node_a", routing_function, {True: "node_b", False: "node_c"})

상태 갱신과 라우팅을 단일 함수에서 결합하고 싶다면 조건부 엣지 대신 Command를 사용하세요.

진입점 (Entry point)

진입점은 그래프가 시작될 때 처음 실행되는 노드(들)입니다. 가상 START 노드에서 실행할 첫 노드로 add_edge 메서드를 사용해 그래프에 어디로 들어갈지 지정할 수 있습니다.

from langgraph.graph import START

graph.add_edge(START, "node_a")

조건부 진입점 (Conditional entry point)

조건부 진입점을 사용하면 커스텀 로직에 따라 서로 다른 노드에서 시작할 수 있습니다. 가상 START 노드에서 add_conditional_edges를 사용하면 됩니다.

from langgraph.graph import START

graph.add_conditional_edges(START, routing_function)

선택적으로 routing_function의 출력을 다음 노드의 이름에 매핑하는 사전을 제공할 수 있습니다.

graph.add_conditional_edges(START, routing_function, {True: "node_b", False: "node_c"})

Send

기본적으로 Nodes와 Edges는 미리 정의되어 있고 동일한 공유 상태에서 동작합니다. 하지만 정확한 엣지를 미리 알 수 없거나, 동시에 여러 버전의 State가 존재하기를 원하는 경우가 있을 수 있습니다. 흔한 예는 map-reduce 디자인 패턴입니다. 이 패턴에서 첫 번째 노드가 객체 리스트를 생성하고, 그 객체들 모두에 다른 노드를 적용하고 싶을 수 있습니다. 객체 수가 사전에 알려지지 않을 수 있고(따라서 엣지 수도 알 수 없음), 다운스트림 노드의 입력 State도 달라야 합니다(생성된 객체마다 하나씩).

이 디자인 패턴을 지원하기 위해 LangGraph는 조건부 엣지에서 Send 객체를 반환하는 것을 지원합니다. Send는 두 인자를 받습니다. 첫 번째는 노드 이름, 두 번째는 그 노드에 전달할 상태입니다.

from langgraph.types import Send

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

graph.add_conditional_edges("node_a", continue_to_jokes)

Command

Command는 그래프 실행을 제어하기 위한 다재다능한 프리미티브입니다. 네 가지 파라미터를 받습니다.

update: 상태 갱신을 적용합니다(노드에서 갱신을 반환하는 것과 유사).
goto: 특정 노드로 이동합니다(조건부 엣지와 유사).
graph: 서브그래프에서 이동할 때 부모 그래프를 대상으로 지정합니다.
resume: 인터럽트 후 실행을 재개할 값을 제공합니다.

Command는 세 가지 맥락에서 사용됩니다.

노드에서 반환(Return from nodes): update, goto, graph를 사용해 상태 갱신과 제어 흐름을 결합합니다.
invoke 또는 stream에 입력(Input to invoke or stream): resume을 사용해 인터럽트 후 실행을 계속합니다.
도구에서 반환(Return from tools): 노드에서 반환하는 것과 유사하게, 도구 내부에서 상태 갱신과 제어 흐름을 결합합니다.

노드에서 반환 (Return from nodes)

update와 goto

노드 함수에서 Command를 반환하면 한 단계에서 상태를 갱신하고 다음 노드로 라우팅합니다.

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

Command를 사용하면 동적 제어 흐름 동작(조건부 엣지와 동일)도 구현할 수 있습니다.

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

상태를 갱신하고 다른 노드로 라우팅해야 할 때는 Command를 사용하세요. 라우팅만 하면 되고 상태 갱신이 필요 없다면 조건부 엣지를 사용하세요.

노드 함수에서 Command를 반환할 때는 노드가 라우팅하는 노드 이름 리스트로 반환 타입 주석을 달아야 합니다. 예: Command[Literal["my_other_node"]]. 이는 그래프 렌더링에 필요하며, LangGraph에 my_node가 my_other_node로 이동할 수 있음을 알려줍니다.

경고: Command는 동적 엣지만 추가합니다. add_edge로 정의된 정적 엣지는 여전히 실행됩니다. 예를 들어 node_a가 Command(goto="my_other_node")를 반환하는데 graph.add_edge("node_a", "node_b")도 있다면, node_b와 my_other_node 둘 다 실행됩니다. 각 노드에 대해 다음 노드로 라우팅할 때 Command 또는 정적 엣지 중 하나만 사용하세요. 둘 다 쓰지 마세요.

Command 사용법의 종단 간 예시는 이 how-to 가이드를 확인하세요.

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
    )

graph를 Command.PARENT로 설정하면 가장 가까운 부모 그래프로 이동합니다.

서브그래프 노드에서 부모 그래프 노드로, 부모·서브그래프 상태 스키마에 모두 공유되는 키에 대한 갱신을 보낼 때는 부모 그래프 상태에서 갱신 중인 키에 리듀서를 정의해야 합니다.

이는 특히 **멀티 에이전트 핸드오프(multi-agent handoffs)**를 구현할 때 유용합니다. 자세한 내용은 "부모 그래프의 노드로 이동하기(Navigate to a node in a parent graph)"를 참고하세요.

invoke 또는 stream에 입력 (Input to invoke or stream)

Command(resume=...)는 invoke()/stream()에 입력으로 쓰도록 의도된 유일한 Command 패턴입니다(선택적으로 update=...와 결합해 재개하면서 상태 변경도 적용). 멀티 턴 대화를 계속하기 위해 Command(update=...)만 단독으로 입력으로 쓰지 마세요. 어떤 Command든 입력으로 전달하면 가장 최근 체크포인트(즉 마지막으로 실행된 스텝, __start__가 아님)에서 재개되므로, 그래프가 이미 끝났다면 멈춘 것처럼 보입니다. 기존 스레드에서 대화를 계속하려면 평범한 입력 dict를 전달하세요.

# WRONG - graph resumes from the latest checkpoint
# (last step that ran), appears stuck
graph.invoke(Command(update={
    "messages": [{"role": "user", "content": "follow up"}]
}), config)

# CORRECT - plain dict restarts from __start__
graph.invoke({
    "messages": [{"role": "user", "content": "follow up"}]
}, config)

resume

Command(resume=...)를 사용해 값을 제공하고 인터럽트 후 그래프 실행을 재개합니다. resume에 전달된 값은 일시 중지된 노드 안의 interrupt() 호출의 반환값이 됩니다.

from typing import TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt


class State(TypedDict):
    messages: list[dict]


def human_review(state: State):
    # Pauses the graph and waits for a value
    answer = interrupt("Do you approve?")
    return {"messages": [{"role": "user", "content": answer}]}


graph = (
    StateGraph(State)
    .add_node("human_review", human_review)
    .add_edge(START, "human_review")
    .add_edge("human_review", END)
    .compile(checkpointer=InMemorySaver())
)

config = {"configurable": {"thread_id": "graph-api-resume"}}

# First run - hits the interrupt and pauses
stream = graph.stream_events({"messages": []}, config, version="v3")
_ = stream.output  # drive the stream to completion
print(stream.interrupts)

# Resume with a value - the interrupt() call returns "yes"
resumed = graph.stream_events(Command(resume="yes"), config, version="v3")
final = resumed.output

인터럽트 패턴(다중 인터럽트, 검증 루프 포함)의 전체 내용은 interrupts 개념 가이드를 참고하세요.

도구에서 반환 (Return from tools)

도구에서 Command를 반환해 그래프 상태를 갱신하고 제어 흐름을 정할 수 있습니다. update로 상태를 수정하고(예: 대화 중 조회한 고객 정보 저장), goto로 도구 완료 후 특정 노드로 라우팅합니다.

도구 내부에서 사용할 때 goto는 동적 엣지를 추가합니다. 도구를 호출한 노드에 이미 정의된 정적 엣지는 여전히 실행됩니다. 각 노드에 대해 도구 기반 동적 라우팅 또는 정적 엣지 중 하나만 사용해 다음 노드로 라우팅하세요. 둘 다 쓰지 마세요.

자세한 내용은 "도구 내부에서 사용하기(Use inside tools)"를 참고하세요.

그래프 마이그레이션 (Graph migrations)

LangGraph는 상태를 추적하는 체크포인터를 사용하더라도 그래프 정의(노드, 엣지, 상태)의 마이그레이션을 쉽게 처리할 수 있습니다.

그래프 끝에 있는(즉 인터럽트되지 않은) 스레드에 대해서는 그래프의 전체 토폴로지(모든 노드·엣지: 제거, 추가, 이름 변경 등)를 변경할 수 있습니다.
현재 인터럽트된 스레드에 대해서는 노드 이름 변경/제거를 제외한 모든 토폴로지 변경을 지원합니다(그 스레드가 이제 존재하지 않는 노드로 들어가려 할 수 있기 때문). 이것이 문제가 된다면 문의해 주세요. 우선순위를 매겨 해결할 수 있습니다.
상태 수정의 경우 키 추가/제거에 대해 완전한 하위·상위 호환성(backwards and forwards compatibility)을 갖습니다.
이름이 바뀐 상태 키는 기존 스레드에서 저장된 상태를 잃습니다.
타입이 호환되지 않는 방식으로 바뀐 상태 키는 변경 전의 상태를 가진 스레드에서 현재 문제를 일으킬 수 있습니다. 문제가 된다면 문의해 주세요.

기술적으로 호환되지만 비즈니스 로직을 바꾸는 변경(예: 도구 집합 재작성, 대화 흐름 재구성)은 Business compatibility를 참고하세요. 이 페이지는 상태에 동작 버전(behavioral version)을 고정해서 기존 스레드는 옛 경로를 유지하고 새 스레드는 최신 버전을 사용하도록 하는 방법을 다룹니다.

런타임 컨텍스트 (Runtime context)

그래프를 만들 때 노드에 전달할 런타임 컨텍스트용 context_schema를 지정할 수 있습니다. 그래프 상태의 일부가 아닌 정보를 노드에 전달할 때 유용합니다. 예를 들어 모델 이름이나 데이터베이스 연결 같은 **의존성(dependency)**을 전달하고 싶을 수 있습니다.

@dataclass
class ContextSchema:
    llm_provider: str = "openai"

graph = StateGraph(State, context_schema=ContextSchema)

그런 다음 invoke 메서드의 context 파라미터로 이 컨텍스트를 그래프에 전달할 수 있습니다.

graph.invoke(inputs, context={"llm_provider": "anthropic"})

컨텍스트는 노드나 조건부 엣지 안에서 접근·사용할 수 있습니다.

from langgraph.runtime import Runtime

def node_a(state: State, runtime: Runtime[ContextSchema]):
    llm = get_llm(runtime.context.llm_provider)
    # ...

구성(configuration)의 전체 설명은 "런타임 구성 추가하기(Add runtime configuration)"를 참고하세요.

재귀 한도 (Recursion limit)

**재귀 한도(recursion limit)**는 그래프가 단일 실행 동안 실행할 수 있는 최대 슈퍼 스텝 수를 설정합니다. 한도에 도달하면 LangGraph는 GraphRecursionError를 발생시킵니다. 버전 1.0.6부터 기본 재귀 한도는 1000 스텝입니다.

재귀 한도는 런타임에 어떤 그래프에든 설정할 수 있으며, config 사전을 통해 invoke/stream에 전달합니다. 중요한 점은 recursion_limit이 독립적인 config 키이며, 다른 모든 사용자 정의 구성처럼 configurable 키 안에 넣어서는 안 된다는 것입니다. 아래 예시를 보세요.

graph.invoke(inputs, config={"recursion_limit": 5}, context={"llm": "anthropic"})

재귀 한도가 어떻게 동작하는지 더 알고 싶으면 "Recursion limit" 문서를 읽어보세요.

재귀 카운터 접근·처리 (Accessing and handling the recursion counter)

현재 스텝 카운터는 어떤 노드 안에서든 config["metadata"]["langgraph_step"]로 접근할 수 있습니다. 이를 통해 재귀 한도에 도달하기 전에 선제적으로 재귀를 처리할 수 있습니다. 그래프 로직 안에서 우아한 성능 저하(graceful degradation) 전략을 구현할 수 있게 해 줍니다.

동작 방식 (How it works)

스텝 카운터는 config["metadata"]["langgraph_step"]에 저장됩니다. LangGraph는 그래프가 실행됨에 따라 이 카운터를 증가시키고, 설정된 recursion_limit을 초과하면 GraphRecursionError를 발생시킵니다.

현재 스텝 카운터 접근 (Accessing the current step counter)

어떤 노드 안에서든 현재 스텝 카운터에 접근해 실행 진행 상황을 모니터링할 수 있습니다.

from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph

def my_node(state: dict, config: RunnableConfig) -> dict:
    current_step = config["metadata"]["langgraph_step"]
    print(f"Currently on step: {current_step}")
    return state

선제적 재귀 처리 (Proactive recursion handling)

LangGraph는 재귀 한도에 도달하기 전에 남은 스텝 수를 추적하는 **RemainingSteps 관리 값(managed value)**을 제공합니다. 이를 통해 그래프 안에서 우아한 성능 저하를 구현할 수 있습니다.

from typing import Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.managed import RemainingSteps

class State(TypedDict):
    messages: Annotated[list, lambda x, y: x + y]
    remaining_steps: RemainingSteps  # Managed value - tracks steps until limit

def reasoning_node(state: State) -> dict:
    # RemainingSteps is automatically populated by LangGraph
    remaining = state["remaining_steps"]

    # Check if we're running low on steps
    if remaining <= 2:
        return {"messages": ["Approaching limit, wrapping up..."]}

    # Normal processing
    return {"messages": ["thinking..."]}

def route_decision(state: State) -> Literal["reasoning_node", "fallback_node"]:
    """Route based on remaining steps"""
    if state["remaining_steps"] <= 2:
        return "fallback_node"
    return "reasoning_node"

def fallback_node(state: State) -> dict:
    """Handle cases where recursion limit is approaching"""
    return {"messages": ["Reached complexity limit, providing best effort answer"]}

# Build graph
builder = StateGraph(State)
builder.add_node("reasoning_node", reasoning_node)
builder.add_node("fallback_node", fallback_node)
builder.add_edge(START, "reasoning_node")
builder.add_conditional_edges("reasoning_node", route_decision)
builder.add_edge("fallback_node", END)

graph = builder.compile()

# RemainingSteps works with any recursion_limit
result = graph.invoke({"messages": []}, {"recursion_limit": 10})

선제적 vs 반응적 접근 (Proactive vs reactive approaches)

재귀 한도를 처리하는 데는 두 가지 주요 접근이 있습니다. 선제적(proactive)(그래프 내부에서 모니터링)과 반응적(reactive)(외부에서 오류를 잡기)입니다.

from typing import Annotated, Literal, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.managed import RemainingSteps
from langgraph.errors import GraphRecursionError

class State(TypedDict):
    messages: Annotated[list, lambda x, y: x + y]
    remaining_steps: RemainingSteps

# Proactive Approach (recommended) - using RemainingSteps
def agent_with_monitoring(state: State) -> dict:
    """Proactively monitor and handle recursion within the graph"""
    remaining = state["remaining_steps"]

    # Early detection - route to internal handling
    if remaining <= 2:
        return {
            "messages": ["Approaching limit, returning partial result"]
        }

    # Normal processing
    return {"messages": [f"Processing... ({remaining} steps remaining)"]}

def route_decision(state: State) -> Literal["agent", END]:
    if state["remaining_steps"] <= 2:
        return END
    return "agent"

# Build graph
builder = StateGraph(State)
builder.add_node("agent", agent_with_monitoring)
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", route_decision)
graph = builder.compile()

# Proactive: Graph completes gracefully
result = graph.invoke({"messages": []}, {"recursion_limit": 10})

# Reactive Approach (fallback) - catching error externally
try:
    result = graph.invoke({"messages": []}, {"recursion_limit": 10})
except GraphRecursionError as e:
    # Handle externally after graph execution fails
    result = {"messages": ["Fallback: recursion limit exceeded"]}

두 접근의 주요 차이점은 다음과 같습니다.

접근	탐지(Detection)	처리(Handling)	제어 흐름(Control Flow)
선제적 (`RemainingSteps` 사용)	한도 도달 전	조건부 라우팅으로 그래프 내부	그래프가 완료 노드까지 계속
반응적 (`GraphRecursionError` 잡기)	한도 초과 후	`try/catch`로 그래프 외부	그래프 실행 종료

선제적 접근의 장점:

그래프 안에서 우아한 성능 저하
중간 상태를 체크포인트에 저장 가능
부분 결과로 더 나은 사용자 경험
그래프가 예외 없이 정상 완료

반응적 접근의 장점:

구현이 더 단순
그래프 로직 수정 불필요
중앙 집중식 오류 처리

기타 사용 가능한 메타데이터 (Other available metadata)

langgraph_step과 함께, 다음 메타데이터도 config["metadata"]에서 사용할 수 있습니다.

def inspect_metadata(state: dict, config: RunnableConfig) -> dict:
    metadata = config["metadata"]

    print(f"Step: {metadata['langgraph_step']}")
    print(f"Node: {metadata['langgraph_node']}")
    print(f"Triggers: {metadata['langgraph_triggers']}")
    print(f"Path: {metadata['langgraph_path']}")
    print(f"Checkpoint NS: {metadata['langgraph_checkpoint_ns']}")

    return state

시각화 (Visualization)

그래프를 시각화할 수 있으면 특히 복잡해질수록 좋습니다. LangGraph에는 그래프를 시각화하는 여러 내장 방법이 있습니다. 자세한 내용은 "그래프 시각화하기(Visualize your graph)"를 참고하세요.

관측성과 트레이싱 (Observability and Tracing)

에이전트를 트레이스·디버그·평가하려면 LangSmith를 사용하세요.

그래프 API 개요 (Graph API Overview)