Ветвление: условные рёбра, Command и Send | Курс LangGraph урок 3
9
Цель урока: развести работу по веткам и запустить неизвестное заранее число одинаковых задач.
Необходимые знания:
1) урок 1 этого курса: StateGraph, узлы и рёбра, START и END, компиляция, шаг выполнения, схема графа из get_graph().draw_mermaid()
2) урок 2 этого курса: схема состояния на TypedDict, редьюсер, Annotated с operator.add
3) Python на уровне джуниора: Literal, pathlib, генераторы списков
Чем закончился урок 2
Состояние из урока 2 выдерживает параллельную работу, но рёбра заданы при сборке, поэтому граф идёт одним путём при любом входе.
А путь должен зависеть от репозитория: четыре файла можно разобрать подробно, каждый по отдельности, а десять тысяч только быстрым разбором. Сколько в репозитории файлов, становится известно только при запуске, поэтому и выбирать путь граф должен тогда же.
Условное ребро выбирает следующий узел по состоянию. Command позволяет узлу самому объявить переход. Send запускает один и тот же узел столько раз, сколько элементов нашлось в списке.
Примеры работают на стенде из урока 1: положите папку mini_repo с файлами app.py, config.py, db.py и utils.py рядом с примерами, листинги файлов есть в уроке 1. Модель и ключ API в этом уроке не нужны.
Условное ребро и функция маршрутизации
Развилку задаёт метод add_conditional_edges: он принимает имя узла, после которого стоит развилка, и функцию маршрутизации. Функция получает состояние графа и возвращает значение, по которому LangGraph выбирает следующий узел.
Возврат функции читается как имя узла: вернули "deep_scan", выполнится узел deep_scan. Вернули список имён, все перечисленные узлы выполнятся одним шагом, параллельно.
Третий аргумент add_conditional_edges, словарь соответствия, переводит возврат функции в имя узла. Тогда функция может вернуть любое хешируемое значение, например признак True или False. Значение, которого нет в словаре, роняет запуск с KeyError. Вместо словаря можно передать список узлов, имя вне списка даёт тот же KeyError.
Развилка от START даёт условную точку входа: первый узел запуска выбирается по входным данным.
Возврат END из функции маршрутизации заканчивает этот путь: следующего узла после развилки нет. Если других задач в графе не осталось, запуск заканчивается. На этом стоит цикл из урока 4: ребро назад, в узел, который уже выполнился, и выход на END по условию.
Пример 01_conditional_edges.py
"""Пример 1 урока 3: условное ребро и функция маршрутизации.
Граф обходит стенд mini_repo и после сбора файлов выбирает, каким разбором идти
дальше: быстрым или подробным. Выбор делает функция маршрутизации. Варианты в
примере: имя узла прямо из функции, словарь соответствия, условная точка
входа и маршрут сразу в два узла.
"""
import operator
from pathlib import Path
from typing import Annotated, Literal
from langgraph.graph import END, START, StateGraph
from typing_extensions import TypedDict
REPO = Path(__file__).parent / "mini_repo"
class ScanState(TypedDict):
root: str
limit: int
files: list[str]
report: 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 quick_scan(state: ScanState) -> dict:
"""Короткий разбор: только число файлов."""
return {"report": [f"быстрый разбор: файлов {len(state['files'])}"]}
def deep_scan(state: ScanState) -> dict:
"""Подробный разбор: считает строки в каждом файле."""
root = Path(state["root"])
total = sum(
len((root / name).read_text(encoding="utf-8").splitlines())
for name in state["files"]
)
return {"report": [f"подробный разбор: строк {total}"]}
def route_by_size(state: ScanState) -> Literal["quick_scan", "deep_scan"]:
"""Файлов больше предела, значит хватит быстрого разбора."""
if len(state["files"]) > state["limit"]:
return "quick_scan"
return "deep_scan"
def is_big(state: ScanState) -> bool:
"""Та же проверка, но возвращает признак."""
return len(state["files"]) > state["limit"]
def route_both(state: ScanState) -> list[str]:
"""Маршрут сразу в два узла: оба пойдут одним шагом."""
return ["quick_scan", "deep_scan"]
def build(route, path_map=None) -> object:
"""Собирает граф, где после сбора файлов стоит условное ребро."""
builder = StateGraph(ScanState)
builder.add_node("collect_files", collect_files)
builder.add_node("quick_scan", quick_scan)
builder.add_node("deep_scan", deep_scan)
builder.add_edge(START, "collect_files")
builder.add_conditional_edges("collect_files", route, path_map)
builder.add_edge("quick_scan", END)
builder.add_edge("deep_scan", END)
return builder.compile()
# 1. Функция маршрутизации вернула имя узла.
graph = build(route_by_size)
print("1. предел 2:", graph.invoke({"root": str(REPO), "limit": 2})["report"])
print(" предел 10:", graph.invoke({"root": str(REPO), "limit": 10})["report"])
# 2. Функция вернула признак, имя узла нашлось в словаре соответствия.
mapped = build(is_big, {True: "quick_scan", False: "deep_scan"})
print("2. предел 2, словарь соответствия:", mapped.invoke({"root": str(REPO), "limit": 2})["report"])
# 3. Условная точка входа: развилка стоит сразу после START.
def choose_start(state: ScanState) -> Literal["collect_files", "quick_scan"]:
"""Список файлов пришёл на входе, значит собирать его заново незачем."""
if state.get("files"):
return "quick_scan"
return "collect_files"
entry = StateGraph(ScanState)
entry.add_node("collect_files", collect_files)
entry.add_node("quick_scan", quick_scan)
entry.add_node("deep_scan", deep_scan)
entry.add_conditional_edges(START, choose_start)
entry.add_conditional_edges("collect_files", route_by_size)
entry.add_edge("quick_scan", END)
entry.add_edge("deep_scan", END)
entry_graph = entry.compile()
print("3. вход без списка:", entry_graph.invoke({"root": str(REPO), "limit": 10})["report"])
print(
" вход со списком:",
entry_graph.invoke({"root": str(REPO), "limit": 2, "files": ["app.py"]})["report"],
)
# 4. Маршрут в два узла сразу: обе правки приходят в одном шаге.
both = build(route_both, ["quick_scan", "deep_scan"])
print("4. маршрут в два узла:", both.invoke({"root": str(REPO), "limit": 2})["report"])
# Вывод:
# 1. предел 2: ['быстрый разбор: файлов 4']
# предел 10: ['подробный разбор: строк 44']
# 2. предел 2, словарь соответствия: ['быстрый разбор: файлов 4']
# 3. вход без списка: ['подробный разбор: строк 44']
# вход со списком: ['быстрый разбор: файлов 1']
# 4. маршрут в два узла: ['подробный разбор: строк 44', 'быстрый разбор: файлов 4']
Функция маршрутизации состояние не меняет: её возврат идёт на выбор узла и в состояние не попадает.
Аннотация возврата у route_by_size и choose_start нужна схеме графа: без аннотации переходы из развилки со схемы пропадают.
Действует она и при запуске. Если третьего аргумента нет, LangGraph читает Literal как словарь соответствия, и имя вне аннотации роняет запуск с KeyError. END это строка "__end__". Поэтому в аннотацию функции, которая может вернуть END, добавьте эту строку: Literal["quick_scan", "deep_scan", "__end__"].
Пункт 4 держится на редьюсере из урока 2: оба узла выполнились одним шагом и прислали правку в ключ report с operator.add. Без редьюсера запуск упал бы с InvalidUpdateError.
Правки узлов, запущенных рёбрами, LangGraph 1.2.16 применяет в алфавитном порядке имён, поэтому правка deep_scan идёт первой. Документация порядок не обещает, подробнее в уроке 2.
Command: обновить состояние и перейти одним возвратом
Условное ребро отделяет решение о переходе от работы узла. Узел collect_files собирает файлы, а их число с пределом сравнивает отдельная функция route_by_size, которая читает список из состояния.
С Command узел возвращает правку состояния и имя следующего узла одним объектом, условные рёбра не нужны.
Параметры Command:
1) update, правка состояния. Работает как обычный словарь из узла и проходит редьюсеры
2) goto, куда идти дальше. Принимает имя узла, объект Send или список, где могут стоять и имена узлов, и объекты Send
3) graph, к какому графу относится переход. Значение Command.PARENT уводит из подграфа в родительский граф, это тема урока 7
4) resume, значение для продолжения после паузы, это урок 6
Узлу с Command нужна аннотация возврата со всеми узлами, куда он может перейти: Command[Literal["quick_scan", "deep_scan"]]. Без неё граф работает, но переходы пропадают со схемы, и компиляция не проверяет имена узлов.
Пример 02_command.py
"""Пример 2 урока 3: Command вместо пары "обновить состояние и перейти".
Та же развилка, что в примере 1, но узел сам объявляет переход. Условных рёбер
в графе нет вовсе. Ещё в примере: статическое ребро рядом с Command
выполняется вместе с ним, а goto на END заканчивает запуск в первом узле.
"""
import operator
from pathlib import Path
from typing import Annotated, Literal
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command
from typing_extensions import TypedDict
REPO = Path(__file__).parent / "mini_repo"
class ScanState(TypedDict):
root: str
limit: int
files: list[str]
report: Annotated[list[str], operator.add]
def quick_scan(state: ScanState) -> dict:
"""Короткий разбор: только число файлов."""
return {"report": [f"быстрый разбор: файлов {len(state['files'])}"]}
def deep_scan(state: ScanState) -> dict:
"""Подробный разбор: считает строки в каждом файле."""
root = Path(state["root"])
total = sum(
len((root / name).read_text(encoding="utf-8").splitlines())
for name in state["files"]
)
return {"report": [f"подробный разбор: строк {total}"]}
def collect_files(state: ScanState) -> Command[Literal["quick_scan", "deep_scan"]]:
"""Собирает файлы и тут же говорит, каким разбором идти дальше."""
names = sorted(path.name for path in Path(state["root"]).glob("*.py"))
goto = "quick_scan" if len(names) > state["limit"] else "deep_scan"
return Command(update={"files": names}, goto=goto)
builder = StateGraph(ScanState)
builder.add_node("collect_files", collect_files)
builder.add_node("quick_scan", quick_scan)
builder.add_node("deep_scan", deep_scan)
builder.add_edge(START, "collect_files")
builder.add_edge("quick_scan", END)
builder.add_edge("deep_scan", END)
graph = builder.compile()
# 1. Переход задан внутри узла, рёбер между тремя узлами нет.
print("1. предел 2:", graph.invoke({"root": str(REPO), "limit": 2})["report"])
print(" предел 10:", graph.invoke({"root": str(REPO), "limit": 10})["report"])
# 2. Статическое ребро рядом с Command: выполняются оба перехода.
mixed = StateGraph(ScanState)
mixed.add_node("collect_files", collect_files)
mixed.add_node("quick_scan", quick_scan)
mixed.add_node("deep_scan", deep_scan)
mixed.add_edge(START, "collect_files")
mixed.add_edge("collect_files", "quick_scan")
mixed.add_edge("quick_scan", END)
mixed.add_edge("deep_scan", END)
print("2. Command и ребро вместе:", mixed.compile().invoke({"root": str(REPO), "limit": 10})["report"])
# 3. Command на END: разбирать нечего, запуск заканчивается в первом узле.
def collect_or_stop(state: ScanState) -> Command[Literal["quick_scan", "__end__"]]:
"""Файлов не нашлось, значит дальше идти незачем."""
names = sorted(path.name for path in Path(state["root"]).glob("*.py"))
if not names:
return Command(update={"report": ["файлов нет, разбор не нужен"]}, goto=END)
return Command(update={"files": names}, goto="quick_scan")
stop = StateGraph(ScanState)
stop.add_node("collect_or_stop", collect_or_stop)
stop.add_node("quick_scan", quick_scan)
stop.add_edge(START, "collect_or_stop")
stop.add_edge("quick_scan", END)
stop_graph = stop.compile()
# Папки docs в стенде нет, поэтому glob по ней вернёт пустой список.
print("3. папки нет:", stop_graph.invoke({"root": str(REPO / "docs"), "limit": 2})["report"])
print(" стенд на месте:", stop_graph.invoke({"root": str(REPO), "limit": 2})["report"])
# Вывод:
# 1. предел 2: ['быстрый разбор: файлов 4']
# предел 10: ['подробный разбор: строк 44']
# 2. Command и ребро вместе: ['подробный разбор: строк 44', 'быстрый разбор: файлов 4']
# 3. папки нет: ['файлов нет, разбор не нужен']
# стенд на месте: ['быстрый разбор: файлов 4']
Command добавляет узлу динамический переход, статические рёбра при этом остаются. В пункте 2 узел collect_files вернул goto="deep_scan", рядом стоит add_edge("collect_files", "quick_scan"), и выполнились оба перехода.
Поэтому держите у узла один способ перехода: статические рёбра или динамический переход (Command или условное ребро). Когда способы смешаны, по коду узла не видно, сколько переходов выполнится.
В пункте 3 goto=END заканчивает запуск в первом узле, когда разбирать нечего.
Если решение о переходе вытекает из работы узла, берите Command. Если оно принимается по готовому состоянию, берите условное ребро.
Что попадает на схему графа
Схема графа из урока 1 строится по переходам, объявленным при сборке. Переход, который функция или узел выбирают при запуске, попадает на схему только по подсказке: аннотации возврата, третьему аргументу add_conditional_edges или параметру destinations у add_node.
Пример 03_picture.py
"""Пример 3 урока 3: что попадает на схему графа.
Одна и та же развилка собрана на условном ребре, на Command и с параметром
destinations, с подсказкой о переходах и без неё. Для каждой сборки печатаются
рёбра схемы графа. В конце показан переход на узел, которого в графе нет.
"""
from typing import Literal
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command
from typing_extensions import TypedDict
class ScanState(TypedDict):
files: list[str]
report: str
def quick_scan(state: ScanState) -> dict:
"""Короткий разбор."""
return {"report": "быстрый разбор"}
def deep_scan(state: ScanState) -> dict:
"""Подробный разбор."""
return {"report": "подробный разбор"}
def edges(builder: StateGraph) -> list[str]:
"""Возвращает рёбра собранного графа в том виде, в каком их строит схема."""
graph = builder.compile().get_graph()
return [
f"{edge.source} -> {edge.target}"
+ (" (условное)" if edge.conditional else "")
for edge in graph.edges
]
def base(collect) -> StateGraph:
"""Каркас из трёх узлов, общий для всех вариантов."""
builder = StateGraph(ScanState)
builder.add_node("collect_files", collect)
builder.add_node("quick_scan", quick_scan)
builder.add_node("deep_scan", deep_scan)
builder.add_edge(START, "collect_files")
builder.add_edge("quick_scan", END)
builder.add_edge("deep_scan", END)
return builder
def collect_plain(state: ScanState) -> dict:
"""Обычный узел без Command, переход задаёт условное ребро."""
return {"files": ["app.py"]}
def route_typed(state: ScanState) -> Literal["quick_scan", "deep_scan"]:
"""Маршрут с аннотацией возврата."""
return "quick_scan"
def route_bare(state):
"""Тот же маршрут, но без аннотации возврата."""
return "quick_scan"
def collect_cmd_typed(state: ScanState) -> Command[Literal["quick_scan", "deep_scan"]]:
"""Command с аннотацией возврата."""
return Command(update={"files": ["app.py"]}, goto="quick_scan")
def collect_cmd_bare(state):
"""Тот же Command, но без аннотации возврата."""
return Command(update={"files": ["app.py"]}, goto="quick_scan")
# 1. Условное ребро, у функции маршрутизации есть аннотация.
typed = base(collect_plain)
typed.add_conditional_edges("collect_files", route_typed)
print("1. условное ребро с аннотацией:", edges(typed))
# 2. То же самое без аннотации и без словаря соответствия.
bare = base(collect_plain)
bare.add_conditional_edges("collect_files", route_bare)
print("2. условное ребро без аннотации:", edges(bare))
# 3. Словарь соответствия возвращает переходы на схему.
mapped = base(collect_plain)
mapped.add_conditional_edges(
"collect_files",
route_bare,
{"quick_scan": "quick_scan", "deep_scan": "deep_scan"},
)
print("3. условное ребро со словарём:", edges(mapped))
# 4. Command с аннотацией и без неё.
cmd_typed = base(collect_cmd_typed)
print("4. Command с аннотацией:", edges(cmd_typed))
cmd_bare = base(collect_cmd_bare)
print(" Command без аннотации:", edges(cmd_bare))
# 5. Тот же Command без аннотации, но с destinations при регистрации узла.
named = StateGraph(ScanState)
named.add_node(
"collect_files",
collect_cmd_bare,
destinations={"quick_scan": "файлов много", "deep_scan": "файлов мало"},
)
named.add_node("quick_scan", quick_scan)
named.add_node("deep_scan", deep_scan)
named.add_edge(START, "collect_files")
named.add_edge("quick_scan", END)
named.add_edge("deep_scan", END)
print("5. Command с destinations:", edges(named))
print()
print(named.compile().get_graph().draw_mermaid())
# 6. Переход на узел, которого в графе нет.
def goto_nowhere_typed(state: ScanState) -> Command[Literal["nowhere"]]:
"""Аннотация называет узел, которого в графе нет."""
return Command(goto="nowhere")
def goto_nowhere_bare(state):
"""То же самое, но без аннотации."""
return Command(goto="nowhere")
broken = StateGraph(ScanState)
broken.add_node("collect_files", goto_nowhere_typed)
broken.add_edge(START, "collect_files")
try:
broken.compile()
except ValueError as error:
print("6. переход в никуда с аннотацией:", type(error).__name__, error)
silent = StateGraph(ScanState)
silent.add_node("collect_files", goto_nowhere_bare)
silent.add_node("quick_scan", quick_scan)
silent.add_edge(START, "collect_files")
silent.add_edge("quick_scan", END)
print(" переход в никуда без аннотации:", silent.compile().invoke({"files": []}))
# Вывод:
# 1. условное ребро с аннотацией: ['__start__ -> collect_files', 'collect_files -> deep_scan (условное)', 'collect_files -> quick_scan (условное)', 'deep_scan -> __end__', 'quick_scan -> __end__']
# 2. условное ребро без аннотации: ['__start__ -> collect_files', 'collect_files -> __end__']
# 3. условное ребро со словарём: ['__start__ -> collect_files', 'collect_files -> deep_scan (условное)', 'collect_files -> quick_scan (условное)', 'deep_scan -> __end__', 'quick_scan -> __end__']
# 4. Command с аннотацией: ['__start__ -> collect_files', 'collect_files -> deep_scan (условное)', 'collect_files -> quick_scan (условное)', 'deep_scan -> __end__', 'quick_scan -> __end__']
# Command без аннотации: ['__start__ -> collect_files', 'collect_files -> __end__']
# 5. Command с destinations: ['__start__ -> collect_files', 'collect_files -> deep_scan (условное)', 'collect_files -> quick_scan (условное)', 'deep_scan -> __end__', 'quick_scan -> __end__']
#
# ---
# config:
# flowchart:
# curve: linear
# ---
# graph TD;
# __start__([<p>__start__</p>]):::first
# collect_files(collect_files)
# quick_scan(quick_scan)
# deep_scan(deep_scan)
# __end__([<p>__end__</p>]):::last
# __start__ --> collect_files;
# collect_files -. файлов мало .-> deep_scan;
# collect_files -. файлов много .-> quick_scan;
# deep_scan --> __end__;
# quick_scan --> __end__;
# classDef default fill:#f2f0ff,line-height:1.2
# ... (ещё две строки classDef)
#
# 6. переход в никуда с аннотацией: ValueError Found edge ending at unknown node `nowhere`
# переход в никуда без аннотации: {'files': []}
#
# В пункте 6 без аннотации LangGraph пишет в журнал (stderr, модуль logging):
# Task collect_files with path ('__pregel_pull', 'collect_files') wrote to
# unknown channel branch:to:nowhere, ignoring it.
В пункте 2 и во второй строке пункта 4 подсказки о переходах нет. От развилки на схеме остаётся одно ребро collect_files -> __end__, а у quick_scan и deep_scan нет ни одной стрелки. Схема показывает граф, которого нет.
Любая из подсказок возвращает на схему оба перехода пунктирными стрелками:
1) аннотация возврата у функции маршрутизации или у узла с Command: -> Literal["quick_scan", "deep_scan"] или -> Command[Literal["quick_scan", "deep_scan"]]
2) третий аргумент add_conditional_edges: ключи словаря соответствия становятся подписями на стрелках, если отличаются от имён узлов, а список даёт стрелки без подписей
3) параметр destinations у add_node: в словаре ключ это имя узла, значение это подпись стрелки, кортеж даёт только имена. На запуск он не влияет, но имена из него проверяет компиляция
Докстринг add_conditional_edges в версии 1.2.16 предупреждает, что без аннотации и третьего аргумента схема считает переход возможным в любой узел. Пример 3 показывает обратное: вместо переходов во все узлы на схеме остаётся одно ребро в __end__.
Аннотация даёт и проверку. В пункте 6 узел объявляет переход на узел nowhere, которого в графе нет, и компиляция падает с ValueError. Без аннотации граф компилируется и запускается, переход не выполняется, а в журнал уходит предупреждение о записи в неизвестный канал branch:to:nowhere.
Такой граф выглядит рабочим: запуск вернул состояние без ошибки, хотя переход из collect_files не выполнился и граф остановился на первом узле. С аннотацией компиляция сверяет с графом имена из Literal ещё до первого запуска. Значение goto она не проверяет: опечатка в нём при верной аннотации тоже уйдёт в журнал предупреждением.
Send: число задач известно только при запуске
Развилка выбирает из узлов, объявленных при сборке. Узел разбора файла надо запустить столько раз, сколько файлов нашлось, а это число известно только при запуске.
Для этого есть Send. Функция маршрутизации возвращает список объектов Send, каждый объект запускает отдельную копию названного узла со своим состоянием. Send принимает имя узла и состояние для копии. Раскладку работы на несколько копий узла или на несколько узлов одним шагом документация называет fan-out.
У копии своё состояние, отдельное от состояния графа: узлу разбора нужны только имя файла и корень репозитория. Копия получает словарь из Send целиком, со всеми ключами. Узел, запущенный ребром, получает только ключи своей схемы, это было в уроке 2.
Что узел ждёт на входе, описывает аннотация state: FileState. В примере 4 её повторяет параметр input_schema у add_node.
Правки копий приходят в общее состояние одним шагом, поэтому ключу, куда они пишут, нужен редьюсер. Применяются они в порядке списка Send.
Пример 04_send.py
"""Пример 4 урока 3: Send и раскладка работы по числу файлов.
Сколько файлов в репозитории, известно только при запуске, поэтому рёбер под
них при сборке нет. Узел разбора запускается по разу на файл объектами Send,
каждая копия получает своё состояние, а находки собираются редьюсером. Дальше
fan-out на ноль элементов и тот же fan-out, собранный на Command.
"""
import operator
from pathlib import Path
from typing import Annotated, Literal
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, Send
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]
card: str
class FileState(TypedDict):
"""Состояние одной копии узла разбора: свой файл и корень стенда."""
root: str
name: str
def collect_files(state: ScanState) -> dict:
"""Собирает имена файлов стенда."""
names = sorted(path.name for path in Path(state["root"]).glob("*.py"))
return {"files": names}
def scan_file(state: FileState) -> dict:
"""Разбирает один файл. Состояние здесь своё, отдельное от общего."""
path = Path(state["root"]) / state["name"]
lines = len(path.read_text(encoding="utf-8").splitlines())
return {"findings": [f"{state['name']}: строк {lines}"]}
def make_card(state: ScanState) -> dict:
"""Сводит находки в одну карточку."""
return {"card": f"разобрано файлов {len(state['findings'])}"}
def fan_out(state: ScanState) -> list[Send]:
"""По одному Send на файл: имя узла и состояние для этой копии."""
return [
Send("scan_file", {"root": state["root"], "name": name})
for name in state["files"]
]
builder = StateGraph(ScanState)
builder.add_node("collect_files", collect_files)
builder.add_node("scan_file", scan_file, input_schema=FileState)
builder.add_node("make_card", make_card)
builder.add_edge(START, "collect_files")
builder.add_conditional_edges("collect_files", fan_out, ["scan_file"])
builder.add_edge("scan_file", "make_card")
builder.add_edge("make_card", END)
graph = builder.compile()
# 1. Четыре файла стенда, значит четыре копии узла разбора.
result = graph.invoke({"root": str(REPO)})
print("1. находок:", len(result["findings"]))
for finding in sorted(result["findings"]):
print(" ", finding)
print(" карточка:", result["card"])
# 2. Схема графа: fan-out рисуется одной пунктирной стрелкой.
print()
print(graph.get_graph().draw_mermaid())
# 3. Пустой fan-out: папки docs в стенде нет, список файлов пустой.
empty = graph.invoke({"root": str(REPO / "docs")})
print("3. пустой fan-out, ключи итога:", sorted(empty))
# 4. Тот же fan-out, но собранный на Command вместо условного ребра.
def collect_and_fan(state: ScanState) -> Command[Literal["scan_file"]]:
"""Собирает файлы и тут же раскладывает их по копиям узла разбора."""
names = sorted(path.name for path in Path(state["root"]).glob("*.py"))
sends = [Send("scan_file", {"root": state["root"], "name": n}) for n in names]
return Command(update={"files": names}, goto=sends)
by_command = StateGraph(ScanState)
by_command.add_node("collect_files", collect_and_fan)
by_command.add_node("scan_file", scan_file, input_schema=FileState)
by_command.add_node("make_card", make_card)
by_command.add_edge(START, "collect_files")
by_command.add_edge("scan_file", "make_card")
by_command.add_edge("make_card", END)
print("4. fan-out на Command:", by_command.compile().invoke({"root": str(REPO)})["card"])
# Вывод:
# 1. находок: 4
# app.py: строк 13
# config.py: строк 6
# db.py: строк 15
# utils.py: строк 10
# карточка: разобрано файлов 4
#
# ---
# config:
# flowchart:
# curve: linear
# ---
# graph TD;
# __start__([<p>__start__</p>]):::first
# collect_files(collect_files)
# scan_file(scan_file)
# make_card(make_card)
# __end__([<p>__end__</p>]):::last
# __start__ --> collect_files;
# collect_files -.-> scan_file;
# scan_file --> make_card;
# make_card --> __end__;
# classDef default fill:#f2f0ff,line-height:1.2
# ... (ещё две строки classDef)
#
# 3. пустой fan-out, ключи итога: ['files', 'findings', 'root']
# 4. fan-out на Command: разобрано файлов 4
Третий аргумент add_conditional_edges здесь список ["scan_file"]: fan-out идёт в один узел, подписывать стрелку нечем. На схеме список даёт ту же пунктирную стрелку, что и словарь.
В пункте 3 список файлов пуст. Ни одна копия scan_file не запустилась, поэтому не выполнился и make_card: ключа card в итоге нет, ошибки тоже нет.
Пустой список проверяйте до fan-out и в этом случае переходите сразу на узел карточки make_card. Если fan-out стоит на условном ребре, впишите make_card и в третий аргумент, ["scan_file", "make_card"], иначе возврат "make_card" уронит запуск с KeyError.
В пункте 4 тот же fan-out собран на Command: список объектов Send стоит в параметре goto, условное ребро не нужно. Так удобно, когда узел сам собирает список и сразу его раскладывает.
Ветки разной длины и итоговый узел
Ветка это цепочка узлов от места, где работа расходится, до итогового узла, где результаты сходятся. В примере 5 работа расходится после collect_files, итоговый узел make_card. Пока ветки одинаковой длины, они заканчиваются в одном шаге, и итоговый узел выполняется один раз.
С ветками разной длины это не так. Ветка из одного узла заканчивается в первом шаге после collect_files, ветка из двух узлов во втором. Итоговый узел выполняется дважды: по разу на каждый шаг, в котором закончилась хотя бы одна ветка.
Дождаться всех веток помогают defer=True у итогового узла и ребро из списка узлов. Разница между ними видна, когда функция маршрутизации выбрала не все ветки.
Пример 05_defer.py
"""Пример 5 урока 3: ветки разной длины и итоговый узел.
Быстрая ветка из одного узла и подробная из двух расходятся после сбора файлов
и сходятся в карточке. Узел карточки подключён отдельными рёбрами, с
defer=True и ребром из списка узлов. Пункты 4 и 5 показывают, что меняется, когда
функция маршрутизации выбрала одну ветку.
"""
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
which: str
files: list[str]
findings: Annotated[list[str], operator.add]
cards: 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, "findings": [f"файлов {len(names)}"]}
def count_lines(state: ScanState) -> dict:
"""Быстрая ветка: одна операция, один узел."""
root = Path(state["root"])
total = sum(
len((root / name).read_text(encoding="utf-8").splitlines())
for name in state["files"]
)
return {"findings": [f"строк {total}"]}
def read_imports(state: ScanState) -> dict:
"""Подробная ветка, первый узел: собирает строки импорта."""
root = Path(state["root"])
found = []
for name in state["files"]:
for line in (root / name).read_text(encoding="utf-8").splitlines():
if line.startswith("import ") or line.startswith("from "):
found.append(line.split()[1])
return {"findings": [f"импортов {len(found)}"]}
def rank_imports(state: ScanState) -> dict:
"""Подробная ветка, второй узел. Работы в нём нет, он удлиняет ветку."""
return {"findings": ["связи разобраны"]}
def make_card(state: ScanState) -> dict:
"""Итоговый узел: печатает полученные находки и дописывает карточку."""
print(" make_card получил находки:", state["findings"])
return {"cards": [f"находок {len(state['findings'])}"]}
def build(*, defer: bool, list_edge: bool, conditional: bool) -> object:
"""Собирает граф с двумя ветками.
Флаги задают способ дождаться веток и то, выбирает ли ветки функция
маршрутизации.
"""
builder = StateGraph(ScanState)
builder.add_node("collect_files", collect_files)
builder.add_node("count_lines", count_lines)
builder.add_node("read_imports", read_imports)
builder.add_node("rank_imports", rank_imports)
builder.add_node("make_card", make_card, defer=defer)
builder.add_edge(START, "collect_files")
if conditional:
def route(state: ScanState) -> list[str]:
"""Выбирает ветки по значению which."""
if state["which"] == "обе":
return ["count_lines", "read_imports"]
return ["count_lines"]
builder.add_conditional_edges(
"collect_files", route, ["count_lines", "read_imports"]
)
else:
builder.add_edge("collect_files", "count_lines")
builder.add_edge("collect_files", "read_imports")
builder.add_edge("read_imports", "rank_imports")
if list_edge:
builder.add_edge(["count_lines", "rank_imports"], "make_card")
else:
builder.add_edge("count_lines", "make_card")
builder.add_edge("rank_imports", "make_card")
builder.add_edge("make_card", END)
return builder.compile()
start = {"root": str(REPO), "which": "обе"}
# 1. Отдельные рёбра: карточка собирается дважды, ветки заканчиваются в разных шагах.
print("1. отдельные рёбра")
print(" карточки:", build(defer=False, list_edge=False, conditional=False).invoke(start)["cards"])
# 2. defer=True: узел ждёт, пока в графе не останется незаконченных задач.
print("2. defer=True")
print(" карточки:", build(defer=True, list_edge=False, conditional=False).invoke(start)["cards"])
# 3. Ребро из списка узлов: тот же итог, но без defer.
print("3. ребро из списка узлов")
print(" карточки:", build(defer=False, list_edge=True, conditional=False).invoke(start)["cards"])
# 4. Функция маршрутизации выбрала одну ветку из двух.
# Ребро из списка узлов ждёт обе и не дожидается.
print("4. выбрана одна ветка, ребро из списка узлов")
one = {"root": str(REPO), "which": "быстрая"}
result = build(defer=False, list_edge=True, conditional=True).invoke(one)
print(" карточки:", result["cards"], "| находки:", result["findings"])
# 5. То же самое, но с defer=True: карточка собирается один раз.
print("5. выбрана одна ветка, defer=True")
print(" карточки:", build(defer=True, list_edge=False, conditional=True).invoke(one)["cards"])
# Вывод:
# 1. отдельные рёбра
# make_card получил находки: ['файлов 4', 'строк 44', 'импортов 3']
# make_card получил находки: ['файлов 4', 'строк 44', 'импортов 3', 'связи разобраны']
# карточки: ['находок 3', 'находок 4']
# 2. defer=True
# make_card получил находки: ['файлов 4', 'строк 44', 'импортов 3', 'связи разобраны']
# карточки: ['находок 4']
# 3. ребро из списка узлов
# make_card получил находки: ['файлов 4', 'строк 44', 'импортов 3', 'связи разобраны']
# карточки: ['находок 4']
# 4. выбрана одна ветка, ребро из списка узлов
# карточки: [] | находки: ['файлов 4', 'строк 44']
# 5. выбрана одна ветка, defer=True
# make_card получил находки: ['файлов 4', 'строк 44']
# карточки: ['находок 2']
| Способ | Ветки одинаковой длины | Ветки разной длины | Выбрана часть веток |
|---|---|---|---|
| отдельные рёбра | выполняется один раз | выполняется по разу на каждый шаг, где закончилась ветка | выполняется по разу на каждый шаг, где закончилась выбранная ветка |
отдельные рёбра плюс defer=True |
выполняется один раз | выполняется один раз, когда закончились все ветки | выполняется один раз по выбранным веткам |
| ребро из списка узлов | выполняется один раз | выполняется один раз | не выполняется вовсе |
Последняя клетка таблицы, это та же ловушка, что с пустым списком Send: итоговый узел молча не выполняется. Ребро add_edge(["count_lines", "rank_imports"], "make_card") ждёт оба узла, а функция маршрутизации выбрала только быструю ветку. rank_imports не выполнился, поэтому не выполнился и make_card. Ошибки нет, ключ cards пуст.
Узел с defer=True ждёт, пока в графе не останется незаконченных задач, в том числе в узлах, от которых к нему нет рёбер. Поэтому он выполняется один раз после всех выбранных веток. Send, адресованный такому узлу, запускает его отдельно, помимо отложенного выполнения по рёбрам.
Практическое задание
Соберите граф разбора репозитория, где узел сбора файлов сам раскладывает работу.
Требования:
1) схема состояния на TypedDict: путь к папке, список файлов, находки с редьюсером, карточки с редьюсером
2) узел сбора файлов возвращает Command: если файлов нет, переход сразу на карточку, иначе fan-out: объект Send в узел разбора на каждый файл и в том же списке goto переход в узел сводки
3) сводка по репозиторию, например число файлов и строк, идёт двумя узлами подряд, разбор файла одним, поэтому ветки разной длины
4) узел разбора одного файла имеет свою схему входа: имя файла и корень
5) карточка собирает все находки и выполняется один раз
Как проверить результат:
1) get_graph().draw_mermaid() печатает пунктиром переходы из узла сбора в разбор, сводку и карточку
2) запуск по стенду mini_repo даёт по находке на файл, находки сводки и одну карточку
3) запуск по папке без файлов Python доходит до карточки и не падает
4) если снять аннотацию возврата у узла сбора, переходы из него пропадают со схемы графа
5) если снять defer=True у карточки, она собирается дважды
Подсказка: в goto можно передать один список, где стоят и объекты Send, и имя узла сводки.
Итоги урока
Условное ребро вызывает функцию маршрутизации, её возврат читается как имя узла, а словарь соответствия переводит в имя любое хешируемое значение. Развилка от START даёт условную точку входа.
Command соединяет правку состояния и переход в одном возврате. Статические рёбра узла выполняются вместе с переходом из Command, поэтому держите у узла один способ перехода.
Без аннотации возврата, третьего аргумента add_conditional_edges или destinations переходы пропадают со схемы графа. Компиляция проверяет имена из аннотации и destinations, но не значение goto.
Send запускает узел по разу на элемент списка, правкам копий нужен редьюсер. При пустом списке не выполняется ни одна копия и узел после них, ошибки нет.
Итоговый узел на отдельных рёбрах выполняется по разу на каждый шаг, где закончилась ветка, а с defer=True один раз в конце. Ребро из списка узлов не выполнит его вовсе, если один из перечисленных узлов не выполнился.
В уроке 4 разберу циклы: ребро назад, предел шагов recursion_limit, RemainingSteps и кеш узла.
Код урока
Примеры этого урока лежат в репозитории курса, папка lesson_03. Закреплённые версии, на которых получен вывод в тексте, лежат в requirements.txt в корне репозитория.
Предыдущий урок: Состояние графа: схема, редьюсеры, каналы
Следующий урок: Циклы и остановка
Подписывайтесь на мой Telegram канал
Если вам нужен ментор и вы хотите научиться разрабатывать AI агентов, пишите, обсудим условия
Авторизуйтесь, чтобы оставить комментарий.
Нет комментариев.
Тут может быть ваша реклама
Пишите info@aisferaic.ru
Похожие туториалы
