Учебник

Мульти-агентные системы: ускорить работу в 5 раз

Ваши менеджеры тратят часы на сбор данных из разных источников, а аналитику приходится ждать днями. Мульти-агентные системы - это когда несколько AI-программ работают параллельно, как команда специалистов. Разбираем три схемы, которые можно внедрить без программиста за вечер. Получите конкретный шаблон для своего бизнеса.

Макс Космов··9 мин чтения

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

Один AI-агент (программа на основе большой языковой модели, которая сама решает, что делать дальше) справляется с задачей в одном окне контекста. Как только задача требует параллельной работы, узкой специализации или роста под нагрузкой - одиночка не вытягивает. Нужна система агентов. Три базовых паттерна закрывают почти все реальные сценарии.

Зачем бизнесу несколько агентов вместо одного

Для владельца бизнеса главный вопрос - не «как красиво устроена архитектура», а «что я с этого получу». Многоагентная система даёт три выгоды, которые видны в деньгах: скорость, качество и устойчивость. Дальше разберём каждую.

Параллельная работа

Задачи, которые не зависят друг от друга, можно делать одновременно. Три агента-исследователя работают по своим темам и заканчивают за время одного. Не за тройное время по очереди. Для задачи «собрать аналитику по 20 компаниям» разница между 5 и 100 минутами.

Разберём на примере стройфирмы (это пример, не реальный кейс автора). У вас есть 10 подрядчиков, по каждому нужно собрать: цены на материалы, сроки, отзывы. Один менеджер делает это за 3 дня. Три AI-агента - за 3 часа. Выгода очевидна.

Специализация

Агент с узким промптом (текстом-инструкцией для модели) и точечным набором инструментов работает лучше универсального. Агент-исследователь заточен под поиск и сжатие текста. Агент-кодер - под код и тесты. Агент-ревьюер - под критику. Каждый делает одно дело хорошо.

Изоляция ошибок

Если один агент завис или вернул мусор - система продолжает работать. Координатор перезапустит сбойного исполнителя, заменит результат или позовёт человека. В монолитном агенте одна ошибка ломает всё.

Есть и побочный плюс: параллельные агенты обходят лимит окна контекста. Задача, которая не влезает в 200 тысяч токенов (токен - это часть слова или символ, которым модель оперирует), режется на куски. Каждый кусок идёт в отдельного агента.

Паттерн 1: Координатор-исполнитель (Orchestrator-Worker)

Это самый распространённый паттерн на проде (в боевой среде). Бизнес-смысл: один начальник распределяет задачи между специалистами и собирает результат. Так работает любая нормальная команда. В коде то же самое.

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

Пользователь -> [Координатор] -> [Исполнитель A] [Исполнитель B] [Исполнитель C]
 ^________________^_______________^______________^
 результаты собираются обратно

Дальше код на LangGraph (фреймворк для построения агентов в виде графа). Он описывает: разбить задачу на 2 подзадачи, отдать их двум исполнителям параллельно, собрать ответы в финальный текст. Вам этот код не понадобится - просто поймите логику.

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

class PipelineState(TypedDict):
 task: str
 subtasks: list[str]
 results: Annotated[list[dict], operator.add]
 final_output: str

def orchestrator(state: PipelineState):
 subtasks = llm.invoke(
 f"Декомпозируй задачу на 3 независимые подзадачи:\n{state['task']}"
 )
 return {"subtasks": parse_subtasks(subtasks)}

def worker_a(state: PipelineState):
 result = research_agent.invoke(state["subtasks"][0])
 return {"results": [{"worker": "A", "output": result}]}

def worker_b(state: PipelineState):
 result = analysis_agent.invoke(state["subtasks"][1])
 return {"results": [{"worker": "B", "output": result}]}

def aggregator(state: PipelineState):
 combined = "\n".join([r["output"] for r in state["results"]])
 final = llm.invoke(f"Обобщи результаты:\n{combined}")
 return {"final_output": final}

graph = StateGraph(PipelineState)
graph.add_node("orchestrator", orchestrator)
graph.add_node("worker_a", worker_a)
graph.add_node("worker_b", worker_b)
graph.add_node("aggregator", aggregator)

