Много рук, один перо: семь правил оркестрации ИИ-агентов
Семь правил для оркестрации ИИ-агентов: когда простой рабочий процесс более эффективен, почему у оркестратора нет инструментов, возвращение типизированных субагентов и условия остановки в текстовом виде.
Источники: Many Hands, One Pen: Orchestrating Agents Without Losing the Plot
Представьте ВМС США в 1942 году, сталкивающиеся с проблемой координации, которая много говорит о современном программном обеспечении. Военные корабли были оснащены радарами, сонарами, средствами радиоперехвата и людьми, наблюдающими за окружающей обстановкой, и каждый отчет поступал на мостик одновременно, но в случайном порядке. Капитан, пытавшийся вести бой, тонул в огромном количестве непрообработанных сигналов.
Ответом ВМС стала комната без окон под названием Центр боевой информации. Каждый отчет с датчиков отмечался на единой общей карте в установленном формате. Операторы, читавшие показатели приборов, не отдавали никаких приказов; их задачей было поддерживать точность карты, в то время как один офицер принимал решение о следующем действии корабля. Информация поступала в большом объеме, решения сужались до одного варианта, а стандартный формат отчетов позволял сравнивать полученные данные между собой.
Системы с несколькими агентами постоянно заново исследуют эту среду. Этап, когда добавление ещё одного агента считалось решением всех проблем, уже прошёл, и 2026 год в основном был посвящён оценке цены этого решения. Единственный подход, который остаётся эффективным, — это тот, который выбрала военно-морская служба: сбор информации параллельно, отчётность в строгом формате, принятие решений в одном месте и чёткое определение критериев завершения работы.
Полезная концепция: большинство задач с несколькими агентами на самом деле не являются проблемами агентов. Это проблемы контекста и управления потоком действий, маскирующиеся под проблемы агентов. Задание обычно проваливается из-за того, что два агента принимают несовместимые решения, которые ни один из них не может увидеть, из-за того, что информация теряет свой смысл при пересечении границ, или потому, что никто не описал, как выглядит завершённая работа.
Далее приведено семь практик, каждая из которых основана на ошибках в уже работающих системах. Вместе они помогают определить, когда обычный рабочий процесс лучше, чем агент, когда оркестратору стоит оплачивать свои токены, как распределять работу в соответствии с границами контекста, почему оркестратору следует не вмешиваться в инструменты, как предоставить субагентам контракт возврата, который можно проверить, и как намеренно завершать выполнение задачи, а не по произволу модели.
Используйте обычные рабочие процессы по умолчанию; автономия должна быть завоевана
Что идет не так. Возьмем процесс возврата средств с четырьмя возможными исходами, каждый из которых уже регламентирован письменным правилом. Команда оборачивает его автономным циклом, используя шесть инструментов и строку целей. В результате тикет от вторника начинает обрабатываться по другому пути, чем тикет от понедельника, причем каждая итерация стоит в десять–пятьдесят раз дороже, чем при использовании старого кода. Точность остается неизменной, поскольку ни один этап процесса не является настолько неоднозначным, чтобы требовать применения интеллектуальных алгоритмов.
Решение. Сначала нарисуйте диаграмму потоков. Все, что можно спланировать заранее, должно находиться в рамках обычного контрольного потока, поэтому простой код сможет обработать каждую из возможных ветвей. Модель следует вызывать только на том этапе, когда входные данные действительно могут трактоваться по-разному. Агент должен получить право на управление ветвью лишь тогда, когда будет доказано, что эту ветвь невозможно описать словами.
REFUND_LIMIT = 50.00
def llm_step(prompt: str) -> str:
"""Called only where the written rules run out."""
return model.complete(prompt).strip().lower()
def handle_refund(ticket):
if ticket.days_since_purchase > 30:
return deny(ticket, reason="outside window")
if ticket.amount <= REFUND_LIMIT:
return approve(ticket) # a rule, not a judgement call
intent = llm_step(
f"Classify as fraud, defect, or remorse:\n{ticket.body}"
)
if intent == "fraud":
return escalate(ticket, queue="risk")
if intent == "defect":
return approve(ticket)
return route_to_human(ticket)
Проблема. Вы можете попасть в лабиринт вложенных условий, которые скрыто отражают правила оценки, которые модель на самом деле не может применить. Когда вы пишете пятое условие для обработки свободного текста, именно такая неоднозначность и является причиной существования данной модели.
Игнорируйте оркестратор, если сбой уже известен
Что идет не так. Четырехэтапный процесс формирования отчета оказывается внутри цикла оркестратора с ограничением в пятнадцать итераций. Каждая итерация тратит время на планирование следующих действий, причем каждый такой запрос приходит к уже известному ответу: загрузить, извлечь, подвести итоги, сформатировать. Это пятнадцать операций обработки по несколько тысяч токенов каждая для того, чтобы снова получить список, который можно было бы ввести за пять строк.
Решение. Используйте простое правило перед написанием кода. Если полный список подзадач известен до начала выполнения, обрабатывайте его с помощью простого цикла с фиксированным распределением задач, а токены сохраняйте для других целей. Оркестрация, основанная на моделях, подходит там, где сама структура задач неизвестна или должна развиваться на основе результатов предыдущих шагов.
# Before: up to 15 planning calls to rediscover a fixed list
def run_orchestrated(doc):
state = {"doc": doc}
for _ in range(15):
decision = orchestrator.plan(state) # 1 LLM call per pass
if decision.action == "finalize":
break
state = WORKERS[decision.worker](state)
return state
# After: 0 planning calls, same four workers, same result
PIPELINE = [fetch, extract, summarize, format_report]
def run_static(doc):
state = {"doc": doc}
for step in PIPELINE: # 0 LLM calls here
state = step(state)
return state
Недостаток. У этого совета есть обратная сторона. Пайплайн, в котором каждый месяц создается новый специальный случай, или где результат второго этапа действительно определяет, будет ли выполняться третий этап, представляет собой настоящее планирование. Удаление планировщика лишь перемещает логику разветвления в какой-то конфигурационный файл, за которым никто не будет присматривать.
Разделяйте агентов по контексту, а не по должностным наименованиям
Что идет не так. Версия многоагентной системы в виде организационной схемы выглядит аккуратно: планировщик, реализатор, тестер, рецензент — работа над одной функцией по порядку. На практике при каждой передаче задачи возникают предположения, которые никто не записывает. Тестер проверяет поведение, которое реализатор вовсе не имел в виду, а рецензент возражает против выбора, сделанного планировщиком намеренно. Вы платите в четыре раза больше из-за таких ошибок, и один агент, выполняющий все четыре функции, справился бы лучше.
Решение. Разделите задачи по линиям контекста. Прежде чем добавлять второго агента, задайте себе вопрос: может ли эта подзадача быть выполнена в окне контекста, не имеющем ничего общего с окном соседней подзадачи? Если ответ «нет», обе подзадачи должны относиться к одному агенту, и именно поэтому агент, разрабатывающий функцию, также должен писать тесты для неё.
from dataclasses import dataclass
@dataclass
class Subtask:
name: str
needs: set[str] # facts this subtask must read
produces: set[str] # facts this subtask decides
def should_split(a: Subtask, b: Subtask) -> bool:
"""Split only when neither side needs what the other decides."""
shared = (a.needs & b.produces) | (b.needs & a.produces)
return not shared
implement = Subtask("implement", {"spec"}, {"api_shape", "error_semantics"})
test = Subtask("test", {"spec", "api_shape", "error_semantics"}, {"cases"})
assert not should_split(implement, test) # one agent writes both
Проблема. Полностью изолированные подзадачи встречаются реже, чем может показаться схема организации, поэтому в тестах часто указывается «один агент», и это правильный ответ, а не отсутствие воображения. Разделение задач оправдано при работе с большим объемом информации, например когда пять агентов читают пять независимых источников, поскольку решение одного не влияет на потребности другого.
Пусть оркестратор выдает решение и никогда не вызывает инструменты
Что идет не так. Оркестратор получает инструмент поиска для чрезвычайных ситуаций. Вскоре он перестает распределять задачи и начинает самостоятельно отвечать на подзадачи. Чистый вывод инструментов накапливается в его окне, из-за чего планирование ухудшается по мере увеличения времени выполнения, когда оно становится особенно важным. Также нарушается возможность отслеживания, поскольку неясно, какой агент сделал какое открытие.
Решение. Любые планы не должны также приносить проблем, поэтому организатору не следует предоставлять никаких инструментов. Он выдает лишь одно компактное, типизированное решение: делегировать, перепланировать или завершить работу, а также указывает рабочий процесс для использования и текст задачи. Окружающий цикл считывает это решение, запускает указанный рабочий процесс и сохраняет результат в общем состоянии, готовом к следующему решению.
from typing import Literal
from pydantic import BaseModel
class OrchestratorDecision(BaseModel):
next_action: Literal["delegate", "replan", "finalize"]
target_worker: str | None = None
task_description: str | None = None
reasoning: str
def orchestrator_step(state):
d = decide(state) # model has zero tools attached
if d.next_action == "finalize":
return finalize(state, d.reasoning)
if d.next_action == "replan":
return state.reset_plan(d.reasoning)
worker = WORKERS[d.target_worker] # workers own every tool
result = worker.run(d.task_description)
return state.record(d.target_worker, result)
Проблема. Без специальных инструментов оркестратор может видеть только то, что возвращают рабочие агенты. Если агент предоставляет лишь краткое резюме, это мешает эффективному планированию. Необходимо предоставить возможность только читать общее состояние, а не использовать инструменты в реальном времени, при этом данные, возвращаемые агентами, должны быть достаточно структурированными для планирования.
Обязательство каждого подагента соблюдать установленный формат ответа
Что идет не так. Возьмем пример исследовательского подагента, который возвращает 2500-словой дневник всех своих попыток, включая неудачные варианты. Если таких агентов семеро, то еще до начала обработки в окно родительского агента попадает около 20 000 токенов. Делегирование задач должно было защитить родительского агента от избыточной нагрузки. Однако в итоге исследования выполняются дважды, а родительский агент еще хуже следует инструкциям именно в тот момент, когда это критично.
Решение. Рассматривайте возврат данных как границу API, при этом каждый подагент должен иметь схему ответа. Заключительное сообщение представляет собой JSON с именованными полями: выводы, источники, неразрешенные вопросы и статус завершения. Проверяйте его по мере получения и возвращайте всё некорректное или слишком длинное для повторной попытки, чтобы оно никогда не дошло до родительского элемента.
from pydantic import BaseModel, Field, ValidationError
class SubagentResult(BaseModel):
findings: list[str] = Field(max_length=5) # hard cap, one sentence each
sources: list[str]
open_questions: list[str] = []
completion_status: str # complete | partial | blocked
def parse_or_retry(worker, task, attempts=2):
for _ in range(attempts):
raw = worker.run(task, response_format=SubagentResult)
try:
return SubagentResult.model_validate_json(raw)
except ValidationError as err:
task = f"{task}\n\nRejected: {err}\nReturn only JSON in the schema."
raise RuntimeError(f"{worker.name} returned no valid result")
Проблема. Шаблоны устанавливают минимальный лимит, и ничего больше. Рабочий процесс всё равно может поместить абзац из 400 слов в одну строку с результатами и предоставить идентичный избыточный формат, маскируемый под JSON. Необходимо ограничить как количество элементов, так и длину каждого поля, а в тексте отклонения указать поле, которое превысило лимит, чтобы процесс повторной попытки знал, что нужно исправить.
Параллелизация чтения и сбор данных через одного агента
Что идёт не так. Два агента одновременно работают над одной функцией, и каждый сообщает о успехе. Один нарисовал фон в стиле Super Mario, а другой — спрайт Flappy Bird, причём оба проходят тесты, написанные для их собственных вариантов дизайна. Объединение данных не может разрешить выборы, о существовании которых не знали агенты, поэтому один или оба результата отклоняются.
Решение. Чтение можно выполнять параллельно, а письмо — нет. Ограничьте каждого работника списком инструментов только для чтения, что позволит расширить этап сбора данных по мере необходимости, и никогда не предоставляйте работнику инструментов для записи. Затем результат будет сгенерирован в одном вызове без дополнительных условий, который соберет результаты всех работников вместе.
import asyncio
READ_TOOLS = ["search_repo", "read_file", "fetch_docs"]
async def gather_context(subtasks):
workers = [Agent(name=t.name, tools=READ_TOOLS) for t in subtasks]
return await asyncio.gather(
*(w.run(t.prompt) for w, t in zip(workers, subtasks))
)
async def build_feature(spec, subtasks):
findings = await gather_context(subtasks) # wide, parallel, read-only
writer = Agent(name="writer", tools=["write_file", "apply_patch"])
return await writer.run(spec, context=findings) # one writer, one pass
Проблема. Этот единственный записчик является узким местом и должен сразу обрабатывать результаты всех работников, поэтому проблема с лимитом токенов возникает задолго до достижения логического предела. Если результат действительно разделяется четко, как в случае с трьмя файлами, не имеющими между собой связей, следует назначить каждому файлу отдельного записчика и соблюдать ту же дисциплину внутри каждого из них.
Запишите условия завершения записи
Что идет не так. Запуск завершается, ошибок не возникает, но работа остается незавершенной. Анализ 1 600 треков многокомпонентных систем в семи фреймворках показал, что 12,4% сбоев связаны с тем, что агенты не знают условий своего останова, еще 6,2% случаев связаны с преждевременным прекращением работы. Таким образом, почти один из пяти сбоев проявляется в четком завершении работы с незавершенным результатом.
Решение. Превратите конечную цель во что-то конкретное, а не в просто ощущение. Установите для каждой задачи четкий список пунктов для её выполнения, а также отметку о подтверждении достижения исходной цели на единственном пути к окончательному состоянию. Согласно тем же исследованиям, использование механизма проверки повысило эффективность ChatDev на 9,4%, а контроль за выполнением высокоуровневых целей — на 15,6%.
CHECKLIST = [
"every requested section exists",
"each claim cites a retrieved source",
"open questions are listed, or explicitly none",
]
def verify_complete(objective, checklist, output) -> bool:
for item in checklist:
verdict = judge(f"Objective: {objective}\nCheck: {item}\n\n{output}")
if not verdict.passed:
log.info("termination blocked by: %s", item)
return False
return judge(f"Does this satisfy the objective?\n{objective}\n\n{output}").passed
def finish(state):
if verify_complete(state.objective, CHECKLIST, state.draft):
return state.done()
return state.keep_working(reason="checklist not satisfied")
Проблема. Если позволить модели, отвечающей за обработку, самой оценивать свой чек-лист, она просто одобрит свою работу. По возможности предпочитайте механические критерии, такие как наличие файла, заполнение поля или корректное разрешение ссылки. Запор, который никогда не открывается, сам по себе является сбоем, поэтому ограничьте количество попыток и выдайте ошибку, когда они исчерпаны.
С чего начать
Если вы можете изменить лишь одну вещь, выберите самое простое решение: зафиксируйте письменно, что означает завершение работы. Чек-лист в сочетании с шагом проверки перед достижением конечного состояния можно создать за один день, он подойдет как для одного агента, так и для всей группы, и позволит избавиться от целого класса скрытых сбоев. Сделайте это заданием на эту неделю.
Далее идет процедура удаления. Изучите ваш оркестратор и проверьте, сводятся ли его параметры к фиксированному списку и функции агрегации. Если да, удалите его: траты на токены и задержки снижаются, а результаты остаются стабильными. Поскольку его легко отменить, это безопасный первый шаг.
Откладывайте использование более сложных решений до тех пор, пока это не станет проблемой. Типизированные контракты возврата оправдывают себя при использовании пяти и более работников, а список разрешенных операций с только чтением имеет значение, когда два агента действительно работают одновременно. Исключением является оркестратор без дополнительных инструментов: в начале он практически бесплатен, но добавление к нему дополнительных функций становится сложным и дорогостоящим.
Системы, которые останутся работоспособными в 2026 году, не будут содержать большое количество агентов. В них связи между агентами являются самой простой и явной частью архитектуры. Меньше участников, более строгие контракты, один инструмент для управления.
Связанные материалы
- Агенты параллельной обработки с git Worktrees, роль дирижера и Antigravity 2.0 — Как координирующий агент, git Worktrees для каждого агента и треки, основанные на задачах, позволяют разработать десяток функций SRE примерно за пять часов, а также уроки по вопросам HITL и правилам слияния кода.
- Агентный подход к RAG с Agno Teams: переписывание запросов, цикл критики и Qdrant — Координирующая команда переписывает запросы, использует цикл критики для получения информации и отвечает только на основе одобренного контекста, применяя гибридные алгоритмы поиска типа dense, BM25 и ColBERT в Qdrant.