AI Orchestration

AI Orchestration (Оркестрация ИИ) это отдельный управляющий слой, который забирает из кода приложения последовательность вызовов языковой модели, состояние задачи и восстановление после сбоя. Пока приложение делает один вызов модели и печатает ответ, оркестрация ему не нужна. Она появляется тогда, когда вызовов много, они зависят друг от друга, между ними ждут человека, а работа обязана пережить перезапуск сервиса.

Что такое оркестрация ИИ

Оркестрация ИИ это не один продукт с единым номером версии, а класс решений. Общего у них ровно одно, а именно вынесенное из прикладного кода управление тем, в каком порядке вызываются модель, инструменты и другие агенты. По состоянию на 2026 год внутри класса различимы три группы.

• Фреймворки оркестрации агентов. Подключаются к приложению как библиотека. Сюда относятся LangGraph (графовый фреймворк от команды LangChain, версия 1.0 вышла в октябре 2025), CrewAI, LlamaIndex Workflows и Microsoft Agent Framework, в котором Microsoft свела вместе свои прежние проекты Semantic Kernel и AutoGen.
• Долгоживущие исполнители рабочих процессов. Temporal (платформа durable execution для процессов, которые живут дольше любого отдельного процесса-воркера) и Apache Airflow (планировщик пакетных DAG, привычный дата-инженерам). Они берут на себя расписание, повторы и хранение состояния для всего конвейера, а не для одного агента.
• Протоколы взаимодействия. MCP (Model Context Protocol) связывает агента с инструментами и источниками данных, A2A (Agent2Agent Protocol) связывает агента с соседним агентом. Оркестраторами они не являются, зато задают словарь, на котором оркестратор строится.

Разбор топологий, по которым собираются такие системы, есть в статье Эволюция ИИ-агентов и топологии рабочих процессов LLM.

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

Архитектура управляющего слоя

Единой эталонной схемы у класса нет, но набор компонентов повторяется от фреймворка к фреймворку. Приложение отдаёт оркестратору задачу и получает результат. Оркестратор держит описание процесса, общее состояние и хранилище снимков этого состояния. Ниже него сидят исполнители, то есть агенты и узлы, которые ходят в модель, в инструменты через MCP и в соседние агентные системы через A2A.

Такое расслоение полезно тем, что каждый слой меняется отдельно от соседних.

Граф рабочего процесса и общее состояние

В графовых фреймворках процесс описывается узлами и рёбрами. Узел это функция, которая получает состояние и возвращает свой кусок обновления. Ребро задаёт, куда управление уходит дальше. В LangGraph конструкция называется StateGraph, а состояние объявляется типизированной структурой, где у каждого поля есть правило слияния. Поле-журнал с редьюсером-склейкой узлы дописывают, а не затирают, поэтому параллельные ветки не вытесняют результаты друг друга.

Важная деталь, которая ломает интуицию новичка. Агенты внутри графа между собой не разговаривают напрямую. Исполнитель кладёт результат в общее состояние, следующий исполнитель его оттуда читает. Прямой вызов одного агента из другого вернул бы всю сложность обратно в код, ради выноса которой оркестратор и заводили.

Инструменты через MCP, соседние агенты через A2A

Оба протокола за год стали инфраструктурой по умолчанию. MCP создан в Anthropic и в декабре 2025 года передан в Agentic AI Foundation под крылом Linux Foundation, к февралю 2026 года суммарные загрузки его SDK для Python и TypeScript перевалили за 97 миллионов в месяц. A2A появился в Google в апреле 2025 года, ушёл в Linux Foundation в июне того же года, до стабильной версии 1.0 дорос в марте 2026 года, а патч 1.0.1 вышел в мае.

Для архитектора это означает простое разделение зон. Оркестратор отвечает за порядок и состояние внутри своей системы, MCP отвечает за доступ к внешнему миру, A2A отвечает за делегирование работы наружу. Смешивать эти роли не стоит, иначе замена одного фреймворка потянет за собой переписывание интеграций.

Как это работает под капотом

Дальше механика показана на LangGraph, потому что он документирован подробнее прочих и на нём сделано демо. У Temporal и Airflow восстановление устроено иначе.

Суперстеп и снимок состояния по thread_id

