Асинхронный Python-клиент Jev: пачки запросов, лимиты, повторы

Когда нужен асинхронный клиент

AsyncTypeSafeClient пригодится в трёх случаях:

• Пакетная обработка. Тысячи тикетов, писем или карточек товаров, и каждый документ нужно оценить.
• Асинхронный веб-сервер. Синхронный вызов заблокирует цикл событий, и сервер перестанет отвечать другим клиентам.
• Агенты. Когда агенту нужно несколько независимых решений по разным данным одновременно.

Если у вас один документ и много вопросов к нему, асинхронность не нужна: задайте все вопросы одним запросом. Базовая работа с SDK описана в статье о Jev на Python, здесь речь о пачках и лимитах.

Базовый вызов

import asyncio

from typesafe_sdk import AsyncTypeSafeClient, Choice, Noul


async def main() -> None:
    async with AsyncTypeSafeClient() as client:
        result = await client.system_one(
            state={"ticket": "С карты списали оплату дважды, верните деньги."},
            questions={
                "refund": Noul(instructions="Клиент просит вернуть деньги?"),
                "tone": Choice(
                    instructions="Какой тон у клиента?",
                    criteria={"calm": None, "frustrated": None, "angry": None},
                ),
            },
        )
        print(result.nouls["refund"].noul, result.choices["tone"].choice)
        print(result.model, result.request_id, result.usage.input_tokens)


asyncio.run(main())

Клиент принимает те же параметры, что и синхронный: api_key, base_url, model, retry, timeout. Ключ, адрес и модель можно задать переменными TYPESAFE_API_KEY, TYPESAFE_BASE_URL и TYPESAFE_DEFAULT_MODEL. Создавайте один клиент на всю работу: он держит сетевые соединения и закрывает их при выходе из async with или по вызову await client.aclose(). В веб-сервере клиент удобно создать при старте приложения и закрыть при остановке.

Правило 1: сначала вопросы, потом параллельность

Прежде чем распараллеливать запросы, проверьте, нельзя ли их объединить. Jev оценивает каждый вопрос независимо, поэтому ответ не зависит от соседних вопросов в запросе. Документ при этом оплачивается один раз.

В рецепте TypeSafe к статье Википедии о GDPR (около 54 000 символов) задали 13 вопросов двумя способами:

Способ	Запросов	Стоимость	Время
Все вопросы одним запросом	1	$0,000497	0,27 с
По вопросу на запрос	13	$0,006090	2,71 с

Итог: в 12,2 раза дешевле и в 10 раз быстрее, ответы совпали в пределах случайного разброса. Время во второй строке — сумма последовательных вызовов. Параллельные запросы сократят разрыв во времени, но 13-кратная разница в токенах останется. Подробнее о приёме — в статье о fan-out запросах.

Параллельность нужна для другого: когда документов много. Тогда один документ со всеми вопросами — один запрос, а запросы к разным документам идут одновременно.

Правило 2: темп важнее числа потоков

По документации TypeSafe, у Jev 1.13 два лимита: 1200 запросов в минуту и 250 000 токенов в секунду. Превышение любого даёт 429. Компания предупреждает, что лимиты сейчас меняются без уведомления.

Семафор ограничивает число одновременных запросов, но не их частоту. Пропускная способность примерно равна числу потоков, делённому на время ответа. При ответе за 0,3 секунды 20 одновременных запросов дают около 66 запросов в секунду. Это почти 4000 в минуту, втрое больше лимита. SDK повторит отклонённые запросы, но пачка пойдёт рывками. Надёжнее задать темп явно и дать воркерам разбирать очередь:

import asyncio
import time

from typesafe_sdk import (
    AsyncTypeSafeClient,
    Choice,
    Noul,
    RetryPolicy,
    TypeSafeAPIError,
    TypeSafeError,
)

QUESTIONS = {
    "complaint": Noul(instructions="Клиент жалуется на проблему?"),
    "topic": Choice(
        instructions="О чём обращение?",
        criteria={
            "billing": "Оплата, списания, возвраты",
            "delivery": "Доставка и сроки",
            "account": "Вход в аккаунт и настройки",
            "other": None,
        },
    ),
}


class RateLimiter:
    """Запускает не больше per_minute запросов в минуту через равные интервалы."""

    def __init__(self, per_minute: int) -> None:
        self.interval = 60 / per_minute
        self.next_start = time.monotonic()
        self.lock = asyncio.Lock()

    async def wait(self) -> None:
        async with self.lock:
            now = time.monotonic()
            self.next_start = max(self.next_start, now)
            delay = self.next_start - now
            self.next_start += self.interval
        await asyncio.sleep(delay)


async def worker(client, limiter, queue, results, failed):
    while True:
        ticket_id, text = await queue.get()
        try:
            await limiter.wait()
            r = await client.system_one(state={"ticket": text}, questions=QUESTIONS)
            results[ticket_id] = {
                "complaint": r.nouls["complaint"].noul,
                "topic": r.choices["topic"].choice,
                "confidence": r.choices["topic"].confidence,
                "tokens": r.usage.input_tokens or 0,
            }
        except TypeSafeAPIError as error:
            failed[ticket_id] = f"HTTP {error.status}, request_id={error.request_id}"
        except TypeSafeError as error:
            failed[ticket_id] = repr(error)
        finally:
            queue.task_done()


