Пакетные запросы к LLM: очередь, стоимость и обработка ошибок

Представьте, что вашему LLM-сервису нужно обработать, допустим, 50 000 договоров или товаров. Можно пойти через синхронный путь — посылать запросы к LLM. Но такой путь через некоторое время упирается и в деньги, и в лимиты. И ограничения с проблемами растут в геометрической прогрессии от количества запросов.

Вот тут нам и помогает Batch API — она появилась не вчера, но стала отраслевым стандартом отправки батчей, групп запросов. OpenAI, Anthropic и Google дают асинхронную пакетную обработку со скидкой и окном завершения до 24 часов. То бишь вы собираете все задачи в один файл, отправляете его провайдеру и ждёте.

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

В этой статье разберём, из чего эта очередь состоит, как считать деньги и почему обрабатывать ошибки в батче — значит обрабатывать новые классы ошибок.

Синхронный API устроен просто: стоит отправить запрос — получил ответ. Пока запросов десяток или сотня, всё хорошо. Но чем ближе к тысячам, тем чаще вы упираетесь в TPM и RPM, а большие контекстные запросы начинают отваливаться ещё до того, как дело доходит до генерации. Батч идёт по отдельному пулу лимитов, и соответственно можно отправить 50 тысяч запросов одним файлом, не выедая себе синхронную квоту. Правда, у этого пула есть свой потолок, в виде того что если новый пакет выводит организацию за лимит enqueued-токенов, API возвращает token_limit_exceeded и просто отказывается создавать джобу.

Поэтому и нужно считать бюджет токенов до отправки, разбирать результат, а также решить, что делать с expired-запросами в батче, отправлять их или нет.

Кстати, если не хочется строить этот пайплайн с нуля — валидацию JSONL, семафор на enqueued-токены, разбор errored и expired, повторную отправку — имеет смысл посмотреть на OpenAI-совместимые шлюзы. Bothub, в частности, принимает тот же JSONL и те же параметры пакета. Архитектура полностью совместима со спецификацией OpenAI Batch API. Эндпоинт — https://openai.bothub.chat/v1, оплата российской картой, без VPN. По реферальной ссылке доступно 300 000 CAPS, так что можно потрогать батч на своих задачах ещё до того, как писать свой первый семафор. Заявленная экономия — 45% на генерациях.

Как устроена очередь

Когда отправляешь пакет в Batch API, поначалу кажется, что дальше можно просто закрыть ноутбук и ждать письма. Ну, отправил файл, провайдер сам разберётся. Но, увы, очередь устроена чуть сложнее, если не уметь, то можно и потерять деньги или данные. Давайте разберём.

Первый уровень — сам пакет, джоба. Вы формируете файл, где каждая строка это отдельный запрос с уникальным custom_id, и загружаете его через Files API. purpose всегда должен быть "batch". Дальше создаётся задача, и результаты приходят не в том порядке, в котором вы отправляли запросы. custom_id позволяет отсортировать вывод.

Второй уровень — пул enqueued-токенов. Вот тут начинается самое неприятное, потому что лимит считается не по файлам, а по общему объёму токенов во всех незавершённых пакетах сразу. У OpenAI это tier-зависимая история — чем выше tier, тем больше токенов можно держать в очереди одновременно. Если новый пакет выводит организацию за этот потолок, API возвращает token_limit_exceeded и просто отказывается создавать джобу.

Третий уровень — окно завершения. 24 часа это SLA, обещание, а не оценка. Внутри окна провайдер сам решает, когда запускать пакет, и обычно это 2–6 часов, потому что батчи стараются ставить в окно низкой загрузки GPU. Anthropic, например, пишет, что большинство пакетов завершается меньше чем за час, но максимальный срок всё равно 24.

У OpenAI незавершённые запросы отменяются, но за уже завершённые деньги списываются.

Вот пример структуры JSONL, который показывает сразу два запроса с разными задачами — перевод и генерация хокку:

{"custom_id": "task-translate-001", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "claude-opus-4.6", "messages": [{"role": "system", "content": "Переведи текст на французский язык."}, {"role": "user", "content": "Привет, как дела?"}], "max_tokens": 100}}
{"custom_id": "task-generate-003", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "claude-opus-4.6", "messages": [{"role": "user", "content": "Напиши короткое хокку про нейросети."}], "max_tokens": 50}}