Выполнение графа разбито на суперстепы. Один суперстеп это отработка узла целиком. После каждого суперстепа компонент-чекпоинтер сохраняет снимок общего состояния, помеченный идентификатором thread_id. Идентификатор привязан к задаче или диалогу, а не к запуску скрипта. Вызвали граф с тем же thread_id, и он продолжил с последнего снимка, вызвали с новым, и он начал с нуля.

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

Пауза на человеке через interrupt_before

Из того же механизма снимков вырастает пауза на человеке. Параметр interrupt_before останавливает граф перед запуском указанного узла, interrupt_after после. Первое нужно для подтверждения необратимого действия, второе для проверки уже полученного результата.

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

Паттерны координации

Способов расставить узлы немного, и выбирают между ними по тому, кто принимает решение о следующем шаге.

В живых системах паттерны сочетаются. Супервизор наверху, конвейер внутри роли, веер на сборе фактов и пауза перед отправкой результата наружу.

Фреймворк или durable-исполнитель

Второй развилкой становится выбор между фреймворком внутри приложения и внешним исполнителем. Разница не в наборе возможностей, а в том, что происходит при падении.

Разница между реплеем и повтором становится осязаемой на цифрах. Упавшая тридцатиминутная задача Airflow при повторе отрабатывает эти тридцать минут заново целиком, потому что частичного прогресса внутри задачи планировщик не помнит. Temporal восстанавливает состояние, прокручивая историю событий, и переделывает только незавершённый шаг. LangGraph занимает середину, его гранулярность равна размеру узла графа, поэтому крупный узел на десять минут работы теряется весь.

Дата-инженерная половина вопроса решается сближением подходов. Airflow оброс инструментами для работы с моделями, разбор есть в статье Обзор Airflow AI SDK.

Когда оркестратор нужен, а когда это лишний слой

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

• Шагов больше трёх и порядок зависит от данных. Ветвление в прикладном коде быстро превращается в нечитаемый набор условий.
• Процесс длиннее одного запроса. Диалог живёт неделю, задача ждёт согласования, воркер перезапускается по деплою.
• Нужна пауза на человеке. Согласование платежа, публикация письма клиенту, изменение в проде.
• Шаги дорогие. Если каждый вызов стоит минуты работы модели, повтор всей цепочки после сбоя обходится ощутимо дороже хранения снимков.
• Нужен разбор полётов. История шагов даёт ответ на вопрос, почему система приняла именно такое решение.

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

Supervisor-граф на LangGraph с чекпоинтом в SQLite

Демо к статье поднимает управляющий слой над локальной моделью qwen2.5:7b в Ollama, сохраняет состояние в SQLite, а потом убивает процесс на середине работы и доигрывает граф из сохранённого снимка. Прогон сделан на macOS 26.5.2 arm64, Python 3.12.13, langgraph 1.2.11, langgraph-checkpoint-sqlite 3.1.1, langchain-ollama 1.1.0, Ollama 0.32.13.

По традиции весь код используемый в статье выкладываем на наш GitHub репозиторий

Первый файл, supervisor_graph.py, собирает граф из трёх узлов. Управляющий узел спрашивает модель, кого звать дальше, и проверяет её ответ по состоянию. Два исполнителя обмениваются результатами только через общее состояние.

# LangGraph 1.2.11, langgraph-checkpoint-sqlite 3.1.1, langchain-ollama 1.1.0,
# Ollama 0.32.13, модель qwen2.5:7b. Прогнано на стенде 2026-08-25.
# Supervisor-граф: управляющий узел выбирает исполнителя, состояние после каждого
# суперстепа уходит в чекпоинтер SQLite.

import os
import sqlite3
import time
from typing import Annotated, TypedDict

from langchain_ollama import ChatOllama
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command

DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "checkpoints.sqlite")
MODEL = "qwen2.5:7b"
ROLES = ("researcher", "writer")


def append(left: list, right: list) -> list:
    """Редьюсер общего состояния: узлы дописывают в журнал, а не затирают его."""
    return (left or []) + (right or [])


class OrchestrationState(TypedDict):
    """Общее состояние графа. Его снимок и попадает в чекпоинтер после каждого шага."""
    task: str
    notes: str
    draft: str
    trace: Annotated[list, append]


llm = ChatOllama(model=MODEL, temperature=0)


