Оркестрировка агентов: Best Practices для многоагентных систем

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

TL;DR — оркестрировка агентов за 90 секунд

Что такое оркестрировка агентов?

Оркестрировка агентов это координация нескольких ИИ-агентов с учетом:

• Распределение задач: кто за что отвечает? какая экспертиза нужна для каждой части?
• Коммуникация: как агенты обмениваются информацией? через состояние, сообщения или события?
• Управление: кто принимает решение о следующем шаге? центральный оркестратор или условные маршруты?
• Обработка ошибок: что происходит при сбое? повтор, fallback, вмешательство человека?
• Управление ресурсами: как оптимизировать LLM-вызовы? кеширование, выбор модели, лимиты токенов?
• Наблюдаемость: как отследить что происходит? логирование, трейсинг, метрики?

Single-Agent vs Multi-Agent — когда нужна оркестрировка?

Single-Agent достаточно для: линейных задач, одной специализации, без итераций, решаемых одним LLM-вызовом.

Multi-Agent необходим для: разных специализаций (поиск + написание + ревью), итераций (Writer ↔ Reviewer), параллелизма, задач выходящих за лимит контекста, разных инструментов, участия человека в процессе.

Паттерны оркестрировки — подробно

1. Hub-and-Spoke (централизованный оркестратор)

Центральный агент-оркестратор решает, какие агенты запускать и когда.

         [Orchestrator]
        /     |      \
[Agent A] [Agent B] [Agent C]

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

def orchestrator(state):
    task = state["task"]
    # LLM решает, какой агент нужен
    response = llm.invoke(f"Какой агент для: {task}? (research/code/write)")
    task_type = response.content.strip().lower()
    return {"task_type": task_type}

def route_to_agent(state):
    return state["task_type"]

# В LangGraph:
workflow.add_conditional_edges("orchestrator", route_to_agent, {
    "research": "research_agent",
    "code": "code_agent",
    "write": "writer_agent"
})

2. Pipeline (последовательный)

Агенты работают поочередно, каждый передает результат следующему. CrewAI по умолчанию использует Process.sequential.

[Agent A] → [Agent B] → [Agent C] → [Output]

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

Когда использовать? Для линейных workflow без ветвлений (поиск → написание → ревью).

3. DAG (Directed Acyclic Graph)

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

[Agent A] ──→ [Agent C] ──→ [Agent E] (Merge)
     └──→ [Agent B] ──→ [Agent D] ──┘

Преимущества: параллелизм — независимые задачи выполняются одновременно. Эффективность — общее время = самый длинный путь. Недостатки: сложно отлаживать, зависимости должны быть четко определены, возможны race conditions при общем состоянии.

Пример с LangGraph (параллельные исследователи):

class ResearchState(TypedDict):
    topic: str
    general_research: str
    technical_research: str
    combined_report: str

def general_researcher(state: ResearchState) -> dict:
    response = llm.invoke(f"Общая информация о: {state['topic']}")
    return {"general_research": response.content}

def technical_researcher(state: ResearchState) -> dict:
    response = llm.invoke(f"Технические детали о: {state['topic']}")
    return {"technical_research": response.content}

def merge_results(state: ResearchState) -> dict:
    combined = f"Общее:\n{state['general_research']}\n\nТехническое:\n{state['technical_research']}"
    return {"combined_report": combined}

# Graph с параллельными узлами
workflow = StateGraph(ResearchState)
workflow.add_node("general", general_researcher)
workflow.add_node("technical", technical_researcher)
workflow.add_node("merge", merge_results)

workflow.set_entry_point("general")
workflow.add_edge("general", "technical")  # Оба параллельно
workflow.add_edge("general", "merge")
workflow.add_edge("technical", "merge")
workflow.add_edge("merge", END)

Когда использовать? Когда независимые подзадачи можно выполнять параллельно (например 3 исследования, которые потом объединяются).

4. Event-Driven (реактивный)

Агенты реагируют на события вместо прямых вызовов. Event-bus распределяет события всем заинтересованным агентам.

[Event Bus]
  ├── [Agent A] (слушает "research.done")
  ├── [Agent B] (слушает "code.reviewed")
  └── [Agent C] (слушает "test.failed")

Преимущества: слабая связанность, масштабируемость, агенты легко добавлять динамически. Недостатки: сложно отследить логику, возможны race conditions, отладка затруднена.

from collections import defaultdict

class EventBus:
    def __init__(self):
        self.subscribers = defaultdict(list)
    
    def subscribe(self, event_type: str, handler):
        self.subscribers[event_type].append(handler)
    
    def publish(self, event_type: str, data: dict):
        results = []
        for handler in self.subscribers[event_type]:
            results.append(handler(data))
        return results

