Как обеспечивать exactly-once delivery между агентами?

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

Exactly-once delivery (доставка ровно один раз) между агентами — это гарантия, что каждое сообщение будет обработано ровно один раз, несмотря на сетевые ошибки, перезапуски или дублирование. В распределённых мультиагентных системах эта гарантия достигается комбинацией идемпотентности обработчиков, дедупликации на основе уникальных идентификаторов, транзакционных механизмов очередей (Kafka, DBOS) и паттерна Saga для компенсации частично выполненных шагов.

1. Термин: Exactly-once delivery (EOD)

Exactly-once delivery — свойство системы обмена сообщениями, при котором каждое сообщение обрабатывается ровно один раз, без потерь и без дублирования.

В распределённых системах (особенно в архитектурах с агентами) идеальная гарантия EOD теоретически недостижима из-за теоремы FLP (невозможность консенсуса в асинхронной сети с отказами). На практике EOD реализуется как эмуляция ровно одного выполнения поверх at-least-once (как минимум один раз) и идентификации дубликатов.

Почему это сложно:

• Сетевые сбои: отправитель не знает, дошло ли сообщение, поэтому повторяет отправку.
• Отказы потребителя: агент может упасть после обработки, но до подтверждения (ack) — сообщение будет переотправлено.
• Конкуренция: несколько экземпляров одного агента могут обработать одно и то же сообщение.

2. Основные проблемы при доставке между агентами

3. Подход 1: Идемпотентный потребитель (Idempotent consumer)

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

Пример: агент, который списывает деньги со счёта. Вместо прямого уменьшения баланса он выполняет операцию:
баланс = баланс - сумма, если ID транзакции не обработан.

Реализация через check-then-act:

class IdempotentPaymentAgent:
    def __init__(self, db):
        self.db = db  # хранилище выполненных транзакций

    def process(self, message: dict):
        tx_id = message['tx_id']
        if self.db.is_processed(tx_id):
            return  # уже обработано, пропускаем
        # бизнес-логика
        user_id = message['user_id']
        amount = message['amount']
        self.db.withdraw(user_id, amount)
        self.db.mark_processed(tx_id)

Плюсы: простота, не требует изменения брокера.
Минусы: требует внешнего хранилища; не защищает от дублей на уровне брокера.

4. Подход 2: Дедупликация с Transaction ID + Redis

Каждое сообщение снабжается уникальным Transaction ID (обычно UUID). Получатель хранит ID обработанных сообщений в быстром хранилище (например, Redis) с TTL (time-to-live), достаточным для покрытия возможных повторов.

import redis, uuid

r = redis.Redis()
TTL = 3600  # час — период, в течение которого возможны дубли

def send_message(agent, payload):
    tx_id = str(uuid.uuid4())
    message = {'tx_id': tx_id, 'payload': payload}
    agent.send(message)

def receive_message(message):
    tx_id = message['tx_id']
    if r.exists(tx_id):
        return  # дубль
    r.setex(tx_id, TTL, 'done')
    process(message['payload'])

Важно: TTL должен быть больше максимального времени повторной отправки. Если сообщение придёт после истечения TTL — будет повторно обработано (нарушение exactly-once). Компромисс: выбирать TTL с запасом (несколько часов до нескольких суток).

5. Подход 3: Kafka Transactions (Exactly-once Semantics — EOS)

Apache Kafka предоставляет встроенную гарантию exactly-once для связки продюсер → консьюмер через транзакции:

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

Ограничения:

• Работает только в рамках одного кластера Kafka.
• Требует поддержки со стороны потребителя (Kafka Consumer API с isolation.level=read_committed).
• Не решает проблему идемпотентности при внешних вызовах (API банка и т.п.).

6. Подход 4: DBOS (Database Operating System)

DBOS — парадигма, в которой весь workflow (включая обмен сообщениями между агентами) исполняется как транзакция в базе данных. Каждое сообщение — это строка со статусом. Дубли исключаются на уровне ACID (атомарность, согласованность, изоляция, долговечность).

