Panduan: pemrosesan batch

Job batch (menerjemahkan katalog produk, mengklasifikasikan tiket dukungan, mengekstrak field dari ribuan dokumen) mengirim jenis request yang sama berulang kali. Pekerjaannya sederhana; bagian yang sulit adalah throughput, kegagalan, dan biaya. Panduan ini membahas ketiganya dengan paket openai resmi dan base URL https://api14.com/api/v1.

Tidak ada endpoint batch terpisah: Batch API milik OpenAI tidak tersedia di api14 (lihat Kompatibilitas OpenAI SDK). Anda menjalankan batch sebagai request biasa dengan konkurensi yang terkendali.

Ketahui batas laju Anda

Batas dihitung dalam request per menit (RPM) per API key, dibatasi oleh paket Anda:

Key dengan batas RPM sendiri mendapat yang lebih rendah dari keduanya. Periksa milik Anda dengan GET /key (rate_limit.requests bernilai null ketika key memakai batas paket), lihat Saldo dan info key. Melampaui batas mengembalikan 429 rate_limited dengan Retry-After: 60.

RPM membatasi berapa banyak request yang dimulai per menit, bukan berapa banyak yang berjalan bersamaan. Jika rata-rata request memakan 4 detik dan batas Anda 100 RPM, sekitar 7 request yang sedang berjalan sudah mencapai batas. Atur laju request berdasarkan waktu mulai, dan jaga konkurensi tetap moderat.

Runner konkuren dengan pengaturan laju (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))

Yang dilakukan kode ini:

• Pengaturan laju: Pacer memulai paling banyak RPM request per menit; semaphore membatasi berapa banyak yang berjalan bersamaan.
• Retry: SDK mengulang 429 dan 5xx dengan exponential backoff, menghormati Retry-After (max_retries=4). 429 dari provider upstream diulang dengan cara yang sama.
• Kode fatal menghentikan job: daily_limit_reached dan insufficient_balance tidak akan hilang dengan mengulang, jadi job keluar alih-alih menghabiskan percobaan.
• Checkpoint: hasil ditambahkan ke file JSONL begitu selesai. Menjalankan ulang skrip melewati id yang sudah selesai, sehingga crash atau job yang dihentikan tidak kehilangan apa pun.

Di Node, struktur yang sama bekerja dengan Promise.all atas sejumlah worker tetap yang mengambil dari antrean bersama, masing-masing menunggu 60_000 / RPM ms di antara pemulaian.

Masukkan beberapa item dalam satu request

Untuk item pendek, satu request per item memboroskan RPM dan mengulang instruksi setiap kali. Klasifikasikan 20–50 item per request dengan structured output, dan jumlah request Anda terbagi dengan faktor yang sama:

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

Pastikan setiap id input kembali; kirim ulang yang hilang. Jika finish_reason adalah "length", batch terlalu besar untuk max_tokens, pecah menjadi lebih kecil. Lihat Structured outputs. Untuk embeddings, kirim hingga 100 string dalam satu array input, lihat Embeddings.

Tekan biaya dengan prompt caching

Dalam batch, instruksi, contoh, dan skema identik di setiap request. Letakkan itu di awal dan item yang bervariasi di akhir, agar provider dapat memakai ulang prefiks yang di-cache:

• OpenAI, DeepSeek, Gemini 2.5, dan lainnya melakukan cache secara otomatis begitu prefiks cukup panjang (1.024 token untuk OpenAI). Pembacaan cache berbiaya sebagian kecil dari input normal.
• Model Anthropic memerlukan breakpoint cache_control eksplisit pada bagian prompt yang statis.

Dengan demikian system prompt yang panjang dan terperinci dengan beberapa contoh kerja bisa berbiaya hampir sama dengan yang pendek, dan biasanya meningkatkan kualitas. Pastikan hit lewat usage.prompt_tokens_details.cached_tokens. Lihat Prompt caching.

Pilih model murah, lalu verifikasi

Untuk terjemahan, klasifikasi, dan ekstraksi, model kecil sering kali sudah cukup baik dengan harga sebagian kecil. Tampilkan model teks berdasarkan harga input:

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

Lalu ukur sebelum berkomitmen:

1. Ambil 50–100 item representatif dan labeli keluaran yang diharapkan secara manual.
2. Jalankan lewat dua atau tiga model kandidat dan bandingkan akurasi serta jumlah usage.cost.
3. Kalikan biaya per item dengan total jumlah item Anda untuk memperkirakan seluruh job, dan pastikan saldo Anda mencukupi dengan GET /credits.

Varian :free memakai endpoint gratis model jika ada, tetapi memiliki batas laju upstream yang lebih ketat, jadi lebih cocok untuk pengujian daripada batch besar. Lihat ID model dan varian.

Sebelum menjalankan skala besar

• Buat key khusus dengan batas USD harian agar bug tidak menguras saldo Anda. Job kemudian berhenti dengan 429 daily_limit_reached. Lihat Batas laju.
• Atur max_tokens pada setiap request.
• Jalankan 1% item pertama, periksa keluaran dan biayanya, lalu mulai sisanya.
• Request yang gagal (4xx atau 5xx apa pun) tidak ditagih, jadi menjalankan ulang yang gagal aman. Lihat Error.