LangGraph Subgraphs: вложенные графы

Зачем вкладывать графы друг в друга?

Представь агента, у которого внутри есть полноценный RAG-блок: достать документы → переранжировать → сгенерировать ответ. Если вписать эти узлы прямо в главный граф вперемешку с маршрутизацией и форматированием, получится плоское «макаронное» полотно: пятнадцать узлов, тридцать рёбер, и не понятно, где заканчивается одна логическая часть и начинается другая.

Те же проблемы решает декомпозиция в коде. Subgraph даёт три вещи:

Инкапсуляция — сложный блок выглядит снаружи как один узел; детали спрятаны.
Переиспользование — один раз собранный RAG-сабграф можно вставить в несколько разных агентов.
Композиция команд — каждый «агент» в мультиагентной системе — это отдельный subgraph, а супервайзер связывает их.

ℹ️ Subgraph — это просто скомпилированный граф

Никакого особого типа нет. Subgraph — это обычный граф, собранный через StateGraph и compile(), который мы используем как узел в другом графе. Раз скомпилированный граф — это Runnable (помнишь урок про compile()?), его можно вставить туда же, куда и обычную функцию-узел.

Граф как узел: модель вложенности

Идея проста: там, где обычно стоит узел-функция, может стоять целый граф. Когда поток управления доходит до такого узла, родитель «проваливается» внутрь сабграфа, тот отрабатывает свой START → ... → END, и управление возвращается родителю.

100%

колёсико — масштаб · зажать и тянуть — перемещение

Снаружи родитель видит просто узел rag — что у него внутри пять шагов, ему неважно. Это и есть инкапсуляция: сложность спрятана за «одним прямоугольником». Главный вопрос подключения — как родитель и сабграф обмениваются состоянием. Здесь два сценария.

Способ 1: общая схема состояния

Самый простой случай — когда у родителя и сабграфа есть общие ключи состояния. Тогда скомпилированный сабграф добавляется как узел напрямую: LangGraph сам передаёт общие ключи внутрь и вливает изменённые обратно.

Сабграф с общими ключами — добавляем как узел напрямую

python

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

# Общая схема: оба графа понимают ключи documents и answer
class State(TypedDict):
    question: str
    documents: list[str]
    answer: str

# --- СABГРАФ ---
def retrieve(state: State) -> dict:
    return {"documents": [f"док про {state['question']}"]}

def generate(state: State) -> dict:
    return {"answer": f"Ответ на основе {len(state['documents'])} док."}

sub = StateGraph(State)
sub.add_node("retrieve", retrieve)
sub.add_node("generate", generate)
sub.add_edge(START, "retrieve")
sub.add_edge("retrieve", "generate")
sub.add_edge("generate", END)
rag_subgraph = sub.compile()        # ← обычная компиляция

# --- РОДИТЕЛЬ ---
parent = StateGraph(State)
parent.add_node("rag", rag_subgraph)   # ← компилированный граф КАК УЗЕЛ
parent.add_edge(START, "rag")
parent.add_edge("rag", END)
graph = parent.compile()

print(graph.invoke({"question": "LangGraph", "documents": [], "answer": ""}))
# сабграф отработал внутри узла "rag" и вернул answer в общее состояние

Поскольку схемы совпадают, ключи «протекают» прозрачно: родитель кладёт question, сабграф читает его, пишет documents и answer, а родитель получает их обратно. Никакого ручного маппинга не нужно.

Способ 2: разные схемы состояния

Чаще сабграф — самостоятельный модуль со своей схемой, не совпадающей с родителем. Например, RAG-сабграф мыслит в терминах query/result, а родитель — в messages. Передавать состояние напрямую нельзя: общих ключей нет. Тогда сабграф вызывают внутри обычного узла-обёртки, который переводит состояние туда и обратно.

Сабграф со своей схемой — обёртка транслирует состояние

python

# У сабграфа СВОЯ схема
class RagState(TypedDict):
    query: str
    result: str

