Архитектура мультиагентных систем

Введение

В предыдущих модулях мы изучали построение отдельных AI-агентов с использованием Claude и инструментов. Однако при разработке сложных систем одного агента часто недостаточно. Этот урок посвящён архитектурным паттернам для построения систем из нескольких взаимодействующих агентов.

Цель урока: понять, когда и как использовать мультиагентные системы, познакомиться с основными паттернами оркестрации и протоколами коммуникации.

Зачем нужны мультиагентные системы

Ограничения одиночного агента

Хотя современные модели, такие как Claude, обладают большой длиной контекста (200K токенов), одного агента часто недостаточно для решения сложных задач:

Проблема контекстного окна

• Обработка больших документов (сотни страниц), анализ множества файлов одновременно
• Хранение полной истории всех взаимодействий пользователя за неделю/месяц
• Каждый токен контекста — это затраты на обработку и задержка в ответе

Проблема специализации

• Один агент должен быть экспертом во всём (анализ кода, работа с БД, документы, API)
• Сложно обновлять логику: изменение в одной части может сломать другие
• Трудно проверить поведение конкретной специальности

Проблема масштабируемости

• Если один агент обрабатывает 10 задач в секунду, система потребляет ресурсы линейно
• Высокая стоимость при больших объёмах
• Задержки растут с увеличением сложности логики

Проблема надёжности

• Сбой одного агента = сбой всей системы
• Нельзя переделать только часть обработки при ошибке
• Сложно повторить только упавший этап

Когда имеет смысл мультиагентная архитектура

Используйте мультиагентные системы, когда:

1. Задача требует разных экспертиз

   • Анализ кода + генерация контрактов + юридическая проверка
   • Исследование + обработка данных + визуализация

2. Объём обработки велик

   • Параллельная обработка множества документов
   • Разделение нагрузки между несколькими агентами

3. Нужна модульность и переиспользуемость

   • Агент проверки кода может использоваться в разных системах
   • Каждый агент тестируется отдельно

4. Требуется избежать контекстного переполнения

   • Разные агенты — разные контексты
   • Каждый фокусируется на своей задаче

НЕ используйте мультиагентные системы, когда:

• Задача простая и одного LLM-вызова достаточно
• Система должна очень быстро отреагировать (мультиагент добавляет latency)
• Нет понятного разделения ответственности между агентами

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

Оркестрация — это организация взаимодействия между агентами. Существует четыре основных паттерна:

Orchestrator (Центральный координатор)

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

graph TD
    U[Запрос пользователя] --> O[Оркестратор]
    O --> W1[Воркер 1 -- анализ]
    O --> W2[Воркер 2 -- обработка]
    O --> W3[Воркер 3 -- проверка]
    W1 --> R[Сборка результатов]
    W2 --> R
    W3 --> R

    style O fill:#4f46e5,color:#fff,stroke:#4338ca
    style W1 fill:#f8fafc,stroke:#e2e8f0
    style W2 fill:#f8fafc,stroke:#e2e8f0
    style W3 fill:#f8fafc,stroke:#e2e8f0
    style R fill:#059669,color:#fff,stroke:#047857

Характеристики:

• Один главный агент направляет работу
• Простота управления и контроля
• Точка отказа: если координатор упадёт, упадёт всё

Когда использовать:

• Ясная иерархия задач
• Количество подагентов известно и фиксировано
• Нужен полный контроль над потоком

Pipeline (Конвейер)

Идея: результат одного агента передаётся следующему по цепочке.

graph LR
    I[Вход] --> A1[Агент 1 -- парсинг]
    A1 --> A2[Агент 2 -- анализ]
    A2 --> A3[Агент 3 -- форматирование]
    A3 --> O[Результат]

    style I fill:#f8fafc,stroke:#e2e8f0
    style A1 fill:#4f46e5,color:#fff,stroke:#4338ca
    style A2 fill:#4f46e5,color:#fff,stroke:#4338ca
    style A3 fill:#4f46e5,color:#fff,stroke:#4338ca
    style O fill:#059669,color:#fff,stroke:#047857

Характеристики:

• Линейный поток: A -> B -> C
• Каждый агент получает результат предыдущего
• Сложно изменять порядок на лету
• Хорошо масштабируется горизонтально

Когда использовать:

• Чёткий порядок обработки
• Выход одного шага — вход следующего
• Примеры: ETL, обработка документов в stages

Swarm (Рой)

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

