Что такое AgentPool и Handoff в multi-agent orchestration?

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

AgentPool — это пул специализированных агентов-исполнителей, управляемый центральным роутером, который распределяет входящие запросы по правилам или по загрузке. Handoff — механизм явной передачи задачи от текущего агента другому агенту с сохранением полного контекста выполнения через sealed (неизменяемый) SendHandle (с лимитом 64 KiB) и строгую границу контекста ContextBoundary. Вместе они образуют основу надёжной многолетней оркестрации в средах типа harness-one, позволяя строить масштабируемые, изолированные и легко тестируемые multi-agent системы.

1. Термины и их место в multi-agent orchestration

Multi-agent orchestration — подход, при котором сложная задача разбивается на подзадачи, каждая из которых выполняется отдельным агентом (автономной сущностью с LLM, памятью, инструментами). Агенты могут работать параллельно, обмениваться сообщениями, делегировать задачи и синхронизироваться.

AgentPool — структурный паттерн, предоставляющий набор готовых агентов, объединённых общей точкой входа (роутером). Вместо того чтобы запускать каждого агента вручную, система передаёт запрос роутеру, который на основе типа запроса, загрузки или политики выбирает подходящего агента из пула.

Handoff — операционный паттерн, реализующий передачу управления от одного агента другому с гарантией, что весь накопленный контекст (история диалога, промежуточные результаты, состояние инструментов) будет передан полностью и без искажений. В harness-one handoff выполняется через SendHandle (неизменяемый дескриптор сообщения) с жёстким ограничением размера в 64 KiB и через MessageTransport (транспорт сообщений) с соблюдением ContextBoundary (границы контекста).

2. AgentPool: устройство и принцип работы

AgentPool решает три главные задачи:

• Масштабирование: можно динамически добавлять/удалять агентов под нагрузкой.
• Специализация: каждый агент может быть настроен на определённый домен (RAG, код, анализ данных).
• Изоляция сбоев: ошибка одного агента не валит всю систему.

2.1 Компоненты AgentPool

2.2 Пример простой реализации на Python (asyncio)

import asyncio
from typing import Protocol, Dict

class Agent(Protocol):
    async def handle(self, request: dict) -> dict: ...

class Router:
    def __init__(self):
        self.routes: Dict[str, Agent] = {}

    def register(self, pattern: str, agent: Agent):
        self.routes[pattern] = agent

    async def dispatch(self, request: dict):
        key = request.get("type", "default")
        agent = self.routes.get(key, self.routes["default"])
        return await agent.handle(request)

class AgentPool:
    def __init__(self, router: Router):
        self.router = router
        self.agents: list[Agent] = []

    async def process(self, request: dict) -> dict:
        return await self.router.dispatch(request)

2.3 Стратегии маршрутизации

• Content-based: роутер анализирует текст запроса (например, содержит «код» → агент-программист).
• Load-balanced: round-robin или least-connections для равномерной нагрузки.
• Priority-queue: запросы с высоким приоритетом обрабатываются быстрее.

3. Handoff: механизм передачи управления и контекста

Handoff — ключевой элемент для построения сложных цепочек агентов (pipeline). Без него каждый агент начинал бы с нуля, теряя историю.

3.1 Составляющие Handoff

• SendHandle — sealed (неизменяемый) объект, который содержит сам запрос и метаданные (например, ID сессии, обратный адрес). После создания его нельзя модифицировать, что гарантирует целостность.
• MessageTransport — слой, отвечающий за доставку SendHandle от одного агента другому. Может быть реализован через in-memory очередь (asyncio.Queue), Redis Pub/Sub, Kafka.
• ContextBoundary — граница, которая определяет, какая часть контекста (история, состояние) может быть передана другому агенту. Обычно включает только те данные, которые явно разрешены политикой безопасности (например, без приватных ключей).

Лимит в 64 KiB для SendHandle — это защита от переполнения памяти и атак типа «сообщение-монстр». При превышении ручка разбивается или отклоняется.

3.2 Пример handoff с asyncio

import asyncio
from dataclasses import dataclass, asdict
from typing import Any

@dataclass(frozen=True)
class SendHandle:  # sealed
    payload: dict
    source: str
    destination: str
    session_id: str

class ContextBoundary:
    @staticmethod
    def strip_context(raw: dict) -> dict:
        # удаляем чувствительные поля
        return {k: v for k, v in raw.items() if not k.startswith("_secret")}

class SimpleMessageTransport:
    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, handle: SendHandle):
        await self.queues[handle.destination].put(handle)

    async def receive(self, agent_id: str) -> SendHandle:
        return await self.queues[agent_id].get()

# Агент A передаёт задачу агенту B
async def agent_a(transport: SimpleMessageTransport, handle: SendHandle):
    # ... обработка ...
    stripped = ContextBoundary.strip_context(handle.payload)
    new_handle = SendHandle(stripped, "A", "B", handle.session_id)
    await transport.send(new_handle)

async def agent_b(transport: SimpleMessageTransport):
    handle = await transport.receive("B")
    # ... продолжение обработки ...

3.3 Отличие от обычного message passing

4. Сценарии совместного использования AgentPool и Handoff

1. Pipeline: Router направляет запрос в AgentPool, агент A обрабатывает первую часть, затем делает Handoff агенту B.
2. Fork/Join: Router запускает несколько агентов параллельно, результаты собираются через Handoff в агрегирующий агент.
3. Supervisor: Главный агент (supervisor) раздаёт подзадачи через Handoff саб-агентам из пула и собирает результаты.

Пример для RAG:

• Router (в составе AgentPool) определяет тип запроса: фактографический, аналитический, мультимодальный.
• Агент-поиск (SearchAgent) выполняет retrieval по векторной БД.
• Через Handoff передаёт найденные документы + историю запроса агенту-генератору (GenerateAgent) для финального ответа.

5. Проблемы и решения

6. Сравнение с другими фреймворками

7. Интеграция с RAG (Agentic RAG)

В контексте Agentic RAG AgentPool может содержать:

• RetrievalAgent — отвечает за поиск чанков.
• RerankAgent — переранжирует результаты.
• GenerateAgent — строит ответ с помощью LLM.
• GuardrailAgent — проверяет факты и фильтрует.

Handoff между ними обеспечивает сохранение исходного запроса и промежуточных результатов. Например, RetrievalAgent находит документы, делает Handoff с контекстом {"query": "...", "docs": [...], "scores": [...]}, RerankAgent возвращает ранжированный список с sealed SendHandle, GenerateAgent выдаёт финальный ответ.

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

Задача: Построить простую multi-agent систему для вопросов по документации (Agentic RAG) с использованием AgentPool и Handoff.

Инструменты:

• Python 3.11+ (asyncio)
• FAISS (векторная БД)
• LLM через OpenAI API (можно mock)
• asyncio.Queue как MessageTransport

Шаги:

1. Реализовать класс Agent с методами handle(query, context).
2. Создать агентов: RouterAgent, SearchAgent, AnswerAgent.
3. Реализовать SendHandle как NamedTuple (sealed).
4. Написать ContextBoundary, исключающий секреты (например, ключи API).
5. Собрать AgentPool с роутером на основе ключевых слов.
6. Реализовать handoff: Router → Search → Answer.
7. Запустить тестовый запрос и проверить, что контекст (история + результаты поиска) передаётся без потерь.

Ожидаемый результат: Рабочий прототип, в котором запрос проходит цепочку из трёх агентов, каждый получает полный контекст, финальный ответ корректен.

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

Навигация

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