Batch API للـ AI: شغّل المهام بالليل بنصف التكلفة

لو عندك 100 ألف مهمة AI غير عاجلة، Batch API يخليك تشغّلها بهدوء خلال الليل وتدفع أقل بدل ما تحرق الميزانية على طلبات متزامنة.

مستوى القارئ: متوسط

المشكلة باختصار

الطريقة الشائعة إنك تبعت كل صف في CSV كطلب API عادي. ده مفهوم لو المستخدم مستني الرد الآن. لكنه بيبقى اختيار غالي لو المهمة أصلًا مش مستعجلة: تصنيف تذاكر دعم، تلخيص مراجعات منتجات، تنظيف أوصاف كتالوج، أو استخراج حقول من مستندات داخلية.

اللي بيحصل فعلاً إن السيرفر عندك يبدأ يعمل queue داخلية، retry logic، rate limit handling، ومراقبة فشل. بعد شوية تلاقي نفسك بنيت نصف Batch API بنفسك. الطريقة دي بتفشل لما العدد يكبر لأنك بتنافس traffic المستخدمين على نفس الـ rate limits، وبتدفع سعر المعالجة المتزامنة رغم إنك موافق تستنى.

الافتراض هنا إن عندك مهام كثيرة، النتيجة مطلوبة خلال ساعات، وليس خلال ثواني. مثال واقعي: متجر عنده 80 ألف مراجعة منتج جديدة في نهاية الأسبوع، وعايز يصنفها إلى شكاوى جودة، شحن، سعر، أو تجربة استخدام قبل اجتماع الاثنين.

الفكرة بمثال بسيط

ركز في التشبيه العملي: بدل ما تقف على الشباك تطلب 10 آلاف ورقة واحدة واحدة، تسيب ملف مرتب فيه كل الطلبات، وترجع بعد ما الموظف يخلصهم. أنت خسرت السرعة اللحظية، لكن كسبت سعر أقل وتنظيم أوضح.

في OpenAI، Batch API مبني على ملف JSONL ترفعه بغرض batch، وبعدها تنشئ batch بنافذة معالجة. وثائق OpenAI تذكر إن المعالجة غير المتزامنة تتم خلال 24 ساعة وبخصم 50% مقارنة بالطلبات المتزامنة. في Anthropic، Message Batches تعمل بفكرة مشابهة، وتدعم مجموعات كبيرة من الطلبات مع نافذة قد تصل إلى 24 ساعة، وتعرض docs التسعير كخصم 50% على input وoutput tokens.

المهم هنا: Batch API مش موديل جديد. هو طريقة تشغيل مختلفة لنفس نوع المهام. بتكسب تكلفة وسعة تشغيل، وبتخسر زمن الاستجابة الفوري.

مثال تنفيذي: ابنِ ملف JSONL

المثال التالي يحوّل ملف مراجعات إلى JSONL مناسب لفكرة batch. استخدمه كنقطة بداية، وعدّل الـ endpoint والـ model حسب مزودك وحسب التوثيق الحالي.

import csv
import json

input_csv = "reviews.csv"
output_jsonl = "batch_reviews.jsonl"

with open(input_csv, newline="", encoding="utf-8") as src, \
     open(output_jsonl, "w", encoding="utf-8") as out:
    reader = csv.DictReader(src)
    for row in reader:
        review_id = row["review_id"]
        review_text = row["review_text"][:4000]

        task = {
            "custom_id": f"review-{review_id}",
            "method": "POST",
            "url": "/v1/responses",
            "body": {
                "model": "gpt-5.4-mini",
                "input": [
                    {
                        "role": "system",
                        "content": "صنف المراجعة إلى: جودة، شحن، سعر، تجربة استخدام. أعد JSON فقط."
                    },
                    {
                        "role": "user",
                        "content": review_text
                    }
                ]
            }
        }
        out.write(json.dumps(task, ensure_ascii=False) + "\n")

print(f"wrote {output_jsonl}")

وجود custom_id مهم جدًا. هو المفتاح اللي هترجع به تربط نتيجة النموذج بالصف الأصلي في قاعدة البيانات. من غيره، هتضطر تعتمد على ترتيب النتائج، وده هش وقت الفشل أو إعادة التشغيل.

القياس: قبل وبعد

خلينا نحط رقم واضح. لو عندك 100 ألف مراجعة، وكل مراجعة تستهلك في المتوسط 900 input tokens و300 output tokens، فأنت بتعالج تقريبًا 90 مليون input tokens و30 مليون output tokens. لو تكلفة التشغيل المتزامن في سيناريوك طلعت 36 دولار لليلة، الخصم 50% يخلي نفس الحمل حوالي 18 دولار. الرقم تقديري، لكنه كفاية يوضح القرار: المكسب يظهر لما الحجم كبير والانتظار مقبول.

خطوات التطبيق بدون تعقيد

1. افصل المهام غير العاجلة عن مسار المستخدم. لا تضع checkout أو chat مباشر داخل batch.
2. اكتب ملف JSONL وفي كل سطر custom_id ثابت.
3. احسب حجم الملف وعدد الطلبات قبل الرفع. راجع حدود المزود لأن OpenAI وAnthropic يختلفان في التفاصيل.
4. ارفع الملف وأنشئ batch بنافذة 24 ساعة إذا كانت مدعومة.
5. اسحب ملف النتائج، وحدث قاعدة البيانات بناءً على custom_id.
6. سجل الطلبات الفاشلة في ملف مستقل، وأعد تشغيلها في batch صغير بدل ما تعيد الكل.

الـ trade-off هنا

أفضل طريقة لاستخدام Batch API هي الأعمال الخلفية المتكررة. بتكسب تكلفة أقل، ضغط أقل على السيرفر، وفصل واضح بين workload المستخدمين وworkload المعالجة. بتخسر الـ streaming، الرد الفوري، وبعض التحكم اللحظي في ترتيب التنفيذ.

لو batch اتأخر أو انتهى قبل ما يخلص كل الطلبات، لازم نظامك يقبل partial results. ده معناه إنك محتاج حالة لكل صف: pending، succeeded، failed، expired. التكلفة الهندسية هنا مش كبيرة، لكنها ضرورية. بدونها هتتعامل مع الملف كأنه نجح بالكامل، وده خطأ إنتاجي واضح.

متى لا تستخدم هذه الطريقة

لا تستخدم Batch API لو المستخدم مستني الإجابة الآن، أو لو القرار حساس ويتطلب مراجعة فورية، أو لو كل طلب يعتمد على نتيجة الطلب السابق. لا تستخدمه أيضًا لو حجمك 200 طلب في اليوم فقط؛ التوفير لن يبرر التعقيد. في الحالة دي، queue بسيطة مع API متزامن ومراقبة جيدة غالبًا تكفي.

مصادر راجعها قبل التنفيذ

• OpenAI Batch API guide: حدود الملفات، JSONL، ونموذج التشغيل غير المتزامن.
• OpenAI Batch API FAQ: نافذة 24 ساعة، الخصم، وحالات انتهاء batch.
• Anthropic batch processing docs: Message Batches، التسعير، وحدود الاستخدام.

الخطوة التالية

اختار مهمة AI واحدة غير عاجلة عندك، مثل تصنيف تذاكر الأسبوع الماضي، وابنِ لها ملف JSONL من 100 صف فقط. لو الربط بالـ custom_id اشتغل بدون فقدان نتائج، كبّرها إلى 10 آلاف صف.