graph TD
    SC[Общий контекст] --- A1[Агент 1]
    SC --- A2[Агент 2]
    SC --- A3[Агент 3]
    A1 <--> A2
    A2 <--> A3
    A1 <--> A3

    style SC fill:#4f46e5,color:#fff,stroke:#4338ca
    style A1 fill:#f8fafc,stroke:#e2e8f0
    style A2 fill:#f8fafc,stroke:#e2e8f0
    style A3 fill:#f8fafc,stroke:#e2e8f0

Характеристики:

• Агенты принимают решения независимо
• Все видят общий контекст
• Эмерджентное поведение
• Сложно предсказать и отладить

Когда использовать:

• Исследование и генерация идей
• Мозговой штурм
• Симуляции сложных систем

Supervisor + Workers (Супервизор и рабочие)

Идея: один супервизор наблюдает за прогрессом, принимает решения о следующих шагах, может перезапустить упавшие задачи.

┌─────────────────────────────┐
│   Supervisor                │
│   (мониторит, решает,       │
│    перезапускает)           │
└──────────┬────────┬─────────┘
           │        │
      ┌────▼───┐ ┌──▼──────┐
      │Worker 1│ │Worker 2 │
      │(retry) │ │(retry)  │
      └────────┘ └─────────┘

Характеристики:

• Надежность благодаря retry механизму
• Супервизор видит весь прогресс
• Автоматическое восстановление при ошибках
• Требует хорошего механизма логирования

Когда использовать:

• Задачи должны быть выполнены гарантированно
• Допускаются повторные попытки
• Примеры: batch processing, long-running workflows

Какой паттерн выбрать

flowchart TD
    START{Задача для мультиагента} --> Q1{Чёткий порядок шагов?}
    Q1 -- Да --> Q2{Нужен контроль над потоком?}
    Q1 -- Нет --> Q3{Агенты автономны?}
    Q2 -- Да --> ORC[Orchestrator]
    Q2 -- Нет --> PIP[Pipeline]
    Q3 -- Да --> SWR[Swarm]
    Q3 -- Нет --> Q4{Нужен retry при ошибках?}
    Q4 -- Да --> SUP[Supervisor + Workers]
    Q4 -- Нет --> ORC

    style START fill:#4f46e5,color:#fff,stroke:#4338ca
    style ORC fill:#059669,color:#fff,stroke:#047857
    style PIP fill:#059669,color:#fff,stroke:#047857
    style SWR fill:#059669,color:#fff,stroke:#047857
    style SUP fill:#059669,color:#fff,stroke:#047857

Протоколы коммуникации между агентами

Agent-to-Agent (A2A) от Google

A2A — это протокол для прямого общения между LLM-агентами, предложенный Google. Идея: агенты обмениваются структурированными сообщениями.

Основные компоненты:

• Agent Card: описание агента для discovery
• Task: задача с lifecycle (создание, выполнение, завершение)
• Artifacts: результаты выполнения задачи
• JSON-RPC 2.0 over HTTP/SSE: транспортный уровень протокола

Пример использования:

# Агент A запрашивает у агента B
request = {
    "action": "analyze_code",
    "input": code_snippet,
    "context": {
        "language": "python",
        "focus": "security"
    }
}

# Агент B обрабатывает и отвечает
response = {
    "result": "findings",
    "confidence": 0.95,
    "metadata": {...}
}

Преимущества:

• Структурированность
• Возможность добавления metadata
• Ясная ответственность

Недостатки:

• Требует договорённости о формате
• Синхронный обмен может быть медленным

Model Context Protocol (MCP) как канал инструментов

MCP (от Anthropic) — это стандартный протокол для подключения инструментов к LLM. Мультиагентные системы могут использовать MCP для того, чтобы агенты вызывали инструменты друг друга.

Архитектура:

┌────────────┐          ┌────────────┐
│  Agent A   │          │  Agent B   │
└──────┬─────┘          └──────┬─────┘
       │                       │
       └───────────────┬───────┘
                       │
                    (MCP)
                       │
                ┌──────┴──────┐
                │             │
            ┌───▼───┐    ┌───▼───┐
            │Tool 1 │    │Tool 2 │
            └───────┘    └───────┘

Пример:

# Agent A использует инструмент от Agent B через MCP
agent_a.call_tool(
    tool_name="agent_b_analyze",
    arguments={"data": input_data}
)

Преимущества:

• Стандартизация
• Безопасность
• Расширяемость

Простые паттерны: Файловая система и Очереди

Файловая система

