Guide: batchbearbetning

Batchjobb (översätta en produktkatalog, klassificera supportärenden, extrahera fält ur tusentals dokument) skickar samma typ av anrop många gånger. Arbetet är enkelt, de svåra delarna är genomströmning, fel och kostnad. Den här guiden täcker alla tre med det officiella paketet openai och base URL https://api14.com/api/v1.

Det finns ingen separat batch-endpoint: OpenAI:s Batch API finns inte i api14 (se Kompatibilitet med OpenAI SDK). Du kör batchar som vanliga anrop med kontrollerad samtidighet.

Känn din hastighetsgräns

Gränser räknas i anrop per minut (RPM) per API-nyckel, begränsade av din plan:

En nyckel med egen RPM-gräns får den lägsta av de två. Kontrollera din med GET /key (rate_limit.requests är null när nyckeln använder planens gräns), se Saldo och nyckelinfo. Att överskrida gränsen returnerar 429 rate_limited med Retry-After: 60.

RPM begränsar hur många anrop som startar per minut, inte hur många som körs samtidigt. Om ett genomsnittligt anrop tar 4 sekunder och din gräns är 100 RPM, når ungefär 7 anrop under körning redan gränsen. Jämna ut anropen efter starttid och håll samtidigheten måttlig.

En jämn, samtidig körare (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           # din plans eller nyckels gräns
CONCURRENCY = 10
FATAL = {"daily_limit_reached", "insufficient_balance", "invalid_api_key", "model_not_allowed"}

class Pacer:
    """Sprider ut anropsstarter jämnt så att vi aldrig överskrider 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)  # loggas, kör igen för att försöka på nytt

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

Vad detta gör:

• Jämn takt: Pacer startar högst RPM anrop per minut, och semaforen begränsar hur många som pågår.
• Återförsök: SDK:n försöker igen vid 429 och 5xx med exponentiell backoff och respekterar Retry-After (max_retries=4). 429 från uppströmsleverantörer hanteras på samma sätt.
• Ödesdigra koder stoppar jobbet: daily_limit_reached och insufficient_balance försvinner inte av återförsök, så jobbet avslutas i stället för att förbruka försök.
• Kontrollpunkter: resultat läggs till i en JSONL-fil allteftersom de blir klara. Om du kör skriptet igen hoppar det över id:n som redan är klara, så en krasch eller ett stoppat jobb förlorar ingenting.

I Node fungerar samma struktur med Promise.all över ett fast antal arbetare som hämtar från en gemensam kö, där varje arbetare väntar 60_000 / RPM ms mellan starter.

Lägg flera poster i ett anrop

För korta poster slösar ett anrop per post på RPM och upprepar instruktionerna varje gång. Klassificera 20–50 poster per anrop med strukturerad utdata, så delar du antalet anrop med samma faktor:

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

Kontrollera att varje indata-id kom tillbaka och skicka om de som saknas. Om finish_reason är "length" var batchen för stor för max_tokens, dela upp den. Se Strukturerade utdata. För embeddings, skicka upp till 100 strängar i en input-array, se Embeddings.

Sänk kostnaden med prompt-cachning

I en batch är instruktionerna, exemplen och schemat identiska i varje anrop. Lägg dem först och den varierande posten sist, så att leverantörer kan återanvända det cachade prefixet:

• OpenAI, DeepSeek, Gemini 2.5 med flera cachar automatiskt när prefixet är tillräckligt långt (1 024 tokens för OpenAI). Cacheläsningar kostar en bråkdel av vanliga indata.
• Anthropic-modeller behöver en explicit cache_control-brytpunkt på den statiska delen av prompten.

En lång, detaljerad systemprompt med några genomarbetade exempel kan därför kosta knappt mer än en kort, och förbättrar vanligtvis kvaliteten. Bekräfta träffar i usage.prompt_tokens_details.cached_tokens. Se Prompt-cachning.

Välj en billig modell och verifiera sedan

För översättning, klassificering och extrahering räcker små modeller ofta till, till en bråkdel av priset. Lista textmodeller efter indatapris:

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

Mät sedan innan du bestämmer dig:

1. Ta 50–100 representativa poster och märk det förväntade resultatet för hand.
2. Kör dem genom två eller tre kandidatmodeller och jämför noggrannhet och summerad usage.cost.
3. Multiplicera kostnaden per post med det totala antalet poster för att uppskatta hela jobbet, och kontrollera att ditt saldo räcker med GET /credits.

:free-varianter använder en modells kostnadsfria endpoint där en sådan finns, men de har strängare hastighetsgränser uppströms, så de passar bättre för tester än stora batchar. Se Modell-id och varianter.

Före en stor körning

• Skapa en dedikerad nyckel med en daglig USD-gräns så att ett fel inte kan tömma ditt saldo. Jobbet stoppas då med 429 daily_limit_reached. Se Hastighetsgränser.
• Ange max_tokens på varje anrop.
• Kör de första 1 % av posterna, granska resultat och kostnad och starta sedan resten.
• Misslyckade anrop (alla 4xx och 5xx) debiteras inte, så det är säkert att köra om misslyckanden. Se Fel.