def supervisor(state: OrchestrationState) -> Command:
    """Управляющий узел. Спрашивает модель, кого звать следующим, и возвращает
    Command с полем goto: это и есть маршрутизация внутри графа."""
    done = []
    if state.get("notes"):
        done.append("researcher")
    if state.get("draft"):
        done.append("writer")

    prompt = (
        "Ты диспетчер конвейера из двух исполнителей.\n"
        f"researcher собирает факты, writer пишет текст по фактам.\n"
        f"Задача: {state['task']}\n"
        f"Уже отработали: {', '.join(done) if done else 'никто'}\n"
        "Ответь ровно одним словом из списка: researcher, writer, done."
    )
    t0 = time.time()
    raw = llm.invoke(prompt).content.strip().lower()
    took = round(time.time() - t0, 2)

    # Проверка предусловий. Локальная модель регулярно предлагает writer до того,
    # как researcher собрал факты. Оркестратор обязан такой маршрут отклонить:
    # порядок шагов это его зона ответственности, а не модели.
    def valid(name: str) -> bool:
        if name == "researcher":
            return not state.get("notes")
        if name == "writer":
            return bool(state.get("notes")) and not state.get("draft")
        return bool(state.get("notes")) and bool(state.get("draft"))

    choice = next((r for r in (*ROLES, "done") if r in raw), None)
    fallback = ""
    if choice is None or not valid(choice):
        rejected = choice or f"ответ '{raw[:30]}'"
        choice = "researcher" if not state.get("notes") else ("writer" if not state.get("draft") else "done")
        fallback = f" (модель предложила {rejected}, маршрут отклонён по предусловию)"

    line = f"supervisor -> {choice} за {took} с{fallback}"
    print("[граф]", line)
    goto = END if choice == "done" else choice
    return Command(goto=goto, update={"trace": [line]})


def researcher(state: OrchestrationState) -> dict:
    """Первый исполнитель. Возвращает только свой кусок состояния."""
    t0 = time.time()
    text = llm.invoke(
        f"Задача: {state['task']}\nДай ровно три коротких тезиса, по одному в строке, без вступления."
    ).content.strip()
    took = round(time.time() - t0, 2)
    print(f"[граф] researcher отработал за {took} с, {len(text)} символов")
    return {"notes": text, "trace": [f"researcher: {took} с"]}


def writer(state: OrchestrationState) -> dict:
    """Второй исполнитель. Видит результат первого через общее состояние,
    а не через прямой вызов: агенты между собой не общаются."""
    t0 = time.time()
    text = llm.invoke(
        f"Задача: {state['task']}\nТезисы:\n{state['notes']}\n"
        "Напиши один абзац на 3-4 предложения по этим тезисам."
    ).content.strip()
    took = round(time.time() - t0, 2)
    print(f"[граф] writer отработал за {took} с, {len(text)} символов")
    return {"draft": text, "trace": [f"writer: {took} с"]}


def build_graph(checkpointer):
    """Сборка графа. Ребра от исполнителей ведут назад в супервизор,
    поэтому управление всегда возвращается в одну точку."""
    g = StateGraph(OrchestrationState)
    g.add_node("supervisor", supervisor)
    g.add_node("researcher", researcher)
    g.add_node("writer", writer)
    g.add_edge(START, "supervisor")
    g.add_edge("researcher", "supervisor")
    g.add_edge("writer", "supervisor")
    return g.compile(checkpointer=checkpointer)