bus = EventBus()
bus.subscribe("research.done", writer_agent)
bus.subscribe("research.done", fact_checker_agent)
bus.subscribe("code.reviewed", refactor_agent)
bus.publish("research.done", {"topic": "ИИ 2026", "result": "..."})

Когда использовать? Для слабо связанных систем, где агенты добавляются динамически и порядок выполнения не зафиксирован.

Сравнение паттернов

Паттерн	Сложность	Параллелизм	Гибкость	Отладка	Применение
Pipeline	Низкая	❌	Низкая	Простая	Линейные workflow
Hub-and-Spoke	Средняя	❌	Средняя	Средняя	Динамическая маршрутизация
DAG	Высокая	✅	Высокая	Сложная	Параллельные задачи
Event-Driven	Высокая	✅	Очень высокая	Очень сложная	Слабо связанные системы

Best Practices — Подробный разбор

1. Определи роли четко (Single Responsibility Principle)

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

Плохо:

agent = Agent(
    role="Researcher, Writer and Tester",
    goal="Research, write and test everything"
)

Хорошо:

researcher = Agent(role="Researcher", goal="Find information")
writer = Agent(role="Writer", goal="Write content based on research")
tester = Agent(role="Tester", goal="Test functionality and report bugs")

2. Состояния делай явными

Используй типизированные States вместо неявной коммуникации. Это делает workflow понятным и устойчивым к ошибкам.

from typing import TypedDict, Optional, List, Annotated
from operator import add

class WorkflowState(TypedDict):
    input: str
    task_type: str
    research_result: Optional[str]
    draft: Optional[str]
    review_feedback: Optional[str]
    final_output: Optional[str]
    error: Optional[str]
    retry_count: int
    messages: Annotated[List[str], add]  # Будет добавляться, не перезаписываться

3. Обработка ошибок и retry-логика

LLM ненадежны, у них случаются rate-limits, таймауты, галлюцинации. Каждый агент нужен с обработкой ошибок:

import logging

logger = logging.getLogger("agent_orchestrator")

def agent_with_retry(agent_func, state, max_retries=3, fallback=None):
    for attempt in range(max_retries):
        try:
            logger.info(f"Agent {agent_func.__name__} - Попытка {attempt + 1}")
            result = agent_func(state)
            if validate_result(result):
                return result
            else:
                logger.warning(f"Agent {agent_func.__name__} - невалидный результат")
                state["error"] = "Invalid result"
        except Exception as e:
            logger.error(f"Agent {agent_func.__name__} - ошибка: {str(e)}")
            state["error"] = str(e)
            state["retry_count"] = attempt + 1
    
    if fallback:
        return fallback(state)
    return {**state, "error": f"Agent failed after {max_retries} retries"}

def validate_result(result):
    if not result:
        return False
    if "error" in result and result["error"]:
        return False
    return True

4. Контроль стоимости

Запросы к LLM стоят денег. При 5 агентах с 3 итерациями каждый, получается 15+ вызовов LLM. На GPT-4o это может быть 5-20$ за запуск.

class CostTracker:
    def __init__(self, max_budget=10.0):
        self.costs = {}
        self.max_budget = max_budget
    
    def track(self, agent_name, tokens, model="gpt-4o"):
        cost_per_1k = {
            "gpt-4o": 0.005,
            "gpt-4o-mini": 0.0003,
            "claude-3.5-sonnet": 0.003,
        }
        rate = cost_per_1k.get(model, 0.005)
        cost = (tokens / 1000) * rate
        self.costs[agent_name] = self.costs.get(agent_name, 0) + cost
        
        if self.total() > self.max_budget:
            raise Exception(f"Budget exceeded: ${self.total():.2f}")
    
    def total(self):
        return sum(self.costs.values())
    
    def report(self):
        lines = [f"Cost Report (Total: ${self.total():.4f})"]
        for agent, cost in sorted(self.costs.items()):
            lines.append(f"  {agent}: ${cost:.4f}")
        return "\n".join(lines)

Стратегии оптимизации стоимости:

• Кеширование: не выполняй одинаковые промпты дважды
• Выбор модели: простые задачи на GPT-4o-mini ($0.30/M), сложные на GPT-4o ($5.00/M)
• Ранний выход: останавливайся когда результат “достаточно хорош”
• Лимиты токенов: устанавливай max_tokens для каждого агента
• Локальные модели: используй Ollama для разработки ($0)

5. Observability и логирование

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

import logging
import time
from functools import wraps

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("agent_orchestrator")

def observed_agent(agent_func):
    @wraps(agent_func)
    def wrapper(state):
        name = agent_func.__name__
        start = time.time()
        logger.info(f"[{name}] START - Input keys: {list(state.keys())}")
        try:
            result = agent_func(state)
            duration = time.time() - start
            logger.info(f"[{name}] DONE - {duration:.2f}s")
            return result
        except Exception as e:
            duration = time.time() - start
            logger.error(f"[{name}] ERROR - {duration:.2f}s - {str(e)}")
            raise
    return wrapper

