Зачем графу ветвление?

Безусловное ребро add_edge("A", "B") говорит: «после A всегда иди в B». Этого хватает для конвейера, но не для агента. Агенту нужно принимать решение по результату шага:

  • LLM попросила вызвать инструмент → идём в узел инструментов; нет → выходим с ответом.
  • Найденного контекста мало → возвращаемся искать ещё; достаточно → генерируем ответ.
  • Ревьюер забраковал черновик → на доработку; одобрил → завершаем.

Все эти «→» зависят от данных, которых нет на момент сборки графа — они появятся только в рантайме, в состоянии. Значит, нужно ребро, которое смотрит на состояние и решает, куда шагнуть. Это и есть conditional edge.

ℹ️ Ветвление — это «if» графа

Вспомни while-агента из начала модуля: вся его логика держалась на if response.tool_calls: .... Условное ребро — это тот же if, но вынесенный из тела шага в структуру графа. Решение «куда дальше» становится видимым на схеме, а не спрятанным внутри функции.

Функция-маршрутизатор

Сердце условного ребра — функция-маршрутизатор (router). Это функция state → имя следующего узла. Она читает состояние и возвращает строку — куда идти дальше. Важнейшее правило: маршрутизатор только выбирает путь и ничего не меняет в состоянии. Это не узел, это «стрелочник».

Маршрутизатор: читает состояние, возвращает имя узла
python
from langgraph.graph import END

def route(state: AgentState) -> str:
    """Решает, куда идти после узла agent. Возвращает имя узла (строку)."""
    last = state["messages"][-1]
    if last.tool_calls:        # LLM попросила инструмент
        return "tools"
    return END                 # ответ готов — завершаем граф

Маршрутизатор может вернуть имя любого узла графа или служебный END. Никаких побочных эффектов: не вызывай здесь LLM, не пиши в состояние. Если внутри router потянуло сделать «работу» — значит, её место в отдельном узле, а сюда вынеси только проверку результата.

⚠️ Router решает, узел делает

Чёткое разделение: узлы меняют состояние, маршрутизаторы выбирают переход. Если смешать — граф станет невозможно ни отлаживать, ни визуализировать, потому что «работа» расползётся по стрелкам. Держи router крошечным и без сайд-эффектов.

add_conditional_edges и карта исходов

Маршрутизатор подключается к узлу через add_conditional_edges. Метод принимает: из какого узла ветвимся, функцию-маршрутизатор и (опционально) карту исходов — словарь «что вернул router → в какой узел идти».

Подключаем условное ребро к узлу agent
python
builder.add_conditional_edges(
    "agent",                        # 1. из какого узла ветвимся
    route,                          # 2. функция-маршрутизатор
    {                               # 3. карта: результат route → узел
        "tools": "tools",
        END: END,
    },
)

Карта исходов делает граф читаемым: по ней сразу видно все возможные переходы из узла agent. Технически третий аргумент можно опустить — тогда LangGraph трактует строку, которую вернул router, прямо как имя узла. Но явная карта надёжнее: она защищает от опечаток и служит документацией ветвления (а для визуализации графа подписывает стрелки).

ℹ️ Ключи карты — это «метки», а не обязательно имена узлов

Удобный приём: router возвращает осмысленные метки ("need_tools", "done"), а карта переводит их в реальные узлы ({"need_tools": "tools", "done": END}). Тогда логику router'а можно читать, не помня названий узлов, а связи задаёт карта.

Циклы и защита от зацикливания

Условные рёбра дают графу то, чего не было у линейного конвейера, — циклы. Если из узла tools провести ребро обратно в agent, получится петля «думать → действовать → думать», основа любого агента. Цикл живёт ровно до тех пор, пока маршрутизатор не вернёт END.

Обратная сторона циклов — риск зациклиться навсегда (router никогда не отдаёт END). На этот случай у LangGraph есть страховка — recursion_limit: максимальное число шагов (супершагов) за один запуск. Превысили — граф бросает GraphRecursionError вместо вечного выполнения.

