Guia: processamento em lote

Jobs em lote (traduzir um catálogo de produtos, classificar tickets de suporte, extrair campos de milhares de documentos) enviam o mesmo tipo de requisição muitas vezes. O trabalho é simples; as partes difíceis são vazão, falhas e custo. Este guia cobre as três usando o pacote oficial openai com a base URL https://api14.com/api/v1.

Não existe um endpoint de lote separado: a Batch API da OpenAI não está disponível no api14 (veja Compatibilidade com o SDK da OpenAI). Você executa lotes como requisições normais com concorrência controlada.

Conheça seu limite de taxa

Os limites são contados em requisições por minuto (RPM) por chave de API, com teto definido pelo seu plano:

Uma chave com RPM próprio recebe o menor dos dois valores. Consulte o seu com GET /key (rate_limit.requests é null quando a chave usa o limite do plano), veja Saldo e informações da chave. Exceder o limite retorna 429 rate_limited com Retry-After: 60.

O RPM limita quantas requisições começam por minuto, não quantas rodam ao mesmo tempo. Se uma requisição média leva 4 segundos e seu limite é de 100 RPM, cerca de 7 requisições em andamento já atingem o limite. Controle o ritmo pelo horário de início e mantenha a concorrência moderada.

Um executor concorrente com ritmo controlado (Python)

import asyncio, json, os, time
from openai import AsyncOpenAI, APIStatusError

client = AsyncOpenAI(base_url="https://api14.com/api/v1", api_key=os.environ["API14_API_KEY"], max_retries=4)

RPM = 100           # o limite do seu plano ou da sua chave
CONCURRENCY = 10
FATAL = {"daily_limit_reached", "insufficient_balance", "invalid_api_key", "model_not_allowed"}

class Pacer:
    """Espaça os inícios das requisições de forma uniforme para nunca exceder o RPM."""
    def __init__(self, rpm):
        self.interval, self.next, self.lock = 60 / rpm, 0.0, asyncio.Lock()
    async def wait(self):
        async with self.lock:
            now = time.monotonic()
            self.next = max(self.next, now)
            delay, self.next = self.next - now, self.next + self.interval
        await asyncio.sleep(delay)

SYSTEM = "Translate the user's text to German. Keep product names and units unchanged. Reply with the translation only."

async def translate(item, pacer, sem):
    async with sem:
        await pacer.wait()
        r = await client.chat.completions.create(
            model="openai/gpt-5-mini",
            messages=[{"role": "system", "content": SYSTEM}, {"role": "user", "content": item["text"]}],
            max_tokens=500,
        )
        return {"id": item["id"], "text": r.choices[0].message.content, "cost": r.usage.cost}

async def main(items, out_path="out.jsonl"):
    done = set()
    if os.path.exists(out_path):
        done = {json.loads(line)["id"] for line in open(out_path)}
    todo = [i for i in items if i["id"] not in done]

    pacer, sem = Pacer(RPM), asyncio.Semaphore(CONCURRENCY)
    with open(out_path, "a") as out:
        for task in asyncio.as_completed([translate(i, pacer, sem) for i in todo]):
            try:
                out.write(json.dumps(await task) + "\n")
            except APIStatusError as e:
                if e.code in FATAL:
                    raise SystemExit(f"Stopping batch: {e.code}")
                print("failed:", e.status_code, e.code)  # registrado; rode de novo para tentar outra vez

items = [json.loads(line) for line in open("items.jsonl")]  # [{"id": "...", "text": "..."}]
asyncio.run(main(items))

O que isso faz:

• Controle de ritmo: o Pacer inicia no máximo RPM requisições por minuto; o semáforo limita quantas ficam em andamento.
• Retentativas: o SDK tenta de novo em 429 e 5xx com backoff exponencial, respeitando Retry-After (max_retries=4). Os 429 vindos do provedor upstream são tratados da mesma forma.
• Códigos fatais interrompem o job: daily_limit_reached e insufficient_balance não se resolvem com novas tentativas, então o job termina em vez de gastar tentativas à toa.
• Checkpoints: os resultados são acrescentados a um arquivo JSONL conforme terminam. Ao rodar o script de novo, os ids já concluídos são ignorados, então uma falha ou um job interrompido não perde nada.

