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

Параллельный запуск мультиагентной системы начинается с выбора модели конкурентности. Для I/O-bound агентов (AI-запросы, ожидание ответов) асинхронный event loop с async/await даст низкую стоимость переключения, но для CPU-bound (тяжёлые вычисления) потребуются отдельные процессы или потоки с блокировками. В распределённом сценарии — Kafka или RabbitMQ для асинхронного обмена и оркестратор (например, Celery или Temporal).

В одном процессе на Python я обычно использую asyncio. Агент — это корутина, которая в цикле читает из очереди входящие сообщения и пишет ответы. Запуск через asyncio.gather:

import asyncio
from typing import AsyncIterator, Callable

class Agent:
    def __init__(self, name: str, inbox: asyncio.Queue, outbox: asyncio.Queue, handler: Callable):
        self.name = name
        self.inbox = inbox
        self.outbox = outbox
        self.handler = handler

    async def run(self):
        while True:
            msg = await self.inbox.get()
            result = await self.handler(msg)
            await self.outbox.put(result)

async def main():
    # агенты общаются через очереди
    queue_a = asyncio.Queue()
    queue_b = asyncio.Queue()

    agent_a = Agent("A", queue_a, queue_b, async_handler_a)
    agent_b = Agent("B", queue_b, queue_a, async_handler_b)

    # параллельный запуск
    await asyncio.gather(agent_a.run(), agent_b.run())

asyncio.run(main())
Ключевые точки. Синхронизация. Очереди решают только прямую передачу. Если агенты разделяют ресурс (общее состояние), нужен asyncio.Lock или акторная модель с изолированной памятью. Graceful shutdown. Добавьте стоп-сигнал через asyncio.Event или обрабатывайте SIGTERM через loop.add_signal_handler, чтобы агенты завершили текущую задачу и закрыли соединения. Обработка ошибок. В агенте try/except + повторная отправка в очередь (с лимитом ретраев). Используйте паттерн Circuit Breaker для внешних вызовов. Масштабирование. Для распределённой версии замените in-memory очереди на Kafka topic’ы — каждый агент читает свою партицию, балансировка через consumer group. Тогда оркестрация не нужна — сообщения сами маршрутизируются.
Из практики: не пытайтесь делать точную параллельность (распараллелить один поток данных). Лучше пусть агенты работают со своими очередями и подтверждают обработку (ack). Это даёт отказоустойчивость и возможность репликации.