4 мар. 2026 г.

Мультиагентные паттерны: оркестраторы, воркеры и конвейеры

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

Мультиагентные системы не являются по своей сути более сложными — они просто структурированы иначе. Ключ в том, чтобы выбрать правильный паттерн для задачи. Три паттерна охватывают большинство случаев использования: оркестратор-воркер, конвейер и параллельный fan-out.

Почему несколько агентов

Обоснование мультиагентных систем сводится к трём практическим причинам.

Специализация. Один агент с 50 инструментами в контексте путается. Специализированный агент с 5 сфокусированными инструментами работает лучше. Разделите по домену — исследование, написание, код, верификация — и каждый агент хорошо делает одно дело.

Параллелизм. Некоторые задачи разбиваются на независимые подзадачи. Анализ 20 документов последовательно — медленно; анализ их одновременно параллельными агентами — быстро.

Верификация. Когда один агент производит вывод, а второй агент независимо его критикует, это улавливает ошибки, которые самопроверка пропускает. У рецензента нет стимула защищать исходный ответ.

Паттерн 1: Оркестратор-воркер

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

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

import anthropic
import json

client = anthropic.Anthropic()

def run_worker(system_prompt: str, task: str) -> str:
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2048,
        system=system_prompt,
        messages=[{"role": "user", "content": task}]
    )
    return response.content[0].text

def orchestrator(user_request: str) -> str:
    # Step 1: plan the work
    plan_response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        system="""You are a planning agent. Given a user request, break it into
2-4 specific subtasks. Return a JSON array of task descriptions only.
Example: ["Research X", "Analyze Y", "Synthesize findings"]""",
        messages=[{"role": "user", "content": user_request}]
    )

    tasks = json.loads(plan_response.content[0].text)

    # Step 2: run each task with a specialized worker
    results = []
    for task in tasks:
        result = run_worker(
            system_prompt="You are a focused execution agent. Complete the assigned task thoroughly.",
            task=task
        )
        results.append({"task": task, "result": result})

    # Step 3: synthesize
    synthesis_prompt = f"""Original request: {user_request}

Worker results:
{json.dumps(results, indent=2)}

Synthesize these results into a cohesive final response."""

    final = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2048,
        messages=[{"role": "user", "content": synthesis_prompt}]
    )
    return final.content[0].text

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

Одна ловушка: оркестраторы могут галлюцинировать подзадачи, которые не имеют смысла. Ограничьте формат вывода (JSON-массив, нумерованный список) и проверяйте его перед запуском воркеров. Блок try/except вокруг разбора JSON с резервным шагом перепланирования обрабатывает это корректно.

Паттерн 2: Конвейер

Агенты образуют последовательную цепочку. Каждый агент преобразует входные данные и передаёт свой вывод следующему. Ни один агент не знает о других — они получают входные данные и производят вывод.

Это простейший паттерн для реализации и осмысления. Он хорошо работает для задач преобразования с чётко определёнными этапами.

def run_pipeline(input_text: str) -> str:
    stages = [
        {
            "name": "Researcher",
            "system": "Extract and organize all key facts from the input. "
                      "Format as a structured list with sources noted where available.",
        },
        {
            "name": "Writer",
            "system": "Transform the research notes into clear, readable prose. "
                      "Maintain all factual content. Target a technical audience.",
        },
        {
            "name": "Editor",
            "system": "Improve clarity and concision. Remove redundancy. "
                      "Do not change facts. Return only the improved text.",
        },
        {
            "name": "Fact Checker",
            "system": "Review for internal consistency. Flag any claims that "
                      "contradict each other or seem unsupported. "
                      "If no issues, return 'VERIFIED: ' followed by the original text.",
        },
    ]

    current = input_text
    for stage in stages:
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=2048,
            system=stage["system"],
            messages=[{"role": "user", "content": current}]
        )
        current = response.content[0].text
        print(f"[{stage['name']}] complete ({len(current)} chars)")

    return current

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

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

Паттерн 3: Параллельный fan-out

Разбейте большие входные данные на независимые фрагменты, обрабатывайте каждый одновременно с отдельными агентами, затем агрегируйте результаты.

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

