서브그래프

서브그래프 (Subgraphs)

이 가이드는 서브그래프 사용의 메커니즘을 설명해요. 서브그래프는 다른 그래프에서 노드로 사용되는 그래프예요.

서브그래프는 이런 데 유용해요.

• 멀티 에이전트 시스템 구축
• 노드 집합을 여러 그래프에서 재사용
• 개발 분산: 서로 다른 팀이 그래프의 서로 다른 부분을 독립적으로 작업하길 원할 때, 각 부분을 서브그래프로 정의할 수 있어요. 서브그래프 인터페이스(입력·출력 스키마)만 존중된다면 부모 그래프는 서브그래프의 세부 내용을 알지 못한 채 구축될 수 있어요.

출처: 문서

본문

설정 (Setup)

pip install -U langgraph

uv를 쓴다면:

uv add langgraph

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

서브그래프 통신 정의 (Define subgraph communication)

서브그래프를 추가할 때 부모 그래프와 서브그래프가 어떻게 통신할지 정의해야 해요.

Pattern	When to use	State schemas
Call a subgraph inside a node	Parent and subgraph have different state schemas (no shared keys), or you need to transform state between them	You write a wrapper function that maps parent state to subgraph input and subgraph output back to parent state
Add a subgraph as a node	Parent and subgraph share state keys—the subgraph reads from and writes to the same channels as the parent	You pass the compiled subgraph directly to add_node—no wrapper function needed

노드 안에서 서브그래프 호출하기 (Call a subgraph inside a node)

부모 그래프와 서브그래프가 다른 상태 스키마를 가지면(공유 키 없음) 노드 함수 안에서 서브그래프를 호출해요. 멀티 에이전트 시스템에서 각 에이전트마다 개인 메시지 이력을 유지하고 싶을 때 흔해요.

노드 함수는 서브그래프를 호출하기 전에 부모 상태를 서브그래프 상태로 변환하고, 반환하기 전에 결과를 다시 부모 상태로 변환해요.

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

class SubgraphState(TypedDict):
    bar: str

# Subgraph

def subgraph_node_1(state: SubgraphState):
    return {"bar": "hi! " + state["bar"]}

subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph = subgraph_builder.compile()

# Parent graph

class State(TypedDict):
    foo: str

def call_subgraph(state: State):
    # Transform the state to the subgraph state
    subgraph_output = subgraph.invoke({"bar": state["foo"]})  # [!code highlight]
    # Transform response back to the parent state
    return {"foo": subgraph_output["bar"]}

builder = StateGraph(State)
builder.add_node("node_1", call_subgraph)
builder.add_edge(START, "node_1")
graph = builder.compile()

전체 예시: 다른 상태 스키마

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

# Define subgraph
class SubgraphState(TypedDict):
    # note that none of these keys are shared with the parent graph state
    bar: str
    baz: str

def subgraph_node_1(state: SubgraphState):
    return {"baz": "baz"}

def subgraph_node_2(state: SubgraphState):
    return {"bar": state["bar"] + state["baz"]}

subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_node(subgraph_node_2)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph_builder.add_edge("subgraph_node_1", "subgraph_node_2")
subgraph = subgraph_builder.compile()

# Define parent graph
class ParentState(TypedDict):
    foo: str

def node_1(state: ParentState):
    return {"foo": "hi! " + state["foo"]}

def node_2(state: ParentState):
    # Transform the state to the subgraph state
    response = subgraph.invoke({"bar": state["foo"]})
    # Transform response back to the parent state
    return {"foo": response["bar"]}