rag_subgraph = build_rag().compile()   # принимает query, возвращает result

# У родителя ДРУГАЯ схема
class ParentState(TypedDict):
    question: str
    answer: str

def call_rag(state: ParentState) -> dict:
    # 1. родитель → сабграф
    sub_in = {"query": state["question"]}
    # 2. запускаем сабграф как обычный Runnable
    sub_out = rag_subgraph.invoke(sub_in)
    # 3. сабграф → родитель
    return {"answer": sub_out["result"]}

parent = StateGraph(ParentState)
parent.add_node("rag", call_rag)       # ← узел-обёртка, а не сам сабграф
parent.add_edge(START, "rag")
parent.add_edge("rag", END)
graph = parent.compile()

✅ Как выбрать способ

Делят так. Общие ключи (сабграф — органичная часть того же потока) → добавляй компилированный граф как узел напрямую. Разные схемы (сабграф — независимый переиспользуемый модуль) → оборачивай в узел-функцию с явным маппингом. Второй вариант чуть многословнее, зато сабграф остаётся автономным: его легко тестировать и вставлять в другие агенты.

Изоляция состояния и просмотр сабграфа

У сабграфа — собственное состояние. В способе 2 это очевидно (схемы разные), но и в способе 1 наружу «протекают» только общие ключи: приватные поля сабграфа остаются внутри и родителю не видны. Это и есть инкапсуляция на уровне состояния.

При включённом checkpointer чекпойнты сабграфа сохраняются в отдельном пространстве имён (checkpoint_ns) внутри того же треда. По умолчанию get_state показывает только верхний уровень; чтобы заглянуть внутрь сабграфа, явно попроси вложенные состояния:

Просмотр состояния вместе с сабграфами

python

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

# Только состояние родителя
graph.get_state(config)

# Состояние родителя + вложенных сабграфов
graph.get_state(config, subgraphs=True)

# Стриминг тоже умеет «проваливаться» внутрь сабграфа
for chunk in graph.stream(inp, config, subgraphs=True):
    print(chunk)     # увидишь шаги и родителя, и сабграфа

⚠️ Не компилируй сабграф со своим checkpointer'ом

Когда сабграф используется внутри родителя, ему не нужен собственный checkpointer — память подключается на верхнем уровне, в compile() родителя, и распространяется на вложенные графы. Передашь checkpointer и туда, и сюда — получишь конфликт. Компилируй сабграф без него: sub.compile().

Сабграфы как мультиагентные команды

Самое мощное применение — мультиагентные системы. Каждый агент (исследователь, кодер, ревьюер) — это отдельный subgraph со своей логикой и инструментами. Сверху стоит граф-супервайзер, который условными рёбрами решает, какому агенту передать работу.

Супервайзер маршрутизирует между агентами-сабграфами

python

researcher = build_researcher().compile()   # агент-исследователь (subgraph)
coder      = build_coder().compile()         # агент-программист (subgraph)

team = StateGraph(TeamState)
team.add_node("supervisor", supervisor_node)
team.add_node("researcher", researcher)      # каждый агент — узел-сабграф
team.add_node("coder", coder)

team.add_edge(START, "supervisor")
team.add_conditional_edges("supervisor", route_to_agent,
                           {"research": "researcher", "code": "coder", "done": END})
team.add_edge("researcher", "supervisor")    # агент отчитался — назад к супервайзеру
team.add_edge("coder", "supervisor")
app = team.compile()

Это прямой задел на модуль про мультиагентные системы: тот же приём «агент = subgraph, супервайзер сверху» лежит в основе паттерна Supervisor, который мы разберём в разделе «Паттерны агентов».

Когда НЕ стоит дробить на сабграфы

Ситуация	Subgraph
Логический блок переиспользуется в нескольких графах	✅ да
Мультиагентная система (агент = модуль)	✅ да
Большой блок засоряет схему — хочется инкапсулировать	✅ да
Два-три простых узла подряд	❌ оверинжиниринг
Просто хочется «красиво сгруппировать» без переиспользования	❌ лишний слой