async def run(tickets: dict[str, str], workers: int = 10, per_minute: int = 1000):
    queue = asyncio.Queue()
    for item in tickets.items():
        queue.put_nowait(item)

    results, failed = {}, {}
    limiter = RateLimiter(per_minute)
    policy = RetryPolicy(max_retries=5, backoff_max=20.0, timeout=120.0)

    async with AsyncTypeSafeClient(retry=policy) as client:
        tasks = [
            asyncio.create_task(worker(client, limiter, queue, results, failed))
            for _ in range(workers)
        ]
        await queue.join()
        for task in tasks:
            task.cancel()
        await asyncio.gather(*tasks, return_exceptions=True)

    return results, failed

Что здесь важно:

• Очередь вместо gather на весь список. Воркеры берут задачи по одной, и в памяти не висят десятки тысяч корутин. Этот же каркас легко переделать на чтение из базы или файла.
• Темп с запасом. 1000 запросов в минуту оставляют место другим сервисам на том же ключе.
• Число воркеров считается как темп, умноженный на время ответа. 1000 запросов в минуту — это около 17 в секунду. При ответе до 0,5 секунды хватит 9 одновременных запросов, 10 воркеров дают небольшой запас.
• Ошибки не роняют пачку. Упавший документ попадает в failed с кодом и request_id, остальные обрабатываются дальше.

Длинные документы и лимит токенов

Второй лимит, 250 000 токенов в секунду, на коротких текстах незаметен. Тикет на 500 токенов при темпе 17 запросов в секунду даёт около 8500 токенов в секунду. Но если state близок к максимуму в 32 000 токенов, в лимит поместится меньше 8 запросов в секунду, то есть меньше 470 в минуту. Для длинных документов считайте per_minute так: 250 000 разделите на средний размер запроса в токенах и умножьте на 60. Получится предел запросов в минуту, от него оставьте запас.

У агрегаторов лимиты свои. При работе через Polza.AI, OpenRouter или Vercel сверяйтесь с их документацией.

Правило 3: повторы и таймауты под сценарий

Повторы настраиваются через RetryPolicy. Значения по умолчанию:

Параметр	По умолчанию	Смысл
max_retries	2	Повторов после первой попытки, 0 отключает повторы
backoff_initial	0,5 с	Первая пауза, дальше удваивается
backoff_max	5 с	Потолок паузы
backoff_jitter	0,25	Доля паузы, которая случайно вычитается
http_statuses	408, 429, 500–599	Какие коды повторять, 529 входит в диапазон
respect_retry_after	да	Учитывать заголовки Retry-After и retry-after-ms
timeout	30 с	Общий бюджет на вызов вместе с паузами

Обрывы соединения и таймауты тоже повторяются по умолчанию. Если следующая пауза не укладывается в бюджет, SDK прекращает попытки и бросает последнюю ошибку.

Политику можно задать клиенту или отдельному вызову, system_one(..., retry=RetryPolicy(...)). Для разных сценариев нужны разные настройки. Значения ниже — пример, подберите свои:

Сценарий	Пример настройки
Ночная пакетная обработка	RetryPolicy(max_retries=6, backoff_max=30.0, timeout=300.0)
Пользователь ждёт ответа	RetryPolicy(max_retries=1, timeout=2.0)
Тесты	RetryPolicy(max_retries=0)

Не путайте два таймаута. Параметр timeout клиента ограничивает одну HTTP-операцию: по умолчанию 10 секунд. RetryPolicy.timeout — бюджет на весь вызов со всеми повторами. Для больших документов первый стоит увеличить: в рецепте TypeSafe со статьёй на 54 000 символов клиенту задают timeout=120.0.

Можно добавить и свои правила повтора: параметр exceptions принимает дополнительные типы исключений, а predicate — функцию, которая решает по конкретной ошибке.

Правило 4: разбирайте упавшие документы по типу ошибки

После всех повторов документы в failed делятся на две группы:

• 429, 5xx, обрывы связи. Временные сбои. Поставьте документы в очередь повторно, позже или с меньшим темпом.
• 401 и 422. Ошибка в ключе или в запросе. Повтор не поможет: проверьте ключ, формат вопросов и размер state.

У HTTP-ошибок SDK, наследников TypeSafeAPIError, есть status, body и request_id. Отдельный тип TypeSafeRateLimitError содержит ещё retry_after_ms, время ожидания от сервера. Разбор всех кодов — в статье об ошибках API Jev.

Считайте токены и пишите версию модели

В каждом ответе есть usage.input_tokens. Сумма по пачке, умноженная на цену, даёт стоимость: при работе напрямую с TypeSafe это $0,042 за миллион входных токенов, выходные бесплатны. Прикинуть бюджет заранее поможет калькулятор стоимости, а из чего складывается цена, описано в статье о ценах Jev.

Логируйте и result.model. Алиас jev-latest переезжает на новые версии, а в поле model приходит конкретная версия, например jev-1.13.0. Если пороги подобраны под одну версию, после обновления их нужно перепроверить.