Multi-Agent Architecture: когда одного AI недостаточно

Один LLM-вызов решает одну задачу. Но продуктовые сценарии редко укладываются в один вызов. Анализ документа, генерация плана, проверка качества, форматирование результата — четыре разных задачи с разными требованиями к модели, контексту и промпту. Попытка уместить всё в один промпт приводит к деградации качества на каждом шаге.

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

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

Один агент vs несколько: где проходит граница

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

Пример: AI-ассистент для travel-приложения. Пользователь пишет “спланируй поездку в Японию на 10 дней”. Один агент должен:

1. Понять намерение и извлечь параметры (NLU)
2. Найти достопримечательности и оценить логистику (поиск + reasoning)
3. Построить маршрут с учётом расстояний и времени (оптимизация)
4. Сгенерировать описания для каждого дня (генерация текста)
5. Проверить результат на ошибки (валидация)

Каждый шаг требует разный промпт, разную температуру, иногда разную модель. NLU работает лучше с temperature: 0. Генерация текста — с temperature: 0.7. Оптимизация маршрута может требовать reasoning-модель вроде o3. Описания дня — дешёвую и быструю Gemini Flash.

Когда все шаги запихнуты в один промпт, модель получает противоречивые инструкции. Результат: длинный промпт на 3000+ токенов, непредсказуемое поведение, невозможность отладить конкретный шаг.

Правило: если задача содержит больше двух шагов с разными требованиями к модели или промпту — пора разделять на агентов.

Три паттерна оркестрации мульти-агентных систем

Sequential Pipeline

Самый простой паттерн. Агенты выстроены в цепочку. Выход одного — вход следующего.

Input → [Agent A] → [Agent B] → [Agent C] → Output
         Extract      Process      Format

Реализация минимальна:

async def pipeline(user_input: str) -> str:
    # Шаг 1: Извлечение параметров
    params = await agent_extract.run(user_input)

    # Шаг 2: Генерация плана
    plan = await agent_plan.run(params)

    # Шаг 3: Валидация и форматирование
    result = await agent_format.run(plan)

    return result

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

Минусы: общее время = сумма всех шагов. Если Agent B упал, весь pipeline падает. Нет параллелизма.

Когда подходит: ETL-подобные задачи, обработка документов, любой workflow с фиксированным порядком шагов.

Parallel Fan-Out / Fan-In

Несколько агентов работают одновременно над разными аспектами задачи. Результаты собираются и объединяются.

              ┌─ [Agent A: Safety] ──┐
Input ────────┤  [Agent B: Quality] ──├───► Merge → Output
              └─ [Agent C: Style]  ──┘

async def parallel_review(code: str) -> dict:
    # Запуск трёх агентов параллельно
    safety, quality, style = await asyncio.gather(
        agent_safety.run(code),
        agent_quality.run(code),
        agent_style.run(code),
    )

    # Объединение результатов
    return merge_reviews(safety, quality, style)

Этот паттерн используется в Claude Concilium для параллельных консультаций с несколькими LLM. Три модели независимо анализируют один и тот же код, результаты сравниваются для поиска консенсуса.

Плюсы: время = время самого медленного агента (не сумма). Независимость: падение одного не блокирует остальных.

Минусы: нужна логика объединения результатов. Агенты не видят результаты друг друга.

Когда подходит: code review, мульти-аспектный анализ, A/B тестирование промптов, валидация с нескольких точек зрения.

Router: динамическая маршрутизация агентов

Агент-маршрутизатор анализирует запрос и направляет к специализированному агенту. Похоже на API gateway в микросервисах.

              ┌─ [Agent: Travel Planning]
Input → [Router] ─┤─ [Agent: Booking]
              └─ [Agent: General Chat]

ROUTES = {
    "travel_plan": agent_travel,
    "booking": agent_booking,
    "general": agent_chat,
}

async def router(user_input: str) -> str:
    # Классификация намерения (дешёвая модель)
    intent = await classifier.run(
        f"Classify intent: {user_input}",
        model="gemini-2.0-flash",
        temperature=0,
    )

    # Маршрутизация к специализированному агенту
    agent = ROUTES.get(intent, agent_chat)
    return await agent.run(user_input)