recursion_limit — предохранитель от бесконечного цикла
python
from langgraph.errors import GraphRecursionError

try:
    result = graph.invoke(
        {"messages": [...]},
        {"recursion_limit": 25},     # не больше 25 шагов за запуск
    )
except GraphRecursionError:
    print("Агент не сошёлся за 25 шагов — обрываем")
Две защиты от вечного цикла

recursion_limit — это аварийный тормоз, а не логика остановки. Хорошая практика — добавить в состояние счётчик (steps) и в самом router'е возвращать END при превышении лимита попыток. Тогда нормальное завершение управляется явно, а recursion_limit ловит только настоящие баги.

Собираем ReAct-агента с ветвлением

Сложим всё вместе. Классический ReAct-цикл выглядит так: узел agent зовёт LLM; условное ребро смотрит, попросила ли модель инструмент; если да — идём в tools и возвращаемся обратно в agent, если нет — выходим в END.

100%
колёсико — масштаб  ·  зажать и тянуть — перемещение
START agent вызов LLM tool_calls? route() tools END нет → END да → tools результат → снова в agent

Обрати внимание: agent → tools → agent образует цикл, а единственный выход наружу — жёлтая ветка «нет → END». Граф будет крутиться, пока LLM просит инструменты, и завершится, как только она сформулирует финальный ответ.

Полный рабочий пример: ReAct-агент на conditional edges
python
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI

# ---- 1. Инструмент ----
def get_weather(city: str) -> str:
    return f"В городе {city} сейчас +18°C, ясно."

# ---- 2. Состояние (reducer из прошлого урока) ----
class State(TypedDict):
    messages: Annotated[list, add_messages]

# ---- 3. LLM с привязанным инструментом ----
llm = ChatOpenAI(model="gpt-4o-mini")
llm_with_tools = llm.bind_tools([get_weather])

# ---- 4. Узлы ----
def agent(state: State) -> dict:
    """Спрашивает LLM по текущей истории."""
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

def tools(state: State) -> dict:
    """Выполняет инструменты, которые попросила LLM."""
    last = state["messages"][-1]
    results = []
    for call in last.tool_calls:
        output = get_weather(**call["args"])
        results.append({"role": "tool", "content": output,
                        "tool_call_id": call["id"]})
    return {"messages": results}

# ---- 5. Маршрутизатор ----
def route(state: State) -> str:
    return "tools" if state["messages"][-1].tool_calls else END

# ---- 6. Сборка графа с ветвлением ----
builder = StateGraph(State)
builder.add_node("agent", agent)
builder.add_node("tools", tools)

builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", route, {"tools": "tools", END: END})
builder.add_edge("tools", "agent")        # ← цикл: после инструментов снова к LLM

graph = builder.compile()

# ---- 7. Запуск ----
result = graph.invoke(
    {"messages": [{"role": "user", "content": "Какая погода в Москве?"}]},
    {"recursion_limit": 10},
)
print(result["messages"][-1].content)
# → "Сейчас в Москве +18°C и ясно."

Проследим поток исполнения:

START   → {messages: [вопрос пользователя]}
agent   → LLM решает вызвать get_weather(city="Москва")
route   → в истории есть tool_calls → "tools"
tools   → выполняем get_weather → дописываем результат в messages
agent   → LLM видит результат, формулирует финальный ответ
route   → tool_calls больше нет → END
END     → возвращаем финальное состояние
ℹ️ Это и есть create_react_agent

LangGraph поставляет готовую функцию create_react_agent(llm, tools), которая собирает ровно такой граф за одну строку. Мы написали его руками, чтобы было видно: внутри нет магии — узлы, условное ребро и цикл. На проде берут готовую функцию, а руками строят графы, когда нужна нестандартная маршрутизация.

Развилка на несколько путей

