# Мультиагентная система: архитектура и паттерны оркестрации AI-агентов

> Как построить мультиагентную систему: паттерны оркестрации (Sequential, Parallel, Classifier+Router), маршрутизация задач, специализация агентов, примеры кода.
> Author: Roman Belov · Published: 2026-03-25 · Source: https://futurecraft.pro/ru/blog/multi-agent-architecture/

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

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

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

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

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

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

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

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

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

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

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

### Sequential Pipeline

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

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

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

```python
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]  ──┘
```

```python
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](/ru/blog/claude-concilium-multi-agent-code-review/) для параллельных консультаций с несколькими LLM. Три модели независимо анализируют один и тот же код, результаты сравниваются для поиска консенсуса.

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

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

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

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

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

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

```python
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-3.5-flash",
        temperature=0,
    )

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

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

Подробнее о маршрутизации моделей по задачам — в статье о [multi-provider LLM архитектуре](/ru/blog/multi-provider-llm-architecture/).

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

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

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

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

Judge-агент заслуживает отдельного внимания. Он не генерирует контент, а оценивает результат других агентов по критериям: полнота, релевантность, безопасность. Подробнее об этом паттерне — в статье о [LLM-as-Judge](/ru/blog/llm-as-judge-automated-quality-gate/).

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

```python
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-3.5-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-6",
        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-5.4",
        temperature=0,
        max_tokens=2048,
        timeout_seconds=15,
    ),
}
```

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

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

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

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

```python
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 серверах](/ru/blog/mcp-production-custom-servers/).

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

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

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

```python
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 отлично подходит для этого:

```python
from langfuse import get_client

langfuse = get_client()

async def run_with_tracing(user_input: str) -> str:
    # Корневой span = trace всего запроса
    with langfuse.start_as_current_observation(
        as_type="span", name="travel-planning", input=user_input
    ) as trace_span:
        # Classifier span
        with langfuse.start_as_current_observation(
            as_type="span", name="classifier"
        ) as span:
            intent = await run_agent("classifier", user_input)
            span.update(output=intent)

        # Planner span
        with langfuse.start_as_current_observation(
            as_type="span", name="planner"
        ) as span:
            plan = await run_agent("planner", user_input)
            span.update(output=plan)

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

        trace_span.update(output=plan)
        return plan
```

В Langfuse видно: какие агенты вызывались, сколько времени занял каждый, сколько токенов потратил, какой результат вернул. Подробнее о настройке — в статье об [LLM Observability с Langfuse](/ru/blog/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 пошёл по альтернативному пути. Без этого отладка превращается в гадание.

```python
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.27/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 по каждому агенту. Менять модели, промпты, архитектуру на основе данных.

## FAQ

### Какой реальный latency-оверхед добавляет слой оркестрации поверх существующих LLM-вызовов?

Сама логика оркестрации добавляет единицы миллисекунд — Python async scheduler работает незаметно. Реальные расходы — это сетевые round-trip: каждая граница агента — отдельный HTTP-вызов со своим P95 tail latency. В последовательном pipeline из 5 агентов при 800 мс P95 на агента накапливается ~4 секунды — не от оркестрации, а от последовательных LLM-вызовов. Перевод независимых шагов в параллельное выполнение (`asyncio.gather`) — самая высокоэффективная оптимизация: обычно снижает суммарный latency pipeline на 40–60%.

### Как управлять накоплением контекста в длинном pipeline, не выходя за бюджет токенов?

Каждый агент должен получать только то, что ему нужно, а не полный вывод всех предыдущих агентов. Planner, который сгенерировал 3 000 токенов, не должен передавать весь этот блок Validator — только структурированное резюме или конкретные поля. Используйте Pydantic-схемы для выборочной передачи полей на каждой границе агентов. На практике обрезка контекста снижает расход токенов на 30–50% за прогон pipeline по сравнению с наивной передачей сырого текста.

### Может ли мульти-агентная система быть экономически эффективной для небольших команд с низкими объёмами запросов?

Да, но архитектура важнее объёма. Связка Classifier + Worker, где дешёвая модель (Gemini Flash, ~$0.0001/вызов, приблизительно) занимается роутингом, а дорогая подключается только для сложной генерации, может стоить меньше за запрос, чем единственный Claude Sonnet на всё — даже при сотнях запросов в день. Точка безубыточности — когда 60%+ запросов может обработать дешёвый специализированный агент, что типично для большинства чат-ботов и задач обработки документов.

---

*Нужна помощь с проектированием мультиагентных систем? Я помогаю стартапам внедрять AI-решения и строить продукты — [belov.works](https://belov.works).*