provod.AI - агенты и API в РФ

Async-вызовы и Batch API в LLM: как сэкономить до 50% и ускорить обработку

Async-вызовы и Batch API в LLM: как сэкономить до 50% и ускорить обработку

Когда у вас 10 запросов в LLM — синхронный for нормально. Когда 1000 — он становится бутылочным горлышком, и пайплайн крутится часами. Когда 100 000 — обычный API становится дорогим, и расходы на токены съедают юнит-экономику. Два классических решения: async-параллельность (asyncio + aiohttp для 50–500 запросов в секунду) и Batch API (off-line режим со скидкой 50% на input/output).

Этот гайд — рабочий код обоих паттернов через единый шлюз provod.ai (Claude Opus 4.8, GPT-5.5, Gemini 3.1 Pro, DeepSeek V4 Pro), расчёт реальной экономии на типовых сценариях, паттерны очередей и retry для production, и чёткие правила «когда что брать». оплата в рублях по договору, полный пакет закрывающих документов, цены в рублях по курсу ЦБ.

TL;DR — два режима, две экономии

Async (через asyncio.gather + AsyncOpenAI):

  • Real-time, ответ за секунды
  • Throughput до 500 RPS на ключ
  • Та же цена, что у обычного API
  • Когда: UI, агенты, real-time чаты

Batch API (через client.batches.create):

  • Offline, SLA до 24 часов (обычно час-два)
  • Скидка 50% на input и output
  • Лимита на размер нет (миллионы запросов в одном файле)
  • Когда: разметка, классификация архива, summary всей базы

Производительный production-стек использует оба.

Часть 1: Async-вызовы через asyncio

Базовый паттерн — параллельное выполнение N запросов через asyncio.gather:

import asyncio from openai import AsyncOpenAI client = AsyncOpenAI( api_key="sk-provod-...", base_url="https://api.provod.ai/v1", ) async def call_one(prompt: str) -> str: response = await client.chat.completions.create( model="claude-sonnet-4-6", messages=[{"role": "user", "content": prompt}], ) return response.choices[0].message.content async def main: prompts = [f"Расскажи короткий факт про число {i}" for i in range(100)] results = await asyncio.gather(*[call_one(p) for p in prompts]) for p, r in zip(prompts, results): print(p, "→", r[:80]) asyncio.run(main)

100 запросов в секунду — обычно реально (упирается в rate limit ключа, не в SDK). Сравнение со синхронным циклом:

  • Сценарий: Sync for (100 запросов) • Время: ~180 сек • Throughput: 0.5 RPS
  • Сценарий: asyncio.gather(100) • Время: ~3.5 сек • Throughput: ~28 RPS
  • Сценарий: asyncio.gather + Semaphore(20) • Время: ~6 сек • Throughput: ~17 RPS

Async ускоряет в 30–50 раз на типовом latency 1–2 секунды на запрос. Но без контроля параллельности вы быстро упрётесь в rate limit.

Async-вызовы и Batch API в LLM: как сэкономить до 50% и ускорить обработку

Rate limit через Semaphore

Если параметр N в gather слишком большой — ловите 429 от API. Решение — asyncio.Semaphore:

async def call_with_semaphore(sem: asyncio.Semaphore, prompt: str) -> str: async with sem: return await call_one(prompt) async def main: sem = asyncio.Semaphore(20) # максимум 20 параллельных запросов prompts = [f"Запрос {i}" for i in range(1000)] results = await asyncio.gather(*[call_with_semaphore(sem, p) for p in prompts])

Semaphore(20) означает: всегда не больше 20 параллельных, как только одна задача завершилась — следующая стартует. Это даёт постоянную нагрузку без всплесков.

Производительный лимит на ключ через provod.ai — обычно 600 RPM (10 RPS), при росте трафика — поднимается через дашборд. Semaphore — это контроль на вашей стороне, чтобы не пропускать 429 в код приложения.

Retry с exponential backoff

Даже с Semaphore периодически прилетят 429 (всплески), 503 (временные сбои API), таймауты. Стандарт — tenacity:

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type from openai import RateLimitError, APIConnectionError, APITimeoutError @retry( stop=stop_after_attempt(5), wait=wait_exponential(multiplier=1, min=2, max=30), retry=retry_if_exception_type((RateLimitError, APIConnectionError, APITimeoutError)), reraise=True, ) async def call_with_retry(prompt: str) -> str: response = await client.chat.completions.create( model="claude-sonnet-4-6", messages=[{"role": "user", "content": prompt}], timeout=120, ) return response.choices[0].message.content

