LangGraph Nodes and Edges

Суть

• Node — Python-функция state -> dict|State: вся бизнес-логика живёт здесь (вызов LLM, обращение к БД, инструмент). Возвращает частичное обновление, движок его применяет.
• Edge — задаёт порядок исполнения:
  • обычное (add_edge("A","B")) — факт «B после A»;
  • условное (add_conditional_edges) — функция-маршрутизатор возвращает имя следующего узла (str) на основе state; условием может быть что угодно, вплоть до решения LLM.

Зачем это нужно

Разделение «логика в узлах / поток в рёбрах» делает архитектуру явной и управляемой: маршрут не зашит в код функций, а описан декларативно — его можно ветвить, замыкать в цикл и визуализировать.

Как работает

1. builder = StateGraph(State) — создаём нескомпилированный граф.
2. builder.add_node("name", fn) — регистрируем узлы.
3. Рёбра: add_edge(START, "A"), add_edge("A","B"), add_edge("B", END); ветвление — add_conditional_edges("from", router_fn, {"key": "node"}).
4. graph = builder.compile() — переводим описание в исполняемый объект (на этом же этапе подключают checkpointer — см. LangGraph Checkpointers).
5. START / END — предопределённые точки входа/выхода.

Пример

from langgraph.graph import StateGraph, START, END

def route_by_number(state: State) -> str:        # функция-маршрутизатор
    return "positive" if state["number"] > 0 else "negative"

builder = StateGraph(State)
builder.add_node("check", check_number)
builder.add_node("positive", handle_positive)
builder.add_node("negative", handle_negative)
builder.add_edge(START, "check")
builder.add_conditional_edges("check", route_by_number,
                              {"positive": "positive", "negative": "negative"})
builder.add_edge("positive", END); builder.add_edge("negative", END)
graph = builder.compile()

Повтор — это ребро, а не цикл внутри узла

Типовая задача: узел вернул слишком короткий результат, нужно попробовать ещё раз. Реализовать это можно двумя способами, и они не равнозначны.

Первый — while внутри узла: узел сам крутится, пока результат не устроит. Работает, но повтор становится невидимым. Снаружи это один длинный шаг: в трассировке одна запись, в состоянии — только финальный результат, между попытками нет чекпоинта, и общий лимит рекурсии графа такой цикл не ограничивает.

Второй — условное ребро, ведущее узел сам в себя:

def route_from_researcher(state) -> str:
    notes = state.get("research_notes") or ""
    if len(notes) < 150 and state.get("retry_count", 0) < 2:
        return "researcher"          # ещё попытка — возврат в тот же узел
    return "writer"                  # достаточно — идём дальше

builder.add_conditional_edges("researcher", route_from_researcher,
                              {"researcher": "researcher", "writer": "writer"})

Каждая попытка становится полноценным супершагом: отдельная запись в трассировке, отдельный чекпоинт, счётчик попыток в состоянии, а не в локальной переменной. Из этого же следует, что повтор попадает под общий лимит рекурсии графа — отдельного предохранителя от бесконечного цикла писать не нужно.

Есть и содержательное отличие: условие повтора уезжает из узла в роутер. Узел отвечает только за «сделать», решение «достаточно ли» принимает маршрутизатор. Это то же разделение «логика в узлах, поток в рёбрах», просто применённое к ретраю, — и оно позволяет менять критерий приёмки, не трогая код узла.

Отличать от RetryPolicy (LangGraph Reliability): та повторяет узел при исключении — сеть отвалилась, API вернул пятисотку. Здесь узел отработал штатно, но результат не устроил по смыслу, и это решение прикладное, а не транспортное.

Две развилки, которые встречаются сразу

StateGraph против готовых обёрток. Обёртка удобна, пока состояние агента — это список сообщений и больше ничего. Как только в состоянии появляется что-то своё (счётчики, промежуточные артефакты, флаги маршрутизации), обёртка начинает мешать: её схема не расширяется, и данные приходится прятать в сообщения. Практически это значит, что StateGraph берут сразу, если заранее известно про хотя бы одно собственное поле, — переезд позже дороже, чем описать схему сначала (LangGraph State).

Счётчик попыток и залипание между запусками. Счётчик — это поле состояния, а состояние переживает перезапуск вместе с чекпоинтом (LangGraph Checkpointers). Отсюда типичная ошибка: счётчик, увеличенный в прошлом прогоне того же треда, остаётся ненулевым, и новый запрос стартует «уже почти исчерпанным». Лечится сбросом на входе в цикл, а не в узле повтора: сброс должен принадлежать шагу, который открывает попытку, а reducer этого поля — заменять значение, а не накапливать его (LangGraph Reducers).

Связано с

• LangGraph State — узлы читают/обновляют state, рёбра задают порядок
• Agent Routing — условные рёбра = реализация роутинга в графе
• LangGraph ReAct Loop — минимальный граф из узлов и условного ребра
• LangGraph — узлы+рёбра+state = три примитива State Machine
• LangGraph Reliability — RetryPolicy на исключение; здесь повтор по смыслу результата
• Pregel Model — почему каждая попытка через ребро становится отдельным супершагом