graph.set_entry_point("orchestrator")
graph.add_conditional_edges("orchestrator", lambda s: ["worker_a", "worker_b"])
graph.add_edge("worker_a", "aggregator")
graph.add_edge("worker_b", "aggregator")
graph.add_edge("aggregator", END)

Когда применять. Задача поддаётся разбивке. Исполнители независимы друг от друга. Нужна агрегация. Большинство реальных сценариев.

Слабое место. Координатор - единая точка отказа. Если он завис или плохо разбил задачу - страдает вся система.

Паттерн 2: Равные между собой (Peer-to-Peer)

Зачем бизнесу этот паттерн. Иногда задача не сводится к чёткому плану. Нужен мозговой штурм, где эксперты обсуждают между собой, спорят, уточняют друг у друга. Code review без жёсткого порядка - типичный пример.

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

[Агент A] <-> [Агент B]
 | |
[Агент C] <-> [Агент D]

AutoGen v0.4 (фреймворк от Microsoft для многоагентных систем) - удобная среда для peer-to-peer через групповые чаты. Дальше код описывает команду из трёх агентов: архитектор, разработчик, эксперт по безопасности. Они общаются между собой, пока кто-то не скажет «APPROVED».

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import SelectorGroupChat
from autogen_agentchat.conditions import TextMentionTermination

architect = AssistantAgent(
 name="architect",
 model_client=model_client,
 system_message="""Ты архитектор. Проектируешь решения, но можешь
 обратиться к developer или security_expert за уточнениями."""
)

developer = AssistantAgent(
 name="developer",
 model_client=model_client,
 system_message="""Ты разработчик. Реализуешь то, что спроектировал architect.
 При вопросах безопасности обращайся к security_expert."""
)

security_expert = AssistantAgent(
 name="security_expert",
 model_client=model_client,
 system_message="""Ты эксперт по безопасности. Рецензируешь решения
 architect и developer, указываешь на уязвимости."""
)

team = SelectorGroupChat(
 [architect, developer, security_expert],
 model_client=model_client,
 termination_condition=TextMentionTermination("APPROVED")
)

Когда применять. Заранее непонятен порядок работы. Нужно творческое взаимодействие или экспертный спор. Типичный сценарий - команда ревьюеров кода без жёсткого протокола.

Главный риск. Без чёткого условия завершения агенты могут обсуждать бесконечно. Всегда ставьте max_turns (предельное число шагов) и TextMentionTermination (стоп-слово).

Паттерн 3: Иерархия

Зачем бизнесу. Если задача большая и делится на несколько доменов - дизайн, код, тесты, документация - один координатор не справится. Нужна корпоративная структура: гендиректор, менеджеры, исполнители. Иерархия масштабирует паттерн координатор-исполнитель на несколько уровней.

Топ-уровень знает только об уровне ниже. Детали реализации спрятаны.

[CEO-агент]
 |
[Менеджер ресёрча] [Менеджер разработки] [Менеджер тестирования]
 | | |
[Ресёрчер 1,2] [Разработчик 1,2,3] [Тестер 1,2]

Дальше код собирает трёхуровневую структуру через механизм handoffs (передача задачи от одного агента другому в OpenAI Agents SDK).

researcher_1 = Agent(name="researcher_1", instructions="Исследуй академические источники")
researcher_2 = Agent(name="researcher_2", instructions="Исследуй индустриальные источники")

research_manager = Agent(
 name="research_manager",
 instructions="Координируй исследование. Делегируй researcher_1 и researcher_2.",
 handoffs=[researcher_1, researcher_2]
)

eng_manager = Agent(
 name="engineering_manager",
 instructions="Координируй разработку. Делегируй developer-агентам.",
 handoffs=[developer_1, developer_2]
)

ceo_agent = Agent(
 name="project_coordinator",
 instructions="""Координируй проект целиком. Используй research_manager
 для сбора данных, engineering_manager для реализации.""",
 handoffs=[research_manager, eng_manager]
)

Каждый менеджер заточен под свою область. Он не перегружен знанием о всей системе. CEO-агент работает на уровне стратегии.

Когда применять. Большие задачи с понятными зонами ответственности. Команды больше 5 агентов. Типичный сценарий - полный цикл разработки фичи: исследование, дизайн, код, тесты, документация.

Как агенты общаются между собой

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

Прямые вызовы функций

Простейший вариант. Агент А вызывает функцию, внутри которой запускается агент Б. Никаких протоколов, чистый Python.