Что важно:

  • 5 attempts — больше обычно бесполезно, проблема не разрешится.
  • Exponential backoff 2 → 4 → 8 → 16 → 30 сек — даёт API время восстановиться.
  • Только на retry-able ошибках — 400 (bad request) или 401 (auth) не retry'ить, это ваши ошибки.
  • reraise=True — после 5 неудачных попыток ошибка пробрасывается в код, не глотается.

Полный production-шаблон async

Собираем всё вместе — паттерн для обработки тысяч задач:

import asyncio from openai import AsyncOpenAI from tenacity import retry, stop_after_attempt, wait_exponential client = AsyncOpenAI( api_key="sk-provod-...", base_url="https://api.provod.ai/v1", ) @retry(stop=stop_after_attempt(5), wait=wait_exponential(min=2, max=30)) async def classify(text: str, sem: asyncio.Semaphore) -> dict: async with sem: response = await client.chat.completions.create( model="gpt-5-4", messages=[ {"role": "system", "content": "Классифицируй текст: positive/neutral/negative."}, {"role": "user", "content": text}, ], temperature=0, timeout=60, ) return { "text": text[:100], "label": response.choices[0].message.content.strip.lower, "tokens": response.usage.total_tokens, } async def process_dataset(texts: list[str], concurrency: int = 20) -> list[dict]: sem = asyncio.Semaphore(concurrency) results = await asyncio.gather( *[classify(t, sem) for t in texts], return_exceptions=True, ) # отделяем успехи от ошибок successes = [r for r in results if not isinstance(r, Exception)] failures = [r for r in results if isinstance(r, Exception)] print(f"Успешно: {len(successes)}, ошибок: {len(failures)}") return successes asyncio.run(process_dataset(my_texts, concurrency=30))

return_exceptions=True — критично: одна упавшая задача не валит весь батч. Анализируете ошибки отдельно, ретраите при необходимости.

Часть 2: Batch API — −50% за оффлайн режим

Async помогает с throughput, но цену за токены не меняет. Batch API даёт скидку 50% на input и output, если согласны ждать до 24 часов. Эта статья — часть pillar-гида: полный технический гид по LLM API на Python — токены, function calling, streaming, RAG, batch.

Архитектура: вы готовите JSONL-файл с тысячами запросов, загружаете на сервер, ждёте окончания, скачиваете JSONL с результатами.

Шаг 1. Готовим JSONL

Каждая строка — один запрос:

import json def make_batch_file(texts: list[str], model: str, path: str): with open(path, "w", encoding="utf-8") as f: for i, text in enumerate(texts): request = { "custom_id": f"task-{i}", "method": "POST", "url": "/v1/chat/completions", "body": { "model": model, "messages": [ {"role": "system", "content": "Классифицируй текст."}, {"role": "user", "content": text}, ], "temperature": 0, }, } f.write(json.dumps(request, ensure_ascii=False) + "\n") make_batch_file(texts, "gpt-5-4", "/tmp/batch.jsonl")

custom_id — ваш идентификатор для сопоставления результата с исходным запросом. Обычно — id записи в БД или индекс.

Шаг 2. Загружаем и стартуем batch

from openai import OpenAI client = OpenAI( api_key="sk-provod-...", base_url="https://api.provod.ai/v1", ) # загружаем файл upload = client.files.create(file=open("/tmp/batch.jsonl", "rb"), purpose="batch") print(f"File ID: {upload.id}") # стартуем batch batch = client.batches.create( input_file_id=upload.id, endpoint="/v1/chat/completions", completion_window="24h", metadata={"task": "classify_tickets", "version": "v3"}, ) print(f"Batch ID: {batch.id}, status: {batch.status}")

completion_window="24h" — SLA, обычно реально завершается за час-два.

Шаг 3. Опрос статуса и скачивание результата

import time def wait_for_batch(batch_id: str, poll_interval: int = 60) -> dict: while True: batch = client.batches.retrieve(batch_id) print(f"Status: {batch.status}, completed: {batch.request_counts.completed}/{batch.request_counts.total}") if batch.status in ("completed", "failed", "expired", "cancelled"): return batch time.sleep(poll_interval) batch = wait_for_batch(batch.id) if batch.status == "completed": output = client.files.content(batch.output_file_id) with open("/tmp/batch_results.jsonl", "wb") as f: f.write(output.content)

Шаг 4. Парсим результаты

results = {} with open("/tmp/batch_results.jsonl") as f: for line in f: record = json.loads(line) custom_id = record["custom_id"] if record.get("error"): results[custom_id] = {"error": record["error"]} else: answer = record["response"]["body"]["choices"][0]["message"]["content"] results[custom_id] = {"answer": answer} # теперь по custom_id сопоставляете с исходными данными

