गाइड: बैच प्रोसेसिंग

बैच जॉब (प्रोडक्ट कैटलॉग का अनुवाद, सपोर्ट टिकट का वर्गीकरण, हज़ारों दस्तावेज़ों से फ़ील्ड निकालना) एक ही तरह का अनुरोध कई बार भेजते हैं। काम सरल है; कठिन हिस्से हैं थ्रूपुट, विफलताएँ और लागत। यह गाइड आधिकारिक openai पैकेज और बेस URL https://api14.com/api/v1 के साथ तीनों को कवर करती है।

कोई अलग बैच एंडपॉइंट नहीं है: OpenAI की Batch API api14 पर उपलब्ध नहीं है (देखें OpenAI SDK संगतता)। आप बैच को नियंत्रित समवर्तीता (concurrency) के साथ सामान्य अनुरोधों के रूप में चलाते हैं।

अपनी रेट लिमिट जानें

सीमाएँ प्रति API कुंजी प्रति मिनट अनुरोध (RPM) में गिनी जाती हैं और आपके प्लान से सीमित होती हैं:

जिस कुंजी की अपनी RPM सीमा हो, उस पर दोनों में से कम सीमा लागू होती है। अपनी सीमा GET /key से जाँचें (जब कुंजी प्लान की सीमा इस्तेमाल करती है तब rate_limit.requests null होता है), देखें बैलेंस और कुंजी जानकारी। सीमा पार होने पर Retry-After: 60 के साथ 429 rate_limited लौटता है।

RPM यह सीमित करता है कि प्रति मिनट कितने अनुरोध शुरू होते हैं, यह नहीं कि एक साथ कितने चलते हैं। यदि औसत अनुरोध में 4 सेकंड लगते हैं और आपकी सीमा 100 RPM है, तो लगभग 7 अनुरोध एक साथ चलने पर ही सीमा आ जाती है। अनुरोधों को उनके शुरू होने के समय के अनुसार गति दें, और समवर्तीता संतुलित रखें।

गति-नियंत्रित, समवर्ती रनर (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           # आपके प्लान या कुंजी की सीमा
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)  # लॉग किया गया; रीट्राई के लिए दोबारा चलाएँ

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

यह क्या करता है:

• गति नियंत्रण: Pacer प्रति मिनट अधिकतम RPM अनुरोध शुरू करता है; सेमाफ़ोर सीमित करता है कि कितने अनुरोध एक साथ चल रहे हैं।
• रीट्राई: SDK 429 और 5xx को exponential backoff के साथ दोबारा आज़माता है और Retry-After का सम्मान करता है (max_retries=4)। अपस्ट्रीम प्रोवाइडर के 429 भी इसी तरह रीट्राई होते हैं।
• घातक कोड जॉब रोक देते हैं: daily_limit_reached और insufficient_balance रीट्राई से ठीक नहीं होंगे, इसलिए जॉब प्रयास बर्बाद करने के बजाय बाहर निकल जाता है।
• चेकपॉइंटिंग: नतीजे पूरे होते ही JSONL फ़ाइल में जोड़े जाते हैं। स्क्रिप्ट दोबारा चलाने पर पूरे हो चुके id छोड़ दिए जाते हैं, इसलिए क्रैश या रुके हुए जॉब में कुछ नहीं खोता।

Node में यही संरचना एक साझा कतार से काम उठाने वाले निश्चित संख्या के वर्कर पर Promise.all के साथ काम करती है, जहाँ हर वर्कर शुरू करने के बीच 60_000 / RPM ms रुकता है।

एक अनुरोध में कई आइटम रखें

छोटे आइटम के लिए हर आइटम पर एक अनुरोध RPM बर्बाद करता है और हर बार निर्देश दोहराता है। स्ट्रक्चर्ड आउटपुट के साथ प्रति अनुरोध 20-50 आइटम वर्गीकृत करें, और अनुरोधों की संख्या उतने ही गुना घट जाती है:

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

जाँचें कि हर इनपुट id लौटा है; जो छूट गए उन्हें दोबारा भेजें। यदि finish_reason "length" है, तो बैच max_tokens के लिए बहुत बड़ा था, उसे बाँट दें। देखें स्ट्रक्चर्ड आउटपुट। एम्बेडिंग के लिए एक input ऐरे में 100 तक स्ट्रिंग भेजें, देखें एम्बेडिंग।

प्रॉम्प्ट कैशिंग से लागत घटाएँ

बैच में निर्देश, उदाहरण और schema हर अनुरोध में एक जैसे होते हैं। उन्हें पहले और बदलने वाले आइटम को आख़िर में रखें, ताकि प्रोवाइडर कैश्ड प्रीफ़िक्स दोबारा इस्तेमाल कर सकें:

• OpenAI, DeepSeek, Gemini 2.5 और अन्य प्रीफ़िक्स पर्याप्त लंबा होते ही अपने आप कैश करते हैं (OpenAI के लिए 1,024 टोकन)। कैश रीड की लागत सामान्य इनपुट का एक छोटा हिस्सा होती है।
• Anthropic मॉडल को प्रॉम्प्ट के स्थिर हिस्से पर स्पष्ट cache_control ब्रेकपॉइंट चाहिए।

इसलिए कुछ उदाहरणों वाला लंबा, विस्तृत सिस्टम प्रॉम्प्ट छोटे प्रॉम्प्ट से बहुत ज़्यादा महँगा नहीं पड़ता और आमतौर पर गुणवत्ता बढ़ाता है। हिट की पुष्टि usage.prompt_tokens_details.cached_tokens में करें। देखें प्रॉम्प्ट कैशिंग।

सस्ता मॉडल चुनें, फिर जाँचें

अनुवाद, वर्गीकरण और डेटा निकासी के लिए छोटे मॉडल अक्सर कीमत के एक अंश पर पर्याप्त अच्छे होते हैं। टेक्स्ट मॉडल को इनपुट कीमत के अनुसार सूचीबद्ध करें:

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

फिर प्रतिबद्ध होने से पहले मापें:

1. 50-100 प्रतिनिधि आइटम लें और अपेक्षित आउटपुट हाथ से लेबल करें।
2. उन्हें दो या तीन उम्मीदवार मॉडल से चलाएँ और सटीकता तथा जोड़े गए usage.cost की तुलना करें।
3. प्रति आइटम लागत को कुल आइटम संख्या से गुणा करके पूरे जॉब का अनुमान लगाएँ, और GET /credits से जाँचें कि आपका बैलेंस पर्याप्त है।

:free वेरिएंट किसी मॉडल का मुफ़्त एंडपॉइंट इस्तेमाल करते हैं जहाँ वह मौजूद हो, लेकिन उन पर अपस्ट्रीम रेट लिमिट सख़्त होती है, इसलिए वे बड़े बैच के बजाय परीक्षण के लिए उपयुक्त हैं। देखें मॉडल id और वेरिएंट।

बड़े रन से पहले

• दैनिक USD सीमा वाली अलग कुंजी बनाएँ ताकि कोई बग आपका बैलेंस खाली न कर सके। तब जॉब 429 daily_limit_reached के साथ रुक जाता है। देखें रेट लिमिट।
• हर अनुरोध पर max_tokens सेट करें।
• पहले 1% आइटम चलाएँ, आउटपुट और लागत जाँचें, फिर बाकी शुरू करें।
• विफल अनुरोधों (कोई भी 4xx या 5xx) पर शुल्क नहीं लगता, इसलिए विफलताओं को दोबारा चलाना सुरक्षित है। देखें त्रुटियाँ।