Что такое agent communication protocol (формат сообщений между агентами)?

Краткий тезис

Agent communication protocol — это стандартизированный формат и набор правил обмена сообщениями между автономными агентами в multi-agent системе. В архитектурах Agentic RAG такой протокол позволяет агентам (planner, retrieval, generator, coordinator) согласованно передавать задачи, обмениваться данными и отслеживать полный цикл выполнения запроса. Чаще всего используется JSON-схема с обязательными полями: agent_id, target_agent, message_type, content, correlation_id и timestamp. Протокол обеспечивает масштабируемость, отказоустойчивость и прозрачность взаимодействий.

1. Понятие agent communication protocol

Agent communication protocol (протокол коммуникации агентов) определяет:

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

В отличие от обычного межсервисного обмена (HTTP/REST, gRPC), протокол для агентов предполагает асинхронность, наличие контекста сессии и возможность длинных цепочек взаимодействий. Каждый агент может выступать как отправителем, так и получателем, а сообщение может быть частью более крупного диалога.

Протокол может быть реализован поверх любого транспорта: RabbitMQ, Kafka, Redis Pub/Sub или прямые HTTP-вызовы с очередями. Главное — единая контрактная схема.

2. Структура сообщения: ключевые поля

Типовое сообщение выглядит как JSON-объект. Рассмотрим каждое поле детально.

{
  "message_id": "msg_001",
  "agent_id": "planner_1",
  "target_agent": "executor_2",
  "message_type": "command",
  "content": {
    "action": "search",
    "query": "погода в Москве",
    "params": {"top_k": 5}
  },
  "correlation_id": "abc123",
  "timestamp": 1700000000,
  "ttl": 60,
  "metadata": {}
}

Расшифровка терминов

• correlation_id — сквозной идентификатор, который прикрепляется к запросу пользователя и передаётся через все сообщения цепочки. Позволяет восстановить полный маршрут (trace) для отладки и мониторинга.
• ttl (Time To Live) — если сообщение не было обработано за указанное время, оно игнорируется. Предотвращает зависание агентов.

3. Типы сообщений (message types)

Стандартный набор включает пять основных типов:

В сложных системах добавляют:

• heartbeat — проверка активности агента.
• acknowledge — подтверждение получения (для гарантированной доставки).
• cancel — отмена ранее отправленного command.

4. Correlation ID: трейсинг цепочки сообщений

Correlation ID — это «нить», связывающая все сообщения в рамках одного пользовательского запроса. В Agentic RAG запрос может проходить через несколько агентов:

1. Пользователь → Gateway
2. Gateway → Planner (разбивает запрос на подзадачи)
3. Planner → Retriever (ищет документы)
4. Retriever → Generator (формирует ответ)
5. Generator → Gateway → Пользователь

Без correlation_id логи разных агентов невозможно связать в единую историю. С ним можно:

• Измерить latency каждого этапа.
• Найти, на каком шаге произошла ошибка.
• Собрать full trace для датасета (вход → цепочка → выход).

Пример в логах:

[corr:abc123] 10:00:00 planner_1 → retriever: command поиск
[corr:abc123] 10:00:05 retriever → generator: response 3 документа
[corr:abc123] 10:00:07 generator → gateway: response текст

5. JSON vs. Protobuf: выбор формата сериализации

Хотя в примерах часто используется JSON, в production-системах с высокой нагрузкой применяют Protocol Buffers (Protobuf). Сравнение:

В Agentic RAG с высоким RPS (>1000) лучше Protobuf. Для небольших систем или прототипов — JSON (гибкость, отладка).

Промежуточный вариант использовать MsgPack или CBOR — бинарные форматы с сохранением структуры, похожие на JSON.

6. Синхронный vs асинхронный обмен

Протокол может работать в двух режимах:

• Синхронный: агент отправляет command и ждёт response (блокирующий вызов). Проще в реализации, но плохо для долгих операций (retrieval с реранжированием).
• Асинхронный: отправитель продолжает работу, получатель отвечает через очередь или callback. Требует управления состоянием (pending requests, таймауты).

В Agentic RAG чаще асинхронный, потому что:

• Retrieval может занимать секунды.
• Планировщик может запускать несколько подзадач параллельно.
• Система должна оставаться отзывчивой.

Пример асинхронного обмена с очередью (RabbitMQ):