Маршрутизатор не ограничен выбором из двух вариантов — он может вернуть любую из нескольких меток. Это удобно для классификации запроса: например, направлять пользователя в разные ветки по типу вопроса.

Маршрутизатор с тремя исходами
python
def classify(state: State) -> str:
    intent = state["intent"]          # его проставил предыдущий узел
    if intent == "code":
        return "code_branch"
    if intent == "search":
        return "search_branch"
    return "chitchat_branch"

builder.add_conditional_edges(
    "router_node",
    classify,
    {
        "code_branch":     "write_code",
        "search_branch":   "web_search",
        "chitchat_branch": "small_talk",
    },
)
Параллельные ветки одним маршрутизатором

Если router вернёт список имён узлов, LangGraph запустит их все параллельно (fan-out). Собрать результаты обратно поможет reducer из прошлого урока. А для динамического распараллеливания «по элементам» есть объект Send — это уже тема урока про map-reduce в разделе «Продвинутые возможности».

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

Ошибка 1: router возвращает то, чего нет в карте

Если функция вернула строку, которой нет среди ключей карты исходов, — будет ошибка маршрутизации. Перечисли в карте все возможные возвраты router'а (включая END).

Ошибка 2: маршрутизатор меняет состояние

Router — не узел. Если внутри него вызвать LLM или записать в состояние, изменения потеряются (возврат router'а трактуется как имя узла, а не как обновление). Любая работа — только в узлах.

Ошибка 3: цикл без выхода в END

Если router в цикле agent ↔ tools никогда не возвращает END — граф крутится до GraphRecursionError. Убедись, что есть условие, по которому петля разрывается (нет tool_calls, превышен счётчик попыток и т. п.).

Ошибка 4: обычное ребро там, где нужно условное

add_edge("agent", "tools") сделает переход безусловным — агент пойдёт в инструменты даже когда уже готов ответить. Для развилки нужен именно add_conditional_edges, а не add_edge.

Шпаргалка

Conditional edges — всё в одном месте
python
from langgraph.graph import StateGraph, START, END

# 1. Маршрутизатор: state -> имя узла (НИЧЕГО не меняет в состоянии!)
def route(state) -> str:
    return "tools" if state["messages"][-1].tool_calls else END

# 2. Подключение: из узла, router, карта исходов
builder.add_conditional_edges("agent", route, {"tools": "tools", END: END})

# 3. Цикл — ребро назад создаёт петлю
builder.add_edge("tools", "agent")          # agent <-> tools

# 4. Защита от зацикливания
graph.invoke(state, {"recursion_limit": 25})   # иначе GraphRecursionError

# 5. Несколько исходов — router возвращает разные метки
def classify(state) -> str:
    return state["intent"]                  # "code" | "search" | "chat"
builder.add_conditional_edges("node", classify,
    {"code": "a", "search": "b", "chat": "c"})

# Правила:
#  • router без сайд-эффектов; работа — в узлах
#  • в карте перечислены ВСЕ возвраты router (вкл. END)
#  • в цикле обязателен путь к END

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

Построй граф с реальной развилкой:

Задание: агент-«угадай число» с циклом

  1. Опиши State с полями secret: int, guess: int и attempts: int.
  2. Сделай узел make_guess, который генерирует случайную догадку и увеличивает attempts.
  3. Напиши маршрутизатор check: если guess == secret — вернуть END, иначе — "make_guess" (повторить).
  4. Собери граф START → make_guess, затем условное ребро от make_guess через check на саму себя или в END. Запусти с recursion_limit и убедись, что цикл завершается при совпадении.
  5. Добавь в маршрутизатор выход в END после 10 попыток (через attempts) — чтобы остановка управлялась логикой, а не только аварийным лимитом.
  6. Со звёздочкой: сделай узел-классификатор, который по чётности secret направляет в разные узлы-«подсказки» (even_hint / odd_hint) перед догадкой.

Что дальше