Rehber: toplu işleme

Toplu işler (bir ürün kataloğunu çevirmek, destek taleplerini sınıflandırmak, binlerce belgeden alan çıkarmak) aynı türden isteği defalarca gönderir. İş basittir; zor kısımlar verim, hatalar ve maliyettir. Bu rehber üçünü de, https://api14.com/api/v1 base URL'iyle resmi openai paketini kullanarak ele alır.

Ayrı bir batch uç noktası yoktur: OpenAI'ın Batch API'si api14'te mevcut değildir (bkz. OpenAI SDK uyumluluğu). Toplu işleri, kontrollü eşzamanlılıkla normal istekler olarak çalıştırırsınız.

Hız limitinizi bilin

Limitler API anahtarı başına dakikalık istek sayısı (RPM) olarak sayılır ve planınızla sınırlanır:

Kendi RPM limiti olan bir anahtar, ikisinden düşük olanı alır. Kendi limitinizi GET /key ile kontrol edin (anahtar plan limitini kullanıyorsa rate_limit.requests değeri null olur), bkz. Bakiye ve anahtar bilgisi. Limiti aşmak, Retry-After: 60 ile 429 rate_limited döndürür.

RPM, dakikada kaç isteğin başladığını sınırlar; aynı anda kaç isteğin çalıştığını değil. Ortalama bir istek 4 saniye sürüyorsa ve limitiniz 100 RPM ise, uçuştaki yaklaşık 7 istek limite zaten ulaşır. İstekleri başlangıç zamanına göre aralıklandırın ve eşzamanlılığı ılımlı tutun.

Aralıklı, eşzamanlı bir çalıştırıcı (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           # your plan or key limit
CONCURRENCY = 10
FATAL = {"daily_limit_reached", "insufficient_balance", "invalid_api_key", "model_not_allowed"}

class Pacer:
    """Spaces request starts evenly so we never exceed 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)  # logged; rerun to retry

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

Bunun yaptıkları:

• Aralıklandırma: Pacer dakikada en fazla RPM istek başlatır; semaphore aynı anda uçuşta kaç istek olacağını sınırlar.
• Yeniden denemeler: SDK 429 ve 5xx hatalarını Retry-After değerine uyarak üstel geri çekilmeyle yeniden dener (max_retries=4). Sağlayıcı kaynaklı 429'lar da aynı şekilde yeniden denenir.
• Ölümcül kodlar işi durdurur: daily_limit_reached ve insufficient_balance yeniden denemekle düzelmez; bu yüzden iş, denemeleri tüketmek yerine çıkar.
• Checkpoint: sonuçlar biter bitmez bir JSONL dosyasına eklenir. Betiği yeniden çalıştırmak tamamlanmış kimlikleri atlar; böylece bir çökme veya durdurulmuş iş hiçbir şey kaybettirmez.

Node'da aynı yapı, paylaşılan bir kuyruktan çeken sabit sayıda worker üzerinde Promise.all ile çalışır; her worker başlangıçlar arasında 60_000 / RPM ms bekler.

Tek istekte birden çok öğe koyun

Kısa öğeler için öğe başına bir istek RPM'i boşa harcar ve yönergeleri her seferinde tekrarlar. Structured output ile istek başına 20 ila 50 öğe sınıflandırın; istek sayısını aynı oranda bölmüş olursunuz:

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

Her girdi kimliğinin geri geldiğini kontrol edin; eksik olanları yeniden gönderin. finish_reason değeri "length" ise toplu iş max_tokens için çok büyüktü, bölün. Bkz. Structured outputs. Embedding'lerde tek bir input dizisinde 100 dizeye kadar geçin, bkz. Embeddings.

Prompt caching ile maliyeti düşürün

Bir toplu işte yönergeler, örnekler ve şema her istekte aynıdır. Bunları en başa, değişen öğeyi sona koyun; böylece sağlayıcılar önbelleğe alınmış öneki yeniden kullanabilir:

• OpenAI, DeepSeek, Gemini 2.5 ve diğerleri, önek yeterince uzun olduğunda otomatik önbellekler (OpenAI için 1.024 token). Önbellek okumaları normal girdinin bir kesrine mal olur.
• Anthropic modelleri, prompt'un statik kısmında açık bir cache_control kesme noktası ister.

Bu yüzden birkaç çalışılmış örnek içeren uzun, ayrıntılı bir sistem prompt'u kısa bir prompt'tan pek pahalı olmayabilir ve genellikle kaliteyi artırır. İsabetleri usage.prompt_tokens_details.cached_tokens içinde doğrulayın. Bkz. Prompt caching.

Ucuz bir model seçin, sonra doğrulayın

Çeviri, sınıflandırma ve veri çıkarma için küçük modeller çoğu zaman fiyatın çok küçük bir kesriyle yeterince iyidir. Metin modellerini girdi fiyatına göre listeleyin:

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

Sonra karar vermeden önce ölçün:

1. 50 ila 100 temsilci öğe alın ve beklenen çıktıyı elle etiketleyin.
2. Bunları iki veya üç aday modelde çalıştırın ve doğruluğu ile toplam usage.cost değerini karşılaştırın.
3. Öğe başına maliyeti toplam öğe sayınızla çarparak tüm işi tahmin edin ve bakiyenizin bunu GET /credits ile karşıladığını kontrol edin.

:free varyantları, varsa bir modelin ücretsiz uç noktasını kullanır, ancak sağlayıcı tarafında daha sıkı hız limitleri vardır; bu yüzden büyük toplu işlerden çok testlere uygundur. Bkz. Model kimlikleri ve varyantlar.

Büyük bir çalıştırmadan önce

• Bir hatanın bakiyenizi tüketmemesi için günlük USD limitli özel bir anahtar oluşturun. İş o zaman 429 daily_limit_reached ile durur. Bkz. Hız limitleri.
• Her istekte max_tokens ayarlayın.
• Öğelerin ilk %1'ini çalıştırın, çıktıyı ve maliyeti inceleyin, sonra geri kalanını başlatın.
• Başarısız istekler (her 4xx veya 5xx) ücretlendirilmez; bu yüzden başarısızları yeniden çalıştırmak güvenlidir. Bkz. Hatalar.