Guía: procesamiento por lotes

Los trabajos por lotes (traducir un catálogo de productos, clasificar tickets de soporte, extraer campos de miles de documentos) envían el mismo tipo de petición muchas veces. El trabajo es sencillo; lo difícil es el rendimiento, los fallos y el coste. Esta guía cubre los tres con el paquete oficial openai y la URL base https://api14.com/api/v1.

No hay un endpoint de lotes aparte: la Batch API de OpenAI no está disponible en api14 (consulta Compatibilidad con el SDK de OpenAI). Los lotes se ejecutan como peticiones normales con concurrencia controlada.

Conoce tu límite de tasa

Los límites se cuentan en peticiones por minuto (RPM) por clave de API, con el tope de tu plan:

Una clave con su propio límite de RPM recibe el menor de los dos. Consulta el tuyo con GET /key (rate_limit.requests es null cuando la clave usa el límite del plan); consulta Saldo e información de la clave. Superar el límite devuelve 429 rate_limited con Retry-After: 60.

El RPM limita cuántas peticiones empiezan por minuto, no cuántas se ejecutan a la vez. Si una petición media tarda 4 segundos y tu límite es de 100 RPM, unas 7 peticiones en curso ya alcanzan el límite. Espacia las peticiones según su hora de inicio y mantén una concurrencia moderada.

Un ejecutor concurrente con 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           # el límite de tu plan o clave
CONCURRENCY = 10
FATAL = {"daily_limit_reached", "insufficient_balance", "invalid_api_key", "model_not_allowed"}

class Pacer:
    """Espacia el inicio de las peticiones de forma uniforme para no superar el 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; vuelve a ejecutar para reintentar

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

Qué hace este código:

• Ritmo: el Pacer inicia como máximo RPM peticiones por minuto; el semáforo limita cuántas están en curso.
• Reintentos: el SDK reintenta los errores 429 y 5xx con retroceso exponencial, respetando Retry-After (max_retries=4). Los 429 del proveedor se reintentan igual.
• Los códigos fatales detienen el trabajo: daily_limit_reached e insufficient_balance no se resuelven reintentando, así que el trabajo termina en lugar de gastar intentos.
• Puntos de control: los resultados se añaden a un archivo JSONL a medida que terminan. Al volver a ejecutar el script se omiten los ids ya completados, así que un fallo o una interrupción no pierde nada.

En Node, la misma estructura funciona con Promise.all sobre un número fijo de workers que toman elementos de una cola compartida, cada uno esperando 60_000 / RPM ms entre inicios.

Agrupa varios elementos en una petición

Para elementos cortos, una petición por elemento desperdicia RPM y repite las instrucciones cada vez. Clasifica 20-50 elementos por petición con una salida estructurada y dividirás el número de peticiones por el mismo factor:

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"]

Comprueba que volvió cada id de entrada y reenvía los que falten. Si finish_reason es "length", el lote era demasiado grande para max_tokens; divídelo. Consulta Salidas estructuradas. Para embeddings, pasa hasta 100 cadenas en un solo array input; consulta Embeddings.

Reduce el coste con la caché de prompts

En un lote, las instrucciones, los ejemplos y el esquema son idénticos en cada petición. Colócalos al principio y deja el elemento variable al final, para que los proveedores puedan reutilizar el prefijo en caché:

• OpenAI, DeepSeek, Gemini 2.5 y otros usan la caché automáticamente cuando el prefijo es lo bastante largo (1.024 tokens en OpenAI). Las lecturas de caché cuestan una fracción de la entrada normal.
• Los modelos de Anthropic necesitan un punto de corte cache_control explícito en la parte estática del prompt.

Un prompt de sistema largo y detallado con unos pocos ejemplos resueltos puede costar así poco más que uno corto, y normalmente mejora la calidad. Confirma los aciertos en usage.prompt_tokens_details.cached_tokens. Consulta Caché de prompts.

Elige un modelo económico y verifica

Para traducción, clasificación y extracción, los modelos pequeños suelen ser suficientes a una fracción del precio. Lista los modelos de texto por precio 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)"'

Luego mide antes de comprometerte:

1. Toma 50-100 elementos representativos y etiqueta a mano el resultado esperado.
2. Pásalos por dos o tres modelos candidatos y compara la precisión y la suma de usage.cost.
3. Multiplica el coste por elemento por el total de elementos para estimar el trabajo completo, y comprueba con GET /credits que tu saldo lo cubre.

Las variantes :free usan el endpoint gratuito de un modelo cuando existe, pero tienen límites de tasa más estrictos en el proveedor, así que sirven más para pruebas que para lotes grandes. Consulta Ids de modelo y variantes.

Antes de una ejecución grande

• Crea una clave dedicada con un límite diario en USD para que un error no pueda vaciar tu saldo. El trabajo se detendrá con 429 daily_limit_reached. Consulta Límites de tasa.
• Establece max_tokens en cada petición.
• Ejecuta el primer 1 % de los elementos, revisa el resultado y el coste, y luego lanza el resto.
• Las peticiones fallidas (cualquier 4xx o 5xx) no se cobran, así que repetir los fallos es seguro. Consulta Errores.