Как выглядит:

• Таблица messages(id, payload, status).
• Агент читает строку со статусом pending, меняет на processing, выполняет бизнес-логику, меняет на done.
• Если агент упал, система обнаруживает незавершённые строки и перезапускает их (с проверкой идемпотентности).

Пример на псевдо‑SQL:

BEGIN;
SELECT * FROM messages WHERE id = :msg_id FOR UPDATE;
-- проверяем, что статус = 'pending', если нет — ROLLBACK
UPDATE messages SET status='processing' WHERE id = :msg_id;
-- вызываем внешний сервис (через outbox pattern или REST с идемпотентностью)
UPDATE messages SET status='done' WHERE id = :msg_id;
COMMIT;

Плюсы: полная гарантия exactly-once, естественное совмещение с базами данных.
Минусы: высокая нагрузка на БД; не все workflow легко укладываются в транзакции.

7. Подход 5: Паттерн SAGA

SAGA — последовательность локальных транзакций, каждая из которых имеет компенсирующее действие для отката. Для exactly-once в SAGA применяют компенсирующие транзакции вместе с идемпотентностью.

Пример: заказ — резервирование товара → списание денег. Если списание не удалось, запускается компенсация «отмена резервирования».

Варианты:

• Orchestration SAGA: центральный оркестратор направляет события агентам и обрабатывает откаты.
• Choreography SAGA: каждый агент публикует события, на которые подписываются другие, и компенсации запускаются по схеме.

Для exactly-one-delivery в SAGA критично, чтобы каждое событие (шаг или компенсация) было доставлено ровно один раз. Обычно это достигается через брокер с подтверждением (Kafka, RabbitMQ с publisher confirms) и идемпотентные обработчики.

8. Сравнение подходов

9. Рекомендации для Agentic RAG

Архитектура Agentic RAG — сложная сеть из LLM-агентов, которые могут вызывать друг друга, обращаться к внешним API, выполнять действия. Exactly-once между ними критично, если есть сайд-эффекты (платежи, отправка писем, изменение состояния).

Практический выбор:

1. Простые сценарии: используйте идемпотентность + Redis дедупликацию. Все внешние вызовы должны быть идемпотентными (например, API с Idempotency-Key).
2. Высокая пропускная способность и много агентов: Kafka transactions (или RabbitMQ with at-least-once + идемпотентность на стороне потребителя).
3. Строгая консистентность с БД: DBOS, если workflow не выходит за пределы одной базы.
4. Длительные цепочки действий: SAGA, где каждый шаг имеет компенсацию, а сообщения между шагами доставляются с дедупликацией.

Ключевой принцип: не пытайтесь добиться идеального exactly-once на уровне транспорта — это дорого и ненадёжно. Лучше сделать потребителя идемпотентным, а транспорту разрешить at-least-once.

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

Задача: Реализовать простую двухагентную систему: AgentA принимает запрос пользователя и отправляет сообщение AgentB для выполнения платежа (имитация). Обеспечить exactly-once обработку платежа.

Инструменты: Python, Redis (через redis-py), FastAPI (для агентов), uuid.

Шаги:

1. Напишите AgentA — REST сервис с эндпоинтом /pay. Он генерирует transaction_id и отправляет POST-запрос на AgentB с этим ID.
2. AgentB в обработчике:
   • Проверяет в Redis, есть ли уже transaction_id со статусом processed. Если есть — возвращает успех (дубль).
   • Если нет — выполняет «платёж» (например, запись в SQLite).
   • Сохраняет transaction_id в Redis с TTL=3600.
   • Возвращает status: ok.
3. Реализуйте механизм повторных попыток в AgentA (retry с exponential backoff) при сетевых ошибках.
4. Напишите тест, который отправляет один запрос много раз (с одинаковым ID) и проверяет, что платеж выполнился ровно один раз (проверьте запись в БД).

Ожидаемый результат: Вы убедитесь, что при любом количестве повторных запросов с одним и тем же transaction_id бизнес-логика выполняется только один раз.

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

Навигация

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