Маршрутизатор использует дешёвую модель для классификации. Специализированные агенты могут использовать дорогие модели с большим контекстом. Экономия: 80%+ запросов обрабатываются дешёвыми агентами.

Подробнее о маршрутизации моделей по задачам — в статье о multi-provider LLM архитектуре.

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

Специализация агентов: кто что делает

Типичное разделение для продуктового стартапа:

Агент	Задача	Модель	Temperature	Prompt size
Classifier	Определение intent / routing	Gemini Flash	0	~200 tokens
Extractor	Извлечение данных из текста	DeepSeek Chat	0	~500 tokens
Planner	Генерация планов, маршрутов	Claude Sonnet / o3	0.3	~1500 tokens
Writer	Генерация текста для пользователя	Gemini Flash	0.7	~800 tokens
Validator	Проверка JSON-схемы, фактов	GPT-4o	0	~400 tokens
Judge	Оценка качества output	Claude Sonnet	0.2	~600 tokens

Judge-агент заслуживает отдельного внимания. Он не генерирует контент, а оценивает результат других агентов по критериям: полнота, релевантность, безопасность. Подробнее об этом паттерне — в статье о LLM-as-Judge.

Каждый агент — это комбинация трёх вещей: системный промпт, выбор модели, набор инструментов (tools/functions). Пример конфигурации:

from dataclasses import dataclass
from typing import Optional

@dataclass
class AgentConfig:
    name: str
    system_prompt: str
    model: str
    temperature: float = 0.0
    max_tokens: int = 4096
    tools: list[str] | None = None
    timeout_seconds: int = 30

AGENTS = {
    "classifier": AgentConfig(
        name="classifier",
        system_prompt="You are an intent classifier. Return one of: travel_plan, booking, general.",
        model="google/gemini-2.0-flash",
        temperature=0,
        max_tokens=50,
        timeout_seconds=5,
    ),
    "planner": AgentConfig(
        name="planner",
        system_prompt="You are a travel planner. Create detailed day-by-day itineraries.",
        model="anthropic/claude-sonnet-4",
        temperature=0.3,
        max_tokens=8192,
        tools=["search_places", "get_distances", "check_opening_hours"],
        timeout_seconds=60,
    ),
    "validator": AgentConfig(
        name="validator",
        system_prompt="Validate the travel plan. Check: dates are consistent, distances are realistic, no duplicate places.",
        model="openai/gpt-4o",
        temperature=0,
        max_tokens=2048,
        timeout_seconds=15,
    ),
}

Протокол общения: как агенты передают данные

Агенты должны обмениваться данными в предсказуемом формате. Два подхода.

Подход 1: Structured Output (JSON)

Каждый агент возвращает JSON по заранее определённой схеме. Следующий агент получает JSON как часть промпта.

from pydantic import BaseModel

class TravelParams(BaseModel):
    destination: str
    duration_days: int
    interests: list[str]
    budget_level: str  # "budget" | "mid" | "luxury"

class DayPlan(BaseModel):
    day: int
    activities: list[str]
    estimated_cost_usd: float

class TravelPlan(BaseModel):
    params: TravelParams
    days: list[DayPlan]
    total_cost_usd: float

# Agent A → structured output
params: TravelParams = await agent_extract.run(
    user_input,
    response_format=TravelParams,
)

# Agent B принимает structured input
plan: TravelPlan = await agent_plan.run(
    f"Create a travel plan based on: {params.model_dump_json()}",
    response_format=TravelPlan,
)

Плюсы: типизация, валидация на каждом шаге, легко логировать и дебажить.

Минусы: overhead на structured output (не все модели поддерживают одинаково хорошо), ригидность схемы.

Подход 2: MCP (Model Context Protocol)

Агенты взаимодействуют через стандартизированный протокол. Каждый агент — MCP-сервер, который экспортирует tools. Оркестратор вызывает tools нужного агента.

