Python: параллельные запросы к Claude через AsyncOpenAI

Классифицировать тысячу обращений или извлечь данные из сотни документов по одному запросу за раз — долго: каждый ответ занимает секунды, и почти всё это время скрипт просто ждёт. AsyncOpenAI с asyncio отправляет параллельные запросы, а семафор не даёт отправить их слишком много сразу. Ниже — рабочие примеры для Zukko: базовая схема, повторы при ошибках, остановка при пустом балансе, подсчёт стоимости и обработка CSV.

Базовый пример: параллельные запросы с семафором

Зависимость одна: pip install openai. Ключ — в переменной окружения ZUKKO_API_KEY.

import asyncio
import os

from openai import AsyncOpenAI

MODEL = "claude-sonnet-5"

client = AsyncOpenAI(
    base_url="https://api.zukko.pro/v1",
    api_key=os.environ["ZUKKO_API_KEY"],
)


async def classify(text: str, semaphore: asyncio.Semaphore) -> str:

asyncio.gather запускает все корутины сразу, но внутрь async with semaphore одновременно попадают только пять. Остальные ждут, пока освободится слот. Результаты возвращаются в том же порядке, что и тексты, даже если ответы пришли вперемешку.

Повторы при 429 и 503

Официальный SDK уже повторяет запросы сам: по умолчанию до двух раз с растущей паузой при кодах 408, 409, 429 и 5xx, а также при сетевых сбоях. Для пакетной обработки число попыток стоит увеличить и задать таймаут:

client = AsyncOpenAI(
    base_url="https://api.zukko.pro/v1",
    api_key=os.environ["ZUKKO_API_KEY"],
    max_retries=5,
    timeout=60.0,
)

Свой цикл повторов поверх этого обычно не нужен. А вот ошибки, которые повторять бессмысленно, нужно обработать явно:

• 401 — неверный ключ;
• 402 — на балансе не хватает денег, все следующие запросы тоже не пройдут;
• 403 — модель не входит в список разрешённых для ключа;
• 413 — текст слишком большой, его нужно сократить или разбить.

Остановка всей обработки при 402

Если баланс закончился, нет смысла отправлять ещё тысячу запросов и собирать тысячу одинаковых ошибок. asyncio.TaskGroup (Python 3.11+) отменяет все оставшиеся задачи, как только одна из них падает:

import asyncio
import os

from openai import AsyncOpenAI, APIStatusError

client = AsyncOpenAI(
    base_url="https://api.zukko.pro/v1",
    api_key=os.environ["ZUKKO_API_KEY"],
    max_retries=5,
)


class StopBatch(Exception):
    pass

Ошибки отдельных текстов (например, 413 на слишком длинном документе) не останавливают пакет — они попадают в результат как строка ERROR ..., и их можно перезапустить позже.

Подсчёт стоимости по usage

В каждом ответе есть response.usage с числом входных (prompt_tokens) и выходных (completion_tokens) токенов. Цены Zukko за 1M токенов: claude-sonnet-5 — $0.3 / $0.3, gpt-5.6-luna — $0.1 / $0.1, glm-5.3-flash — $0.04 / $0.04.

PRICES = {  # $ за 1M токенов: (вход, выход)
    "claude-sonnet-5": (0.3, 0.3),
    "gpt-5.6-luna": (0.1, 0.1),
    "glm-5.3-flash": (0.04, 0.04),
}


def cost_usd(model: str, usage) -> float:
    price_in, price_out = PRICES[model]
    return (usage.prompt_tokens * price_in + usage.completion_tokens * price_out) / 1_000_000

Пример оценки заранее: 10 000 текстов по 200 входных токенов и 5 выходных — это 2 000 000 входных и 50 000 выходных токенов.

Цены могут меняться — сверяйте их со страницей «Модели» в кабинете. Какая модель даст приемлемую точность на ваших текстах, проверьте сами: разметьте вручную 50–100 примеров и сравните ответы двух-трёх моделей с разметкой.

Точную стоимость каждого запроса шлюз возвращает в заголовке X-Zukko-Cost-Micros (в микродолларах). Заголовки доступны через with_raw_response:

raw = await client.chat.completions.with_raw_response.create(
    model="claude-sonnet-5",
    messages=[{"role": "user", "content": "Тема одним словом: акции выросли"}],
    max_tokens=10,
)
completion = raw.parse()
micros = int(raw.headers.get("X-Zukko-Cost-Micros", "0"))
print(completion.choices[0].message.content, f"${micros / 1_000_000:.6f}")

Подробнее о заголовках и лимитах ключа — в разделе о лимитах ключей и кэше↗.

Обработка CSV и сохранение результатов

Типичная задача: файл с колонкой text, нужно добавить колонку с категорией.

import asyncio
import csv
import os

from openai import AsyncOpenAI, APIStatusError, APIConnectionError

client = AsyncOpenAI(
    base_url="https://api.zukko.pro/v1",
    api_key=os.environ["ZUKKO_API_KEY"],
    max_retries=5,
)


async def process_row(row: dict, semaphore: asyncio.Semaphore) -> dict:

Каждая строка обрабатывается независимо: сетевой сбой или ошибка на одном тексте не роняют весь файл. Строки со статусом ERROR легко отфильтровать и прогнать повторно.

Как не упереться в 429

1. Ограничьте параллелизм семафором. Начните с 5 и поднимайте, пока не появятся ответы 429.
2. Оставьте повторы SDK (max_retries): короткие всплески 429 он переживёт сам.
3. Проверьте лимиты ключа в кабинете↗: если дневной лимит расходов исчерпан, ключ тоже отвечает 429, и повторы не помогут до следующих суток.

Частые вопросы

Какой размер семафора выбрать?

Начните с 5. Если ответов 429 нет, а обработка идёт медленно, попробуйте 10. Если 429 появились, вернитесь к меньшему значению. Оптимум зависит от длины запросов и нагрузки, поэтому подбирается опытным путём.

Как прервать обработку по Ctrl+C?

asyncio.run при Ctrl+C сам отменяет главную задачу, а вместе с ней — все ожидающие запросы. Чтобы не потерять уже готовые результаты, сохраняйте их пакетами (см. следующий вопрос).

Как сохранять прогресс, если обработка прервётся?

Делите список на пакеты и записывайте результат после каждого:

async def run_in_batches(texts: list[str], batch_size: int = 100):
    semaphore = asyncio.Semaphore(5)
    for start in range(0, len(texts), batch_size):
        batch = texts[start:start + batch_size]
        results = await asyncio.gather(*(classify(t, semaphore) for t in batch))
        save_results(start, results)  # ваша функция: файл или база
        print(f"Готово {start + len(batch)} из {len(texts)}")

При перезапуске начинайте с первого несохранённого пакета.

Можно ли то же самое сделать синхронно?

Можно через OpenAI и concurrent.futures.ThreadPoolExecutor, но на сотнях параллельных запросов асинхронный клиент проще и экономнее по ресурсам. Оба варианта клиента описаны в документации по Python↗.