Работа с LLM из Python: клиент, повторы и контроль стоимости

Задача: дёргать LLM-модель из бэкенда, где на неё завязана бизнес-логика. Наивный вызов client.chat.completions.create(...) в одну строку работает ровно до первого 429 Too Many Requests, первого сетевого таймаута или первого счёта на несколько тысяч рублей за месяц, который вы не ждали. LLM-провайдер — это внешний HTTP-сервис с лимитами, нестабильной задержкой и поминутной тарификацией по токенам, и относиться к нему надо как к базе данных под нагрузкой, а не как к локальной функции.

Разберём три вещи, которые отличают прод-клиент от прототипа: как устроен вызов и где он падает, как правильно повторять неудачные запросы, и как считать деньги до того, как придёт счёт. Примеры — на официальном SDK openai (версия 1.x, API стабилизировался начиная с 1.0). Логика повторов и подсчёта переносится на любого провайдера с OpenAI-совместимым API.

Что очевидное решение делает не так

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

• Ограничения скорости. Провайдер возвращает 429, когда вы превысили лимит запросов в минуту (RPM) или токенов в минуту (TPM). Без повтора запрос просто теряется.
• Сетевые сбои и таймауты. Генерация длинного ответа может занять десятки секунд. Дефолтный таймаут HTTP-клиента способен оборвать запрос на середине, и вы заплатите за prompt-токены, не получив ответа.
• Непредсказуемая стоимость. Один запрос с большим контекстом и длинным ответом стоит в сотни раз дороже короткого. Без учёта токенов вы узнаёте цену постфактум.

Термин: токены

Токен — единица, на которую модель разбивает текст. Примерно 3–4 символа русского текста или 0,75 английского слова. Тарификация идёт отдельно за входные (prompt) и выходные (completion) токены, причём выходные обычно дороже. Точное разбиение зависит от токенизатора конкретной модели.

Базовый клиент с таймаутом

Начнём с явной настройки таймаута и клиента как переиспользуемого объекта — создавать новый OpenAI() на каждый запрос дорого из-за пересоздания HTTP-пула.

import os
from openai import OpenAI

# Клиент создаётся один раз на процесс.
# Ключ читаем из переменной окружения, не из кода.
client = OpenAI(
    api_key=os.environ["OPENAI_API_KEY"],
    timeout=30.0,   # секунд на весь запрос
    max_retries=0,  # повторы отключаем — сделаем свои
)

def ask(prompt: str, model: str = "gpt-4o-mini") -> str:
    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
    )
    return response.choices[0].message.content

Мы выставили max_retries=0 намеренно. SDK умеет повторять сам, но его логика — чёрный ящик: вы не контролируете, что именно повторяется и с какой задержкой. Для прода лучше собрать повторы явно.

Повторы: что повторять, а что нет

Главная ошибка — повторять всё подряд. Повтор ошибки 400 Bad Request (кривой запрос) или 401 (неверный ключ) бесполезен: результат не изменится, а вы потратите время и, возможно, деньги. Повторять стоит только временные сбои.

Экспоненциальная задержка с джиттером

Повторять сразу — плохо: если провайдер перегружен, вы добавите ему нагрузки. Задержка должна расти (1с, 2с, 4с…), а к ней добавляется случайный джиттер, чтобы сотня ваших воркеров не долбила API синхронно после общего сбоя.

import time
import random
from openai import (
    RateLimitError,
    APITimeoutError,
    APIConnectionError,
    InternalServerError,
)

# Ошибки, которые имеет смысл повторять
RETRYABLE = (
    RateLimitError,
    APITimeoutError,
    APIConnectionError,
    InternalServerError,
)

def ask_with_retry(prompt: str, model: str = "gpt-4o-mini",
                   max_attempts: int = 5) -> str:
    for attempt in range(max_attempts):
        try:
            response = client.chat.completions.create(
                model=model,
                messages=[{"role": "user", "content": prompt}],
            )
            return response.choices[0].message.content
        except RETRYABLE as err:
            if attempt == max_attempts - 1:
                raise  # попытки кончились — пробрасываем ошибку
            # base 2^attempt секунд плюс случайный джиттер до 1с
            delay = 2 ** attempt + random.uniform(0, 1)
            time.sleep(delay)
    raise RuntimeError("недостижимо")

