Параллельные tools AI-агента: как вернуть честный частичный ответ

AI-ассистент интернет-магазина получает вопрос: «Эта модель есть в наличии, когда её привезут и сколько дней даётся на возврат?» Чтобы ответить, ему нужны три источника: остатки, расчёт доставки и правила возврата.

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

Сложность появляется, когда один источник не отвечает. Если сервис доставки недоступен, мы всё ещё можем сообщить остаток и правила возврата. А дату доставки пока назвать не можем. Для нашего примера договоримся: показывать полученные сведения можно, оформлять заказ без расчёта доставки — нельзя. Посмотрим, как выразить это правило в коде.

Почему общий успех здесь слишком грубый

В Python несколько асинхронных запросов удобно объединить в TaskGroup. Но у этой группы есть важное поведение: если один запрос заканчивается обычным исключением, она отменяет остальные незавершённые запросы. Весь вызов завершается ошибкой. Это удобно, когда нужны все результаты сразу.

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

Решение о частичном ответе приходится принимать нам. Для оформления заказа мы требуем остаток и расчёт доставки. Для справки достаточно показать полученные сведения и отдельно написать, чего узнать не удалось. Сам TaskGroup об этих правилах магазина ничего не знает.

Сначала сохраняем исход каждой ветки

Запишем для каждого запроса два значения: полученные данные и статус выполнения. Если сервис не ответил, сохраним ошибку как отдельный результат. Тогда код сборки сообщения увидит, какие сведения доступны. В примере ниже такую запись представляет BranchResult, а функция capture перехватывает ошибку запроса.

Отдельный статус нужен не для красоты. value=None может быть настоящим ответом API, а status="error" означает, что ответа вообще нет. Если заменить сбой нулём, пустой строкой или None, следующая стадия агента начнёт рассуждать о выдуманном факте.

Один дедлайн на всю группу

Зададим срок ожидания для всей группы запросов. Когда он истечёт, отменим незавершённые задачи и запишем для них статус timeout. Если сбор данных — лишь один этап работы агента, этот срок нужно выбирать с учётом времени на остальные этапы: иначе каждый из них будет начинать ожидание заново.

asyncio.wait возвращает завершённые и ожидающие задачи отдельно. Ожидающие задачи он сам не отменяет, поэтому эта часть остаётся в нашем коде. Повторный gather после cancel() забирает результаты отмены: так фоновые задачи не остаются жить после ответа пользователю.

Полный запускаемый пример

Для запуска нужен Python 3.11 или новее. Здесь служба доставки завершается ошибкой, а две другие ветки успевают вернуть данные. Для короткого запуска задержки измеряются сотыми долями секунды, но решение не зависит от этих чисел.

import asyncio
from dataclasses import dataclass

async def call_tool(name: str):
    delays = {"stock": 0.01, "returns": 0.02, "delivery": 0.03}
    await asyncio.sleep(delays[name])
    if name == "delivery":
        raise RuntimeError("delivery service unavailable")
    return {"stock": "in stock", "returns": "30 days"}[name]

async def fragile_run():
    try:
        async with asyncio.TaskGroup() as group:
            stock = group.create_task(call_tool("stock"))
            delivery = group.create_task(call_tool("delivery"))
            returns = group.create_task(call_tool("returns"))
        return stock.result(), delivery.result(), returns.result()
    except* RuntimeError as errors:
        print("Naive run: whole operation failed:", errors.exceptions[0])

@dataclass
class BranchResult:
    status: str
    value: str | None = None
    error: str | None = None

async def capture(name: str):
    try:
        return BranchResult(status="ok", value=await call_tool(name))
    except Exception as error:
        return BranchResult(status="error", error=str(error))

async def resilient_run():
    names = ("stock", "delivery", "returns")
    tasks = {name: asyncio.create_task(capture(name)) for name in names}
    done, pending = await asyncio.wait(tasks.values(), timeout=0.1)
    for task in pending:
        task.cancel()
    await asyncio.gather(*pending, return_exceptions=True)
    return {
        name: task.result() if task in done else BranchResult(status="timeout")
        for name, task in tasks.items()
    }

def build_answer(results: dict[str, BranchResult]) -> str:
    parts = []
    stock = results["stock"]
    delivery = results["delivery"]
    returns = results["returns"]
    parts.append(
        "available"
        if stock.status == "ok" and stock.value == "in stock"
        else "availability unknown"
    )
    parts.append(
        f"return period {returns.value}"
        if returns.status == "ok"
        else "return period unknown"
    )
    parts.append(
        f"delivery {delivery.value}"
        if delivery.status == "ok"
        else "delivery date unknown"
    )
    return "; ".join(parts)

async def main():
    await fragile_run()
    results = await resilient_run()
    print("Branch statuses:", {name: result.status for name, result in results.items()})
    print("Answer:", build_answer(results))
    print("Checkout allowed:", results["stock"].status == results["delivery"].status == "ok")

asyncio.run(main())

Запуск даёт:

Naive run: whole operation failed: delivery service unavailable
Branch statuses: {'stock': 'ok', 'delivery': 'error', 'returns': 'ok'}
Answer: available; return period 30 days; delivery date unknown
Checkout allowed: False

Пользователь видит остаток, правила возврата и сообщение о том, что дату доставки узнать не удалось. Оформление заказа заблокировано до успешного расчёта доставки. Так мы используем доступные данные, не выдавая сбой сервиса за отсутствие товара или доставки.

Что стоит проверить до подключения модели

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

После этого модель получает уже подготовленную картину: факты, пробелы и ограничения действия. Ей не приходится угадывать, означает ли пустая дата отсутствие доставки или сбой сервиса.

От трёх корутин к рабочему агенту

На курсе AI Agents PRO: LangGraph, AutoGen и LLMOps в продакшне вы соберёте агента из нескольких шагов и подключите внешние сервисы. Дальше разберёте, когда повторять неудачный запрос, как не выполнить одно действие дважды и куда отправлять задания, которые не удалось завершить. Итоговый сервис запускается через API; по журналам и проверочным сценариям можно проследить его работу.

Посмотреть программу AI Agents PRO. Для начала можно открыть урок про FSM, DAG и параллельные ветви агента.

Следующий шаг: AI-агенты в приложениях: состояние, инструменты и сбои. По смежным вопросам: Как остановить AI-агента, если пользователь отменил задачу, Устаревший ответ инструмента AI-агента: как не перезаписать новое решение.

Источник: официальная документация Python по asyncio, TaskGroup и wait.