Агенты обмениваются данными через файлы в общей директории.

# Agent A пишет результат
with open("/shared/task_1_output.json", "w") as f:
    json.dump(results, f)

# Agent B читает
with open("/shared/task_1_output.json", "r") as f:
    data = json.load(f)

Плюсы: простота, персистентность. Минусы: медленно, проблемы с конкурентностью.

Очереди сообщений

Агенты отправляют сообщения в очередь (Redis, RabbitMQ, Kafka).

import redis

queue = redis.Redis()

# Agent A отправляет
queue.rpush("task_queue", json.dumps({"task": "analyze", "data": ...}))

# Agent B получает
message = queue.lpop("task_queue")
task = json.loads(message)

Плюсы: надежность, асинхронность, масштабируемость. Минусы: требует инфраструктуры, дополнительная сложность.

Примеры реальных систем

Claude Code с подагентами (Task tool)

Claude Code — инструмент Anthropic для автоматизации разработки. Внутри используется:

• Главный агент: понимает запрос, планирует
• Task tool: создаёт подзадачи для специализированных подагентов
• Подагенты: анализируют код, пишут тесты, генерируют документацию

Архитектура:

• Orchestrator pattern
• MCP для подключения специализированных инструментов
• Контроль главного агента над всем процессом

Cursor Background Agents

Cursor IDE использует фоновых агентов для:

• Анализа кода во время редактирования
• Подсказок на основе контекста
• Поиска связанных файлов

Архитектура:

• Multiple workers, каждый анализирует свой аспект
• Shared context (открытые файлы, выделения)
• Swarm-like поведение

AutoGen от Microsoft

AutoGen (0.4+) — фреймворк для создания мультиагентных систем на Python. В версии 0.4 API полностью переработан: агенты объединяются в команды, общение управляется через условия завершения, а конфигурация моделей вынесена в отдельный клиент.

Ключевые концепции:

• AssistantAgent: агент с подключённой LLM (из autogen_agentchat.agents)
• RoundRobinGroupChat / SelectorGroupChat: команды агентов с разными стратегиями общения
• OpenAIChatCompletionClient: клиент модели (из autogen_ext.models.openai)
• TextMentionTermination: условие завершения диалога по ключевому слову

Пример:

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination
from autogen_ext.models.openai import OpenAIChatCompletionClient

model_client = OpenAIChatCompletionClient(model="gpt-5.5")

planner = AssistantAgent(
    "planner",
    model_client=model_client,
    system_message="Ты планировщик проекта. Декомпозируй задачу на шаги.",
)

developer = AssistantAgent(
    "developer",
    model_client=model_client,
    system_message="Ты разработчик. Реализуй код по плану.",
)

termination = TextMentionTermination("ЗАВЕРШЕНО")
team = RoundRobinGroupChat([planner, developer], termination_condition=termination)

# Запуск
import asyncio
result = asyncio.run(team.run(task="Создай REST API для управления задачами"))

Особенности:

• Командная работа агентов через RoundRobinGroupChat и SelectorGroupChat
• Асинхронный API (await team.run(task=...))
• Гибкая конфигурация моделей и условий завершения

CrewAI

CrewAI — высокоуровневый фреймворк для мультиагентных систем, фокусируется на “командной” работе.

Структура:

from crewai import Agent, Task, Crew

# Определяем агентов
researcher = Agent(
    role="Researcher",
    goal="Find interesting facts",
    llm=claude_model
)

writer = Agent(
    role="Writer",
    goal="Write engaging article",
    llm=claude_model
)

# Определяем задачи
research_task = Task(
    description="Research AI trends",
    agent=researcher
)

write_task = Task(
    description="Write article",
    agent=writer
)

# Создаём crew (команду)
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task]
)

result = crew.kickoff()

Преимущества:

• Высокоуровневый синтаксис
• Роли и цели агентов ясно определены
• Pipeline или граф зависимостей между задачами

LangGraph

LangGraph от LangChain — фреймворк для построения графовых мультиагентных систем.

Концепция:

• Узлы = агенты или функции
• Ребра = переходы и зависимости
• Граф = явная логика оркестрации

Пример:

from langgraph.graph import StateGraph
from langchain_anthropic import ChatAnthropic

from typing import TypedDict

class AgentState(TypedDict):
    input: str
    result_1: str
    result_2: str

llm = ChatAnthropic(model="claude-sonnet-4-6")

def agent_1(state):
    # Обработка
    return {"result_1": "..."}

