Практика стабильности прокси-API: таймауты, повторные попытки, очереди лимитов и разрыв потоковой передачи

Сначала разделите ошибки на три категории

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

Категория	Типичные проявления	Стратегия обработки
Временные и восстанавливаемые	429 rate limit, 503 upstream_busy, сброс соединения, таймаут	Повторные попытки с экспоненциальной задержкой, с ограничением общего числа попыток
Проблемы самого запроса	400 (сумма input и max_tokens превышает 100 000, тело запроса слишком велико), 403 content_blocked	Не повторять, исправить запрос или сообщить пользователю
Проблемы со статусом аккаунта	401 недействительный API-ключ, 402 no_credit	Не повторять, отправить оповещение ответственным, пополнить баланс или сменить ключ

Эту таблицу лучше вынести в комментарии к коду. Один из распространённых инцидентов: после исчерпания баланса API стабильно возвращает 402, а ваша логика повторных попыток не различает коды состояния. В результате каждый запрос повторяется пять раз, что увеличивает трафик в пять раз и превращает логи в сплошной красный цвет.

Таймауты следует настраивать многоуровнево

Установки одного общего таймаута недостаточно. Рекомендуется разделить таймауты на три уровня:

• Таймаут соединения: время на установление TCP и TLS-соединения. Обычно достаточно 5–10 секунд; если соединение не установлено, следует быстро завершить запрос.
• Таймаут чтения: время ожидания данных ответа. Для непотоковых запросов он должен покрывать время генерации всего текста. Для потоковых запросов это максимальный интервал между блоками данных; 10–30 секунд — разумный диапазон.
• Общий бизнес-таймаут: максимальное время ожидания, допустимое вашим бизнесом. Обеспечивается через asyncio.wait_for или на уровне шлюза, включая все повторные попытки.

Чем длиннее вывод, тем дольше выполняется непотоковый запрос. Если вы установите max_tokens на максимум 32 000 и зададите общий таймаут 30 секунд, при генерации длинных текстов таймауты будут возникать постоянно. Надёжнее всегда использовать потоковую передачу для длинных выводов: это даёт мгновенную обратную связь и делает смысл таймаута более ясным. Параметр timeout в SDK может принимать число или детальную конфигурацию; см. документацию вашей версии.

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

Три ключевых принципа экспоненциальной задержки: задержка удваивается с каждой неудачей, установка верхнего предела и добавление случайного джиттера. Без джиттера группа задач, завершившихся неудачей одновременно, попытается повторить запросы в тот же момент и снова перегрузит только что восстановленный сервис. В приведённой функции мы отключаем встроенную в SDK повторную обработку и управляем ею централизованно, повторяя запросы только при 429, 503 и ошибках соединения:

复制import asyncio
import os
import random

from openai import AsyncOpenAI, APIConnectionError, APIStatusError, APITimeoutError

client = AsyncOpenAI(
    base_url="https://api.llmzhongzhuan.com/v1",
    api_key=os.environ["API_KEY"],
    timeout=60.0,
    max_retries=0,  # 重试由下面的函数统一控制
)

RETRY_STATUS = {429, 503}


async def chat_with_retry(messages, max_attempts=5, base_delay=1.0, cap=20.0):
    for attempt in range(1, max_attempts + 1):
        try:
            resp = await client.chat.completions.create(
                model="uncensored", messages=messages, max_tokens=800
            )
            return resp
        except APIStatusError as e:
            if e.status_code not in RETRY_STATUS or attempt == max_attempts:
                raise  # 400/401/402/403 重试没有意义
            reason = f"HTTP {e.status_code}"
        except (APIConnectionError, APITimeoutError) as e:
            if attempt == max_attempts:
                raise
            reason = type(e).__name__
        delay = min(cap, base_delay * 2 ** (attempt - 1))
        delay = random.uniform(0, delay)  # 全抖动，避免所有任务同时醒来
        print(f"第 {attempt} 次失败（{reason}），{delay:.1f}s 后重试")
        await asyncio.sleep(delay)

