Як 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 розпізнає. Він виникає природно щоразу, коли кілька каналів оновлюються в одному суперкроці.