Формат простой: custom_id, method, url и body с моделью и сообщениями. Выглядит почти как обычный запрос к Chat Completions, только упакованный в строку. Вот как это выглядит в Python, если использовать официальный SDK OpenAI с baseURL на Bothub:

from openai import OpenAI

client = OpenAI(
    api_key='ВАШ_API_КЛЮЧ_BOTHUB',
    base_url='https://openai.bothub.chat/v1'
)

file = client.files.create(
    file=open("my_tasks.jsonl", "rb"),
    purpose="batch"
)
print(file.id)  # сохраните этот ID

Это ровно тот же код, что и для OpenAI Batch API, — архитектура Bothub полностью совместима со спецификацией, поэтому SDK не нужно переписывать. Дальше создаётся сам batch:

batch = client.batches.create(
    input_file_id=file.id,
    endpoint="/v1/chat/completions",
    completion_window="24h"
)

completion_window пока всегда "24h", других вариантов нет. После создания остаётся поллинг — периодически дёргать статус, пока status не станет completed или failed. Для пакетов до 1000 строк опрашивать раз в 30 секунд, для крупных — раз в 2–5 минут. Если проверять статус каждые пять секунд, можно поймать 429 на сам polling, и это будет выглядеть как «пакет ещё не готов», хотя на самом деле вы просто упёрлись в rate limit на чтение.

И вот тут самое время вспомнить про третий уровень, окно. Потому что когда пакет завершается, вы скачиваете output-файл, и в нём может быть не всё. Часть строк будет с result.type = "succeeded", часть с "errored", часть с "expired". Если ваш пайплайн просто читает content из каждой строки и складывает в базу, строки с ошибками молча выпадут.

Решить можно примерно так:

results = client.files.content(batch.output_file_id)

for line in results.text.strip().split("\n"):
    row = json.loads(line)
    custom_id = row["custom_id"]

    if row["result"]["type"] == "succeeded":
        content = row["result"]["message"]["content"]
        save(custom_id, content)
    elif row["result"]["type"] == "errored":
        # разбираем ошибку и решаем, повторять или чинить запрос
        handle_error(custom_id, row["result"]["error"])
    elif row["result"]["type"] == "expired":
        # запрос не успел выполниться — только в новый пакет
        retry_later(custom_id, row["body"])

custom_id здесь работает как первичный ключ: по нему вы находите исходный запрос, чтобы понять, что именно чинить или переотправлять.

Для отслеживания подробных расходов по операции передавайте флаг include_usage=true в строке запроса:

batch = client.batches.retrieve("АЙДИ_БАТЧА")
print("Текущий статус:", batch.status)

if batch.status == "completed":
    print("ID готового файла:", batch.output_file_id)

А чтобы скачать результат, используйте идентификатор файла из поля output_file_id, чтобы получить итоговый документ.

import requests

def get_result():
    response = requests.get(
        "https://openai.bothub.chat/v1/files/АЙДИ_ФАЙЛА",
        headers={"Authorization": "Bearer ВАШ_API_КЛЮЧ_BOTHUB"}
    )

    link = response.text
    print(link)

get_result()

Можно отменить батч через функцию client.batches.cancel("АЙДИ_БАТЧА"), а получить список всех пакетов — через функцию batches = client.batches.list(limit=20).

Обработка ошибок

В случае батчей лучше сразу задуматься об обработке ошибок, тут она разительно отличается от синхронных запросов.

Ошибки до создания пакета

Пакет не создаётся или сразу падает в failed. Причины скучные — лишняя запятая в JSONL, неэкранированный символ, забытый custom_id, неверный endpoint, файл больше лимита. У Anthropic, кстати, параметры запроса проверяются только после завершения всего пакета — то бишь вы можете отправить 100 тысяч строк, прождать час и узнать, что в трёх из них кривой max_tokens. У OpenAI формат проверяется до запуска.

Нужно валидировать JSONL локально, до загрузки. Это дешевле, чем ловить failed через API. Минимальная проверка — распарсить каждую строку, убедиться, что custom_id уникален и не пустой, а body содержит model и messages.

import json