Если провайдер вернул заголовок Retry-After, он приоритетнее вашей формулы — там указано, через сколько секунд лимит сбросится. У ошибок SDK доступ к заголовкам ответа есть через атрибут исключения; проверяйте наличие у вашей версии SDK в документации, имена атрибутов между минорными версиями менялись.

Повтор без верхнего предела попыток — это не отказоустойчивость, а способ незаметно сжечь бюджет и заблокировать воркер на минуты. Всегда ставьте max_attempts и общий дедлайн на операцию.

Считаем стоимость по токенам

Ответ API содержит поле usage с фактическим числом токенов. Это единственный точный источник — не оценка, а то, за что вы заплатите.

# Цены — в долларах за 1 млн токенов.
# АКТУАЛЬНЫЕ ставки сверяйте на странице тарифов провайдера,
# они меняются и зависят от модели.
PRICING = {
    "gpt-4o-mini": {"input": 0.15, "output": 0.60},
}

def ask_with_cost(prompt: str, model: str = "gpt-4o-mini"):
    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
    )
    usage = response.usage
    price = PRICING[model]
    cost = (
        usage.prompt_tokens / 1_000_000 * price["input"]
        + usage.completion_tokens / 1_000_000 * price["output"]
    )
    return {
        "text": response.choices[0].message.content,
        "prompt_tokens": usage.prompt_tokens,
        "completion_tokens": usage.completion_tokens,
        "cost_usd": round(cost, 6),
    }

Цифры в PRICING — заглушка структуры, а не действующий прайс. Ставки за токены меняются, различаются по моделям и по регионам. Берите актуальные значения со страницы тарифов провайдера и обновляйте их у себя, иначе учёт разъедется с реальным счётом.

Оценка стоимости до запроса

Иногда нужно узнать цену prompt-части заранее — например, чтобы отсечь слишком большой контекст. Число входных токенов можно посчитать локально библиотекой tiktoken для моделей OpenAI:

import tiktoken

# Кодировка зависит от модели; для актуального соответствия
# смотрите документацию tiktoken.
encoding = tiktoken.encoding_for_model("gpt-4o-mini")
tokens_in = len(encoding.encode(prompt))
print(f"prompt: {tokens_in} токенов")

Число выходных токенов заранее не известно — его можно только ограничить сверху параметром max_tokens в вызове. Это же главный рычаг контроля затрат: модель не сгенерирует больше, чем вы разрешили, а значит верхняя граница стоимости ответа предсказуема.

Типичные ошибки

• Считать стоимость по своей оценке токенов, а не по полю usage. Локальный подсчёт tiktoken годится для прикидки перед запросом, но фактурой служит usage из ответа.
• Повторять таймаут вслепую. Если запрос дошёл и модель начала генерацию, повтор оплатится дважды. Для неидемпотентных операций взвесьте, готовы ли платить за дубль.
• Один клиент без пула под высокой параллельностью. Синхронный OpenAI блокирует поток на время запроса. Под нагрузкой берите AsyncOpenAI и await, либо разносите вызовы по воркерам.
• Ключ в коде или в логах. Только переменная окружения или менеджер секретов. И не логируйте полный prompt, если в нём персональные данные.

Что настроить дальше

Клиент с таймаутом, повторами и учётом токенов — это минимум. Следующие шаги, когда нагрузка вырастет:

• Перейти на AsyncOpenAI, если запросов много и они идут параллельно.
• Складывать cost_usd и число токенов в метрики (Prometheus, лог) — так вы увидите дорогие запросы до счёта.
• Добавить кеш ответов для повторяющихся prompt: самый дешёвый запрос тот, который не ушёл в API.
• Ограничить max_tokens по месту — это верхняя граница цены ответа.

Стриминг ответов и клиентский rate-limiter, который не даёт упереться в лимит провайдера, — отдельные темы, здесь мы их не разворачивали.