Подробности API — в официальной документации Batch у OpenAI и в Message Batches у Anthropic.

Async-вызовы и Batch API в LLM: как сэкономить до 50% и ускорить обработку

Экономика: реальные числа

Пример сценария: классификация 100 000 тикетов, средний 1500 input + 500 output токенов.

Через обычный async API

  • Модель: Claude Opus 4.7 • Цена: 350/1790 ₽ • Стоимость: 142 000 ₽
  • Модель: Claude Sonnet 4.6 • Цена: 210/1070 ₽ • Стоимость: 85 000 ₽
  • Модель: GPT-5.5 • Цена: 350/2150 ₽ • Стоимость: 160 000 ₽
  • Модель: GPT-5.4 • Цена: 170/1070 ₽ • Стоимость: 78 500 ₽
  • Модель: Gemini 3.1 Pro • Цена: 140/860 ₽ • Стоимость: 64 000 ₽
  • Модель: DeepSeek V4 Pro • Цена: 30/60 ₽ • Стоимость: 7 500 ₽

Через Batch API (−50%)

  • Модель: Claude Opus 4.7 • Цена batch: 175/895 ₽ • Стоимость: 71 000 ₽ • Экономия: 71 000 ₽
  • Модель: Claude Sonnet 4.6 • Цена batch: 105/535 ₽ • Стоимость: 42 500 ₽ • Экономия: 42 500 ₽
  • Модель: GPT-5.5 • Цена batch: 175/1075 ₽ • Стоимость: 80 000 ₽ • Экономия: 80 000 ₽
  • Модель: GPT-5.4 • Цена batch: 85/535 ₽ • Стоимость: 39 250 ₽ • Экономия: 39 250 ₽

Если у вас есть оффлайн-процессинг — Batch это просто бесплатные −50% к расходам. На больших объёмах это миллионы рублей экономии в год.

Async-вызовы и Batch API в LLM: как сэкономить до 50% и ускорить обработку

Когда брать async, когда batch — дерево решений

Задача поступила: ┌─────────────────────────────────┐ │ Нужен ответ за секунды? │ └────┬────────────────────┬───────┘ │ да │ нет ▼ ▼ ┌──────┐ ┌─────────────────┐ │ async│ │ Объём > 1000 │ │ (UI, │ │ запросов? │ │ агент,│ └─────┬──────┬────┘ │ чат) │ │ да │ нет └──────┘ ▼ ▼ ┌──────┐ ┌──────┐ │batch │ │async │ │−50% │ │быстро│ └──────┘ └──────┘

Правила:

  • Real-time UI (чат, ассистент) → async + streaming
  • Агент с tool calls → async (нужно несколько roundtrips)
  • Embedding большой базы → batch (вместо 100K параллельных async)
  • Ночная переклассификация → batch
  • A/B тест промтов на датасете → batch
  • Если хочется и того, и того → async с фоновой очередью + batch для архивных задач

Production-паттерн: async + batch в одной системе

Архитектура зрелого LLM-сервиса:

# real-time эндпоинт — async @app.post("/chat") async def chat(req: ChatRequest): return await async_chat_via_streaming(req) # background задачи через очередь (Celery/RQ/Dramatiq) @celery.task def reclassify_all_tickets: tickets = db.query("SELECT id, text FROM tickets WHERE status='new'").all batch_id = submit_to_batch(tickets, model="claude-sonnet-4-6") schedule_check_batch(batch_id, after_minutes=60) @celery.task def check_batch(batch_id: str): batch = client.batches.retrieve(batch_id) if batch.status == "completed": results = download_and_parse(batch.output_file_id) save_to_db(results) elif batch.status in ("in_progress", "validating"): # ещё не готов — перепланировать schedule_check_batch(batch_id, after_minutes=30) else: alert_team(f"Batch {batch_id} failed: {batch.status}")

Та же база токенов, тот же ключ provod.ai, один счёт от юр.лица. Через единый шлюз биллинг един — async расходы и batch расходы видны в одном дашборде.

Async-вызовы и Batch API в LLM: как сэкономить до 50% и ускорить обработку

Распространённые ошибки

1. Использовать sync for на 1000+ запросов. На latency 1.5 сек — это 25 минут вместо 30 секунд через async.

2. Делать async без Semaphore. Получаете шторм 429-х, ваши попытки retry усугубляют ситуацию.

3. Использовать обычный API там, где нужен Batch. На 100K запросов это лишние 40-80K ₽ за прогон. На 10 прогонов в месяц — почти 1М ₽ в год.