Дополнительно: LangSmith для детального трейсинга - показывает каждый вызов LLM, переход состояния и вычисление токенов визуально:

import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls-..."

6. Человек в цикле

Для критических решений человек должен подтвердить. LangGraph это поддерживает нативно с помощью interrupt_before:

app = workflow.compile(
    checkpointer=memory,
    interrupt_before=["publisher"]  # Паузирует перед publisher
)

# Первое выполнение: идет до publisher
result = app.invoke(initial_state, config={"configurable": {"thread_id": "1"}})

# Человек проверяет и одобряет
approval = input("Approve? (y/n): ")
result = app.invoke(
    {"approved": approval == "y"},
    config={"configurable": {"thread_id": "1"}}
)

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

Антипаттерны - чего избегай

1. God Agent

Один агент делает всё. Промпт становится огромным, LLM теряет фокус, результаты плохие. Сложно отлаживать, дорого в работе. Решение: раздели на 3-5 специализированных агентов.

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

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

MAX_ITERATIONS = 5

def route_after_review(state):
    if state.get("approved"):
        return "publish"
    if state.get("iteration_count", 0) >= MAX_ITERATIONS:
        return "human_review"  # Экстренный выход
    return "writer"

3. Отсутствие валидации

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

def validate_research(result: str) -> bool:
    if not result or len(result) < 100:
        return False
    if "http" not in result:  # Хотя бы один источник
        return False
    return True

4. Слишком много агентов

Больше агентов, это больше координационных издержек, выше стоимость, больше точек отказа. Правило большого пальца: 3-5 агентов. Если нужно больше, раздели на несколько независимых crews.

5. Отсутствие контроля стоимости

Без отслеживания затрат multi-agent системы могут стоить сотни долларов за запуск. Решение: используй CostTracker с лимитом бюджета (см. выше).

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

Функция	LangGraph	CrewAI	Кастомное
Pipeline	✅	✅ (sequential)	✅
Hub-and-Spoke	✅ (conditional edges)	✅ (hierarchical)	✅
DAG/Параллель	✅ (native)	❌	✅ (asyncio)
Event-Driven	❌ (не нативно)	❌	✅ (custom bus)
Циклы	✅ (conditional edges)	Ограничено	✅
Human-in-the-Loop	✅ (interrupt)	Вручную	✅
Checkpointing	✅ (native)	❌	✅ (custom)

Ключевые моменты для экзамена

• Оркестрация: координация нескольких агентов с распределением задач, обменом информацией, управлением и обработкой ошибок
• 4 паттерна: Pipeline (последовательный), Hub-and-Spoke (централизованный), DAG (параллельный), Event-Driven (событийный)
• Лучшие практики: четкие роли (SRP), явно типизированные состояния, повторные попытки с max_retries, контроль затрат через CostTracker, видимость системы с Logging + LangSmith, участие человека для критических решений
• Антипаттерны: God Agent, бесконечные циклы, отсутствие валидации, избыток агентов, отсутствие контроля затрат
• Выбор фреймворка: LangGraph для сложных графов, CrewAI для быстрого прототипирования, собственные решения для Event-Driven архитектуры

Часто задаваемые вопросы

Когда нужна мультиагентная система вместо одного агента? Когда задача требует разных областей expertise, нужна итерация (Writer ↔ Reviewer), возможна параллелизация или задача слишком сложна для одного промпта.

Сколько агентов оптимально? 3-5 для большинства случаев. Больше 8 трудно координировать и это дорого. При сложных требованиях лучше разбить на несколько независимых Crews или графов.

Какой паттерн оркестрации лучше? Pipeline для простых линейных задач, Hub-and-Spoke для динамической маршрутизации, DAG для параллельных операций, Event-Driven для слабо связанных систем. Большинство production-систем используют комбинацию нескольких подходов.

Как избежать бесконечных циклов? Всегда устанавливайте максимальное число итераций. Реализуйте аварийный выход, который при превышении числа итераций передает управление Human-in-the-Loop или завершает выполнение.

Как контролировать затраты в мультиагентных системах?

1. CostTracker с лимитом бюджета. 2) Дешевые модели для простых задач. 3) Кеширование одинаковых промптов. 4) Ранняя остановка. 5) Локальные модели для разработки.

Нужен ли LangSmith в production? Настоятельно рекомендуется. Без tracing вы слепы: не видите, сколько времени нужно каждому агенту и где возникают ошибки. У LangSmith есть бесплатный уровень для небольших проектов.

Рекомендуемая литература