builder = StateGraph(ParentState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
graph = builder.compile()

stream = graph.stream_events({"foo": "foo"}, version="v3")
for event in stream:
    if event["method"] == "updates":
        print(event["params"]["namespace"], event["params"]["data"])

[] {'node_1': {'foo': 'hi! foo'}}
['node_2:577b710b-64ae-31fb-9455-6a4d4cc2b0b9'] {'subgraph_node_1': {'baz': 'baz'}}
['node_2:577b710b-64ae-31fb-9455-6a4d4cc2b0b9'] {'subgraph_node_2': {'bar': 'hi! foobaz'}}
[] {'node_2': {'foo': 'hi! foobaz'}}

전체 예시: 다른 상태 스키마 (서브그래프 2단계)

이것은 서브그래프가 2단계인 예시예요: parent -> child -> grandchild.

# Grandchild graph
from typing_extensions import TypedDict
from langgraph.graph.state import StateGraph, START, END

class GrandChildState(TypedDict):
    my_grandchild_key: str

def grandchild_1(state: GrandChildState) -> GrandChildState:
    # NOTE: child or parent keys will not be accessible here
    return {"my_grandchild_key": state["my_grandchild_key"] + ", how are you"}


grandchild = StateGraph(GrandChildState)
grandchild.add_node("grandchild_1", grandchild_1)

grandchild.add_edge(START, "grandchild_1")
grandchild.add_edge("grandchild_1", END)

grandchild_graph = grandchild.compile()

# Child graph
class ChildState(TypedDict):
    my_child_key: str

def call_grandchild_graph(state: ChildState) -> ChildState:
    # NOTE: parent or grandchild keys won't be accessible here
    grandchild_graph_input = {"my_grandchild_key": state["my_child_key"]}
    grandchild_graph_output = grandchild_graph.invoke(grandchild_graph_input)
    return {"my_child_key": grandchild_graph_output["my_grandchild_key"] + " today?"}

child = StateGraph(ChildState)
# We're passing a function here instead of just compiled graph (`grandchild_graph`)
child.add_node("child_1", call_grandchild_graph)
child.add_edge(START, "child_1")
child.add_edge("child_1", END)
child_graph = child.compile()

# Parent graph
class ParentState(TypedDict):
    my_key: str

def parent_1(state: ParentState) -> ParentState:
    # NOTE: child or grandchild keys won't be accessible here
    return {"my_key": "hi " + state["my_key"]}

def parent_2(state: ParentState) -> ParentState:
    return {"my_key": state["my_key"] + " bye!"}

def call_child_graph(state: ParentState) -> ParentState:
    child_graph_input = {"my_child_key": state["my_key"]}
    child_graph_output = child_graph.invoke(child_graph_input)
    return {"my_key": child_graph_output["my_child_key"]}

parent = StateGraph(ParentState)
parent.add_node("parent_1", parent_1)
# We're passing a function here instead of just a compiled graph (`child_graph`)
parent.add_node("child", call_child_graph)
parent.add_node("parent_2", parent_2)

parent.add_edge(START, "parent_1")
parent.add_edge("parent_1", "child")
parent.add_edge("child", "parent_2")
parent.add_edge("parent_2", END)

parent_graph = parent.compile()

stream = parent_graph.stream_events({"my_key": "Bob"}, version="v3")
for event in stream:
    if event["method"] == "updates":
        print(event["params"]["namespace"], event["params"]["data"])

[] {'parent_1': {'my_key': 'hi Bob'}}
['child:2e26e9ce-602f-862c-aa66-1ea5a4655e3b', 'child_1:781bb3b1-3971-84ce-810b-acf819a03f9c'] {'grandchild_1': {'my_grandchild_key': 'hi Bob, how are you'}}
['child:2e26e9ce-602f-862c-aa66-1ea5a4655e3b'] {'child_1': {'my_child_key': 'hi Bob, how are you today?'}}
[] {'child': {'my_key': 'hi Bob, how are you today?'}}
[] {'parent_2': {'my_key': 'hi Bob, how are you today? bye!'}}

서브그래프를 노드로 추가하기 (Add a subgraph as a node)

부모 그래프와 서브그래프가 상태 키를 공유하면 컴파일된 서브그래프를 add_node에 직접 전달할 수 있어요. 래퍼 함수가 필요 없어요. 서브그래프는 부모의 상태 채널에서 자동으로 읽고 써요. 예를 들어 멀티 에이전트 시스템에서 에이전트들은 종종 공유 messages 키로 통신해요.

서브그래프가 부모 그래프와 상태 키를 공유한다면 다음 단계로 그래프에 추가할 수 있어요.

1. 서브그래프 워크플로(아래 예시의 subgraph_builder)를 정의하고 컴파일해요.
2. 부모 그래프 워크플로를 정의할 때 컴파일된 서브그래프를 add_node 메서드에 전달해요.

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

class State(TypedDict):
    foo: str

# Subgraph

def subgraph_node_1(state: State):
    return {"foo": "hi! " + state["foo"]}

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

# Parent graph

builder = StateGraph(State)
builder.add_node("node_1", subgraph)  # [!code highlight]
builder.add_edge(START, "node_1")
graph = builder.compile()

전체 예시: 공유 상태 스키마

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

# Define subgraph
class SubgraphState(TypedDict):
    foo: str  # shared with parent graph state
    bar: str  # private to SubgraphState

def subgraph_node_1(state: SubgraphState):
    return {"bar": "bar"}

def subgraph_node_2(state: SubgraphState):
    # note that this node is using a state key ('bar') that is only available in the subgraph
    # and is sending update on the shared state key ('foo')
    return {"foo": state["foo"] + state["bar"]}

subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_node(subgraph_node_2)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph_builder.add_edge("subgraph_node_1", "subgraph_node_2")
subgraph = subgraph_builder.compile()

# Define parent graph
class ParentState(TypedDict):
    foo: str

def node_1(state: ParentState):
    return {"foo": "hi! " + state["foo"]}

builder = StateGraph(ParentState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", subgraph)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
graph = builder.compile()

stream = graph.stream_events({"foo": "foo"}, version="v3")
for event in stream:
    if event["method"] == "updates" and not event["params"]["namespace"]:
        print(event["params"]["data"])

{'node_1': {'foo': 'hi! foo'}}
{'node_2': {'foo': 'hi! foobar'}}

서브그래프 영속성 (Subgraph persistence)

서브그래프를 쓸 때, 호출 사이에 그 내부 데이터를 어떻게 처리할지 결정해야 해요. 전문 서브에이전트로 작업을 위임하는 고객 지원 봇을 생각해 보세요. "빌링 전문가" 서브에이전트가 고객의 이전 질문을 기억해야 할까요, 아니면 호출될 때마다 새로 시작해야 할까요?

.compile()의 checkpointer 파라미터가 서브그래프 영속성을 제어해요.

Mode	checkpointer=	Behavior
Per-invocation	None (default)	Each call starts fresh and inherits the parent's checkpointer to support interrupts and durable execution within a single call.
Per-thread	True	State accumulates across calls on the same thread. Each call picks up where the last one left off.
Stateless	False	No checkpointing at all—runs like a plain function call. No interrupts or durable execution.

Per-invocation은 대부분의 애플리케이션에 올바른 선택이에요. 서브에이전트가 독립적인 요청을 처리하는 멀티 에이전트 시스템도 포함해서요. 서브에이전트가 여러 대화 턴에 걸친 기억(예: 여러 대화에서 컨텍스트를 쌓는 리서치 어시스턴트)이 필요할 때는 per-thread를 사용해요.

부모 그래프가 서브그래프 영속성 기능(인터럽트, 상태 조사, per-thread 메모리)을 사용하려면 checkpointer와 함께 컴파일돼야 해요. 영속성을 참고해요.

아래 예시는 에이전트를 만드는 흔한 방법인 LangChain의 create_agent를 사용해요. create_agent는 내부적으로 LangGraph 그래프를 만들므로, 모든 서브그래프 영속성 개념이 직접 적용돼요. raw LangGraph StateGraph로 만든다면 같은 패턴과 구성 옵션이 적용돼요 — 자세한 내용은 Graph API를 참고해요.

Stateful

Stateful 서브그래프는 부모 그래프의 checkpointer를 상속해 인터럽트, 영속성, 상태 조사를 가능하게 해요. 두 stateful 모드는 상태가 얼마나 오래 유지되는지가 달라요.

Per-invocation (기본값)

대부분의 애플리케이션, 특히 서브에이전트가 도구로 호출되는 멀티 에이전트 시스템에서 권장되는 모드예요. 인터럽트, 영속성, 병렬 호출을 지원하면서 각 호출을 격리 상태로 유지해요.

서브그래프에 대한 각 호출이 독립적이고 서브에이전트가 이전 호출에서 아무것도 기억할 필요가 없을 때 per-invocation 영속성을 사용해요. 이것이 가장 흔한 패턴이에요. 특히 "이 고객의 주문을 조회해줘"나 "이 문서를 요약해줘" 같은 일회성 요청을 처리하는 서브에이전트가 있는 멀티 에이전트 시스템에서요.

checkpointer를 생략하거나 None으로 설정해요. 각 호출은 새로 시작하지만, 단일 호출 안에서 서브그래프는 부모의 checkpointer를 상속해 interrupt()로 일시정지·재개할 수 있어요.

다음 예시는 두 서브에이전트(과일 전문가, 채소 전문가)를 외부 에이전트의 도구로 감싼 것이에요.

from langchain.agents import create_agent
from langchain.tools import tool
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command, interrupt

@tool
def fruit_info(fruit_name: str) -> str:
    """Look up fruit info."""
    return f"Info about {fruit_name}"

@tool
def veggie_info(veggie_name: str) -> str:
    """Look up veggie info."""
    return f"Info about {veggie_name}"

# Subagents - no checkpointer setting (inherits parent)
fruit_agent = create_agent(
    model="gpt-5.4-mini",
    tools=[fruit_info],
    prompt="You are a fruit expert. Use the fruit_info tool. Respond in one sentence.",
)

veggie_agent = create_agent(
    model="gpt-5.4-mini",
    tools=[veggie_info],
    prompt="You are a veggie expert. Use the veggie_info tool. Respond in one sentence.",
)

# Wrap subagents as tools for the outer agent
@tool
def ask_fruit_expert(question: str) -> str:
    """Ask the fruit expert. Use for ALL fruit questions."""
    response = fruit_agent.invoke(
        {"messages": [{"role": "user", "content": question}]},
    )
    return response["messages"][-1].content

@tool
def ask_veggie_expert(question: str) -> str:
    """Ask the veggie expert. Use for ALL veggie questions."""
    response = veggie_agent.invoke(
        {"messages": [{"role": "user", "content": question}]},
    )
    return response["messages"][-1].content

# Outer agent with checkpointer
agent = create_agent(
    model="gpt-5.4-mini",
    tools=[ask_fruit_expert, ask_veggie_expert],
    prompt=(
        "You have two experts: ask_fruit_expert and ask_veggie_expert. "
        "ALWAYS delegate questions to the appropriate expert."
    ),
    checkpointer=MemorySaver(),
)

인터럽트 (Interrupts)

각 호출은 interrupt()로 일시정지·재개할 수 있어요. 진행 전에 사용자 승인을 요구하도록 도구 함수에 interrupt()를 추가해요.

@tool
def fruit_info(fruit_name: str) -> str:
    """Look up fruit info."""
    interrupt("continue?")  # [!code highlight]
    return f"Info about {fruit_name}"

from langgraph.types import Command

config = {"configurable": {"thread_id": "1"}}

# Stream events - the subagent's tool calls interrupt()
stream = agent.stream_events(
    {"messages": [{"role": "user", "content": "Tell me about apples"}]},
    config=config,
    version="v3",
)
output = stream.output  # drive the stream to completion
# stream.interrupts contains pending interrupts (and stream.interrupted is True)

# Resume - approve the interrupt
resumed = agent.stream_events(Command(resume=True), config=config, version="v3")
final = resumed.output

Example trace 보기: 이 예시의 공개 LangSmith run을 열어보세요.

멀티 턴 (Multi-turn)

각 호출은 서브에이전트 상태로 새로 시작해요. 서브에이전트는 이전 호출을 기억하지 않아요.

config = {"configurable": {"thread_id": "1"}}

# First call
response = agent.invoke(
    {"messages": [{"role": "user", "content": "Tell me about apples"}]},
    config=config,
)
# Subagent message count: 4

# Second call - subagent starts fresh, no memory of apples
response = agent.invoke(
    {"messages": [{"role": "user", "content": "Now tell me about bananas"}]},
    config=config,
)
# Subagent message count: 4 (still fresh!)

여러 서브그래프 호출 (Multiple subgraph calls)

같은 서브그래프에 대한 여러 호출은 충돌 없이 동작해요. 각 호출이 자신만의 체크포인트 네임스페이스를 갖기 때문이에요.

config = {"configurable": {"thread_id": "1"}}

# LLM calls ask_fruit_expert for both apples and bananas
response = agent.invoke(
    {"messages": [{"role": "user", "content": "Tell me about apples and bananas"}]},
    config=config,
)
# Subagent message count: 4 (apples - fresh)
# Subagent message count: 4 (bananas - fresh)

Per-thread

서브에이전트가 이전 상호작용을 기억해야 할 때 per-thread 영속성을 사용해요. 예를 들어 여러 대화에 걸쳐 컨텍스트를 쌓는 리서치 어시스턴트, 이미 편집한 파일을 추적하는 코딩 어시스턴트 같은 경우죠. 서브에이전트의 대화 이력과 데이터는 같은 스레드의 호출들에 걸쳐 누적돼요. 각 호출은 이전 호출이 멈춘 곳에서 이어가요.

checkpointer=True로 컴파일하면 이 동작을 켤 수 있어요.

Per-thread 서브그래프는 병렬 도구 호출을 지원하지 않아요. LLM이 per-thread 서브에이전트를 도구로 접근할 수 있으면, 그 도구를 병렬로 여러 번 호출하려 할 수 있어요(예: 과일 전문가에게 사과와 바나나를 동시에 묻기). 그러면 두 호출이 같은 네임스페이스에 쓰기 때문에 체크포인트 충돌이 발생해요.

아래 예시는 이를 방지하기 위해 LangChain의 ToolCallLimitMiddleware를 사용해요. 순수 LangGraph StateGraph로 만든다면 병렬 도구 호출을 스스로 막아야 해요. 예를 들어 모델이 병렬 도구 호출을 비활성화하도록 구성하거나, 같은 서브그래프가 병렬로 여러 번 호출되지 않게 하는 로직을 추가하세요.

다음 예시는 checkpointer=True로 컴파일한 과일 전문가 서브에이전트를 사용해요.

from langchain.agents import create_agent
from langchain.agents.middleware import ToolCallLimitMiddleware
from langchain.tools import tool
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command, interrupt

@tool
def fruit_info(fruit_name: str) -> str:
    """Look up fruit info."""
    return f"Info about {fruit_name}"

# Subagent with checkpointer=True for persistent state
fruit_agent = create_agent(
    model="gpt-5.4-mini",
    tools=[fruit_info],
    prompt="You are a fruit expert. Use the fruit_info tool. Respond in one sentence.",
    checkpointer=True,  # [!code highlight]
)

# Wrap subagent as a tool for the outer agent
@tool
def ask_fruit_expert(question: str) -> str:
    """Ask the fruit expert. Use for ALL fruit questions."""
    response = fruit_agent.invoke(
        {"messages": [{"role": "user", "content": question}]},
    )
    return response["messages"][-1].content

# Outer agent with checkpointer
# Use ToolCallLimitMiddleware to prevent parallel calls to per-thread subagents,
# which would cause checkpoint conflicts.
agent = create_agent(
    model="gpt-5.4-mini",
    tools=[ask_fruit_expert],
    prompt="You have a fruit expert. ALWAYS delegate fruit questions to ask_fruit_expert.",
    middleware=[  # [!code highlight]
        ToolCallLimitMiddleware(tool_name="ask_fruit_expert", run_limit=1),  # [!code highlight]
    ],  # [!code highlight]
    checkpointer=MemorySaver(),
)

인터럽트 (Interrupts)

Per-thread 서브에이전트도 per-invocation처럼 interrupt()를 지원해요. 사용자 승인을 요구하도록 도구 함수에 interrupt()를 추가해요.

@tool
def fruit_info(fruit_name: str) -> str:
    """Look up fruit info."""
    interrupt("continue?")  # [!code highlight]
    return f"Info about {fruit_name}"

from langgraph.types import Command

config = {"configurable": {"thread_id": "1"}}

# Stream events - the subagent's tool calls interrupt()
stream = agent.stream_events(
    {"messages": [{"role": "user", "content": "Tell me about apples"}]},
    config=config,
    version="v3",
)
output = stream.output  # drive the stream to completion
# stream.interrupts contains pending interrupts (and stream.interrupted is True)

# Resume - approve the interrupt
resumed = agent.stream_events(Command(resume=True), config=config, version="v3")
final = resumed.output

Example trace 보기: 이 예시의 공개 LangSmith run을 열어보세요.

멀티 턴 (Multi-turn)

상태가 호출들에 걸쳐 누적돼요. 서브에이전트가 이전 대화를 기억해요.

config = {"configurable": {"thread_id": "1"}}

# First call
response = agent.invoke(
    {"messages": [{"role": "user", "content": "Tell me about apples"}]},
    config=config,
)
# Subagent message count: 4

# Second call - subagent REMEMBERS apples conversation
response = agent.invoke(
    {"messages": [{"role": "user", "content": "Now tell me about bananas"}]},
    config=config,
)
# Subagent message count: 8 (accumulated!)

여러 서브그래프 호출 (Multiple subgraph calls)

여러 서로 다른 per-thread 서브그래프(예: 과일 전문가와 채소 전문가)가 있으면 각각 자신의 저장 공간이 필요해요. 그래야 체크포인트가 서로 덮어쓰지 않아요. 이를 **네임스페이스 격리(namespace isolation)**라고 해요.

노드 안에서 서브그래프를 호출한다면 LangGraph는 호출 순서(첫 번째 호출, 두 번째 호출 등)에 따라 네임스페이스를 할당해요. 즉 호출 순서를 바꾸면 어느 서브그래프가 어느 상태를 로드하는지 뒤섞일 수 있어요. 이를 피하려면 각 서브에이전트를 고유한 노드 이름을 가진 자체 StateGraph로 감싸세요. 그러면 각 서브그래프가 안정적이고 고유한 네임스페이스를 갖게 돼요.

from langgraph.graph import MessagesState, StateGraph

def create_sub_agent(model, *, name, **kwargs):
    """Wrap an agent with a unique node name for namespace isolation."""
    agent = create_agent(model=model, name=name, **kwargs)
    return (
        StateGraph(MessagesState)
        .add_node(name, agent)  # unique name → stable namespace  # [!code highlight]
        .add_edge("__start__", name)
        .compile()
    )

fruit_agent = create_sub_agent(
    "gpt-5.4-mini", name="fruit_agent",
    tools=[fruit_info], prompt="...", checkpointer=True,
)
veggie_agent = create_sub_agent(
    "gpt-5.4-mini", name="veggie_agent",
    tools=[veggie_info], prompt="...", checkpointer=True,
)

config = {"configurable": {"thread_id": "1"}}

# First call - LLM calls both fruit and veggie experts
response = agent.invoke(
    {"messages": [{"role": "user", "content": "Tell me about cherries and broccoli"}]},
    config=config,
)
# Fruit subagent message count: 4
# Veggie subagent message count: 4

# Second call - both agents accumulate independently
response = agent.invoke(
    {"messages": [{"role": "user", "content": "Now tell me about oranges and carrots"}]},
    config=config,
)
# Fruit subagent message count: 8 (remembers cherries!)
# Veggie subagent message count: 8 (remembers broccoli!)

노드로 추가된 서브그래프는 자동으로 이름 기반 네임스페이스를 얻으므로 이 래퍼가 필요 없어요.

Stateless

서브에이전트를 체크포인팅 오버헤드 없이 평범한 함수 호출처럼 실행하고 싶을 때 사용해요. 서브그래프는 일시정지·재개할 수 없고 내구성 있는 실행(durable execution)의 이점도 없어요. checkpointer=False로 컴파일해요.

체크포인팅이 없으면 서브그래프는 내구성 있는 실행이 없어요. 프로세스가 실행 중에 크래시하면 서브그래프는 복구할 수 없고 처음부터 다시 실행해야 해요.

subgraph_builder = StateGraph(...)
subgraph = subgraph_builder.compile(checkpointer=False)  # [!code highlight]

Checkpointer 참조 (Checkpointer reference)

.compile()의 checkpointer 파라미터로 서브그래프 영속성을 제어해요.

subgraph = builder.compile(checkpointer=False)  # or True / None

Feature	Per-invocation (default)	Per-thread	Stateless
checkpointer=	None	True	False
Interrupts (HITL)	✅	✅	❌
Multi-turn memory	❌	✅	❌
Multiple calls (different subgraphs)	✅	⚠️ (Calls to multiple per-thread subgraphs in the same node can cause namespace conflicts. Workarounds are available.)	✅
Multiple calls (same subgraph)	✅	❌	✅
State inspection	⚠️ (State inspection with per-invocation persistence is available for the current invocation only (while interrupted). Each invocation starts fresh, so there is no accumulated state to inspect after the invocation completes.)	✅	❌

• 인터럽트 (HITL): 서브그래프는 interrupt()로 실행을 일시정지하고 사용자 입력을 기다린 뒤, 멈춘 지점에서 재개할 수 있어요.
• 멀티 턴 메모리: 서브그래프는 같은 스레드의 여러 호출에 걸쳐 상태를 유지해요. 각 호출은 새로 시작하는 대신 이전 호출이 멈춘 곳에서 이어가요.
• 여러 호출(다른 서브그래프): 여러 서로 다른 서브그래프 인스턴스를 체크포인트 네임스페이스 충돌 없이 단일 노드 안에서 호출할 수 있어요.
• 여러 호출(같은 서브그래프): 같은 서브그래프 인스턴스를 단일 노드 안에서 여러 번 호출할 수 있어요. Stateful 영속성에서는 이 호출들이 같은 체크포인트 네임스페이스에 써서 충돌해요. 대신 per-invocation 영속성을 사용하세요.
• 상태 조사: 서브그래프의 상태는 디버깅·모니터링용으로 get_state(config, subgraphs=True)로 접근할 수 있어요.

서브그래프 상태 보기 (View subgraph state)

영속성을 켜면 subgraphs 옵션으로 서브그래프 상태를 조사할 수 있어요. stateless 체크포인팅(checkpointer=False)에서는 서브그래프 체크포인트가 저장되지 않으므로 서브그래프 상태를 사용할 수 없어요.

서브그래프 상태를 보려면 LangGraph가 서브그래프를 **정적으로 발견(discover)**할 수 있어야 해요. 즉 노드로 추가되거나 노드 안에서 호출돼야 해요. 도구 함수 안이나 다른 간접 경로(예: subagents 패턴)에서 호출되면 동작하지 않아요. 인터럽트는 중첩과 무관하게 최상위 그래프로 전파돼요.

Per-invocation

현재 호출에 대해서만 서브그래프 상태를 반환해요. 각 호출은 새로 시작돼요.

from langgraph.graph import START, StateGraph
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
from typing_extensions import TypedDict

class State(TypedDict):
    foo: str

# Subgraph
def subgraph_node_1(state: State):
    value = interrupt("Provide value:")
    return {"foo": state["foo"] + value}

subgraph_builder = StateGraph(State)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph = subgraph_builder.compile()  # inherits parent checkpointer

# Parent graph
builder = StateGraph(State)
builder.add_node("node_1", subgraph)
builder.add_edge(START, "node_1")

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

config = {"configurable": {"thread_id": "1"}}

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

# View subgraph state for the current invocation
subgraph_state = graph.get_state(config, subgraphs=True).tasks[0].state  # [!code highlight]

# Resume the subgraph
graph.invoke(Command(resume="bar"), config)

Per-thread

이 스레드의 모든 호출에 걸쳐 누적된 서브그래프 상태를 반환해요.

from langgraph.graph import START, StateGraph, MessagesState
from langgraph.checkpoint.memory import MemorySaver

# Subgraph with its own persistent state
subgraph_builder = StateGraph(MessagesState)
# ... add nodes and edges
subgraph = subgraph_builder.compile(checkpointer=True)  # [!code highlight]

# Parent graph
builder = StateGraph(MessagesState)
builder.add_node("agent", subgraph)
builder.add_edge(START, "agent")

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

config = {"configurable": {"thread_id": "1"}}

graph.invoke({"messages": [{"role": "user", "content": "hi"}]}, config)
graph.invoke({"messages": [{"role": "user", "content": "what did I say?"}]}, config)

# View accumulated subgraph state (includes messages from both invocations)
subgraph_state = graph.get_state(config, subgraphs=True).tasks[0].state  # [!code highlight]

서브그래프 출력 스트리밍 (Stream subgraph outputs)

중첩된 그래프 실행을 관찰하려면 이벤트 스트리밍을 권장해요. stream.subgraphs 프로젝션은 각 중첩 실행을 발견하고, 네임스페이스 문자열을 파싱하지 않고도 path, messages, values를 노출해요.

stream = graph.stream_events({"foo": "foo"}, version="v3")  # [!code highlight]

for subgraph in stream.subgraphs:
    print(subgraph.graph_name, subgraph.path)

    for snapshot in subgraph.values:
        print(subgraph.path, snapshot)

raw 프로토콜 이벤트가 필요하다면 스트림을 직접 순회하고 event["method"]와 event["params"]["namespace"]로 필터링해요.

stream = graph.stream_events({"foo": "foo"}, version="v3")
for event in stream:
    if event["method"] == "updates":
        print(event["params"]["namespace"], event["params"]["data"])

서브그래프에서 스트리밍

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

# Define subgraph
class SubgraphState(TypedDict):
    foo: str
    bar: str

def subgraph_node_1(state: SubgraphState):
    return {"bar": "bar"}

def subgraph_node_2(state: SubgraphState):
    # note that this node is using a state key ('bar') that is only available in the subgraph
    # and is sending update on the shared state key ('foo')
    return {"foo": state["foo"] + state["bar"]}

subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_node(subgraph_node_2)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph_builder.add_edge("subgraph_node_1", "subgraph_node_2")
subgraph = subgraph_builder.compile()

# Define parent graph
class ParentState(TypedDict):
    foo: str

def node_1(state: ParentState):
    return {"foo": "hi! " + state["foo"]}

builder = StateGraph(ParentState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", subgraph)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
graph = builder.compile()

stream = graph.stream_events({"foo": "foo"}, version="v3")  # [!code highlight]
for event in stream:
    if event["method"] == "updates":
        print(event["params"]["namespace"], event["params"]["data"])

[] {'node_1': {'foo': 'hi! foo'}}
['node_2:e58e5673-a661-ebb0-70d4-e298a7fc28b7'] {'subgraph_node_1': {'bar': 'bar'}}
['node_2:e58e5673-a661-ebb0-70d4-e298a7fc28b7'] {'subgraph_node_2': {'foo': 'hi! foobar'}}
[] {'node_2': {'foo': 'hi! foobar'}}

더 알아보기 (Learn more)

• 멀티 에이전트 시스템: 서브그래프로 멀티 에이전트를 구축하는 방법을 다뤄요.
• Graph API: 노드·엣지·상태·리듀서를 다뤄요.
• 인터럽트: human-in-the-loop와 interrupt()를 다뤄요.
• 영속성: checkpointer와 내구성 있는 실행을 다뤄요.