def agent_2(state):
    # Использует результат agent_1
    return {"result_2": "..."}

graph = StateGraph(AgentState)
graph.add_node("agent_1", agent_1)
graph.add_node("agent_2", agent_2)
graph.add_edge("agent_1", "agent_2")
graph.add_edge("agent_2", "__end__")

compiled = graph.compile()
result = compiled.invoke({"input": "..."})

Преимущества:

• Полный контроль над графом
• Видимость всей логики
• Легко добавлять условные переходы и циклы

Практический пример: Basic Orchestrator

Реализуем простой паттерн Orchestrator на Python:

import json
from typing import Any, Dict, List
from anthropic import Anthropic

client = Anthropic()

class WorkerAgent:
    """Рабочий агент, специализирующийся на одной задаче"""

    def __init__(self, name: str, role: str):
        self.name = name
        self.role = role
        self.conversation_history = []

    def process(self, task: str, context: str = "") -> str:
        """Обработать задачу"""
        system_prompt = f"""You are {self.role}.
        Your task is to: {self.role}
        Respond in Russian.
        Be concise and focus on your specialty."""

        messages = self.conversation_history.copy()
        messages.append({
            "role": "user",
            "content": f"Контекст: {context}\n\nЗадача: {task}"
        })

        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=500,
            system=system_prompt,
            messages=messages
        )

        result = response.content[0].text
        self.conversation_history.append(
            {"role": "user", "content": messages[-1]["content"]}
        )
        self.conversation_history.append(
            {"role": "assistant", "content": result}
        )

        return result


class OrchestratorAgent:
    """Главный координатор, управляющий рабочими агентами"""

    def __init__(self):
        self.workers: Dict[str, WorkerAgent] = {}
        self.conversation_history = []

    def register_worker(self, name: str, role: str) -> None:
        """Зарегистрировать рабочего агента"""
        self.workers[name] = WorkerAgent(name, role)
        print(f"Зарегистрирован агент: {name} ({role})")

    def plan_and_execute(self, user_request: str) -> Dict[str, Any]:
        """
        1. Спланировать, какие агенты нужны
        2. Выполнить их по очереди
        3. Собрать результаты
        """

        # Шаг 1: Планирование
        plan_prompt = f"""You are a project coordinator.
        Given the user request, create a JSON plan of which workers
        to call and in what order.

        Available workers: {list(self.workers.keys())}

        User request: {user_request}

        Respond ONLY with valid JSON in this format:
        {{
            "plan": [
                {{"worker": "worker_name", "task": "specific task"}},
                ...
            ],
            "reasoning": "brief explanation"
        }}

        Respond in Russian."""

        plan_response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=800,
            messages=[{"role": "user", "content": plan_prompt}]
        )

        try:
            plan_text = plan_response.content[0].text
            import re
            json_match = re.search(r'\{.*\}', plan_text, re.DOTALL)
            plan_data = json.loads(json_match.group())
        except Exception as e:
            print(f"Ошибка парсинга плана: {e}")
            return {"error": "Failed to parse plan"}

        print(f"\nПлан выполнения:\n{plan_data['reasoning']}\n")

        # Шаг 2: Выполнение
        results = {}
        previous_results = ""

        for step in plan_data["plan"]:
            worker_name = step["worker"]
            task = step["task"]

            if worker_name not in self.workers:
                print(f"Агент {worker_name} не найден")
                continue

            print(f"Выполняю: {task}")

            result = self.workers[worker_name].process(
                task=task,
                context=previous_results
            )
            results[worker_name] = result
            previous_results += f"\n{worker_name}: {result}"
            print(f"Результат получен\n")

        return {
            "plan": plan_data,
            "results": results
        }


# Пример использования
if __name__ == "__main__":
    orchestrator = OrchestratorAgent()

    orchestrator.register_worker(
        "analyzer",
        "Code analyst who identifies issues and suggests improvements"
    )
    orchestrator.register_worker(
        "documentor",
        "Technical writer who creates clear documentation"
    )
    orchestrator.register_worker(
        "tester",
        "QA specialist who designs test cases"
    )

    user_request = (
        "I have a Python function for calculating fibonacci. "
        "Please analyze it, create documentation, "
        "and suggest test cases."
    )

    print("=" * 60)
    print(f"Пользовательский запрос:\n{user_request}\n")
    print("=" * 60)

    result = orchestrator.plan_and_execute(user_request)

    print("\n" + "=" * 60)
    print("ИТОГОВЫЕ РЕЗУЛЬТАТЫ:")
    print("=" * 60)
    for agent_name, output in result["results"].items():
        print(f"\n{agent_name}:\n{output}\n")