503 означает временную перегрузку сервера; рекомендуется повторить запрос через несколько секунд. Приведённые параметры (базовая задержка 1 с, предел 20 с, максимум 5 попыток) покрывают этот сценарий. Если 429 возникает часто, проблема не в повторных попытках, а в отсутствии контроля скорости на стороне отправителя — см. следующий раздел.

Очередь параллельных запросов при лимите 300 в минуту

Лимит составляет 300 запросов на ключ в минуту, что в среднем равно 5 запросам в секунду. Пакетная обработка часто даёт сбой здесь: если отправить тысячи запросов через asyncio.gather одновременно, первые десятки мгновенно исчерпают квоту, а остальные получат 429.

Надёжное решение — двухуровневый контроль. Семафор ограничивает количество одновременных запросов, предотвращая исчерпание соединений и памяти. Регулятор ограничивает скорость отправки, равномерно распределяя запросы. Оба уровня необходимы: только семафор может пропустить более 300 запросов в минуту при быстрых ответах; только регулятор позволит накопить много ожидающих запросов при медленных ответах. В примере скорость установлена на 270 запросов в минуту, оставляя 10 % запаса на несколько экземпляров, повторные попытки и погрешность часов:

复制import asyncio
import os
import time

from openai import AsyncOpenAI

client = AsyncOpenAI(
    base_url="https://api.llmzhongzhuan.com/v1",
    api_key=os.environ["API_KEY"],
    timeout=60.0,
    max_retries=0,
)

MAX_IN_FLIGHT = 8          # 同时在途的请求数
PER_MINUTE = 270           # 留 10% 余量，低于每分钟 300 次的上限


class Pacer:
    """把请求的发出时刻均匀铺开：每 60/PER_MINUTE 秒放行一个。"""

    def __init__(self, per_minute):
        self.interval = 60.0 / per_minute
        self.next_at = 0.0
        self.lock = asyncio.Lock()

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


sem = asyncio.Semaphore(MAX_IN_FLIGHT)
pacer = Pacer(PER_MINUTE)


async def summarize(idx, text):
    async with sem:            # 控制并发
        await pacer.wait()     # 控制速率
        resp = await client.chat.completions.create(
            model="uncensored",
            messages=[{"role": "user", "content": f"用两句话概括：{text}"}],
            max_tokens=120,
        )
        return idx, resp.choices[0].message.content


async def main():
    texts = [f"第 {i} 条工单的正文……" for i in range(1000)]
    tasks = [summarize(i, t) for i, t in enumerate(texts)]
    done = 0
    for coro in asyncio.as_completed(tasks):
        idx, out = await coro
        done += 1
        if done % 100 == 0:
            print(f"已完成 {done} 条")


asyncio.run(main())

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

Что делать при разрыве потоковой передачи

Потоковое соединение более хрупкое, чем обычные запросы: оно длится дольше, проходит через больше сетевых устройств, и любой таймаут простоя может его разорвать. Принцип обработки: «уже полученные данные — это актив», не следует отбрасывать всё при разрыве.

Приведённая реализация накапливает полученные фрагменты. При захвате исключений соединения или таймаута уже полученный текст отправляется как сообщение assistant для продолжения генерации. Это эвристическое восстановление: продолжение может быть неидеальным, оно подходит для длинных текстов и диалогов, но не для строгих структур, таких как JSON. При разрыве структурированного вывода надёжнее выполнить полный повторный запрос. Обратите внимание, что в конце потоковой передачи автоматически добавляется фрагмент usage; код также его извлекает:

复制import asyncio
import os

from openai import AsyncOpenAI, APIConnectionError, APITimeoutError

client = AsyncOpenAI(
    base_url="https://api.llmzhongzhuan.com/v1",
    api_key=os.environ["API_KEY"],
    timeout=30.0,   # 流式场景下，这是相邻数据块之间允许的最长静默
    max_retries=0,
)