def main():
    conn = sqlite3.connect(DB_PATH, check_same_thread=False)
    checkpointer = SqliteSaver(conn)
    graph = build_graph(checkpointer)

    thread_id = "demo-supervisor"
    config = {"configurable": {"thread_id": thread_id}}
    task = "Чем оркестрация ИИ отличается от обычного вызова языковой модели из кода приложения"

    print(f"=== прогон графа, thread_id={thread_id} ===")
    t0 = time.time()
    # durability='sync' пишет снимок состояния до возврата управления:
    # так чекпоинт переживёт обрыв процесса ровно на этом шаге.
    for step, snapshot in enumerate(
        graph.stream({"task": task, "notes": "", "draft": "", "trace": []},
                     config, stream_mode="values", durability="sync"), start=1):
        print(f"[суперстеп {step}] notes={len(snapshot.get('notes',''))} симв., "
              f"draft={len(snapshot.get('draft',''))} симв., шагов в журнале {len(snapshot.get('trace', []))}")
    total = round(time.time() - t0, 2)

    final = graph.get_state(config).values
    print(f"=== граф отработал за {total} с ===")
    print("--- журнал маршрутизации ---")
    for line in final["trace"]:
        print(" ", line)
    print("--- готовый текст ---")
    print(final["draft"])

    history = list(graph.get_state_history(config))
    print(f"--- чекпоинтов в SQLite по thread_id={thread_id}: {len(history)} ---")
    for snap in reversed(history):
        nxt = snap.next or ("END",)
        print(f"  {snap.config['configurable']['checkpoint_id'][:8]}  следующий узел: {','.join(nxt)}")
    print(f"размер файла чекпоинтов: {os.path.getsize(DB_PATH)} байт")
    conn.close()


if __name__ == "__main__":
    main()

Прогон занял 22,95 секунды на шесть суперстепов, и самое интересное в нём случилось на первом шаге.

# вывод прогона supervisor_graph.py, LangGraph 1.2.11, модель qwen2.5:7b, 25.08.2026
[граф] supervisor -> researcher за 0.71 с (модель предложила writer, маршрут отклонён по предусловию)
[граф] researcher отработал за 7.27 с, 276 символов
[граф] supervisor -> writer за 0.5 с
[граф] writer отработал за 13.93 с, 552 символов
[граф] supervisor -> done за 0.51 с
=== граф отработал за 22.95 с ===
--- чекпоинтов в SQLite по thread_id=demo-supervisor: 7 ---
размер файла чекпоинтов: 4096 байт

Модель на роли супервизора в обоих прогонах предложила сразу звать writer, хотя фактов ещё не было. Управляющий узел этот маршрут отклонил по предусловию. Результат неудобный, но показательный, потому что так и выглядит граница ответственности. Порядок шагов держит оркестратор, а модель лишь предлагает вариант. Семь чекпоинтов на шесть суперстепов объясняются стартовым снимком входа, и вся история графа уместилась в 4096 байт, то есть в одну страницу SQLite.

Второй файл, resume_demo.py, проверяет главное обещание управляющего слоя. Он запускает граф отдельным процессом, убивает его сигналом KILL сразу после того, как первый исполнитель сохранил результат, и продолжает работу в родительском процессе с тем же thread_id.

# LangGraph 1.2.11, langgraph-checkpoint-sqlite 3.1.1, langchain-ollama 1.1.0,
# Ollama 0.32.13, модель qwen2.5:7b. Прогнано на стенде 2026-08-25.
# Обрыв процесса посреди графа и возобновление с того же thread_id:
# фаза 1 идёт отдельным процессом, который убивается сигналом KILL,
# фаза 2 поднимает работу из чекпоинта, не переделывая уже сделанное.

import os
import signal
import sqlite3
import subprocess
import sys
import time

from langgraph.checkpoint.sqlite import SqliteSaver

from supervisor_graph import build_graph

HERE = os.path.dirname(os.path.abspath(__file__))
DB_PATH = os.path.join(HERE, "resume.sqlite")
THREAD_ID = "demo-resume"
TASK = "Зачем оркестратору ИИ хранить состояние вне процесса приложения"
CONFIG = {"configurable": {"thread_id": THREAD_ID}}


def open_graph():
    conn = sqlite3.connect(DB_PATH, check_same_thread=False)
    return conn, build_graph(SqliteSaver(conn))


def phase1_worker():
    """Дочерний процесс. Доходит до первого сохранённого шага исполнителя,
    печатает маркер и дальше ждёт, пока его убьют."""
    conn, graph = open_graph()
    for snapshot in graph.stream(
        {"task": TASK, "notes": "", "draft": "", "trace": []},
        CONFIG, stream_mode="values", durability="sync",
    ):
        # как только researcher положил свой результат в состояние,
        # снимок этого состояния уже лежит в SQLite
        if snapshot.get("notes"):
            print("CHECKPOINT_SAVED", flush=True)
            time.sleep(60)
    conn.close()