4. Не парсить ошибки batch results. Каждая запись может быть либо response, либо error. Если игнорировать error — потеряете 0.1-1% записей молча.

5. Не использовать custom_id осмысленно. Если это просто индекс — после реорганизации файла теряете связь с исходными данными. Используйте id из вашей БД.

6. Запускать batch без оценки стоимости заранее. Считайте ожидаемый input/output через tokenizer («Как считать токены в LLM»), умножайте на batch-ставки, сверяйте с бюджетом — до запуска.

Оплата и закрывающие документы

Async и Batch — это одни и те же модели через тот же шлюз provod.ai. Юрлицо-исполнитель — российское юр.лицо, резидент РФ. Сервисная комиссия 5% берётся только при пополнении баланса, на токены наценки нет. Batch-скидка 50% видна непосредственно в дашборде по статье «Batch usage». Полный пакет закрывающих документов (договор-оферта, счёт на оплату, акт оказанных услуг, счёт-фактура, УПД) приходит через ЭДО — Диадок, СБИС, Контур. Подробнее — на странице «Тарифы».

Что дальше

Async-вызовы — это переход с 0.5 RPS до 30+ на одном ключе и Semaphore'е. Batch API — это −50% к цене токенов за готовность подождать час-два. Зрелый production-стек использует оба: real-time async для пользовательских интерфейсов, batch для оффлайн-обработки и архивных задач. На объёме 100K+ запросов в месяц экономия от Batch измеряется сотнями тысяч рублей. Полезные следующие шаги: «Function calling и tool use» для async-агентов, «Embeddings и векторный поиск» для batch-индексации больших баз, «Streaming LLM-ответов» для real-time UI. Если нужно прикинуть стоимость на вашем трафике или подключить ключ через юрлицо — напишите команде provod.ai в Telegram.

📚 Главный гайд по теме: Лучшая нейросеть 2026: какую LLM выбрать под задачу — связанные материалы и обзор всей категории.

FAQ

В чём разница между async-вызовами и Batch API?

Async — параллельное выполнение N независимых запросов через обычный endpoint, real-time. Batch — режим со скидкой 50%: загружаете JSONL с тысячами запросов, ждёте час-сутки, скачиваете результаты. Async для UI и агентов, Batch для оффлайн обработки.

Какие модели поддерживают Batch API в 2026?

OpenAI Batch — для GPT-5.5, GPT-5.4 и большинства моделей. Anthropic Message Batches — для Claude Opus 4.8 и Sonnet 4.6. Google Vertex Batch Prediction — для Gemini 3.1 Pro и Flash. Везде скидка 50%, SLA до 24ч.

Когда брать async, когда Batch?

Async — real-time UI, агенты с tool calls, 10–1000 параллельных запросов с результатом за секунды. Batch — объём от 1000, латентность не критична, нужно −50%. Типичные batch-юзкейсы: разметка датасета, классификация архива, embeddings корпуса.

Как избежать rate limit при async?

asyncio.Semaphore(N), где N — ваш RPS-лимит. tenacity-retry с exponential backoff на 429. Глобальный timeout. Опциональный jitter. Лимит через provod.ai — 600 RPM на ключ, поднимается через дашборд.

Сколько стоит 100K документов через Batch?

100K документов × 2000 токенов на Claude Sonnet 4.6 обычным API — 85 000 ₽. Через Batch со скидкой 50% — 42 500 ₽. Экономия 42 500 ₽ за один прогон. На GPT-5.5 экономия 80 000 ₽.

Можно ли смешивать async и Batch?

Да, это правильный production-паттерн. Real-time запросы через async/streaming, фоновые задачи через Batch. Один ключ provod.ai, один счёт от российское юр.лицо, единый дашборд расхода.

Async-вызовы и Batch API в LLM: как сэкономить до 50% и ускорить обработку

provod.ai — российский LLM API-агрегатор

Один OpenAI-совместимый endpoint ко всем флагманам: OpenAI (GPT-5.5, GPT-5.4), Anthropic (Claude Opus 4.8, Sonnet 4.6), Google (Gemini 3.1 Pro, 3.5 Flash), DeepSeek V4 Pro, Qwen 3.6 Plus.

Цены 1-в-1 с провайдером по курсу ЦБ — без наценки на токены. Оплата в рублях по договору, полный пакет закрывающих документов (договор-оферта, счёт, акт, счёт-фактура, УПД 5.03 через ЭДО). Без VPN — легальный B2B-сервис в России.

Если статья была полезной — попробуйте provod.ai: главная страница · каталог моделей · документация