Práticas de estabilidade da API de proxy: timeout, retry, fila de limite e desconexão de streaming

Classifique os erros em três categorias

O primeiro passo para estabilidade não é aumentar retries, mas saber o que deve ser retry. Classificamos falhas comuns em três grupos:

Categoria	Sintoma típico	Tratamento
Recuperação imediata	429 rate limit, 503 upstream_busy, conexão reset, timeout	Retry com backoff exponencial, limitando o número total de tentativas
Erro na requisição	400 (input + max_tokens > 100.000, payload grande), 403 content_blocked	Não faça retry; corrija a requisição ou retorne ao usuário
Problema de conta	401 chave inválida, 402 no_credit	Não faça retry; alerte o responsável, recarregue ou troque a chave

Esta tabela deve estar no código. Um incidente comum: após o saldo acabar, a API retorna 402 estável. Se seu retry não distingue status, cada requisição tenta 5 vezes, ampliando o tráfego em 5x e enchendo logs de erros.

Configure timeouts em camadas

Um único timeout total não basta. Divida-o em três camadas:

• Timeout de conexão: tempo para estabelecer TCP/TLS. 5-10s bastam; falhe rápido se não conectar.
• Timeout de leitura: tempo esperando dados. Para não-streaming, cubra a geração completa. Para streaming, defina como intervalo máximo entre chunks (10-30s é razoável).
• Timeout total de negócio: tempo máximo tolerável pelo seu negócio, garantido por asyncio.wait_for ou gateway, incluindo todos os retries.

Textos longos aumentam o tempo de resposta não-streaming. Se ajustar max_tokens para 32.000 com apenas 30s de timeout, você terá timeouts frequentes. Use streaming para textos longos: dá feedback imediato e clarifica o timeout. O parâmetro timeout do SDK aceita número ou config detalhada.

Retry com backoff exponencial para 429 e 503

Três regras: delay dobra com falhas, defina limite máximo, adicione jitter. Sem jitter, tarefas falhadas tentam retry juntas, sobrecarregando o serviço recuperado. O código abaixo desativa o retry do SDK e controla apenas 429, 503 e erros de conexão:

复制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 indica servidor ocupado. Tente novamente após alguns segundos. O exemplo usa base 1s, limite 20s, máx 5 tentativas, cobrindo este cenário. Se 429 for frequente, o problema é falta de controle de taxa no emissor, não a falta de retry.

Fila de requisições simultâneas sob limite de 300/min

300 req/min equivalem a 5 req/s. Batch processes falham aqui: usar asyncio.gather para enviar milhares de requisições de uma vez esgota o limite nos primeiros e gera 429 no resto.

Use duas camadas: semáforo limita requisições em voo (evita estouro de memória/conexões); rate limiter controla a taxa de saída. Sem semáforo, requisições rápidas podem exceder 300/min. Sem rate limiter, requisições lentas acumulam. Exemplo usa 270 req/min para margem de 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())

Para múltiplos processos/máquinas compartilhando chave, o rate limiter em memória não basta, pois o limite é agregado por chave. Use token bucket compartilhado (ex: Redis) ou uma fila de saída unificada para coordenar.

O que fazer ao interromper o streaming

A conexão de streaming é mais frágil que uma requisição normal: ela dura mais tempo, passa por mais equipamentos de rede e qualquer tempo de inatividade em qualquer etapa pode cortá-la. O princípio de tratamento é “o conteúdo já recebido é um ativo”; não descarte tudo apenas porque houve uma interrupção.

A implementação abaixo acumula os fragmentos já recebidos. Ao capturar exceções de conexão ou tempo limite, ela solicita novamente o texto existente como uma mensagem do assistant, permitindo que o modelo continue a geração. É uma tentativa de recuperação de melhor esforço; a continuidade do texto gerado pode não ser perfeita, sendo adequada para textos longos e diálogos, mas não para saídas com estrutura rígida, como JSON. Se uma saída estruturada for interrompida, a abordagem mais segura é refazer tudo do início. Observe que, ao final do streaming, um fragmento de usage é anexado automaticamente; o código também o captura:

复制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)

Ao continuar, lembre-se de que o texto já enviado também conta como tokens de entrada, então cada continuação tem um custo adicional. É necessário limitar o número de max_resume.

Degradação, disjuntor e checklist pré-lançamento

A nova tentativa resolve falhas transitórias. Se a falha persistir por alguns minutos, continuar tentando apenas acumulará requisições. Recomenda-se adicionar uma camada de disjuntor acima da nova tentativa: quando o número de falhas consecutivas atingir um limite (por exemplo, dez falhas em um minuto), pause o envio de requisições por um curto período. Durante esse intervalo, utilize a lógica de degradação, como retornar um resultado em cache, informar ao usuário para tentar novamente mais tarde ou recolocar a tarefa na fila para processamento posterior. Após o tempo de resfriamento, libere apenas algumas requisições de sondagem; se tiverem sucesso, retorne ao volume total.

Planeje a degradação. Para chat, mostre aviso de ocupado; para batch, registre falhas para processamento posterior. Garanta que erros não sejam silenciosos, registrando pelo menos o request id.

Por fim, segue uma lista de verificação pré-lançamento: o SDK está configurado para não realizar novas tentativas implícitas, deixando apenas a sua implementação? As respostas 400, 401, 402 e 403 estão sendo ignoradas? O tempo limite total inclui todas as novas tentativas? O processamento em lote possui controle de taxa e não executa todas as requisições simultaneamente de uma vez? A interrupção do streaming está sendo tratada? O usage está sendo persistido? Há um alerta configurado para o saldo? Após passar por cada item da lista, realize o teste de carga, e não o contrário.

Monitore e concilie usando usage

Cada resposta traz usage (último chunk no streaming). Persista-o com o nome da função: monitoramento de baixo custo. Preços: input $0,25/1M tokens, output $1,00/1M. Estime custo local:

复制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}"]
        )

Após a persistência, recomenda-se verificar diariamente três métricas: a mediana de tokens de entrada por função (para detectar se os prompts estão crescendo silenciosamente); a proporção de tokens de saída (como o preço do token de saída é quatro vezes maior que o de entrada, esta costuma ser a maior parte da fatura); e a proporção entre 429 e 503 (se essa proporção aumentar, significa que a taxa de envio ou a estratégia de nova tentativa precisa ser ajustada). Adicione também um alerta de saldo para notificar o responsável antes que o erro 402 ocorra. Para mais detalhes sobre a integração, consulte a configuração do framework. As perguntas frequentes estão reunidas na seção Perguntas frequentes.