# Отправка команды (publish)
channel.basic_publish(
    exchange='agent_exchange',
    routing_key='retriever_queue',
    body=json.dumps({
        "correlation_id": "abc123",
        "message_type": "command",
        "content": {"action": "search", "query": "..."}
    })
)

# Получение ответа (consume) в callback
def on_response(ch, method, properties, body):
    msg = json.loads(body)
    if msg['correlation_id'] == expected_corr:
        process(msg)

7. Стандартизация протоколов: A2A, MCP и другие

Сейчас активно развиваются открытые спецификации для агентов:

• Agent-to-Agent (A2A) от Google — протокол взаимодействия между агентами на основе JSON, поддерживает discovery, negotiation, streaming.
• Model Context Protocol (MCP) от Anthropic — протокол для интеграции LLM с инструментами (tools), фактически определяет взаимодействие «агент–инструмент».
• Semantic Kernel (Microsoft) — фреймворк, использующий собственный протокол на основе FunctionCall (JSON).

Вопрос 591 не требует знания конкретных стандартов, но важно понимать, что сообщество движется к унификации, и ваш протокол должен быть совместим с ними or легко адаптироваться.

8. Безопасность и верификация сообщений

Протокол должен включать механизмы защиты:

• Аутентификация отправителя (токен, цифровая подпись).
• Валидация схемы — проверка, что сообщение соответствует контракту (JSON Schema, Pydantic модели).
• Защита от таймаутов — обработка ttl, повторная отправка (retry) с idempotency.
• Audit log — запись всех сообщений для разбора инцидентов.

Пример валидации с Pydantic:

from pydantic import BaseModel, Field
from typing import Any, Optional

class AgentMessage(BaseModel):
    message_id: str
    agent_id: str
    target_agent: str
    message_type: str  # "command" | "query" | ...
    content: dict[str, Any]
    correlation_id: str
    timestamp: int
    ttl: Optional[int] = 60
    metadata: dict = {}

# При получении:
msg = AgentMessage(**raw_json)

9. Протокол в контексте Agentic RAG

Архитектура Agentic RAG часто включает таких агентов:

• Gateway — приём запросов от пользователя, генерация correlation_id.
• Planner — разбивает сложный запрос на подзадачи (DAG).
• Retrieval Agent — поиск по векторной базе + гибридный поиск.
• Reranker — реранжирование результатов.
• Generator — LLM, формирующая ответ с учётом найденных документов.
• Memory Agent — хранение истории диалога.

Протокол связывает их в единую систему. Типичный сценарий:

1. Gateway → Planner: {message_type: "command", content: {"action": "plan", "query": "..."}}
2. Planner → Retrieval Agent: {message_type: "command", content: {"action": "search", "subquery": "..."}}
3. Retrieval Agent → Reranker: {message_type: "command", content: {"action": "rerank", "docs": [...]}}
4. Reranker → Generator: {message_type: "command", content: {"action": "generate"}}
5. Generator → Gateway: {message_type: "response", content: {"answer": "..."}}

Все сообщения имеют один correlation_id, что позволяет логировать и отлаживать.

10. Плюсы и минусы применения стандартизированного протокола

Пет-проект для закрепления

Задача Реализовать простую multi-agent систему для ответа на вопросы, используя собственный JSON-протокол.

Инструменты

• Python 3.11+
• RabbitMQ (через pika) или Redis (через redis-py)
• Pydantic для валидации схем
• LangChain (опционально) для агентов

Шаги:

1. Определить JSON-схему сообщения (как в разделе 2).
2. Написать класс AgentMessage с валидацией.
3. Реализовать три агента:
   • Planner — получает запрос пользователя, отправляет команду на поиск.
   • Retriever — имитирует поиск (возвращает фиктивные документы).
   • Generator — принимает документы и формирует ответ.
4. Настроить очередь RabbitMQ: exchange agent_exchange, routing keys: planner, retriever, generator.
5. Добавить correlation_id и записывать все сообщения в лог с отметкой времени.
6. Написать тест: отправить один запрос и убедиться, что цепочка из трёх шагов отработала.
7. Измерить время каждого шага.

Ожидаемый результат Работающий прототип, который выводит в лог:

[INFO] [corr:abc] Gateway → Planner: command
[INFO] [corr:abc] Planner → Retriever: command
[INFO] [corr:abc] Retriever → Generator: response
[INFO] [corr:abc] Generator → Gateway: response

Связь с другими вопросами

Навигация

• Предыдущий: 590
• Следующий: 592
• Индекс: 00. Индекс разборов