Как LangGraph выполняет граф: от graph.invoke() до цикла Pregel

LangGraph кажется простым в освоении. add_node(), add_edge(), add_conditional_edges() и graph.invoke() понятны сразу, и рабочие графы собираются за считаные часы. Но после нескольких проектов обычно начинает донимать вопрос: что на самом деле происходит при вызове graph.invoke()? Как фреймворк выбирает, какой узел запустить, обрабатывает условные рёбра, обновляет каналы и запускает узлы параллельно? Где здесь повторные попытки, чекпоинты, прерывания и стриминг? Большинство руководств объясняют, как пользоваться LangGraph, и пропускают внутреннее устройство. Этот разбор проходит путь вызова от публичного API вглубь движка Pregel, чтобы шаг за шагом показать, как планируются задачи и движется состояние.

Почему Pregel

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

Цикл повторяется, пока не останется задач, готовых к запуску, или пока выполнение не будет прервано, не завершится ошибкой или не упрётся в лимит рекурсии.

Шаг 1: graph.invoke()

Всё начинается с этого вызова:

result = graph.invoke(state)

Можно было бы ожидать, что эта функция выполняет граф, но это не так. graph.invoke() крошечная. Она существует как удобный API, который выполняет граф до конца и возвращает итоговое состояние, а настоящую работу передаёт stream():

graph.invoke()
        │
        ▼
     stream()

Причина в том, что в LangGraph уже есть потоковый движок, который может отдавать обновления после каждого шага Pregel. Вместо того чтобы поддерживать два пути выполнения, invoke() внутри потребляет поток и возвращает последнее значение. Дальше настоящее выполнение происходит внутри stream().

Шаг 2: stream()

Если invoke() — публичный фасад, то stream() — слой оркестрации. Прежде чем сможет запуститься хоть один узел, он готовит всю среду выполнения:

• Заполнение конфигурации по умолчанию
• Настройку менеджеров колбэков и чекпоинтинга
• Подключение обработчиков стриминга
• Создание объекта Runtime
• Построение движка, который выполняет работу

Последний пункт важнее всего. Во время этой настройки LangGraph создаёт два объекта, которые управляют всем остальным запуском:

SyncPregelLoop
PregelRunner

Они делят выполнение на две отдельные задачи:

SyncPregelLoop
        │
        ├── decides WHAT should run
        │
PregelRunner
        │
        └── executes it

Разделение планирования и выполнения означает, что ни одно не зависит от другого.

Шаг 3: SyncPregelLoop

Когда инициализация закончена, выполнение входит в цикл Pregel, где каждая итерация — это один суперстеп Pregel. Каждая итерация проходит один и тот же жизненный цикл:

Determine runnable nodes
        │
Execute those nodes
        │
Apply every write together
        │
Trigger the next set of nodes

Важно, что записи применяются не сразу. Все узлы на шаге видят одинаковый снимок состояния, а записи фиксируются одним пакетом только после того, как завершится последний готовый к запуску узел. Именно это единственное проектное решение делает параллельное выполнение детерминированным.

Шаг 4: tick()

Первая функция, которая выполняется на каждом шаге Pregel, — tick(). Часто думают, что она выполняет узлы графа, но это не так. Она только готовит работу для текущей итерации и выполняет несколько служебных задач:

• Проверяет, не пора ли остановить выполнение из-за лимита рекурсии, прерывания или завершения
• Восстанавливает ожидающие записи при возобновлении из чекпоинта
• Отправляет отладочные события
• Вызывает prepare_next_tasks()

Затем она возвращает булево значение. True означает, что работа ещё осталась, а False — что выполнение закончено. В этот момент ни один узел ещё не запускался. tick() лишь определила, что могло бы запуститься.

Шаг 5: prepare_next_tasks()

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

PUSH-задачи — явные. Они возникают из конструкций вроде Send, goto и call(), то есть что-то напрямую попросило запустить другой узел.

PULL-задачи — неявные. Их никто не запрашивает. Они срабатывают потому, что один из входных каналов узла изменился на предыдущем шаге. Для каждого узла-кандидата LangGraph проверяет, получил ли какой-нибудь триггерный канал более новую версию, и если да, узел становится готовым к выполнению. Найдя готовые узлы, планировщик передаёт каждый из них в:

prepare_single_task()

Шаг 6: prepare_single_task()

Пока известно лишь, что узел X должен запуститься. Эта функция превращает абстрактное решение в то, что движок может выполнить, создавая PregelExecutableTask. Код идёт по одному из двух путей.

Для PUSH явно запланированные задачи обрабатываются одной из этих функций, в зависимости от того, как они были созданы:

prepare_push_task_send()

or

prepare_push_task_functional()

Путь PULL делает заметно больше. Сначала _triggers() подтверждает, что узел должен запуститься. Она не просто проверяет, есть ли в канале данные. Она сравнивает версии каналов:

current version  > last version consumed

Выполнение запускают только более новые версии, поэтому узлы не перезапускаются снова и снова на неизменном состоянии. Когда узел выбран, _proc_input() готовит его вход из каналов графа, управляемых значений, scratchpad и, где возможно, закэшированных входов. Если есть маппер, вход преобразуется в схему, которую ожидает узел, например:

TypedDict

↓

Pydantic Model

Наконец LangGraph собирает метаданные выполнения, конфигурацию среды, информацию о чекпоинте и настройки повторов и упаковывает всё это в PregelExecutableTask. На этом работа планировщика заканчивается, и задача готова к запуску.

Шаг 7: PregelExecutableTask

Считайте это пакетом, который передаётся исполнителю. В нём есть всё необходимое для запуска одного узла: runnable узла, подготовленный вход, конфигурация среды, политика повторов, writers, пространство имён чекпоинта, метаданные выполнения и информация о кэше. Ничто внутри не решает, когда начнётся выполнение. Это лишь содержимое, нужное, когда выполнение уже началось.

Шаг 8: PregelRunner.tick()

Здесь наконец и происходит выполнение. Если готова только одна задача, например:

Node A

LangGraph напрямую вызывает следующее:

run_with_retry()

Когда несколько задач независимы, как эти:

Node B
Node C
Node D

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

Стоит отметить: параллелизм существует только в пределах одного шага Pregel. Последовательные зависимости по-прежнему соблюдаются, потому что следующий шаг не может начаться, пока не завершится каждая задача текущего.

Шаг 9: run_with_retry()

Именно здесь впервые вызывается Python-функция вашего узла. Все предыдущие шаги были планированием. Эта функция выполняет runnable и занимается всем, что его окружает. Её обязанности: вызвать узел, стереть устаревшие записи перед любым повтором, применить политику повторов с экспоненциальной задержкой, обработать ParentCommand и GraphInterrupt и записать каждую запись, которую выдаёт узел. Когда выполнение заканчивается, управление возвращается в цикл Pregel, где after_tick() фиксирует все записи вместе, прежде чем начнётся следующий шаг.

Подробный пример с параллельными узлами

Возьмём граф, в котором два узла действительно работают параллельно, чтобы стало видно, как это решают версии каналов:

                ┌──→ fetch_weather ──┐
START → classify┤                     ├──→ respond → END
                └──→ fetch_news ─────┘

Здесь classify разветвляется на fetch_weather и fetch_news, которые не зависят друг от друга, и оба ведут в respond. Ни один не зависит от результата другого — только от результата classify. Вот версии каналов в начале:

channel: "input"           version: 1   (last consumed by classify: 0)
channel: "weather_data"    version: 0   (last consumed by respond: 0)
channel: "news_data"       version: 0   (last consumed by respond: 0)

• _triggers() сравнивает триггерный канал каждого узла с тем, что он потребил в последний раз.
• classify следит за "input". Версия — 1, последняя потреблённая — 0, значит, 1 > 0, и она считается более новой.
• fetch_weather и fetch_news следят за каналом, в который classify ещё не писал, поэтому они не срабатывают.
• В итоге на этом шаге PULL-задачей становится только classify.

На шаге 1 запускается classify и записывает такой результат:

{"needs_weather": true, "needs_news": true}

Это повышает версии двух нижестоящих каналов:

channel: "needs_weather"   version: 0 → 1
channel: "needs_news"      version: 0 → 1

• prepare_next_tasks() снова просматривает граф. fetch_weather следит за "needs_weather", где версия 1 > 0 (последняя потреблённая), поэтому он срабатывает.
• fetch_news следит за "needs_news", где 1 > 0 тоже выполняется, поэтому он срабатывает.
• Оба становятся PULL-задачами на одном и том же шаге Pregel, поскольку оба условия стали истинными от одной записи.
• respond следит за "weather_data" и "news_data", и ни один из них пока не изменился, поэтому respond не срабатывает.

Вот в этот момент и возникает параллелизм. Два узла срабатывают от записей одного и того же суперстепа, поэтому PregelRunner.tick() отправляет оба как объекты Future и выполняет их одновременно.

На шаге 2 fetch_weather и fetch_news работают одновременно, и оба записывают:

channel: "weather_data"    version: 0 → 1
channel: "news_data"       version: 0 → 1

• Хотя два узла стартуют и завершаются независимо, их записи не применяются, пока не завершатся оба future. Это правило «применять все записи вместе» из раздела о цикле.
• Когда пакет зафиксирован, prepare_next_tasks() снова обходит граф. respond слушает "weather_data" и "news_data", и каждый из них ушёл дальше последней потреблённой версии (0), поэтому он срабатывает.
• respond — единственная PULL-задача на этом шаге.

На шаге 3 запускается respond, и никаких изменений каналов за этим не следует. Ниже по графу это ничего не запускает, а значит, tick() возвращает False, и цикл останавливается.

Всего получается три шага Pregel, но на шаге 1 два узла были запланированы одновременно по одной-единственной причине: prepare_next_tasks() обнаружила две версии каналов, повышенные одним пакетом записей. Сравните с чисто последовательным графом, где каждый шаг даёт одну готовую задачу, поскольку за раз меняется только один канал. Правило сравнения версий в обоих случаях одно и то же. Параллелизм — не особый случай, который LangGraph распознаёт. Он возникает естественным образом всякий раз, когда несколько каналов обновляются в одном суперстепе.