Подграфы: слой как отдельный граф | Курс LangGraph урок 7
6
Цель урока: вынести часть работы в отдельный граф, который собирают, проверяют и чинят без остального агента.
Необходимые знания:
1) урок 7 курса по LangChain: поток, режимы updates и messages, форма chunk при version="v2"
2) урок 1 этого курса: поток stream в режиме updates, create_agent собирает граф LangGraph
3) урок 2 этого курса: схема состояния, редьюсер operator.add, ключ без редьюсера, схемы входа и выхода input_schema и output_schema
4) урок 3 этого курса: Command с параметрами update и goto, статическое ребро рядом с Command, проверка имён из аннотации Command[Literal[...]] при компиляции, Send и копии узла
5) урок 5 этого курса: чекпойнтер, thread_id, get_state с полями next и tasks, задача как одно выполнение узла, повтор от прошлого checkpoint
6) урок 6 этого курса: interrupt(), возобновление вызовом с Command(resume=...), узел с паузой выполняется с первой строки, правила паузы 1 и 4
Чем закончился урок 6
В уроке 6 граф останавливался перед выполнением файла и ждал ответа человека. Все узлы стояли в одном графе: в задании урока 6 это collect_files, pick_file, review_run и run_file.
Агент, который разбирает чужой репозиторий, состоит из слоёв. Слой, это часть агента со своей задачей: карта репозитория, выполнение файлов, проверка гипотез. В каждом слое несколько узлов. Допустим, все слои собраны в один StateGraph. Тогда в его схеме лежат ключи всех слоёв, и отдельным вызовом слой не проверить.
В этом уроке слой становится подграфом. Подграф, это скомпилированный граф, который работает внутри другого графа: узлом или вызовом из функции узла. Граф, внутри которого работает подграф, называется родительским. Подграф собирается и проверяется сам по себе. Родительскому графу нужны только вход и выход подграфа. Поэтому слои могут писать разные люди: пока вход и выход слоя не меняются, родительский граф не зависит от узлов слоя.
Примеры работают на стенде из урока 1: положите папку mini_repo с файлами app.py, config.py, db.py и utils.py рядом с примерами, листинги файлов есть в уроке 1. Модель и ключ API в этом уроке не нужны.
Подграф как узел: общие ключи
Способ связи подграфа с родительским графом выбирается по схемам состояния:
1) у подграфа и родителя есть общие ключи, и переименовывать их не нужно. Тогда скомпилированный подграф передаётся в add_node вместо функции узла
2) общих ключей нет или ключи надо переименовать. Тогда подграф вызывается методом invoke внутри функции узла
Схема слоя карты ScanState делит с родителем ключи root и findings.
Пример 01_subgraph_node.py
"""Пример 1 урока 7: подграф как узел родительского графа.
Слой карты, это граф из узлов collect_files и scan_files. Его можно вызвать
отдельно, как обычный граф. Родительский граф добавляет скомпилированный слой
узлом scan. Подграф и родитель делят ключи root и findings, ключ files
есть только у подграфа. Перед подграфом узел start_note пишет в findings
строку о начале обхода. Слой собирается в двух вариантах: без схем входа и
выхода и со схемами ScanInput и ScanOutput.
"""
import operator
from pathlib import Path
from typing import Annotated
from langgraph.graph import END, START, StateGraph
from typing_extensions import TypedDict
REPO = Path(__file__).parent / "mini_repo"
INPUT_FINDINGS: list[list[str]] = []
class ScanState(TypedDict):
root: str
files: list[str]
findings: Annotated[list[str], operator.add]
class ScanInput(TypedDict):
root: str
class ScanOutput(TypedDict):
findings: list[str]
def collect_files(state: ScanState) -> dict:
"""Собирает имена файлов стенда и запоминает findings на входе подграфа."""
INPUT_FINDINGS.append(list(state["findings"]))
names = sorted(path.name for path in Path(state["root"]).glob("*.py"))
return {"files": names}
def scan_files(state: ScanState) -> dict:
"""Считает строки в каждом файле."""
findings = []
for name in state["files"]:
lines = len((Path(state["root"]) / name).read_text(encoding="utf-8").splitlines())
findings.append(f"{name}: строк {lines}")
return {"findings": findings}
def build_layer(with_io: bool):
"""Слой карты: без схем входа и выхода или со схемами ScanInput и ScanOutput."""
if with_io:
builder = StateGraph(ScanState, input_schema=ScanInput, output_schema=ScanOutput)
else:
builder = StateGraph(ScanState)
builder.add_node("collect_files", collect_files)
builder.add_node("scan_files", scan_files)
builder.add_edge(START, "collect_files")
builder.add_edge("collect_files", "scan_files")
builder.add_edge("scan_files", END)
return builder.compile()
class RepoState(TypedDict):
root: str
findings: Annotated[list[str], operator.add]
report: str
def start_note(state: RepoState) -> dict:
"""Пишет в находки строку о начале обхода."""
return {"findings": ["обход начат"]}
def make_report(state: RepoState) -> dict:
"""Сводит находки в одну строку отчёта."""
return {"report": f"находок {len(state['findings'])}"}
def build_parent(layer):
"""Родительский граф: start_note, слой узлом scan, make_report."""
builder = StateGraph(RepoState)
builder.add_node("start_note", start_note)
builder.add_node("scan", layer)
builder.add_node("make_report", make_report)
builder.add_edge(START, "start_note")
builder.add_edge("start_note", "scan")
builder.add_edge("scan", "make_report")
builder.add_edge("make_report", END)
return builder.compile()
# 1. Слой карты вызывается сам по себе, как обычный граф.
scan_layer = build_layer(with_io=False)
alone = scan_layer.invoke({"root": str(REPO)})
print("1. подграф отдельно, ключи результата:", sorted(alone))
print(" находки:", alone["findings"])
# 2 и 3. Тот же слой узлом родительского графа, без схем и со схемами.
for number, with_io in ((2, False), (3, True)):
INPUT_FINDINGS.clear()
graph = build_parent(build_layer(with_io))
result = graph.invoke({"root": str(REPO)})
title = "со схемами входа и выхода" if with_io else "без схем входа и выхода"
print(f"{number}. слой {title}")
print(" findings на входе подграфа:", INPUT_FINDINGS[0])
print(" ключи результата родителя:", sorted(result))
print(" находки родителя:", result["findings"][:3], "...")
print(" отчёт:", result["report"])
# Вывод:
# 1. подграф отдельно, ключи результата: ['files', 'findings', 'root']
# находки: ['app.py: строк 13', 'config.py: строк 6', 'db.py: строк 15', 'utils.py: строк 10']
# 2. слой без схем входа и выхода
# findings на входе подграфа: ['обход начат']
# ключи результата родителя: ['findings', 'report', 'root']
# находки родителя: ['обход начат', 'обход начат', 'app.py: строк 13'] ...
# отчёт: находок 6
# 3. слой со схемами входа и выхода
# findings на входе подграфа: []
# ключи результата родителя: ['findings', 'report', 'root']
# находки родителя: ['обход начат', 'app.py: строк 13', 'config.py: строк 6'] ...
# отчёт: находок 5
Пункт 1: слой карты вызван отдельно, как любой граф. Так слой проверяют тестом до того, как он встанет в родительский граф.
Пункт 2: подграф получил на входе значение findings родителя со строкой "обход начат". На вход подграфа приходят значения родителя по ключам, которые есть в схеме подграфа. Без схемы выхода подграф возвращает всё своё состояние, как в пункте 1. Родитель берёт из результата только ключи своей схемы. Ключ files есть только у подграфа, поэтому в результат родителя он не попал.
Строка "обход начат" стоит в находках дважды. Подграф вернул родителю итоговые значения общих ключей. Список findings вернулся целиком, вместе со строкой, полученной на входе. Родитель записал его правкой узла scan, и редьюсер operator.add дописал список к прежнему значению.
Пункт 3: слой собран со схемой входа ScanInput и схемой выхода ScanOutput из урока 2. Схема входа пропустила в подграф только root, поэтому findings внутри подграфа начался с пустого списка. Схема выхода вернула родителю только findings, и строка "обход начат" стоит один раз.
Схемы входа и выхода задают интерфейс слоя: что слой берёт у родителя и что отдаёт ему. Ключ files в интерфейс не входит, его получают только узлы слоя.
Без схем строка не задваивается ещё в двух случаях:
1) у общего ключа в родителе нет редьюсера. Тогда список из подграфа заменяет прежнее значение
2) подграф вызывается внутри узла, и родителю уходит только то, что вернула функция узла. Это второй способ
Подграф внутри узла: функция-обёртка между схемами
Второй способ нужен, когда у подграфа и родителя нет общих ключей или ключи надо переименовать. Подграф вызывается методом invoke в функции узла родителя. Документация называет такую функцию wrapper function, по-русски функция-обёртка. Обычно она переводит состояние родителя во вход подграфа, а результат подграфа в правку узла.
Пример 02_subgraph_in_node.py
"""Пример 2 урока 7: вызов подграфа внутри узла, функция-обёртка между схемами.
У родительского графа своя схема: repo и notes, общих ключей со слоем карты
scan_layer у неё нет. Узел run_scan вызывает scan_layer.invoke. Узел
start_note пишет в notes строку о начале обхода, как в примере 1.
"""
import operator
from pathlib import Path
from typing import Annotated
from langgraph.graph import END, START, StateGraph
from typing_extensions import TypedDict
REPO = Path(__file__).parent / "mini_repo"
class ScanState(TypedDict):
root: str
files: list[str]
findings: Annotated[list[str], operator.add]
def collect_files(state: ScanState) -> dict:
"""Собирает имена файлов стенда."""
names = sorted(path.name for path in Path(state["root"]).glob("*.py"))
return {"files": names}
def scan_files(state: ScanState) -> dict:
"""Считает строки в каждом файле."""
findings = []
for name in state["files"]:
lines = len((Path(state["root"]) / name).read_text(encoding="utf-8").splitlines())
findings.append(f"{name}: строк {lines}")
return {"findings": findings}
scan_builder = StateGraph(ScanState)
scan_builder.add_node("collect_files", collect_files)
scan_builder.add_node("scan_files", scan_files)
scan_builder.add_edge(START, "collect_files")
scan_builder.add_edge("collect_files", "scan_files")
scan_builder.add_edge("scan_files", END)
scan_layer = scan_builder.compile()
class ProjectState(TypedDict):
repo: str
notes: Annotated[list[str], operator.add]
def start_note(state: ProjectState) -> dict:
"""Пишет в заметки строку о начале обхода."""
return {"notes": ["обход начат"]}
def run_scan(state: ProjectState) -> dict:
"""Функция-обёртка: вход подграфа из состояния родителя и результат обратно."""
scan_input = {"root": state["repo"], "files": [], "findings": []}
scan_result = scan_layer.invoke(scan_input)
return {"notes": [f"карта: {line}" for line in scan_result["findings"]]}
builder = StateGraph(ProjectState)
builder.add_node("start_note", start_note)
builder.add_node("run_scan", run_scan)
builder.add_edge(START, "start_note")
builder.add_edge("start_note", "run_scan")
builder.add_edge("run_scan", END)
graph = builder.compile()
result = graph.invoke({"repo": str(REPO), "notes": []})
print("ключи результата:", sorted(result))
for line in result["notes"]:
print(" ", line)
# Вывод:
# ключи результата: ['notes', 'repo']
# обход начат
# карта: app.py: строк 13
# карта: config.py: строк 6
# карта: db.py: строк 15
# карта: utils.py: строк 10
Функция-обёртка run_scan передала подграфу путь к стенду под ключом root и вернула родителю только новые строки. Поэтому "обход начат" стоит один раз, а ключей подграфа в результате нет.
Код функции-обёртки правят при каждом изменении схемы слоя. Подграфу как узлу такой код не нужен, но подграф зависит от имён ключей родителя и от редьюсеров родителя.
Агент из create_agent тоже скомпилированный граф, урок 1. Его можно добавить узлом, если у родителя есть ключ messages. Тогда агент и родитель делят историю сообщений. Если агенту нужна своя история, вызовите его внутри узла и верните родителю последний ответ. Всё, что дальше сказано о чекпойнтере подграфа, действует и на такого агента.
Пауза внутри подграфа
Слой выполнения из урока 6 тоже выносится в подграф. Чекпойнтер передаётся только родителю.
Checkpoints подграфа сохраняет чекпойнтер родителя, отдельно от checkpoints самого родителя. Для этого у каждого checkpoint есть пространство имён, ключ checkpoint_ns в config. У родителя это пустая строка. У подграфа это имя узла родителя, двоеточие и id задачи, в которой выполняется подграф.
Пример 03_pause_in_subgraph.py
"""Пример 3 урока 7: пауза внутри подграфа и возобновление.
Слой выполнения run_layer, это подграф из узлов plan_run, review_run и
run_file. Узел review_run ставит паузу перед выполнением файла, как в уроке 6.
У подграфа нет своего чекпойнтера, у родителя InMemorySaver. Две сборки
родителя: подграф узлом run_layer и подграф внутри узла call_run_layer.
Список CALLS показывает, какие узлы выполнялись при каждом вызове.
"""
import subprocess
import sys
from pathlib import Path
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt
from typing_extensions import TypedDict
REPO = Path(__file__).parent / "mini_repo"
CALLS: list[str] = []
class RunState(TypedDict):
root: str
chosen: str
output: str
def plan_run(state: RunState) -> dict:
"""Предлагает файл для выполнения."""
CALLS.append("plan_run")
return {"chosen": "db.py"}
def review_run(state: RunState) -> dict:
"""Ставит паузу с предложением. Ответ "да" оставляет файл, другой ответ убирает."""
CALLS.append("review_run")
answer = interrupt({"файл": state["chosen"]})
return {"chosen": state["chosen"] if answer == "да" else ""}
def run_file(state: RunState) -> dict:
"""Выполняет файл стенда отдельным процессом."""
CALLS.append("run_file")
if not state["chosen"]:
return {"output": "не выполнен"}
done = subprocess.run(
[sys.executable, state["chosen"]],
cwd=state["root"],
capture_output=True,
text=True,
encoding="utf-8",
)
return {"output": f"код {done.returncode}"}
run_builder = StateGraph(RunState)
run_builder.add_node("plan_run", plan_run)
run_builder.add_node("review_run", review_run)
run_builder.add_node("run_file", run_file)
run_builder.add_edge(START, "plan_run")
run_builder.add_edge("plan_run", "review_run")
run_builder.add_edge("review_run", "run_file")
run_builder.add_edge("run_file", END)
run_layer = run_builder.compile()
def call_run_layer(state: RunState) -> dict:
"""Вызывает подграф внутри узла. Код до invoke выполняется заново при каждом возобновлении."""
CALLS.append("call_run_layer")
return run_layer.invoke(state)
def build(as_node: bool):
"""Родитель: подграф узлом run_layer или внутри узла call_run_layer."""
builder = StateGraph(RunState)
if as_node:
builder.add_node("run_layer", run_layer)
name = "run_layer"
else:
builder.add_node("call_run_layer", call_run_layer)
name = "call_run_layer"
builder.add_edge(START, name)
builder.add_edge(name, END)
return builder.compile(checkpointer=InMemorySaver())
START_STATE = {"root": str(REPO), "chosen": "", "output": ""}
CONFIG = {"configurable": {"thread_id": "run-1"}}
for number, as_node in enumerate((True, False), start=1):
graph = build(as_node)
print(f"{number}. подграф {'узлом' if as_node else 'внутри узла'}")
CALLS.clear()
result = graph.invoke(START_STATE, CONFIG)
print(" пауза:", result["__interrupt__"][0].value, "| выполнены:", CALLS)
parent = graph.get_state(CONFIG)
print(" родитель: next", parent.next, "| state задачи:", type(parent.tasks[0].state).__name__)
task = graph.get_state(CONFIG, subgraphs=True).tasks[0]
inner = task.state
namespace = inner.config["configurable"]["checkpoint_ns"]
print(" подграф: next", inner.next, "| chosen", inner.values["chosen"])
print(" пространство имён:", namespace.split(":")[0] + ":<id задачи>")
CALLS.clear()
result = graph.invoke(Command(resume="да"), CONFIG)
print(" после ответа выполнены:", CALLS, "|", result["chosen"], result["output"])
print(" задач в get_state после конца:", len(graph.get_state(CONFIG, subgraphs=True).tasks))
# Вывод:
# 1. подграф узлом
# пауза: {'файл': 'db.py'} | выполнены: ['plan_run', 'review_run']
# родитель: next ('run_layer',) | state задачи: dict
# подграф: next ('review_run',) | chosen db.py
# пространство имён: run_layer:<id задачи>
# после ответа выполнены: ['review_run', 'run_file'] | db.py код 0
# задач в get_state после конца: 0
# 2. подграф внутри узла
# пауза: {'файл': 'db.py'} | выполнены: ['call_run_layer', 'plan_run', 'review_run']
# родитель: next ('call_run_layer',) | state задачи: dict
# подграф: next ('review_run',) | chosen db.py
# пространство имён: call_run_layer:<id задачи>
# после ответа выполнены: ['call_run_layer', 'review_run', 'run_file'] | db.py код 0
# задач в get_state после конца: 0
Пауза из подграфа дошла до результата родителя под ключом __interrupt__, как в уроке 6. Ответ тоже передаётся родителю: вызовом с Command(resume=...) и тем же thread_id.
Строка "родитель": в next стоит узел с подграфом. В поле state его задачи лежит словарь, config подграфа с checkpoint_ns. По нему видно, что в узле стоит подграф и под каким пространством имён искать его checkpoints. Состояние подграфа отдаёт вызов get_state(config, subgraphs=True). Тогда в поле state задачи лежит StateSnapshot подграфа, такой же объект, какой get_state возвращает в уроке 5, с полями next и values. Дальше этот объект называется снимком подграфа, это строка "подграф".
LangGraph находит подграф внутри функции узла, если функция берёт его из глобальной переменной, как run_layer в call_run_layer, или из переменной замыкания, как layer в фабрике def make_node(layer). Подграф из словаря (LAYERS["scan"].invoke(...)), из атрибута self.layer или из аргумента functools.partial так не находит. Пауза и возобновление при этом работают, но в поле state задачи None, а get_state с пространством имён такого подграфа в config падает с ValueError. Чтобы состояние подграфа читалось, держите слой в переменной модуля или в замыкании фабрики узла. Поток с subgraphs=True, он разобран ниже, отдаёт события подграфа во всех этих случаях.
При возобновлении подграф выполнялся со своего последнего checkpoint. После шага с plan_run подграф записал checkpoint, как любой граф с чекпойнтером в уроке 5. Поэтому plan_run второй раз не выполнялся, а review_run выполнился с первой строки, как узел с паузой в уроке 6.
Во второй сборке узел call_run_layer выполнился заново с первой строки, как любой узел родителя, в котором случилась пауза. Подграф, вызванный методом invoke, выполнялся со своего последнего checkpoint, поэтому plan_run и здесь не повторился. Код функции-обёртки до вызова invoke выполняется при каждом возобновлении. Значит, к нему относится правило 4 из урока 6: побочные действия до паузы должны выдерживать повтор.
Правило 1 из урока 6, без except Exception вокруг паузы, действует и здесь. Пауза выходит из invoke подграфа исключением GraphInterrupt, наследником Exception. Поэтому try с except Exception вокруг вызова подграфа перехватит паузу, и запуск закончится без ответа человека.
После конца запуска задач в get_state родителя нет, и вызов get_state(config, subgraphs=True) снимка подграфа не отдаёт. Checkpoints подграфа при этом остаются в чекпойнтере под пространством имён с id задачи. Состояние подграфа законченного запуска отдаёт get_state с таким checkpoint_ns в config. Для подграфа внутри функции узла это работает, когда слой взят из глобальной переменной или замыкания. Значение checkpoint_ns берут при паузе из config в поле state задачи родителя. Та же строка лежит в ns событий потока, он разобран ниже. В следующем запуске у задачи с подграфом новый id, поэтому по умолчанию подграф начинает с пустого состояния. Другие варианты задаёт параметр checkpointer подграфа, он в следующем разделе.
Слой может выполняться несколькими копиями из Send, по копии на файл. Тогда у каждой копии своя пауза со своим id, и ответ на них передаётся словарём, как в примере 4 урока 6. get_state(config, subgraphs=True) в этом случае отдаёт снимок подграфа в каждой задаче.
В истории родителя подграф занимает один шаг, его внутренних шагов там нет. Повтор из урока 5 от checkpoint родителя перед узлом подграфа выполняет подграф заново с первого узла.
Чекпойнтер подграфа: None, True, False
Подграф, как любой граф, получает параметр checkpointer в compile. Значения для подграфа:
1) None, значение по умолчанию. Подграф записывает checkpoints чекпойнтером родителя в пространство имён с id задачи. Пауза и возобновление работают. В следующем запуске у задачи с подграфом новый id, и подграф начинает с пустого состояния
2) True. Состояние подграфа копится по thread: следующий запуск в том же thread получает состояние подграфа из прошлого
3) False. Подграф не сохраняет checkpoints и выполняется как вызов обычной функции
Для паузы, памяти между запусками и чтения состояния подграфа у родителя должен быть чекпойнтер.
Пример 04_checkpointer_modes.py
"""Пример 4 урока 7: параметр checkpointer у подграфа, None, True и False.
Подграф review_layer записывает файл в список visited, которого нет у родителя,
и ставит паузу с вопросом о файле. Скрипт запускает родителя с InMemorySaver
два раза в одном thread: для app.py и для db.py. В каждом запуске подграф,
стоящий узлом layer, встаёт на паузу, и скрипт отвечает "да".
Для каждого значения checkpointer видно, отдаёт ли get_state снимок подграфа
на паузе, какие его узлы выполняются при возобновлении и что попадает в
summary во втором запуске.
"""
import operator
from typing import Annotated
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt
from typing_extensions import TypedDict
CALLS: list[str] = []
class ReviewState(TypedDict):
file: str
visited: Annotated[list[str], operator.add]
summary: str
def note_file(state: ReviewState) -> dict:
"""Записывает файл в список visited."""
CALLS.append("note_file")
return {"visited": [state["file"]]}
def ask_human(state: ReviewState) -> dict:
"""Ставит паузу и после ответа перечисляет все файлы из visited."""
CALLS.append("ask_human")
interrupt({"файл": state["file"]})
return {"summary": ", ".join(state["visited"])}
class ParentState(TypedDict):
file: str
summary: str
def build(mode):
"""Подграф с заданным checkpointer, узлом layer в родителе с InMemorySaver."""
review_builder = StateGraph(ReviewState)
review_builder.add_node("note_file", note_file)
review_builder.add_node("ask_human", ask_human)
review_builder.add_edge(START, "note_file")
review_builder.add_edge("note_file", "ask_human")
review_builder.add_edge("ask_human", END)
review_layer = review_builder.compile(checkpointer=mode)
builder = StateGraph(ParentState)
builder.add_node("layer", review_layer)
builder.add_edge(START, "layer")
builder.add_edge("layer", END)
return builder.compile(checkpointer=InMemorySaver())
for number, mode in enumerate((None, True, False), start=1):
graph = build(mode)
config = {"configurable": {"thread_id": f"review-{number}"}}
print(f"{number}. checkpointer={mode}")
for name in ("app.py", "db.py"):
graph.invoke({"file": name, "summary": ""}, config)
task = graph.get_state(config, subgraphs=True).tasks[0]
CALLS.clear()
result = graph.invoke(Command(resume="да"), config)
print(
f" {name}: подграф на паузе {type(task.state).__name__}, "
f"при возобновлении {CALLS}, summary: {result['summary']}"
)
if mode is True:
print(" задач в get_state после конца:", len(graph.get_state(config, subgraphs=True).tasks))
layer_config = {"configurable": {"thread_id": "review-2", "checkpoint_ns": "layer"}}
saved = graph.get_state(layer_config)
print(" visited по checkpoint_ns='layer':", saved.values["visited"])
# Вывод:
# 1. checkpointer=None
# app.py: подграф на паузе StateSnapshot, при возобновлении ['ask_human'], summary: app.py
# db.py: подграф на паузе StateSnapshot, при возобновлении ['ask_human'], summary: db.py
# 2. checkpointer=True
# app.py: подграф на паузе StateSnapshot, при возобновлении ['ask_human'], summary: app.py
# db.py: подграф на паузе StateSnapshot, при возобновлении ['ask_human'], summary: app.py, db.py
# задач в get_state после конца: 0
# visited по checkpoint_ns='layer': ['app.py', 'db.py']
# 3. checkpointer=False
# app.py: подграф на паузе NoneType, при возобновлении ['note_file', 'ask_human'], summary: app.py
# db.py: подграф на паузе NoneType, при возобновлении ['note_file', 'ask_human'], summary: db.py
При None второй запуск начал подграф с пустого состояния, в summary один db.py. На паузе снимок подграфа доступен, при возобновлении выполнился только ask_human.
При True состояние прошлого запуска сохранилось: во втором summary стоят оба файла. Пространство имён такого подграфа, это имя узла layer, без id задачи. Накопленное состояние читается вызовом get_state с "checkpoint_ns": "layer" в config, последняя строка сборки 2. get_state(config, subgraphs=True) после конца запуска состояния подграфа не отдаёт: tasks пустой. Пример на странице документации про подграфы читает tasks[0].state после двух законченных запусков. В 1.2.16 такой код падает с IndexError.
Ограничения True:
1) вызов графа верхнего уровня, у которого нет родителя, с checkpointer=True падает с RuntimeError, значение годится только подграфу
2) две копии одного подграфа с True в одном шаге, например из Send, пишут в одно пространство имён. Ошибки нет, но сохраняется состояние только одной копии
3) пространство имён берётся из имени узла. Если переименовать узел layer, следующий запуск начнёт подграф с пустого состояния, а прежнее останется под старым именем
При False вызов get_state(config, subgraphs=True) снимка подграфа не отдаёт: на паузе в поле state задачи None. Поток с subgraphs=True отдаёт события такого подграфа с пространством имён layer:<id>, как при None. Страница документации про подграфы пишет, что с False паузы нет. В langgraph 1.2.16 пауза срабатывает, и ответ доходит до ask_human. Пауза выходит из такого подграфа исключением, как из обычной функции. Чекпойнтер родителя записывает эту паузу по id задачи узла layer. Своих checkpoints у подграфа нет, поэтому при возобновлении узел layer выполняется заново, и подграф начинает с первого узла: note_file выполнился второй раз.
None |
True |
False |
|
|---|---|---|---|
| пауза и возобновление | да | да | пауза есть, при возобновлении подграф выполняется с первого узла |
| память между запусками в thread | нет | да | нет |
| две копии одного подграфа в шаге | работают | сохраняется состояние одной копии | работают |
состояние подграфа в get_state |
на паузе, после конца по checkpoint_ns с id задачи |
на паузе, после конца по checkpoint_ns |
нет |
Слоям агента подходит None: слой получает вход и отдаёт результат, состояние прошлых запусков ему не нужно. True нужен слою, который ведёт историю между запусками, например переписку с человеком. False подходит короткому слою без пауз, шаги которого сохранять незачем.
Поток из подграфа и пространство имён
Без параметра subgraphs поток родителя отдаёт события только своих узлов. Параметр subgraphs=True у stream добавляет события изнутри подграфов. При version="v2" пространство имён события лежит в поле ns. У родителя это пустой кортеж. У подграфа с checkpointer по умолчанию и с False это кортеж из одной строки: имя узла родителя, двоеточие и id задачи, например ('scan:<id>',). У подграфа с True в этой строке нет двоеточия и id задачи, например ('layer',).
Пример 05_stream_namespaces.py
"""Пример 5 урока 7: поток из подграфа и пространство имён в событиях.
Родительский граф из примера 1 без start_note: узел scan, это слой карты
scan_layer, после него make_report. Поток в режиме updates вызывается два
раза: без subgraphs и с subgraphs=True. У каждого события печатается поле ns.
"""
import operator
from pathlib import Path
from typing import Annotated
from langgraph.graph import END, START, StateGraph
from typing_extensions import TypedDict
REPO = Path(__file__).parent / "mini_repo"
class ScanState(TypedDict):
root: str
files: list[str]
findings: Annotated[list[str], operator.add]
def collect_files(state: ScanState) -> dict:
"""Собирает имена файлов стенда."""
names = sorted(path.name for path in Path(state["root"]).glob("*.py"))
return {"files": names}
def scan_files(state: ScanState) -> dict:
"""Считает строки в каждом файле."""
findings = []
for name in state["files"]:
lines = len((Path(state["root"]) / name).read_text(encoding="utf-8").splitlines())
findings.append(f"{name}: строк {lines}")
return {"findings": findings}
scan_builder = StateGraph(ScanState)
scan_builder.add_node("collect_files", collect_files)
scan_builder.add_node("scan_files", scan_files)
scan_builder.add_edge(START, "collect_files")
scan_builder.add_edge("collect_files", "scan_files")
scan_builder.add_edge("scan_files", END)
scan_layer = scan_builder.compile()
class RepoState(TypedDict):
root: str
findings: Annotated[list[str], operator.add]
report: str
def make_report(state: RepoState) -> dict:
"""Сводит находки в одну строку отчёта."""
return {"report": f"находок {len(state['findings'])}"}
builder = StateGraph(RepoState)
builder.add_node("scan", scan_layer)
builder.add_node("make_report", make_report)
builder.add_edge(START, "scan")
builder.add_edge("scan", "make_report")
builder.add_edge("make_report", END)
graph = builder.compile()
def short(namespace: tuple[str, ...]) -> list[str]:
"""Заменяет id задачи в пространстве имён меткой: id меняется от запуска к запуску."""
return [part.split(":")[0] + ":<id>" for part in namespace]
for number, subgraphs in enumerate((False, True), start=1):
print(f"{number}. subgraphs={subgraphs}")
for chunk in graph.stream(
{"root": str(REPO)},
stream_mode="updates",
subgraphs=subgraphs,
version="v2",
):
for node, update in chunk["data"].items():
print(f" ns {short(chunk['ns'])} узел {node} ключи {sorted(update)}")
# Вывод:
# 1. subgraphs=False
# ns [] узел scan ключи ['findings', 'root']
# ns [] узел make_report ключи ['report']
# 2. subgraphs=True
# ns ['scan:<id>'] узел collect_files ключи ['files']
# ns ['scan:<id>'] узел scan_files ключи ['findings']
# ns [] узел scan ключи ['findings', 'root']
# ns [] узел make_report ключи ['report']
В пункте 1 подграф виден одним событием узла scan, это правка узла с ключами root и findings. Схемы выхода у слоя в этом примере нет, поэтому подграф вернул все свои ключи, а родитель взял из них ключи своей схемы. В пункте 2 перед этим событием стоят события узлов подграфа с пространством имён scan:<id>.
Без subgraphs=True слой не даёт событий, пока не закончит работу. Слой, который разбирает сотни файлов, надолго оставляет поток пустым. С subgraphs=True приложение показывает, какой узел слоя уже выполнился.
Отбирайте события по имени узла, это часть ns до двоеточия. У подграфа внутри подграфа в ns два элемента: сначала узел родителя, потом узел первого подграфа, внутри которого стоит второй.
Если агент из create_agent добавлен узлом, subgraphs=True нужен и в режиме messages. Без него сообщения агента приходят в поток родителя все сразу, после того как агент закончит работу. С ним каждое сообщение приходит, как только выполнится узел агента, который его вернул. Ответ модели, которая отдаёт текст потоком, при этом приходит по частям.
Страница документации про подграфы показывает поток методом stream_events(..., version="v3"). В langgraph 1.2.16 вариант version="v3" помечен как экспериментальный, поэтому в примерах стоит stream.
Переход в родительский граф: Command.PARENT
Подграф заканчивает работу, когда выполнять в нём больше нечего, и родитель идёт по рёбрам из узла, в котором стоит подграф. Иногда слой должен остановиться на середине и передать управление другому узлу родителя. Например, слой проверки нашёл файл, который нельзя выполнять без человека.
Тогда узел подграфа возвращает Command с graph=Command.PARENT, урок 3. Параметр goto называет узел ближайшего родительского графа, а update становится правкой узла, в котором стоит подграф, в примере ниже это check.
Пример 06_parent_command.py
"""Пример 6 урока 7: переход из подграфа в узел родителя, Command.PARENT.
Слой проверки check_layer собирает файлы стенда узлом collect_files. Узел
check_files ищет файлы из RISKY и возвращает Command с graph=Command.PARENT:
при опасном файле переход в узел ask_human родителя, иначе в make_report.
Сборки родителя: без ребра из узла check и с ребром check -> make_report.
Ещё два пункта: вызов слоя без родителя падает, аннотация с узлом родителя не
проходит компиляцию подграфа.
"""
import operator
from pathlib import Path
from typing import Annotated, Literal
from langgraph.errors import ParentCommand
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command
from typing_extensions import TypedDict
REPO = Path(__file__).parent / "mini_repo"
CALLS: list[str] = []
RISKY = {"db.py"}
class CheckState(TypedDict):
root: str
files: list[str]
findings: Annotated[list[str], operator.add]
def collect_files(state: CheckState) -> dict:
"""Собирает имена файлов стенда и пишет их число в находки."""
CALLS.append("collect_files")
names = sorted(path.name for path in Path(state["root"]).glob("*.py"))
return {"files": names, "findings": [f"файлов: {len(names)}"]}
def check_files(state: CheckState) -> Command:
"""Выбирает узел родителя: ask_human при опасном файле, иначе make_report."""
CALLS.append("check_files")
risky = [name for name in state["files"] if name in RISKY]
if risky:
return Command(
update={"findings": [f"{name}: нужен человек" for name in risky]},
goto="ask_human",
graph=Command.PARENT,
)
return Command(
update={"findings": ["опасных файлов нет"]},
goto="make_report",
graph=Command.PARENT,
)
check_builder = StateGraph(CheckState)
check_builder.add_node("collect_files", collect_files)
check_builder.add_node("check_files", check_files)
check_builder.add_edge(START, "collect_files")
check_builder.add_edge("collect_files", "check_files")
check_layer = check_builder.compile()
class RepoState(TypedDict):
root: str
findings: Annotated[list[str], operator.add]
def ask_human(state: RepoState) -> dict:
"""В полном агенте здесь стоит пауза из урока 6."""
CALLS.append("ask_human")
return {}
def make_report(state: RepoState) -> dict:
"""Узел отчёта."""
CALLS.append("make_report")
return {}
def build(edge_after_check: bool):
"""Родитель со слоем check. Ребро из check добавляется по флагу."""
builder = StateGraph(RepoState)
builder.add_node("check", check_layer)
builder.add_node("ask_human", ask_human)
builder.add_node("make_report", make_report)
builder.add_edge(START, "check")
if edge_after_check:
builder.add_edge("check", "make_report")
builder.add_edge("ask_human", "make_report")
builder.add_edge("make_report", END)
return builder.compile()
START_STATE = {"root": str(REPO), "findings": []}
# 1. Опасный файл есть: переход в ask_human.
CALLS.clear()
result = build(edge_after_check=False).invoke(START_STATE)
print("1. выполнены:", CALLS)
print(" находки:", result["findings"])
# 2. Опасных файлов нет: переход прямо в make_report.
RISKY.clear()
CALLS.clear()
result = build(edge_after_check=False).invoke(START_STATE)
print("2. выполнены:", CALLS)
print(" находки:", result["findings"])
# 3. У узла check есть ребро в make_report, и оно тоже срабатывает.
RISKY.add("db.py")
CALLS.clear()
build(edge_after_check=True).invoke(START_STATE)
print("3. с ребром check -> make_report выполнены:", CALLS)
# 4. Слой с Command.PARENT сам по себе.
try:
check_layer.invoke(START_STATE)
except ParentCommand as error:
print("4. отдельный вызов слоя:", type(error).__name__)
# 5. Аннотация возврата с узлом родителя.
def check_annotated(state: CheckState) -> Command[Literal["ask_human"]]:
"""Тот же переход с аннотацией возврата."""
return check_files(state)
annotated = StateGraph(CheckState)
annotated.add_node("check_files", check_annotated)
annotated.add_edge(START, "check_files")
try:
annotated.compile()
except ValueError as error:
print("5. компиляция подграфа:", error)
# Вывод:
# 1. выполнены: ['collect_files', 'check_files', 'ask_human', 'make_report']
# находки: ['db.py: нужен человек']
# 2. выполнены: ['collect_files', 'check_files', 'make_report']
# находки: ['опасных файлов нет']
# 3. с ребром check -> make_report выполнены: ['collect_files', 'check_files', 'ask_human', 'make_report', 'make_report']
# 4. отдельный вызов слоя: ParentCommand
# 5. компиляция подграфа: Found edge ending at unknown node `ask_human`
Пункты 1 и 2: check_files выбрал узел родителя по списку RISKY. В находках только строка из update. Строка "файлов: 4", которую collect_files пишет в findings подграфа, в родителя не попала: подграф передал управление родителю, не вернув своего состояния. Всё, что нужно родителю, кладите в update. Переход работает и из подграфа, вызванного внутри узла. Тогда код функции-обёртки после invoke не выполняется.
Пункт 3: ребро check -> make_report сработало вместе с переходом из Command, и make_report выполнился дважды. Это поведение Command из урока 3: переход добавляется к рёбрам узла и их не отменяет. Условное ребро из узла check тоже срабатывает. Отсюда правило: у узла с подграфом, из которого выходят переходом Command.PARENT, исходящих рёбер не ставят, и все выходы из слоя идут такими переходами. В основной сборке примера у check рёбер нет, оба выхода из check_files идут в родителя.
Пункт 4: вызов слоя без родителя падает с ParentCommand. В тесте отдельного слоя переход ловят этим исключением. ParentCommand наследует Exception, как GraphInterrupt, и except Exception в функции-обёртке его тоже перехватит.
Пункт 5: аннотация Command[Literal["ask_human"]] не прошла компиляцию подграфа. Компиляция проверяет имена из аннотации, урок 3, а узла ask_human в подграфе нет. Поэтому check_files аннотирован типом Command без списка узлов. Страницы документации про Graph API и про использование Graph API аннотируют такой узел Command[Literal["other_subgraph"]] с узлом родителя. В 1.2.16 эта аннотация тоже не проходит компиляцию подграфа. Полный пример на странице документации про использование Graph API обходится без неё.
Страница документации про Graph API пишет, что общему ключу, в который идёт правка из update такого перехода, обязательно нужен редьюсер в родителе. В 1.2.16 при одной правке ключа за шаг ошибки без редьюсера нет: правка узла check заменяет значение ключа. Если в одном шаге ключ правят две задачи, например две копии слоя из Send, без редьюсера запуск падает с InvalidUpdateError, как в уроке 2.
Command.PARENT связывает слой с родителем теснее, чем общие ключи: слой ссылается на имена узлов родителя. В примере 6 check_files последний узел слоя, поэтому тот же выбор можно сделать без Command.PARENT. Тогда слой пишет признак в общий ключ, а родитель выбирает следующий узел условным ребром из check. Так и делайте, если слою достаточно сообщить результат. Command.PARENT оправдан, когда решение принимает узел в середине слоя и оставшиеся узлы слоя выполнять не надо.
Слой агента как подграф
Каждый слой агента собирается своим StateGraph и проверяется вызовом invoke на стенде, без родителя. В родительском графе слой занимает один узел.
Способ связи выбирается по интерфейсу слоя:
1) слой карты берёт путь к репозиторию и отдаёт находки. Ключи у него с родителем общие, поэтому он стоит узлом со схемами входа и выхода, как в пункте 3 примера 1
2) слой выполнения держит результат в своих ключах chosen и output. Родителю результат нужен под другим ключом: одной строкой в runs, которого в схеме слоя нет. Ключи надо переименовать, поэтому функция-обёртка вызывает слой методом invoke, как в примере 2. Пауза стоит внутри слоя, как в примере 3
3) слой проверки гипотез может передать управление другому узлу родителя, например узлу с вопросом человеку. Если выбор делает последний узел слоя, хватает признака в общем ключе и условного ребра родителя. Command.PARENT оправдан, когда выбор делает узел в середине слоя и оставшиеся узлы слоя выполнять не надо
Все три слоя компилируются с checkpointer по умолчанию, None. SqliteSaver из урока 5 передаётся одному родителю, и checkpoints слоёв записываются в тот же файл под своими пространствами имён. Пауза в слое выполнения переживает закрытие соединения так же, как пауза в уроке 6. Новый процесс открывает файл, собирает граф заново и возобновляет запуск, а подграф выполняется со своего checkpoint.
Слой с паузой тоже проверяется отдельно. Один StateGraph слоя компилируется дважды: для родителя с checkpointer по умолчанию и для теста с InMemorySaver, потому что паузе нужен чекпойнтер, а у отдельного слоя родителя нет. Тест запускает слой с thread_id, получает паузу и отвечает на неё, как в уроке 6.
Практическое задание
Соберите родительский граф из двух слоёв: карты и выполнения.
Требования:
1) слой карты map_layer, это подграф из узлов collect_files и scan_files из примера 1 без схем входа и выхода, в родителе он стоит узлом map
2) в схеме родителя ключ findings без редьюсера: findings: list[str]
3) слой выполнения run_layer из примера 3 со своей схемой RunState. Его вызывает узел родителя run_step: функция-обёртка передаёт подграфу root и возвращает родителю строку ": " в ключ runs с редьюсером operator.add
4) рёбра родителя: START -> map -> run_step -> END, чекпойнтер InMemorySaver
5) вход запуска: {"root": <путь к стенду>, "findings": ["обход начат"], "runs": []}, на паузу скрипт отвечает "да"
Как проверить результат:
1) первый вызов вернул __interrupt__ со значением {"файл": "db.py"}
2) на паузе get_state(config, subgraphs=True) даёт задачу run_step, у её снимка подграфа next равен ("review_run",)
3) после ответа в findings пять строк, "обход начат" среди них одна, а runs равен ["db.py: код 0"]
4) при возобновлении узел run_step выполнился, а plan_run нет
5) map_layer.invoke({"root": <путь к стенду>})["findings"] совпадает с findings родителя без первой строки
6) поток stream(..., stream_mode="updates", subgraphs=True, version="v2") в новом thread отдаёт события с пространствами имён, которые начинаются с map: и run_step:. На паузу в этом thread отвечать не нужно, события run_step: приходят до неё
7) если поставить у findings в родителе редьюсер operator.add, проверки 3 и 5 перестают сходиться
Подсказка: отметки выполнения узлов собирайте списком CALLS, как в примере 3.
Итоги урока
Подграф, это скомпилированный граф внутри родительского, узлом или вызовом из функции узла. Его можно вызвать и проверить отдельно.
Подграф как узел без схем входа и выхода возвращает общие ключи целиком. Если у такого ключа в родителе редьюсер operator.add, полученное на входе задваивается, а без редьюсера список заменяет прежнее значение. Схемы входа и выхода задают интерфейс слоя и убирают дубль. Когда общих ключей нет или их надо переименовать, подграф вызывается внутри узла функцией-обёрткой.
Пауза из подграфа доходит до результата родителя, ответ передаётся вызовом родителя с Command(resume=...). При возобновлении подграф с checkpointer по умолчанию выполняется со своего checkpoint, а функция-обёртка заново с первой строки. Снимок подграфа на паузе отдаёт get_state(config, subgraphs=True). Для подграфа внутри функции узла слой должен лежать в глобальной переменной или замыкании.
checkpointer подграфа: с None по умолчанию каждый запуск начинает подграф с пустого состояния, True копит состояние по thread, False не сохраняет checkpoints подграфа. Паузе, памяти между запусками и чтению состояния подграфа нужен чекпойнтер родителя. С False состояние подграфа не читается, а пауза в 1.2.16 срабатывает. При возобновлении такой подграф выполняется заново с первого узла.
stream(..., subgraphs=True) добавляет события подграфов с пространством имён в поле ns. Command(graph=Command.PARENT) передаёт управление узлу родителя без состояния подграфа. Рёбра узла с подграфом срабатывают вместе с переходом. Аннотация с узлом родителя не проходит компиляцию подграфа.
В уроке 8 разберу падение узла: повторные попытки, таймауты, обработчик ошибки, остановку по запросу и отладку графа без внешнего сервиса.
Код урока
Примеры этого урока лежат в репозитории курса, папка lesson_07. Закреплённые версии, на которых получен вывод в тексте, лежат в requirements.txt в корне репозитория.
Предыдущий урок: Пауза и человек в цикле
Следующий урок: Когда узел падает: повторы, таймауты и отладка графа
Подписывайтесь на мой Telegram канал
Если вам нужен ментор и вы хотите научиться разрабатывать AI агентов, пишите, обсудим условия
Авторизуйтесь, чтобы оставить комментарий.
Нет комментариев.
Тут может быть ваша реклама
Пишите info@aisferaic.ru
Похожие туториалы
Когда узел падает: повторы, таймауты и отладка графа | Курс LangGraph урок 8
0,0
3
Checkpointer: сохранение, продолжение и возврат назад | Курс LangGraph урок 5
5,0
6