def phase2_parent():
    """Родитель. Запускает фазу 1, убивает её на середине графа
    и продолжает тот же прогон в своём процессе."""
    if os.path.exists(DB_PATH):
        os.remove(DB_PATH)

    print("=== фаза 1: прогон в отдельном процессе ===")
    proc = subprocess.Popen(
        [sys.executable, os.path.abspath(__file__), "worker"],
        stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, cwd=HERE,
    )
    t0 = time.time()
    for line in proc.stdout:
        print("  [pid %d] %s" % (proc.pid, line.rstrip()))
        if line.startswith("CHECKPOINT_SAVED"):
            break
    # честный обрыв: SIGKILL не даёт процессу ничего доделать и ничего дописать
    os.kill(proc.pid, signal.SIGKILL)
    proc.wait()
    print(f"процесс {proc.pid} убит сигналом KILL через {round(time.time()-t0,2)} с, "
          f"код возврата {proc.returncode}")

    conn, graph = open_graph()
    state = graph.get_state(CONFIG)
    print("--- что осталось в SQLite после обрыва ---")
    print(f"  notes: {len(state.values.get('notes',''))} симв., "
          f"draft: {len(state.values.get('draft',''))} симв.")
    print(f"  следующий узел: {','.join(state.next) if state.next else 'END'}")
    print(f"  чекпоинтов по thread_id={THREAD_ID}: {len(list(graph.get_state_history(CONFIG)))}")

    print("=== фаза 2: возобновление с того же thread_id ===")
    # None вместо входа означает «продолжай с сохранённого состояния»
    t0 = time.time()
    final = graph.invoke(None, CONFIG, durability="sync")
    print(f"граф доигран за {round(time.time()-t0,2)} с")
    print("--- журнал маршрутизации за оба процесса ---")
    for line in final["trace"]:
        print(" ", line)
    print("--- готовый текст ---")
    print(final["draft"])
    conn.close()


if __name__ == "__main__":
    if len(sys.argv) > 1 and sys.argv[1] == "worker":
        phase1_worker()
    else:
        phase2_parent()

Вывод второго демо показывает, что именно пережило убийство процесса.

# вывод прогона resume_demo.py, LangGraph 1.2.11, модель qwen2.5:7b, 25.08.2026
процесс 15239 убит сигналом KILL через 7.57 с, код возврата -9
--- что осталось в SQLite после обрыва ---
  notes: 186 симв., draft: 0 симв.
  следующий узел: supervisor
  чекпоинтов по thread_id=demo-resume: 4
=== фаза 2: возобновление с того же thread_id ===
[граф] supervisor -> writer за 0.5 с
[граф] writer отработал за 12.27 с, 520 символов
граф доигран за 13.29 с

Код возврата минус девять означает, что процесс действительно убит и ничего доделать не успел. Тем не менее в базе остались 186 символов собранных фактов и указание, что следующим должен отработать супервизор. Возобновление обошлось в 13,29 секунды против 22,95 секунды полного прогона, а шаг сбора фактов, стоивший 4,65 секунды, повторно не выполнялся. На игрушечном примере экономия выглядит скромно, но замените локальную семимиллиардную модель на серию платных вызовов, и та же арифметика начнёт считаться в деньгах.

Одна грабля из прогона стоит отдельного упоминания. При режиме записи по умолчанию честного обрыва не получалось вовсе, потому что снимок уходил в базу асинхронно и убитый процесс не оставлял состояния. Демо заработало только после явного переключения на синхронную запись и в потоковом вызове, и в обычном.

Заключение

Оркестрация ИИ решает не задачу «сделать агента умнее», а задачу «сделать процесс с моделью управляемым и переживающим сбой». Три вещи стоит запомнить. Состояние живёт вне процесса приложения и привязано к идентификатору задачи, поэтому перезапуск и ожидание человека перестают быть проблемой. Гранулярность восстановления определяется размером шага, и по ней фреймворки заметно расходятся между собой. Решение о порядке шагов принимает управляющий слой, а не модель, что демо подтвердило буквально на первом же вызове.

Референсные ссылки

• LangGraph persistence, официальная документация по чекпоинтам и thread_id
• AI agent frameworks, сравнение фреймворков оркестрации на 2026 год
• Agent2Agent, история и версии протокола межагентного взаимодействия
• Temporal vs Airflow, реплей истории событий против повтора задачи
• Durable execution for agent runtimes, чекпоинтинг и восстановление агентных рантаймов