Em Node, a mesma estrutura funciona com Promise.all sobre um número fixo de workers que consomem uma fila compartilhada, cada um esperando 60_000 / RPM ms entre os inícios.

Coloque vários itens em uma requisição

Para itens curtos, uma requisição por item desperdiça RPM e repete as instruções todas as vezes. Classifique de 20 a 50 itens por requisição com uma saída estruturada e o número de requisições cai pelo mesmo fator:

schema = {
    "type": "object",
    "properties": {"labels": {"type": "array", "items": {
        "type": "object",
        "properties": {"id": {"type": "string"}, "label": {"type": "string", "enum": ["billing", "bug", "feature", "other"]}},
        "required": ["id", "label"], "additionalProperties": False,
    }}},
    "required": ["labels"], "additionalProperties": False,
}

tickets = "\n".join(f"{t['id']}: {t['text']}" for t in batch_of_30)
r = await client.chat.completions.create(
    model="openai/gpt-5-mini",
    messages=[{"role": "system", "content": "Label each ticket."}, {"role": "user", "content": tickets}],
    response_format={"type": "json_schema", "json_schema": {"name": "labels", "strict": True, "schema": schema}},
)
labels = json.loads(r.choices[0].message.content)["labels"]

Confirme que todos os ids de entrada voltaram e reenvie os que faltarem. Se finish_reason for "length", o lote era grande demais para max_tokens: divida-o. Veja Saídas estruturadas. Para embeddings, envie até 100 strings em um único array input, veja Embeddings.

Reduza o custo com prompt caching

Em um lote, as instruções, os exemplos e o schema são idênticos em toda requisição. Coloque-os primeiro e o item variável por último, para que os provedores reaproveitem o prefixo em cache:

• OpenAI, DeepSeek, Gemini 2.5 e outros fazem cache automaticamente quando o prefixo é longo o bastante (1,024 tokens na OpenAI). Leituras de cache custam uma fração da entrada normal.
• Os modelos da Anthropic exigem um ponto de corte cache_control explícito na parte estática do prompt.

Um prompt de sistema longo e detalhado, com alguns exemplos resolvidos, pode então custar pouco mais que um curto e geralmente melhora a qualidade. Confirme os acertos em usage.prompt_tokens_details.cached_tokens. Veja Prompt caching.

Escolha um modelo barato e depois verifique

Para tradução, classificação e extração, modelos pequenos costumam ser bons o bastante por uma fração do preço. Liste os modelos de texto por preço de entrada:

curl -s "https://api14.com/api/v1/models?category=text" \
  | jq -r '.data | sort_by(.pricing.prompt | tonumber) | .[:15][] | "\(.id)  \(.pricing.prompt)  \(.pricing.completion)"'

Depois meça antes de se comprometer:

1. Pegue de 50 a 100 itens representativos e rotule a saída esperada manualmente.
2. Rode-os em dois ou três modelos candidatos e compare a precisão e a soma de usage.cost.
3. Multiplique o custo por item pelo total de itens para estimar o job completo e confirme com GET /credits que seu saldo cobre o valor.

As variantes :free usam o endpoint gratuito de um modelo quando ele existe, mas têm limites de taxa upstream mais rígidos, então servem mais para testes do que para lotes grandes. Veja Ids de modelos e variantes.

Antes de uma execução grande

• Crie uma chave dedicada com limite diário em USD para que um bug não esvazie seu saldo. O job então para com 429 daily_limit_reached. Veja Limites de taxa.
• Defina max_tokens em toda requisição.
• Rode o primeiro 1% dos itens, inspecione a saída e o custo, e só depois inicie o restante.
• Requisições que falham (qualquer 4xx ou 5xx) não são cobradas, então reexecutar as falhas é seguro. Veja Erros.