Протоколы коммуникации мультиагентных систем

Updated Jul 2026

Почему коммуникация важна

Когда агенты обмениваются информацией, протокол коммуникации определяет, будет система работать или нет. Плохие протоколы приводят к потере контекста, каскадным сбоям и агентам, работающим несогласованно. Хорошие протоколы обеспечивают чистую передачу данных, распространение ошибок и распределённую координацию.

Протокол сообщений

from dataclasses import dataclass

@dataclass
class AgentMessage:
    sender: str           # "code_analyzer"
    receiver: str         # "test_generator" or "broadcast"
    message_type: str     # "analysis_complete" | "tests_ready" | "error"
    payload: dict         # Structured data
    priority: int         # 0=low, 1=normal, 2=high, 3=critical
    timestamp: float
    correlation_id: str   # Traces related messages across agents

Типы сообщений

Пример потока сообщений

messages = [
    AgentMessage(
        sender="code_analyzer",
        receiver="test_generator",
        message_type="analysis_complete",
        payload={
            "module": "auth/login.py",
            "functions_to_test": ["authenticate", "validate_token"],
            "complexity_hints": {"authenticate": "high", "validate_token": "medium"},
            "existing_test_count": 3,
            "suggested_test_count": 8
        },
        priority=1,
        timestamp=time.time(),
        correlation_id="sprint-42-auth-refactor"
    ),
    AgentMessage(
        sender="test_generator",
        receiver="test_runner",
        message_type="tests_ready",
        payload={
            "test_file": "tests/test_auth_login.py",
            "test_count": 8,
            "dependencies": ["pytest", "pytest-asyncio", "httpx"]
        },
        priority=1,
        timestamp=time.time(),
        correlation_id="sprint-42-auth-refactor"
    )
]

Идентификаторы корреляции для трассировки

correlation_id необходим для отладки мультиагентных систем. Он связывает все сообщения, относящиеся к одной задаче:

# All messages for the auth-refactor task share the same correlation ID
# This allows you to:
# 1. Filter logs by correlation_id to see the full pipeline
# 2. Measure total pipeline time (first message to last)
# 3. Identify where bottlenecks occur
# 4. Replay a specific task for debugging

def get_pipeline_trace(correlation_id: str) -> list[AgentMessage]:
    """Get all messages for a specific task, ordered by timestamp."""
    return sorted(
        [m for m in message_store if m.correlation_id == correlation_id],
        key=lambda m: m.timestamp
    )

Распространение ошибок

Когда один агент выходит из строя, система должна обработать это корректно. Существуют три стратегии:

Стратегия 1: Пропустить и продолжить

async def run_pipeline(self, modules: list[str]):
    for module in modules:
        try:
            analysis = await self.code_analyzer.analyze(module)
            tests = await self.test_generator.generate(analysis)
            results = await self.test_runner.run(tests)

            if results.has_failures:
                fixed = await self.test_fixer.fix(tests, results)
                results = await self.test_runner.run(fixed)

        except AgentTimeoutError as e:
            self.log.warning(f"Agent timeout on {module}: {e}")
            self.dead_letter_queue.append(module)  # Retry later
            continue

        except Exception as e:
            self.log.error(f"Unexpected failure on {module}: {e}")
            self.alert_human(module, e)  # Escalate
            continue

Используйте, когда: Модули независимы. Сбой одного не должен блокировать остальные.

Стратегия 2: Быстрый отказ

async def run_pipeline(self, modules: list[str]):
    for module in modules:
        try:
            analysis = await self.code_analyzer.analyze(module)
            tests = await self.test_generator.generate(analysis)
            results = await self.test_runner.run(tests)
        except AgentBudgetExceeded as e:
            self.log.error(f"Budget exceeded on {module}: {e}")
            break  # Stop processing ALL modules

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

Стратегия 3: Резервные агенты

async def analyze_with_fallback(self, module: str):
    """Try the primary analyzer, fall back to a simpler one."""
    try:
        return await self.primary_analyzer.analyze(module)
    except AgentTimeoutError:
        self.log.warning(f"Primary analyzer timed out, using fallback")
        return await self.fallback_analyzer.analyze(module)

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

Паттерны коммуникации

Паттерн 1: Конвейер (последовательный)

Analyzer -> Generator -> Runner -> Fixer -> Runner

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

Паттерн 2: Публикация-подписка

class MessageBus:
    def __init__(self):
        self.subscribers = defaultdict(list)

    def subscribe(self, message_type: str, handler: Callable):
        self.subscribers[message_type].append(handler)

    def publish(self, message: AgentMessage):
        for handler in self.subscribers[message.message_type]:
            handler(message)

# Usage
bus = MessageBus()
bus.subscribe("analysis_complete", test_generator.on_analysis_complete)
bus.subscribe("analysis_complete", coverage_tracker.on_analysis_complete)
bus.subscribe("tests_ready", test_runner.on_tests_ready)
bus.subscribe("error", alerting_service.on_error)

Несколько агентов могут реагировать на одно событие. Coverage Tracker и Test Generator оба получают результат анализа.

Паттерн 3: Запрос-ответ

class AgentProxy:
    """Synchronous request-response to another agent."""

    async def request(self, target_agent: str, action: str, data: dict) -> dict:
        request_id = str(uuid4())
        self.send(AgentMessage(
            sender=self.agent_id,
            receiver=target_agent,
            message_type=f"request:{action}",
            payload={"request_id": request_id, **data},
            priority=2,
        ))

        # Wait for response with timeout
        response = await self.wait_for_response(request_id, timeout=30)
        return response.payload

Мониторинг мультиагентной коммуникации

class CommunicationMonitor:
    """Track message flow health across agents."""

    def __init__(self):
        self.message_count = defaultdict(int)
        self.error_count = defaultdict(int)
        self.latencies = defaultdict(list)

    def on_message(self, message: AgentMessage):
        key = f"{message.sender}->{message.receiver}"
        self.message_count[key] += 1

        if message.message_type == "error":
            self.error_count[message.sender] += 1

    def report(self) -> dict:
        return {
            "total_messages": sum(self.message_count.values()),
            "messages_by_channel": dict(self.message_count),
            "errors_by_agent": dict(self.error_count),
            "error_rate": sum(self.error_count.values()) / max(sum(self.message_count.values()), 1),
        }

Ключевой вывод

Протоколы коммуникации мультиагентных систем определяют надёжность системы. Используйте структурированные сообщения с идентификаторами корреляции для трассировки, реализуйте стратегии распространения ошибок (пропуск, быстрый отказ, резервный агент) в зависимости от модели зависимостей и мониторьте здоровье потока сообщений для раннего обнаружения узких мест и сбоев.