서브그래프 사용하기 (Subgraphs)

문서 인덱스: 전체 문서 인덱스를 https://docs.langchain.com/llms.txt 에서 가져올 수 있습니다. 이후 페이지를 탐색할 때 이 파일을 사용해 사용 가능한 모든 페이지를 찾아보세요.

이 가이드는 서브그래프(Subgraph) 사용의 메커니즘을 설명합니다. 서브그래프는 다른 그래프에서 노드로 사용되는 그래프입니다.

서브그래프는 다음과 같은 경우에 유용합니다:

멀티 에이전트 시스템 구축
여러 그래프에서 노드 집합 재사용

설정

pip install -U langgraph

uv add langgraph

LangGraph 개발을 위한 LangSmith 설정 LangSmith에 가입하면 LangGraph 프로젝트의 문제를 빠르게 발견하고 성능을 개선할 수 있습니다. LangSmith를 사용하면 trace 데이터로 LangGraph로 구축한 LLM 앱을 디버깅·테스트·모니터링할 수 있습니다. LangSmith 시작 방법 에서 자세히 알아보세요.

서브그래프 통신 정의하기

서브그래프를 추가할 때는 부모 그래프(parent graph)와 서브그래프가 어떻게 통신하는지 정의해야 합니다.

노드 내부에서 서브그래프 호출하기

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

class SubgraphState(TypedDict):
    bar: str

# 서브그래프

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

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

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

# 서브그래프 정의
class SubgraphState(TypedDict):
    # 부모 그래프 상태와 공유되지 않는 키들

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

# 손자 그래프 (Grandchild graph)
from typing_extensions import TypedDict
from langgraph.graph.state import StateGraph, START, END

class GrandChildState(TypedDict):
    my_grandchild_key: str

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

서브그래프를 노드로 추가하기

예를 들어 멀티 에이전트 시스템에서 에이전트는 종종 공유 메시지 키로 통신합니다.

서브그래프 워크플로(subgraph_builder)를 정의하고 컴파일합니다.
컴파일된 서브그래프를 부모 그래프 워크플로를 정의할 때 add_node 메서드에 전달합니다.

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

class State(TypedDict):
    foo: str

# 서브그래프

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

# 부모 그래프

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

# 서브그래프 정의
class SubgraphState(TypedDict):
    foo: str  # 부모 그래프 상태와 공유
    bar: str  # SubgraphState에만 비공개

    # 그리고 공유 상태 키('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()

# 부모 그래프 정의

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

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

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

서브그래프 영속성 기능(인터럽트, 상태 검사, 스레드별 메모리)이 작동하려면 부모 그래프가 체크포인터와 함께 컴파일되어야 합니다. persistence 참고.

아래 예시들은 에이전트를 구축하는 일반적인 방법인 LangChain의 create_agent를 사용합니다. create_agent는 내부적으로 LangGraph 그래프를 생성하므로 모든 서브그래프 영속성 개념이 그대로 적용됩니다. 원시 LangGraph StateGraph로 구축하는 경우에도 동일한 패턴과 구성 옵션이 적용됩니다. 자세한 내용은 Graph API를 참고하세요.

상태 유지 (Stateful)

상태 유지 서브그래프는 부모 그래프의 체크포인터를 상속받아 인터럽트, 영속성, 상태 검사를 지원합니다. 두 상태 유지 모드는 상태가 보존되는 기간이 다릅니다.

호출별 (Per-invocation, 기본값)

대부분의 애플리케이션, 특히 서브에이전트가 도구(tool)로 호출되는 멀티 에이전트 시스템에 권장되는 모드입니다. 인터럽트, 영속성, 병렬 호출을 지원하면서 각 호출을 격리시킵니다.

각 호출을 격리하면서 처리하는 가장 일반적인 패턴으로, 특히 서브에이전트가 "고객 주문 조회"나 "문서 요약" 같은 일회성 요청을 처리하는 멀티 에이전트 시스템에서 유용합니다.

무상태 (Stateless)

평범한 함수 호출처럼 체크포인트 오버헤드 없이 서브에이전트를 실행하려면 이 모드를 사용하세요. 서브그래프는 일시 중지/재개할 수 없으며 내구적 실행의 이점도 얻지 못합니다. checkpointer=False로 컴파일합니다.

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

체크포인터 참조

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

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

기능	호출별 (기본값)	스레드별	무상태
`checkpointer=`	`None`	`True`	`False`
인터럽트 (HITL)	✅	✅	❌
다중 턴 메모리	❌	✅	❌

인터럽트 (HITL): 서브그래프는 interrupt()를 사용해 실행을 일시 중지하고 사용자 입력을 기다린 뒤, 중단된 지점부터 재개할 수 있습니다.
다중 턴 메모리: 서브그래프는 동일한 thread 내 여러 호출에 걸쳐 상태를 유지합니다. 각 호출은 새로 시작하는 대신 이전 호출이 끝난 지점부터 이어집니다.

서브그래프 상태 보기

영속성을 활성화하면 subgraphs 옵션을 사용해 서브그래프 상태를 검사할 수 있습니다. 무상태 체크포인팅(checkpointer=False)에서는 서브그래프 체크포인트가 저장되지 않으므로 서브그래프 상태를 사용할 수 없습니다.

서브그래프 상태를 보려면 LangGraph가 서브그래프를 정적으로 발견할 수 있어야 합니다. 즉, 서브그래프가 노드로 추가되거나 노드 내부에서 호출되어야 합니다. 서브그래프가 도구(tool) 함수나 다른 간접 호출(예: 서브에이전트 패턴) 내부에서 호출될 때는 작동하지 않습니다. 인터럽트는 중첩과 관계없이 여전히 최상위 그래프로 전파됩니다.

호출별 (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

스레드별 (Per-thread)

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

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

# 자체 영구 상태를 가진 서브그래프
subgraph_builder = StateGraph(MessagesState)
# ... 노드와 엣지 추가

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

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

# 누적된 서브그래프 상태 보기 (두 호출의 메시지 포함)
subgraph_state = graph.get_state(config, subgraphs=True).tasks[0].state  # [!code highlight]

서브그래프 출력 스트리밍

중첩된 그래프 실행을 관찰하려면 이벤트 스트리밍을 권장합니다. 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:

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

# 서브그래프 정의
class SubgraphState(TypedDict):
    foo: str
    bar: str

def subgraph_node_1(state: SubgraphState):