seen = set()
with open("my_tasks.jsonl") as f:
    for i, line in enumerate(f, 1):
        row = json.loads(line)  # упадёт на кривом JSON — и хорошо
        cid = row.get("custom_id")
        if not cid or cid in seen:
            raise ValueError(f"строка {i}: пустой или повторный custom_id")
        seen.add(cid)

Можно воспользоваться Pydantic-моделями для валидации DTO.

Ошибки на уровне очереди

Пакет не стартует или стартует частично, token_limit_exceeded. Вы отправили новую джобу, а он выводит организацию за потолок enqueued-токенов, и API отказывается его создавать.

Считать бюджет токенов до отправки. Если бюджет не влезает — ставить пакеты в собственную очередь на стороне приложения или ждать, пока завершатся предыдущие.

import time

def submit_with_backoff(client, file_id, max_attempts=5):
    delay = 30
    for attempt in range(max_attempts):
        try:
            return client.batches.create(
                input_file_id=file_id,
                endpoint="/v1/chat/completions",
                completion_window="24h"
            )
        except Exception as e:
            if "token_limit_exceeded" not in str(e):
                raise
            if attempt == max_attempts - 1:
                raise
            time.sleep(delay)
            delay *= 2  # 30 → 60 → 120 → ...

Ошибки внутри пакета

Самая коварная категория. Пакет completed, вы скачиваете output, а там часть строк с result.type = "succeeded", часть с "errored", часть с "expired". У Anthropic счётчики errored, canceled и expired в request_counts показывают ноль, пока обработка всего пакета не завершится — то бишь до последней строки вы не узнаете, сколько именно провалилось.

У OpenAI ошибки пишутся в отдельный error-файл, у Anthropic — в тот же results_url, но с type: "errored" и объектом ошибки внутри.

В таком случае можно парсить каждую (или не каждую, не стоит забывать про оптимизацию) строку и раскладывать по трём корзинам. invalid_request — чинить запрос и переотправлять только его; server error — повторять с задержкой; expired — только в новый пакет, потому что окно уже закрыто. И не переотправлять весь пакет из-за трёх упавших строк: успешные вы уже оплатили.

import json

def triage_results(raw_text):
    done, retry, fix = [], [], []
    for line in raw_text.strip().split("\n"):
        row = json.loads(line)
        cid = row["custom_id"]
        rtype = row["result"]["type"]

        if rtype == "succeeded":
            done.append((cid, row["result"]["message"]["content"]))
        elif rtype == "errored":
            err_type = row["result"]["error"]["type"]
            if err_type == "invalid_request_error":
                fix.append((cid, row["result"]["error"]))
            else:  # rate_limit, timeout, api_error и прочее
                retry.append(cid)
        elif rtype == "expired":
            retry.append(cid)
    return done, retry, fix

custom_id здесь работает как первичный ключ.

Ошибки опроса статуса

Отдельная ловушка, о которой в туториалах молчат. Если дёргать статус пакета каждые пять секунд, можно поймать 429 — и наивный retry-цикл решит, что «пакет ещё не готов», хотя на самом деле вы упёрлись в rate limit на сам polling. Выглядит одинаково, лечится по-разному.

Тут стоит заводить экспоненциальный backoff на polling. Для пакетов до 1000 строк разумно начинать с 30 секунд, для крупных — с 2–5 минут. И логировать created_at пакета, коли он висит дольше нескольких часов, стоит проверить, не застрял ли он в validating.

Заключение

Батч — это другой контракт. Вы платите меньше, но получаете не мгновенный ответ, а обещание вернуть результат в течение суток. Вместе с этим обещанием приходит обязанность самому разбирать, что вернулось: succeeded, errored или expired. Синхронный API такой обязанности не накладывает — там ошибка видна сразу.

Очередь — это ресурс с лимитом. Enqueued-токены, количество одновременных пакетов, окно завершения — вещи которые нужно изучить заранее.

50% скидки — это потолок. Настоящая цифра зависит от того, сколько запросов ушло в expired, сколько вы переотправили и был ли кеш системного промпта. В хорошем сценарии экономия глубже, чем 50%, потому что cached-input тоже дешевеет. В плохом — вы платите дважды за одни и те же строки.