@tool
def call_research_agent(topic: str) -> str:
 return research_agent.run(topic)

Подходит для систем в рамках одного приложения.

MCP (Model Context Protocol)

MCP (Model Context Protocol, протокол подключения внешних инструментов к моделям) - стандарт от Anthropic. Агент выступает MCP-клиентом и вызывает инструменты MCP-серверов по единому правилу. Удобно для интеграции с экосистемой готовых MCP-серверов.

A2A (Agent-to-Agent)

A2A (Agent-to-Agent protocol, протокол связи между агентами) - открытый стандарт Google и Microsoft. Агент публикует карточку с описанием своих возможностей. Другие агенты находят его и вызывают через REST или gRPC. Нужен для распределённых систем, где агенты живут на разных машинах или написаны на разных фреймворках.

{
 "name": "ResearchAgent",
 "description": "Агент для технических исследований",
 "version": "1.0",
 "capabilities": {
 "streaming": true,
 "pushNotifications": false
 },
 "skills": [
 {"id": "web-research", "name": "Web Research"},
 {"id": "summarization", "name": "Document Summarization"}
 ]
}

Выбор прост. Прямые вызовы - для монолита внутри одного приложения. MCP - для подключения внешних инструментов. A2A - для общения между сервисами и командами.

Надёжность: как не потерять деньги на сбоях

В проде агенты падают, зависают, дублируют действия. Если не подстелить соломки, один сбой может отправить клиенту три письма вместо одного, списать деньги дважды, удалить нужную запись. Дальше три практики, которые закрывают 80% боли.

Идемпотентность

Идемпотентность - умное слово про простое правило. Задача даёт один и тот же результат при повторном запуске. Это критично для повторных попыток. Если исполнитель упал и координатор перезапустил его, второй запуск не должен создать дубликат записи в базе или отправить второе письмо.

from functools import wraps
import hashlib

def idempotent_task(task_id_fn):
 def decorator(func):
 @wraps(func)
 def wrapper(*args, **kwargs):
 task_id = task_id_fn(*args, **kwargs)
 if task_completed(task_id):
 return get_cached_result(task_id)
 result = func(*args, **kwargs)
 save_result(task_id, result)
 return result
 return wrapper
 return decorator

@idempotent_task(lambda topic, **kw: hashlib.md5(topic.encode()).hexdigest())
def research_topic(topic: str) -> str:
 return research_agent.run(topic)

Стратегия повторов

Повторные попытки нужны для временных сбоев: лимит запросов в минуту, сетевая ошибка. Логические ошибки повторять бессмысленно. Дальше код ждёт всё дольше между попытками: 1 секунду, 2, 4. И сразу падает на логических ошибках.

import time

def run_with_retry(agent_fn, task, max_retries=3):
 for attempt in range(max_retries):
 try:
 return agent_fn(task)
 except RateLimitError:
 wait = 2 ** attempt
 time.sleep(wait)
 except ValueError as e:
 raise
 raise MaxRetriesExceeded(f"Не удалось выполнить за {max_retries} попыток")

Обнаружение зависших

Ставим таймаут на каждый вызов агента и пишем в лог, если он висит дольше нормы. Дальше код прерывает агента, если тот не уложился в отведённое время.

import asyncio

async def run_with_timeout(coro, timeout_seconds: int):
 try:
 return await asyncio.wait_for(coro, timeout=timeout_seconds)
 except asyncio.TimeoutError:
 log_stuck_agent(task_id=...)
 raise AgentTimeoutError(f"Агент завис: превышен лимит {timeout_seconds}с")

Сравнение фреймворков на одной задаче

Задача. «Проанализируй компанию X. Соберёт данные три агента: финансы, продукт, команда. Четвёртый напишет отчёт.»

LangGraph - граф с параллельными ветками. Около 80 строк кода. Полный контроль над состоянием. Явное ветвление и сборка. Лучшая отладка через LangSmith. Сложнее на старте.

CrewAI - три исполнителя плюс writer через Process.hierarchical. Около 50 строк. Декларативное описание ролей. Менеджер-LLM сам координирует. Меньше контроля над деталями. Проще на старте.