Orchestrator (Claude Code)
    │
    ├── MCP: agent-planner
    │     └── tool: create_plan(destination, days, interests)
    │
    ├── MCP: agent-validator
    │     └── tool: validate_plan(plan_json)
    │
    └── MCP: agent-writer
          └── tool: write_description(day_plan)

MCP даёт стандартный интерфейс для подключения агентов. Агент-сервер можно написать на любом языке, запустить локально или удалённо. Подробнее о создании production MCP-серверов — в статье о кастомных MCP серверах.

На практике MCP лучше подходит для dev-time агентов (помощь разработчику в IDE), а structured output — для runtime-агентов (обработка пользовательских запросов в production).

Оркестратор: центральный компонент системы

Оркестратор решает: какого агента вызвать, в каком порядке, что делать с ошибками. Минимальная реализация:

import asyncio
import logging
from typing import Any

logger = logging.getLogger(__name__)

class Orchestrator:
    def __init__(self, agents: dict[str, AgentConfig]):
        self.agents = agents

    async def run_agent(self, name: str, input_data: str) -> Any:
        """Запуск одного агента с таймаутом и retry."""
        config = self.agents[name]

        for attempt in range(3):
            try:
                result = await asyncio.wait_for(
                    call_llm(
                        model=config.model,
                        system=config.system_prompt,
                        user=input_data,
                        temperature=config.temperature,
                        max_tokens=config.max_tokens,
                        tools=config.tools,
                    ),
                    timeout=config.timeout_seconds,
                )
                logger.info(f"Agent {name} completed", extra={
                    "agent": name,
                    "attempt": attempt + 1,
                    "model": config.model,
                })
                return result

            except asyncio.TimeoutError:
                logger.warning(f"Agent {name} timeout, attempt {attempt + 1}")
                continue
            except Exception as e:
                logger.error(f"Agent {name} failed: {e}")
                if attempt == 2:
                    raise

        raise RuntimeError(f"Agent {name} failed after 3 attempts")

    async def run_pipeline(self, user_input: str) -> str:
        """Sequential pipeline: classify → plan → validate → format."""

        # Шаг 1: Классификация
        intent = await self.run_agent("classifier", user_input)

        if intent == "general":
            return await self.run_agent("writer", user_input)

        # Шаг 2: Извлечение параметров
        params = await self.run_agent("extractor", user_input)

        # Шаг 3: Генерация плана
        plan = await self.run_agent("planner", params)

        # Шаг 4: Параллельная валидация
        validation, quality_score = await asyncio.gather(
            self.run_agent("validator", plan),
            self.run_agent("judge", plan),
        )

        # Шаг 5: Если валидация провалена — retry с feedback
        if not validation.get("is_valid"):
            plan = await self.run_agent("planner",
                f"Previous plan had issues: {validation['issues']}. "
                f"Original request: {params}. Fix the plan."
            )

        return plan

Ключевые решения в оркестраторе:

• Retry с контекстом. При повторной попытке агент получает информацию о том, что пошло не так. Не слепой retry, а retry с feedback от валидатора.
• Параллелизм где возможно. Validator и Judge работают одновременно — экономия 30-50% времени.
• Graceful degradation. Если Judge недоступен, pipeline продолжает без оценки качества. Лучше ответ без score, чем отсутствие ответа.

Observability: видеть, что происходит внутри

Мульти-агентная система без observability — чёрный ящик. Три обязательных компонента.

1. Трейсинг

Каждый запрос пользователя — это trace. Каждый вызов агента — span внутри trace. Langfuse отлично подходит для этого:

from langfuse import Langfuse

langfuse = Langfuse()

async def run_with_tracing(user_input: str) -> str:
    trace = langfuse.trace(
        name="travel-planning",
        input=user_input,
    )

    # Classifier span
    classifier_span = trace.span(name="classifier")
    intent = await run_agent("classifier", user_input)
    classifier_span.end(output=intent)

    # Planner span
    planner_span = trace.span(name="planner")
    plan = await run_agent("planner", user_input)
    planner_span.end(output=plan)

    # ... остальные агенты

    trace.update(output=plan)
    return plan