Что происходит:

1. Регистрация: создаём OrchestratorAgent и регистрируем специалистов
2. Планирование: оркестратор просит Claude спланировать последовательность вызовов
3. Выполнение: идём по плану, вызывая агентов и передавая контекст
4. Результаты: собираем всё в один output

Антипаттерны: когда НЕ использовать мультиагент

Проблема 1: Избыточная сложность

# ПЛОХО: 5 агентов для простой задачи
for simple_query in queries:
    task = create_task(simple_query)
    # Вызываем 5 агентов подряд...
    # Latency = 5x

# ХОРОШО: Один агент с инструментами
response = single_agent.answer(simple_query)

Мультиагент добавляет latency. Если задача простая — один агент с хорошими инструментами быстрее.

Проблема 2: Отладка становится адом

Agent A вызывает Agent B
Agent B вызывает инструмент Tool_1
Tool_1 возвращает ошибку
Agent B передаёт это Agent C
Agent C неправильно интерпретирует ошибку
Agent A получает странный результат

Где ошибка? Неясно.

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

Проблема 3: Денормализация контекста

# ПЛОХО: каждый агент получает свою копию контекста
context_copy_1 = context.copy()  # 100KB
context_copy_2 = context.copy()  # 100KB
context_copy_3 = context.copy()  # 100KB
# Итого: 300KB вместо 100KB

# ХОРОШО: общий контекст (Swarm) или передача по ссылке
shared_context = context  # Все читают один источник

Проблема 4: Галлюцинации множатся

Agent A "генерирует" неправильный ID (галлюцинация)
    |
Agent B использует этот ID
    |
Agent C получает ошибку и генерирует свою галлюцинацию
    |
Результат полностью неправильный

Решение: валидировать outputs между агентами.

Сравнительная таблица фреймворков

Диаграмма полной системы

┌─────────────────────────────────────────────────────────────┐
│                    Пользовательский запрос                  │
└──────────────────────┬──────────────────────────────────────┘
                       │
                       ▼
         ┌─────────────────────────────┐
         │   Orchestrator Agent        │
         │  (Claude Opus 4.7)          │
         │                             │
         │ - Парсинг запроса           │
         │ - Планирование workflow     │
         │ - Координация агентов       │
         └─┬────────────┬──────────┬──┘
           │            │          │
      ┌────▼──┐    ┌───▼───┐  ┌──▼───┐
      │Worker1│    │Worker2│  │Worker3│
      │CODE   │    │DATA   │  │DOCS   │
      │ANALYST│    │ENGINEER│  │WRITER │
      └────┬──┘    └───┬───┘  └──┬───┘
           │           │          │
           └───────────┼──────────┘
                       │
              ┌────────▼─────────┐
              │  Результаты      │
              │  - Анализ кода   │
              │  - Обработка дан-│
              │  - Документация  │
              └──────────────────┘

Best Practices

1. Ясные контракты между агентами

   • Определите точно, какой input/output ожидается
   • Используйте JSON schema для валидации

2. Логирование на каждом уровне

   • Какой агент обрабатывает
   • Что он получил на вход
   • Что вернул на выходе
   • Сколько токенов потратил

3. Graceful degradation

   • Если один агент упал, система должна продолжить
   • Используйте retry, fallback, partial results

4. Тестирование в изоляции

   • Тестируйте каждого агента отдельно
   • Mock-тестируйте communication между агентами

5. Мониторинг в production

   • Следите за latency каждого агента
   • Alert на ошибки и галлюцинации
   • Анализируйте стоимость (токены, API calls)

Дальнейшее изучение

• Google A2A Paper: “Communicating with Large Language Models as Agents”
• LangGraph Documentation: https://docs.langchain.com/oss/python/langgraph/overview
• CrewAI Framework: https://docs.crewai.com
• AutoGen: https://microsoft.github.io/autogen/
• Model Context Protocol: https://modelcontextprotocol.io/

Резюме

Мультиагентные системы — это мощный инструмент для:

• Разделения сложных задач на специализированные подзадачи
• Масштабирования обработки
• Модульности и переиспользования компонентов

Главные паттерны:

1. Orchestrator — когда нужен центральный контроль
2. Pipeline — для чёткого порядка обработки
3. Swarm — для автономных параллельных работ
4. Supervisor+Workers — для надёжности и recovery

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