Multi-Agent Orchestration: Координация LLM-агентов в сложных системах

llmagentsorchestrationmulti-agentarchitecture
← Back to Blog

Введение

Multi-Agent Orchestration — это подход к построению LLM-систем, при котором несколько специализированных агентов координируют свои действия для решения сложных задач. Вместо одного универсального LLM, система использует сеть специализированных агентов с разными ролями и компетенциями.

Почему один LLM недостаточно

Сценарий Один LLM Мультиагентная система
Сложная задача Деградация качества Сохранение точности
Масштабирование Вертикальное Горизонтальное
Отказоустойчивость Единая точка отказа Изоляция сбоев
Специализация Общая Узкая и глубокая
Стоимость Высокая Оптимизированная

Архитектурные паттерны

1. Supervisor Pattern

                    ┌─────────────┐
                    │  Supervisor │
                    │  (Planner)  │
                    └──────┬──────┘
                           │
              ┌────────────┼────────────┐
              ↓            ↓            ↓
        ┌──────────┐ ┌──────────┐ ┌──────────┐
        │ Worker 1 │ │ Worker 2 │ │ Worker 3 │
        │Research  │ │ Analyze  │ │ Generate │
        └────┬─────┘ └────┬─────┘ └────┬─────┘
             │             │             │
             └─────────────┼─────────────┘
                           ↓
                    ┌─────────────┐
                    │  Synthesizer│
                    │  (Writer)   │
                    └─────────────┘

from typing import List, Dict, Any
from dataclasses import dataclass
import asyncio

@dataclass
class Task:
    id: str
    description: str
    agent_type: str
    dependencies: List[str]
    result: Any = None

class SupervisorAgent:
    """Координирует работу рабочих агентов"""
    
    def __init__(self, agents: Dict[str, Any]):
        self.agents = agents
        self.task_queue: List[Task] = []
        self.completed: Dict[str, Task] = {}
    
    def plan(self, goal: str) -> List[Task]:
        """Декомпозиция цели на задачи"""
        # LLM-планировщик разбивает задачу
        planning_prompt = f"""
        Разбейте следующую задачу на подзадачи:
        {goal}
        
        Для каждой подзадачи определите:
        1. Тип агента (research, analyze, generate, review)
        2. Описание задачи
        3. Зависимости от других задач
        """
        return self._parse_plan(planning_prompt)
    
    async def execute(self, tasks: List[Task]) -> Dict[str, Any]:
        """Параллельное выполнение с учётом зависимостей"""
        while tasks or self.completed:
            # Найти задачи, готовые к выполнению
            ready = [
                t for t in tasks
                if all(dep in self.completed for dep in t.dependencies)
            ]
            
            if not ready:
                break
            
            # Выполнить готовые задачи параллельно
            coroutines = []
            for task in ready:
                agent = self.agents[task.agent_type]
                context = self._build_context(task)
                coroutines.append(self._run_agent(agent, task, context))
            
            results = await asyncio.gather(*coroutines)
            for result in results:
                self.completed[result.id] = result
                tasks.remove(result)
        
        return self._synthesize_results()

2. Swarm Pattern

class SwarmAgent:
    """Агент в роевой архитектуре — автономный, но кооперирующий"""
    
    def __init__(self, role: str, model: str, capabilities: List[str]):
        self.role = role
        self.model = model
        self.capabilities = capabilities
        self.memory: List[Dict] = []
    
    def propose(self, problem: str) -> Dict:
        """Предлагает решение"""
        proposal = self._generate_solution(problem)
        self.memory.append({
            "type": "proposal",
            "content": proposal,
            "role": self.role
        })
        return proposal
    
    def evaluate(self, proposal: Dict) -> float:
        """Оценивает чужое предложение"""
        evaluation = self._assess_proposal(proposal)
        self.memory.append({
            "type": "evaluation",
            "target_role": proposal.get("role"),
            "score": evaluation
        })
        return evaluation

3. Pipeline Pattern

Input → [Agent 1] → [Agent 2] → [Agent 3] → Output
         Filter      Transform   Generate

class PipelineAgent:
    """Агент в конвейерной архитектуре"""
    
    def __init__(self, stage: int, total_stages: int, agent_func):
        self.stage = stage
        self.total_stages = total_stages
        self.agent_func = agent_func
    
    async def process(self, data: Dict) -> Dict:
        result = await self.agent_func(data)
        result["stage"] = self.stage
        result["completed"] = self.stage == self.total_stages
        return result

# Создание конвейера
pipeline = [
    PipelineAgent(1, 4, lambda d: sanitize_input(d)),
    PipelineAgent(2, 4, lambda d: extract_entities(d)),
    PipelineAgent(3, 4, lambda d: generate_response(d)),
    PipelineAgent(4, 4, lambda d: format_output(d)),
]

Communication Protocols

Shared Blackboard

class Blackboard:
    """Общее информационное пространство для агентов"""
    
    def __init__(self):
        self.entries: List[Dict] = []
        self.lock = asyncio.Lock()
    
    async def append(self, entry: Dict):
        async with self.lock:
            entry["timestamp"] = datetime.now().isoformat()
            entry["author"] = entry.get("agent_id")
            self.entries.append(entry)
    
    async def query(self, filter_fn) -> List[Dict]:
        return [e for e in self.entries if filter_fn(e)]
    
    async def get_recent(self, n: int, agent_id: str = None) -> List[Dict]:
        filtered = self.entries
        if agent_id:
            filtered = [e for e in filtered if e.get("author") != agent_id]
        return filtered[-n:]

Message Queue

class AgentMessageQueue:
    """Очередь сообщений между агентами"""
    
    def __init__(self):
        self.queues: Dict[str, asyncio.Queue] = {}
    
    def register(self, agent_id: str):
        self.queues[agent_id] = asyncio.Queue()
    
    async def send(self, from_id: str, to_id: str, message: Dict):
        if to_id in self.queues:
            message["from"] = from_id
            message["timestamp"] = time.time()
            await self.queues[to_id].put(message)
    
    async def receive(self, agent_id: str) -> Dict:
        return await self.queues[agent_id].get()

Framework Comparison

Фреймворк Язык Оркестрация Память Функции
LangGraph Python/JS State machine Встроенная Да
AutoGen Python Group chat Встроенная Да
CrewAI Python Roles Встроенная Да
DSPy Python Compiler Внешняя Да
Microsoft AutoGen Python Conversational Встроенная Да

Best Practices

  1. Чёткие границы ответственности — каждый агент должен иметь однозначную роль
  2. Минимизация коммуникации — избыточные запросы между агентами замедляют систему
  3. Fallback механизмы — при сбое агента должен быть план Б
  4. Логирование на каждом этапе — критично для отладки мультиагентных систем
  5. Тестирование изолированно — каждый агент должен проходить тесты отдельно

Заключение

Multi-Agent Orchestration позволяет строить сложные, масштабируемые и отказоустойчивые LLM-системы. Выбор архитектуры зависит от конкретной задачи: Supervisor для сложных декомпозируемых задач, Swarm для креативных задач, Pipeline для последовательной обработки.