Условные рёбра, команда Command для обновления состояния и маршрутизации, Send для параллельных веток.
Когда выбор следующего шага зависит от состояния, используют условные рёбра или команду
Command.
Развилку в графе описывают двумя способами, и выбор между ними — это вопрос того, где должно жить решение.
Условное ребро выносит решение наружу: узел делает работу, отдельная функция-маршрутизатор смотрит на состояние и называет следующий узел. Схема графа остаётся декларативной, маршрутизацию видно на диаграмме.
Command переносит решение внутрь узла: тот сам возвращает и обновление состояния, и имя следующего узла. Кода меньше, шаг один, но развилка перестаёт быть видимой на схеме без чтения тела функции.
Оба способа поддерживаются полноценно, и в одном графе их можно смешивать.
Маршрутизатор — это функция от состояния, возвращающая имя узла. Аннотация Literal в возвращаемом типе не украшение: по ней LangGraph понимает список достижимых узлов и строит корректную схему графа.
from typing import Literal
from langgraph.graph import StateGraph, START, END
def route(state: State) -> Literal["search", "answer"]:
if state["needs_search"]:
return "search"
return "answer"
builder.add_conditional_edges("classify", route)Если маршрутизатор возвращает не имена узлов, а произвольные метки (например, категории), их сопоставляют с узлами через словарь:
builder.add_conditional_edges("classify", route, {
"urgent": "handle_urgent",
"normal": "handle_normal",
"spam": END,
})Такой словарь удобен, когда метки приходят из бизнес-логики и не совпадают с именами узлов. Значением может быть и END, что даёт досрочное завершение ветки.
Маршрутизатор можно повесить и на START: тогда точка входа выбирается по входным данным.
Условное ребро — это то, из чего собран цикл любого агента. Узел модели отвечает, маршрутизатор смотрит, попросила ли модель вызвать инструменты, и либо отправляет в узел инструментов, либо завершает работу:
from typing import Literal
from langgraph.graph import StateGraph, START, END, MessagesState
def call_model(state: MessagesState):
return {"messages": [model_with_tools.invoke(state["messages"])]}
def should_continue(state: MessagesState) -> Literal["tools", "__end__"]:
last_message = state["messages"][-1]
if last_message.tool_calls:
return "tools"
return END
builder = StateGraph(MessagesState)
builder.add_node("model", call_model)
builder.add_node("tools", tool_node)
builder.add_edge(START, "model")
builder.add_conditional_edges("model", should_continue)
builder.add_edge("tools", "model") # обратное ребро замыкает цикл
graph = builder.compile()Ребро от tools обратно к model и создаёт цикл. Именно эту конструкцию собирает create_agent, и понимание её устройства позволяет достроить свои шаги, например проверку ответа перед возвратом пользователю.
Command объединяет обновление состояния и переход. Возвращаемый тип с Literal перечисляет возможные цели, чтобы граф знал о рёбрах заранее:
from typing import Literal
from langgraph.types import Command
def classify(state: State) -> Command[Literal["billing", "support"]]:
category = "billing" if "счёт" in state["text"] else "support"
return Command(
update={"category": category},
goto=category,
)Без Command тот же результат потребовал бы двух сущностей: узла, записывающего категорию, и маршрутизатора, читающего её обратно из состояния. Здесь решение и запись происходят вместе.
Соберём граф, который разбирает запрос и направляет его в один из обработчиков:
from typing import Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
class State(TypedDict):
query: str
category: str
response: str
def classify(state: State) -> Command[Literal["weather", "math", "general"]]:
query = state["query"].lower()
if "погод" in query:
return Command(update={"category": "weather"}, goto="weather")
if any(word in query for word in ["сколько", "посчитай"]):
return Command(update={"category": "math"}, goto="math")
return Command(update={"category": "general"}, goto="general")
def handle_weather(state: State):
return {"response": "Сегодня солнечно, +25°C"}
def handle_math(state: State):
return {"response": "Результат: 42"}
def handle_general(state: State):
return {"response": "Могу помочь с погодой и математикой."}
builder = StateGraph(State)
builder.add_node("classify", classify)
builder.add_node("weather", handle_weather)
builder.add_node("math", handle_math)
builder.add_node("general", handle_general)
builder.add_edge(START, "classify")
builder.add_edge("weather", END)
builder.add_edge("math", END)
builder.add_edge("general", END)
graph = builder.compile()
result = graph.invoke({"query": "Какая погода?", "category": "", "response": ""})Ребро от classify не добавлено ни одной строкой: переходы описаны внутри узла через goto, а тип возврата сообщил графу, куда этот узел может вести.
Подграф по умолчанию не знает о внешнем мире и, завершившись, возвращает управление туда, откуда был вызван. Чтобы отправить выполнение к узлу родителя, указывают область действия команды:
def escalate(state: State) -> Command:
return Command(
update={"escalated": True},
goto="human_review",
graph=Command.PARENT,
)Это основной механизм передачи управления в мультиагентных системах: субагент, поняв, что задача не его, отправляет работу координатору или другому специалисту.
Инструмент тоже может вернуть Command. Тогда после его выполнения состояние обновится, а граф перейдёт в указанный узел. Не забывайте класть в обновление ToolMessage с идентификатором вызова, иначе история сообщений останется незакрытой:
from langchain.messages import ToolMessage
from langchain.tools import ToolRuntime, tool
from langgraph.types import Command
@tool
def transfer_to_billing(runtime: ToolRuntime) -> Command:
"""Передать обращение специалисту по счетам."""
return Command(
goto="billing_agent",
graph=Command.PARENT,
update={
"messages": [
ToolMessage(content="Передаю в биллинг.", tool_call_id=runtime.tool_call_id)
]
},
)Такой инструмент называют инструментом передачи управления: модель вызывает его как обычный инструмент, а фактически это переключение агента.
Условные рёбра выбирают одну ветку из нескольких известных заранее. Когда число веток становится известно только во время выполнения (по документу на каждый найденный источник, по подзадаче на каждый пункт плана), применяют Send.
Send создаёт ветку с собственным входным состоянием, а не копией общего. Маршрутизатор возвращает список таких отправлений:
from langgraph.types import Send
def fan_out(state: State):
return [Send("summarize_one", {"doc": doc}) for doc in state["documents"]]
builder.add_conditional_edges("plan", fan_out, ["summarize_one"])Узел summarize_one получит состояние вида {"doc": ...}, а не общее состояние графа. Результаты веток вернутся в общее состояние, поэтому поле, в которое они пишут, обязано иметь редуктор:
import operator
from typing import Annotated
class State(TypedDict):
documents: list[str]
summaries: Annotated[list[str], operator.add]Это классическая схема map-reduce: Send раскладывает работу, редуктор собирает результаты обратно.
Условные рёбра стоит предпочесть, когда развилка отражает структуру процесса и её полезно видеть на схеме, а также когда один маршрутизатор используется после нескольких узлов.
Command удобнее, когда решение и обновление состояния неразделимы, когда переход нужен из инструмента и когда речь о передаче управления между агентами.
Send нужен только там, где количество параллельных веток заранее неизвестно. Для фиксированного числа веток достаточно нескольких обычных рёбер.
Маршрутизатор возвращает имя узла, которого нет в графе. Ошибка всплывает во время выполнения, а не при компиляции, поэтому аннотация Literal со списком целей окупается.
Забытая аннотация возвращаемого типа у узла с Command. Граф не узнаёт о возможных переходах, и на схеме появляются висящие узлы.
Использование Send там, где хватило бы обычных рёбер. Динамические ветки сложнее отлаживать, и без необходимости их лучше не заводить.
Отсутствие редуктора у поля, куда пишут ветки Send. Симптом тот же, что и при обычном параллелизме: ошибка конфликтующего обновления.
Далее: Потоковая передача данных