Зачем графу ветвление?
Безусловное ребро add_edge("A", "B") говорит: «после A всегда иди в B». Этого хватает для конвейера, но не для агента. Агенту нужно принимать решение по результату шага:
- LLM попросила вызвать инструмент → идём в узел инструментов; нет → выходим с ответом.
- Найденного контекста мало → возвращаемся искать ещё; достаточно → генерируем ответ.
- Ревьюер забраковал черновик → на доработку; одобрил → завершаем.
Все эти «→» зависят от данных, которых нет на момент сборки графа — они появятся только в рантайме, в состоянии. Значит, нужно ребро, которое смотрит на состояние и решает, куда шагнуть. Это и есть conditional edge.
Вспомни while-агента из начала модуля: вся его логика держалась на if response.tool_calls: .... Условное ребро — это тот же if, но вынесенный из тела шага в структуру графа. Решение «куда дальше» становится видимым на схеме, а не спрятанным внутри функции.
Функция-маршрутизатор
Сердце условного ребра — функция-маршрутизатор (router). Это функция state → имя следующего узла. Она читает состояние и возвращает строку — куда идти дальше. Важнейшее правило: маршрутизатор только выбирает путь и ничего не меняет в состоянии. Это не узел, это «стрелочник».
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 крошечным и без сайд-эффектов.
add_conditional_edges и карта исходов
Маршрутизатор подключается к узлу через add_conditional_edges. Метод принимает: из какого узла ветвимся, функцию-маршрутизатор и (опционально) карту исходов — словарь «что вернул router → в какой узел идти».
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 вместо вечного выполнения.
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.
Обрати внимание: agent → tools → agent образует цикл, а единственный выход наружу — жёлтая ветка «нет → END». Граф будет крутиться, пока LLM просит инструменты, и завершится, как только она сформулирует финальный ответ.
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 → возвращаем финальное состояние
LangGraph поставляет готовую функцию create_react_agent(llm, tools), которая собирает ровно такой граф за одну строку. Мы написали его руками, чтобы было видно: внутри нет магии — узлы, условное ребро и цикл. На проде берут готовую функцию, а руками строят графы, когда нужна нестандартная маршрутизация.
Развилка на несколько путей
Маршрутизатор не ограничен выбором из двух вариантов — он может вернуть любую из нескольких меток. Это удобно для классификации запроса: например, направлять пользователя в разные ветки по типу вопроса.
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 в разделе «Продвинутые возможности».
Типичные ошибки
Если функция вернула строку, которой нет среди ключей карты исходов, — будет ошибка маршрутизации. Перечисли в карте все возможные возвраты router'а (включая END).
Router — не узел. Если внутри него вызвать LLM или записать в состояние, изменения потеряются (возврат router'а трактуется как имя узла, а не как обновление). Любая работа — только в узлах.
Если router в цикле agent ↔ tools никогда не возвращает END — граф крутится до GraphRecursionError. Убедись, что есть условие, по которому петля разрывается (нет tool_calls, превышен счётчик попыток и т. п.).
add_edge("agent", "tools") сделает переход безусловным — агент пойдёт в инструменты даже когда уже готов ответить. Для развилки нужен именно add_conditional_edges, а не add_edge.
Шпаргалка
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
Практическое задание
Построй граф с реальной развилкой:
Задание: агент-«угадай число» с циклом
- Опиши
Stateс полямиsecret: int,guess: intиattempts: int. - Сделай узел
make_guess, который генерирует случайную догадку и увеличиваетattempts. - Напиши маршрутизатор
check: еслиguess == secret— вернутьEND, иначе —"make_guess"(повторить). - Собери граф
START → make_guess, затем условное ребро отmake_guessчерезcheckна саму себя или вEND. Запусти сrecursion_limitи убедись, что цикл завершается при совпадении. - Добавь в маршрутизатор выход в
ENDпосле 10 попыток (черезattempts) — чтобы остановка управлялась логикой, а не только аварийным лимитом. - Со звёздочкой: сделай узел-классификатор, который по чётности
secretнаправляет в разные узлы-«подсказки» (even_hint/odd_hint) перед догадкой.