دليل: المعالجة الدفعية

مهام الدفعات (ترجمة كتالوج منتجات، أو تصنيف تذاكر الدعم، أو استخراج حقول من آلاف المستندات) ترسل النوع نفسه من الطلبات مرات كثيرة. العمل بسيط، والصعب هو الإنتاجية والإخفاقات والتكلفة. يغطي هذا الدليل الثلاثة باستخدام حزمة openai الرسمية مع Base URL https://api14.com/api/v1.

لا توجد نقطة نهاية دفعية منفصلة: Batch API من OpenAI غير متاحة في api14 (راجع التوافق مع OpenAI SDK). تشغّل الدفعات كطلبات عادية بتزامن مضبوط.

اعرف حد الطلبات

تُحتسب الحدود بعدد الطلبات في الدقيقة (RPM) لكل مفتاح API، وتحدها خطتك:

المفتاح الذي له حد RPM خاص يأخذ الأدنى من الاثنين. تحقق من حدك عبر GET /key (يكون rate_limit.requests يساوي null عندما يستخدم المفتاح حد الخطة)، راجع الرصيد ومعلومات المفتاح. تجاوز الحد يُعيد 429 rate_limited مع Retry-After: 60.

يحدد 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           # 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))

ماذا يفعل هذا الكود:

• تنظيم الإيقاع: يبدأ Pacer على الأكثر RPM طلبًا في الدقيقة، ويحدد semaphore عدد الطلبات قيد التنفيذ.
• إعادة المحاولة: يعيد SDK محاولة 429 و5xx بتراجع أُسّي يحترم Retry-After (max_retries=4). وتُعاد محاولة 429 الصادرة من المزوّد بالطريقة نفسها.
• الرموز القاتلة توقف المهمة: daily_limit_reached وinsufficient_balance لن تزولا بإعادة المحاولة، لذا تخرج المهمة بدل استنفاد المحاولات.
• نقاط الحفظ: تُلحَق النتائج بملف JSONL عند اكتمالها. إعادة تشغيل السكربت تتخطى المعرّفات المنجزة، فلا يضيع شيء عند انهيار أو إيقاف المهمة.

في Node تعمل البنية نفسها بـ Promise.all على عدد ثابت من العمال يسحبون من طابور مشترك، وينتظر كل منهم 60_000 / RPM ملّي ثانية بين البدايات.

ضع عدة عناصر في طلب واحد

للعناصر القصيرة، طلب لكل عنصر يهدر 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"]

تحقق من عودة كل معرّف إدخال، وأعد إرسال المفقود منها. إذا كان finish_reason يساوي "length" فالدفعة كانت أكبر من max_tokens، فقسّمها. راجع المخرجات المنظمة. وللـ embeddings مرّر حتى 100 سلسلة في مصفوفة input واحدة، راجع Embeddings.

خفّض التكلفة بالتخزين المؤقت للمطالبات

في الدفعة تكون التعليمات والأمثلة والمخطط متطابقة في كل طلب. ضعها أولًا والعنصر المتغير أخيرًا، ليعيد المزوّدون استخدام البادئة المخزنة:

• OpenAI وDeepSeek وGemini 2.5 وغيرها تخزّن مؤقتًا تلقائيًا متى صارت البادئة طويلة بما يكفي (1,024 رمزًا لدى OpenAI). قراءة الذاكرة المؤقتة تكلف جزءًا من الإدخال العادي.
• نماذج 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 النقطة المجانية للنموذج حيث توجد، لكن حدود الطلبات لدى المزوّد عليها أشد، فتناسب الاختبارات أكثر من الدفعات الكبيرة. راجع معرّفات النماذج والمتغيرات.

قبل تشغيل كبير

• أنشئ مفتاحًا مخصصًا بـ حد يومي بالدولار حتى لا يستنزف خطأ برمجي رصيدك. عندها تتوقف المهمة بالخطأ 429 daily_limit_reached. راجع حدود الطلبات.
• اضبط max_tokens في كل طلب.
• شغّل أول 1% من العناصر، وافحص الناتج والتكلفة، ثم ابدأ الباقي.
• الطلبات الفاشلة (أي 4xx أو 5xx) لا تُحاسَب، فإعادة تشغيل الفاشل آمنة. راجع الأخطاء.