Типичные ошибки

❌ Ошибка 1: разные схемы, но сабграф добавлен напрямую

Если у родителя и сабграфа нет общих ключей, а ты добавил add_node("x", subgraph) напрямую — данные не передадутся (нечему «протекать»). Нужна обёртка-функция с маппингом состояния.

❌ Ошибка 2: свой checkpointer у сабграфа

Компиляция сабграфа с отдельным checkpointer'ом при использовании внутри родителя ведёт к конфликту персистентности. Память — только на верхнем уровне.

❌ Ошибка 3: ожидать приватные поля сабграфа в родителе

Наружу выходят только общие ключи (способ 1) или то, что вернула обёртка (способ 2). Внутренние поля сабграфа родителю не видны — это by design. Хочешь значение наверх — положи его в общий/выходной ключ.

❌ Ошибка 4: дробление ради дробления

Сабграф добавляет слой косвенности. Заворачивать два простых узла в отдельный граф — усложнять без выгоды. Дроби, когда есть переиспользование, мультиагентность или реальная инкапсуляция.

Шпаргалка

Subgraphs — всё в одном месте

python

# Subgraph = обычный compiled граф, используемый как узел

subgraph = sub_builder.compile()      # БЕЗ своего checkpointer

# Способ 1: ОБЩАЯ схема (есть общие ключи) — добавляем граф напрямую
parent.add_node("rag", subgraph)

# Способ 2: РАЗНАЯ схема — обёртка с маппингом состояния
def call_sub(state: ParentState) -> dict:
    sub_out = subgraph.invoke({"query": state["question"]})   # родитель → сабграф
    return {"answer": sub_out["result"]}                      # сабграф → родитель
parent.add_node("rag", call_sub)

# Просмотр вложенного состояния
graph.get_state(cfg, subgraphs=True)
graph.stream(inp, cfg, subgraphs=True)

# Когда применять:
#  ✅ переиспользование, мультиагентность, инкапсуляция большого блока
#  ❌ два-три простых узла — не нужен лишний слой

# Память (checkpointer) — ТОЛЬКО на верхнем уровне, распространяется вниз
graph = parent.compile(checkpointer=saver)

Практическое задание

Собери вложенную архитектуру:

Задание: RAG как переиспользуемый сабграф

Собери сабграф retrieve → generate со схемой {query, result} и скомпилируй его отдельно. Проверь, что он работает сам по себе через invoke.
Сделай родительский граф со схемой {question, answer} и подключи сабграф через узел-обёртку (способ 2) с маппингом question → query и result → answer.
Перепиши на способ 1: дай родителю и сабграфу общую схему и добавь сабграф как узел напрямую. Сравни, сколько кода ушло.
Подключи MemorySaver на уровне родителя и выведи get_state(cfg, subgraphs=True) — найди в выводе состояние сабграфа.
Со звёздочкой: собери мини-команду из двух агентов-сабграфов и супервайзера, который условным ребром выбирает, кому передать запрос, и завершается в END.

Что дальше

🧵

Thread management

Несколько одновременных сессий

🗺️

Map-reduce паттерн

Параллельная обработка через Send

👥

Supervisor

Мультиагентная команда из сабграфов

🧩

Все уроки модуля 03

LangGraph целиком

Subgraphs:вложенные графы для сложной логики

Зачем вкладывать графы друг в друга?

Граф как узел: модель вложенности

Способ 1: общая схема состояния

Способ 2: разные схемы состояния

Изоляция состояния и просмотр сабграфа

Сабграфы как мультиагентные команды

Когда НЕ стоит дробить на сабграфы

Типичные ошибки

Шпаргалка

Практическое задание

Задание: RAG как переиспользуемый сабграф

Что дальше

Thread management

Map-reduce паттерн

Supervisor

Все уроки модуля 03

Subgraphs:
вложенные графы для сложной логики