Subgraph

이 가이드는 subgraph 사용의 메커니즘을 설명합니다. Subgraph는 다른 graph에서 node로 사용되는 graph입니다. Subgraph는 다음과 같은 경우에 유용합니다:

멀티 에이전트 시스템 구축
여러 graph에서 node 집합 재사용
개발 분산: 서로 다른 팀이 graph의 다른 부분을 독립적으로 작업하려는 경우, 각 부분을 subgraph로 정의할 수 있으며, subgraph 인터페이스(입력 및 출력 schema)가 준수되는 한 parent graph는 subgraph의 세부 사항을 알 필요 없이 구축될 수 있습니다

Subgraph를 추가할 때 parent graph와 subgraph가 통신하는 방법을 정의해야 합니다:

node에서 graph 호출 — subgraph는 parent graph의 node 내부에서 호출됩니다
graph를 node로 추가 — subgraph가 parent의 node로 직접 추가되고 parent와 state key를 공유합니다

Setup

pip install -U langgraph

LangGraph 개발을 위한 LangSmith 설정 LangSmith에 가입하여 LangGraph 프로젝트의 문제를 빠르게 발견하고 성능을 개선하세요. LangSmith를 사용하면 trace 데이터를 활용하여 LangGraph로 구축된 LLM 앱을 디버그, 테스트 및 모니터링할 수 있습니다 — 시작하는 방법에 대한 자세한 내용은 여기를 참조하세요.

Invoke a graph from a node

Subgraph를 구현하는 간단한 방법은 다른 graph의 node 내부에서 graph를 호출하는 것입니다. 이 경우 subgraph는 parent graph와 완전히 다른 schema를 가질 수 있습니다(공유 key 없음). 예를 들어, 멀티 에이전트 시스템의 각 에이전트에 대해 개별 메시지 기록을 유지하고 싶을 수 있습니다. 애플리케이션이 이러한 경우라면 subgraph를 호출하는 node 함수를 정의해야 합니다. 이 함수는 subgraph를 호출하기 전에 입력(parent) state를 subgraph state로 변환하고, node에서 state 업데이트를 반환하기 전에 결과를 다시 parent state로 변환해야 합니다.

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"]})  
    # 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()

전체 예제: 다른 state schema

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

for chunk in graph.stream({"foo": "foo"}, subgraphs=True):
    print(chunk)

((), {'node_1': {'foo': 'hi! foo'}})
(('node_2:9c36dd0f-151a-cb42-cbad-fa2f851f9ab7',), {'grandchild_1': {'my_grandchild_key': 'hi Bob, how are you'}})
(('node_2:9c36dd0f-151a-cb42-cbad-fa2f851f9ab7',), {'grandchild_2': {'bar': 'hi! foobaz'}})
((), {'node_2': {'foo': 'hi! foobaz'}})

전체 예제: 다른 state schema (두 단계의 subgraph)

이것은 두 단계의 subgraph가 있는 예제입니다: 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()

for chunk in parent_graph.stream({"my_key": "Bob"}, subgraphs=True):
    print(chunk)

((), {'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 graph as a node

Parent graph와 subgraph가 schema의 공유 state key(channel)를 통해 통신할 수 있는 경우, graph를 다른 graph의 node로 추가할 수 있습니다. 예를 들어, 멀티 에이전트 시스템에서 에이전트들은 종종 공유 messages key를 통해 통신합니다. SQL agent graph

Subgraph가 parent graph와 state key를 공유하는 경우, 다음 단계에 따라 graph에 추가할 수 있습니다:

Subgraph workflow(subgraph_builder)를 정의하고 컴파일합니다
Parent graph workflow를 정의할 때 add_node 메서드에 컴파일된 subgraph를 전달합니다

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)  
builder.add_edge(START, "node_1")
graph = builder.compile()

전체 예제: 공유 state schema

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

for chunk in graph.stream({"foo": "foo"}):
    print(chunk)

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

Add persistence

Parent graph를 컴파일할 때만 checkpointer를 제공하면 됩니다. LangGraph는 자동으로 checkpointer를 child subgraph에 전파합니다.

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

class State(TypedDict):
    foo: str

# Subgraph

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

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)
builder.add_edge(START, "node_1")

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

Subgraph가 자체 메모리를 가지도록 하려면 적절한 checkpointer 옵션으로 컴파일할 수 있습니다. 이는 멀티 에이전트 시스템에서 에이전트가 내부 메시지 기록을 추적하도록 하려는 경우 유용합니다:

subgraph_builder = StateGraph(...)
subgraph = subgraph_builder.compile(checkpointer=True)

View subgraph state

persistence를 활성화하면 적절한 메서드를 통해 graph state를 검사(checkpoint)할 수 있습니다. Subgraph state를 보려면 subgraphs 옵션을 사용할 수 있습니다. graph.get_state(config)를 통해 graph state를 검사할 수 있습니다. Subgraph state를 보려면 graph.get_state(config, subgraphs=True)를 사용할 수 있습니다.

중단되었을 때만 사용 가능 Subgraph state는 subgraph가 중단되었을 때만 볼 수 있습니다. Graph를 재개하면 subgraph state에 액세스할 수 없습니다.

중단된 subgraph state 보기

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

# 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)
parent_state = graph.get_state(config)

# This will be available only when the subgraph is interrupted.
# Once you resume the graph, you won't be able to access the subgraph state.
subgraph_state = graph.get_state(config, subgraphs=True).tasks[0].state

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

이것은 subgraph가 중단되었을 때만 사용할 수 있습니다. Graph를 재개하면 subgraph state에 액세스할 수 없습니다.

Stream subgraph outputs

스트리밍된 출력에 subgraph의 출력을 포함하려면 parent graph의 stream 메서드에서 subgraphs 옵션을 설정할 수 있습니다. 이렇게 하면 parent graph와 모든 subgraph의 출력이 스트리밍됩니다.

for chunk in graph.stream(
    {"foo": "foo"},
    subgraphs=True, 
    stream_mode="updates",
):
    print(chunk)

Subgraph에서 스트리밍

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

for chunk in graph.stream(
    {"foo": "foo"},
    stream_mode="updates",
    subgraphs=True, 
):
    print(chunk)

((), {'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'}})

Edit the source of this page on GitHub.

Connect these docs programmatically to Claude, VSCode, and more via MCP for real-time answers.

LangGraph v1.0

Get started

Capabilities

Production

LangGraph APIs

Setup

Invoke a graph from a node

Add a graph as a node

Add persistence

View subgraph state

Stream subgraph outputs

LangGraph v1.0

Get started

Capabilities

Production

LangGraph APIs

​Setup

​Invoke a graph from a node

​Add a graph as a node

​Add persistence

​View subgraph state

​Stream subgraph outputs

Setup

Invoke a graph from a node

Add a graph as a node

Add persistence

View subgraph state

Stream subgraph outputs