async def stream_once(messages):
    parts, usage = [], None
    stream = await client.chat.completions.create(
        model="uncensored", messages=messages, max_tokens=600, stream=True
    )
    try:
        async for chunk in stream:
            if chunk.usage:                      # 最后一个分片携带 usage
                usage = chunk.usage
            if chunk.choices and chunk.choices[0].delta.content:
                parts.append(chunk.choices[0].delta.content)
    except (APIConnectionError, APITimeoutError):
        return "".join(parts), None, False       # 中断：返回已收到的部分
    return "".join(parts), usage, True


async def stream_with_resume(question, max_resume=2):
    messages = [{"role": "user", "content": question}]
    text = ""
    for _ in range(max_resume + 1):
        part, usage, finished = await stream_once(messages)
        text += part
        if finished:
            return text, usage
        # 把已生成部分作为 assistant 消息，请模型接着写
        messages = [
            {"role": "user", "content": question},
            {"role": "assistant", "content": text},
            {"role": "user", "content": "请从上一条回复的结尾处继续，不要重复已写的内容。"},
        ]
    return text, None


if __name__ == "__main__":
    out, usage = asyncio.run(stream_with_resume("用三段话说明为什么日志要带 request id。"))
    print(out)
    print("usage:", usage)

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

Деградация, разрыв связи и проверка перед запуском

Повторные попытки решают временные сбои. Если сбой длится несколько минут, продолжение попыток приведёт к накоплению запросов. Рекомендуется добавить уровень разрыва связи: при достижении порога непрерывных ошибок (например, 10 ошибок за минуту) временно приостановите отправку запросов и перейдите к логике деградации: верните кэшированный результат, сообщите пользователю попробовать позже или верните задачу в очередь. После периода охлаждения отправляйте несколько тестовых запросов и, если они успешны, восстановите полную нагрузку.

Деградацию также следует продумать заранее. Для чата с пользователем можно вернуть дружелюбное сообщение о занятости. Для офлайн-пакетной обработки следует записать неудачные элементы в таблицу ошибок и выполнить их повторный запуск после завершения основного процесса, а не ждать бесконечно в основном потоке. В любом случае убедитесь, что ошибки не поглощаются молча: хотя бы одна запись в лог с request id должна сохраниться.

Наконец, чек-лист перед запуском: отключены ли скрытые повторные попытки SDK и оставлена ли только ваша логика; отменяются ли запросы при 400, 401, 402, 403; учитываются ли все повторные попытки в общем таймауте; есть ли контроль скорости для пакетной обработки вместо одновременного запуска; обрабатывается ли разрыв потоковой передачи; сохраняется ли usage; настроено ли оповещение о балансе. Пройдите этот чек-лист, и только затем проводите нагрузочное тестирование.

Мониторинг и сверка по usage

Каждый ответ содержит usage, в потоковом режиме — в последнем фрагменте. Сохранение usage вместе с именем функции — самый дешёвый способ мониторинга. Стоимость ввода на нашем сайте составляет $0,25 за миллион токенов, вывода — $1,00 за миллион токенов. Вы можете оценить стоимость каждого запроса локально:

复制import csv
import time

LOG = "usage_log.csv"


def record_usage(feature, resp):
    u = resp.usage
    cost = u.prompt_tokens * 0.25 / 1_000_000 + u.completion_tokens * 1.00 / 1_000_000
    with open(LOG, "a", newline="", encoding="utf-8") as f:
        csv.writer(f).writerow(
            [int(time.time()), feature, u.prompt_tokens, u.completion_tokens, f"{cost:.6f}"]
        )

После сохранения данных на диск рекомендуется ежедневно отслеживать три метрики: медианное количество входных токенов по функциям (чтобы заметить скрытое разрастание промптов), долю выходных токенов (так как цена вывода в четыре раза выше, это обычно основная статья расходов) и соотношение 429 к 503 (рост доли указывает на необходимость корректировки скорости отправки или стратегии повторных попыток). Добавьте также предупреждение о балансе, чтобы уведомить ответственных о необходимости пополнить баланс до появления 402. Подробнее о настройках подключения см. в конфигурации фреймворка, а ответы на частые вопросы собраны в разделе FAQ.