OpenAI Agents SDK - агент сортировки плюс три специалиста через handoffs. Около 60 строк. Механизм передачи задачи естественно ложится на последовательную специализацию. Встроенная защита от вредных запросов. Привязка к моделям OpenAI.

Выбор зависит от вашего контекста. Если уже используете LangChain - берите LangGraph. Если нужен быстрый прототип с ролями - CrewAI. Если прод на OpenAI с защитными ограничениями - Agents SDK.

Антипаттерны: на чём горят команды

Это не теория. Это реальные грабли. Если вы продакт и видите у себя один из этих симптомов - значит, проект уже теряет деньги или вот-вот начнёт.

Бог-координатор

Координатор, который знает всё и делает всё. Разбирает задачу, следит за прогрессом, собирает результаты, проверяет качество, пишет итог. Нарушает принцип единственной ответственности. Любое изменение в одной части ломает другие. Решение - разделить на координатора (только планирование) и отдельного агента-агрегатора.

Болтливые агенты

Агенты, которые шлют друг другу кучу мелких сообщений. Каждое сообщение - вызов LLM, задержка и деньги. Вместо 10 уточняющих вопросов - один полный запрос с контекстом. Правило простое. Если агент отправляет больше 3 сообщений подряд без реального прогресса - это болтливый паттерн.

Размытые границы

Агент получает все инструменты подряд и сам решает, что использовать. Результат непредсказуем. Каждый агент должен иметь минимальный набор инструментов под свою задачу - принцип минимальных привилегий.

Бесконечные циклы

Бесконечные циклы без явных условий завершения. Всегда: max_turns, max_iterations, timeout. Все три, а не одно из трёх.

Состояние в промпте

Передача всей истории взаимодействий в промпт каждого агента. При длинных сессиях контекст взрывается. Используйте структурированное состояние (TypedDict, Pydantic) и передавайте только нужные поля.

Частые вопросы

Когда peer-to-peer лучше координатор-исполнитель?

Peer-to-peer лучше, когда нет заранее известного плана и агенты должны самоорганизоваться. Примеры: мозговой штурм, экспертные обсуждения, итеративное улучшение без заданного порядка шагов. Координатор-исполнитель лучше, когда задача поддаётся разбивке и нужна предсказуемость.

Как предотвратить бесконечный цикл вызовов между агентами?

Три барьера одновременно. Лимит шагов в каждом агенте. Общий таймаут на весь прогон. Явное условие завершения в промпте (ключевое слово «ГОТОВО» или структурированный ответ с полем completed: bool). Один барьер ненадёжен - ставьте все три.

A2A-протокол обязателен или хватит простых HTTP-вызовов?

A2A необязателен для большинства проектов. Прямые HTTP-вызовы между агентами работают отлично, если они в одной инфраструктуре. A2A даёт стандартизацию обнаружения (через AgentCard) и совместимость между фреймворками. Нужен только при межсистемной интеграции или когда агенты пишутся разными командами.

Как тестировать многоагентную систему: подменять агентов или интеграционные тесты?

Оба уровня нужны. Юнит-тесты: заменяйте исполнителей на заглушки, тестируйте логику координатора отдельно. Интеграционные: запускайте на реальных LLM с дешёвыми моделями (gpt-4o-mini), ограниченным числом шагов. Записывайте эталонные прогоны и сравнивайте при регрессии.

Как оценить стоимость прогона до выкатки в прод?

Профилирование на 5-10 тестовых запросах с включённой трассировкой (LangSmith, Langfuse). Смотрите среднее число шагов, токены на шаг, доля обращений к дорогим и дешёвым моделям. Дальше умножайте на ожидаемое число запросов в сутки. Оптимизация - заменять дорогие модели на дешёвые там, где не критично качество.

Что дальше

Ваш следующий шаг: возьмите одну задачу из вашего бизнеса, которая сейчас отнимает больше всего времени у вас или менеджера. Например, сбор заявок из мессенджеров или анализ конкурентов. Разбейте её на 2-3 независимые части. Запустите прототип через готовый сервис вроде CrewAI (бесплатный план на 10 запросов хватит). Если нужна помощь с выбором инструмента - пишите в канал ниже.

AI Компас (t.me/kosmoslab_ai) - канал для предпринимателей в РФ и СНГ, которые применяют AI в своём бизнесе без программиста. Разбираем инструменты и схемы - без курсов и теории.