В Langfuse видно: какие агенты вызывались, сколько времени занял каждый, сколько токенов потратил, какой результат вернул. Подробнее о настройке — в статье об LLM Observability с Langfuse.

2. Метрики

Минимальный набор метрик для мульти-агентной системы:

Метрика	Что показывает	Alert threshold
agent.latency_p95	Скорость каждого агента	> 10s для classifier
agent.error_rate	Процент ошибок по агентам	> 5%
agent.token_cost	Стоимость по агентам	> $X/day budget
pipeline.success_rate	Доля успешных pipeline	< 95%
pipeline.retry_rate	Как часто нужен retry	> 20%
judge.quality_score	Средняя оценка качества	< 0.7

3. Логирование решений

Каждое решение оркестратора логируется: почему выбран этот агент, почему произошёл retry, почему pipeline пошёл по альтернативному пути. Без этого отладка превращается в гадание.

logger.info("routing_decision", extra={
    "trace_id": trace.id,
    "intent": intent,
    "selected_agent": "planner",
    "reason": "intent=travel_plan, confidence=0.95",
    "fallback_agent": "writer",
})

Стоимость мульти-агентной системы vs один большой промпт

Больше агентов = больше LLM-вызовов = больше расходов. Но грамотная архитектура выходит дешевле одного большого промпта.

Сравнение для задачи “спланируй поездку”:

Подход	Вызовов	Модели	Tokens (input)	Tokens (output)	Примерная стоимость
Один агент	1	Claude Sonnet	~3000	~2000	~$0.025
Pipeline (5 агентов)	5	Mix	~2500 total	~1800 total	~$0.008

Pipeline дешевле, потому что:

• Classifier использует Flash (~$0.0001 за вызов)
• Extractor использует DeepSeek ($0.00014/1M input tokens)
• Только Planner использует дорогую модель, и его промпт короче (параметры уже извлечены)

Экономия 60-70% на LLM-затратах при правильном разделении задач и маршрутизации моделей.

Ошибки при проектировании мульти-агентных систем

Over-engineering. Три агента для задачи, которую решает один промпт с structured output. Если задача линейна и не требует разных моделей — один агент проще и надёжнее.

Отсутствие fallback. Агент-планировщик недоступен — весь pipeline мёртв. Каждый критический агент должен иметь fallback: альтернативная модель, упрощённый промпт, кэш предыдущих результатов.

Слишком длинные цепочки. Каждый агент добавляет латентность и точку отказа. Pipeline из 8 агентов при P95 латентности 3 секунды на агента даёт 24 секунды на ответ. Пользователь не будет ждать.

Игнорирование контекстного окна. Передача полного output предыдущего агента в промпт следующего. Output планировщика на 5000 токенов + системный промпт валидатора на 500 токенов — лишние расходы. Передавать только то, что нужно текущему агенту.

Нет circuit breaker. Один агент вернул некорректный JSON. Следующий агент получил сломанный input, вернул ошибку. Оркестратор сделал retry. Три попытки, три ошибки, потрачены токены. Circuit breaker: если агент вернул невалидный output, не передавать его дальше, а сразу сообщить об ошибке.

С чего начать

1. Определить bottleneck. Взять текущий single-agent pipeline. Найти шаг, который чаще всего ломается или даёт плохое качество. Выделить его в отдельного агента.
2. Начать с двух агентов. Classifier + Worker. Classifier определяет тип задачи на дешёвой модели. Worker обрабатывает на подходящей. Это уже даёт маршрутизацию и экономию.
3. Добавить structured output. Pydantic-схемы для входов и выходов каждого агента. Валидация на каждом шаге. Без этого отладка мульти-агентной системы невозможна.
4. Подключить трейсинг. Langfuse, LangSmith или аналог. Видеть каждый вызов, каждый промпт, каждый результат. Без observability мульти-агентная система — чёрный ящик.
5. Добавить Judge-агента. Автоматическая оценка качества на выходе pipeline. Если score ниже порога — retry с feedback. Это замыкает цикл качества.
6. Оптимизировать по метрикам. Смотреть на latency, cost, quality score по каждому агенту. Менять модели, промпты, архитектуру на основе данных.