Условная маршрутизация и команда Command

Когда нужно выбрать, куда идти дальше, на основе состояния — используйте условные рёбра или команду Command.

#Условные рёбра

Условное ребро вызывает функцию-маршрутизатор после выполнения узла. Функция принимает текущее состояние и возвращает имя следующего узла:

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

def route(state: State) -> Literal["node_a", "node_b"]:
    if state["needs_search"]:
        return "node_a"
    return "node_b"

builder = StateGraph(State)
builder.add_conditional_edges(START, route)

Можно передать словарь соответствия для маппинга возвращаемых значений на имена узлов:

builder.add_conditional_edges("classifier", route, {
    "urgent": "handle_urgent",
    "normal": "handle_normal",
})

#Команда Command

Command — универсальный примитив для управления выполнением графа. Позволяет обновить состояние и выбрать следующий узел в одном шаге, совмещая логику узла и маршрутизатора.

from typing import Literal
from langgraph.types import Command

def my_node(state: State) -> Command[Literal["other_node"]]:
    return Command(
        update={"foo": "bar"},
        goto="other_node",
    )

Условная логика внутри узла через Command — это альтернатива условным рёбрам. Узел сам решает, куда перейти, на основе состояния.

#Навигация к родительскому графу

При работе с подграфами можно перейти к узлу в родительском графе, указав graph=Command.PARENT. Это полезно для мультиагентных систем, где субагент возвращает управление координатору.

#Возврат из инструментов

Command можно возвращать из инструментов для обновления состояния и маршрутизации после завершения вызова инструмента.

#Практический пример: маршрутизатор запросов

from typing import TypedDict, Literal
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")
    elif 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": ""})

#Send для параллельных веток

Для map-reduce паттернов используется Send — он создаёт динамические параллельные ветки с разными входными данными:

from langgraph.types import Send

def continue_to_jokes(state):
    return [Send("generate_joke", {"subject": s}) for s in state["subjects"]]