import asyncio

async def analyze_document(doc: str, index: int) -> dict:
    """Analyze a single document asynchronously."""
    system = """Analyze this document and return a JSON object with:
- "sentiment": positive/negative/neutral
- "key_topics": list of 3-5 main topics
- "summary": 2-3 sentence summary
- "flags": list of any concerns (empty list if none)"""

    # asyncio.to_thread lets you call synchronous code in a thread pool
    result = await asyncio.to_thread(
        lambda: client.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=512,
            system=system,
            messages=[{"role": "user", "content": doc}]
        ).content[0].text
    )
    return {"index": index, **json.loads(result)}

async def parallel_analysis(documents: list[str]) -> dict:
    # Fan out: analyze all documents concurrently
    tasks = [analyze_document(doc, i) for i, doc in enumerate(documents)]
    analyses = await asyncio.gather(*tasks)

    # Aggregate with a dedicated synthesis agent
    synthesis_input = json.dumps({
        "document_count": len(documents),
        "analyses": analyses
    })

    aggregate_result = await asyncio.to_thread(
        lambda: client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            system="Synthesize document analyses into: overall sentiment distribution, "
                   "top themes across all documents, and notable patterns in flags. "
                   "Return as JSON.",
            messages=[{"role": "user", "content": synthesis_input}]
        ).content[0].text
    )

    return {
        "individual": analyses,
        "aggregate": json.loads(aggregate_result)
    }

Этап агрегации — это место, где большинство реализаций срезает углы. Не конкатенируйте результаты — передавайте их агенту, который понимает задачу агрегации. Конкатенация строк из 20 анализов бесполезна; синтезированное резюме — ценно.

Обратите внимание на выбор модели: используйте claude-haiku-4-5-20251001 для высокообъёмного анализа на уровне документов, где важна скорость и задача проста, и claude-sonnet-4-6 для синтеза, где важнее суждение, а не пропускная способность.

Выбор правильного паттерна

Ситуация	Паттерн
Структура задачи неизвестна до анализа	Оркестратор-воркер
Чётко определённые этапы преобразования	Конвейер
Большие входные данные, независимые фрагменты	Параллельный fan-out
Требуется независимая верификация	Оркестратор или конвейер с этапом проверки
Минимизация задержки для больших входных данных	Параллельный fan-out

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

Практические соображения

Стоимость. Мультиагентные системы умножают API-вызовы. Конвейер из 4 этапов может стоить 4× одного вызова плюс накладные расходы на синтез. Стратегически смешивайте модели: используйте Opus для оркестрации и планирования там, где важно суждение, Haiku — для высокообъёмных задач выполнения.

Распространение ошибок. Заранее определите, как каждый агент обрабатывает сбои. Варианты: распространить ошибку (остановиться), вернуть объект ошибки (пусть оркестратор решает) или повторить с изменённым промптом (корректное восстановление, добавляет задержку). Для большинства производственных систем возврат структурированных объектов ошибок и предоставление оркестратору решать — это правильное значение по умолчанию.

Трассировка. Мультиагентная система, в которой вы не можете видеть, что сделал каждый агент, — это кошмар отладки. Логируйте каждый вызов агента с: входными данными, выводом, моделью, задержкой и количеством токенов. Помечайте каждый вызов идентификатором трассировки, чтобы вы могли восстановить полный путь выполнения.

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

Что строить дальше

Паттерны здесь — это фундамент. То, что вы строите на них, зависит от вашей задачи:

Добавьте использование инструментов воркерам — позвольте специализированным агентам вызывать API, запрашивать базы данных или запускать код
Добавьте контрольные точки с участием человека там, где оркестратор делает паузу перед высокорисковыми действиями
Добавьте память, сохраняя выводы агентов в векторное хранилище, которое будущие агенты могут запрашивать
Добавьте оценку, направляя выводы через агента-судью перед возвратом их пользователю

Мультиагентные системы — это место, где сейчас происходит интересная инженерия ИИ. Паттерны просты; суждение — в правильном их применении к вашей конкретной задаче.