كيفية أتمتة توليد الصور على نطاق واسع

أسهل طريقة لأتمتة توليد الصور على نطاق واسع دون إدارة عدة واجهات برمجة تطبيقات مختلفة هي فصل سير العمل عن مزود النموذج. ضع كل طلب صورة في طابور واحد، وجّه كل مهمة إلى معرّف نموذج حالي، وأرسل الطلبات المتوافقة عبر مفتاح CometAPI واحد وعنوان الأساس المتوافق مع OpenAI https://api.cometapi.com/v1.

يبني هذا الدرس هذا الخط الأنبوبي باستخدام Python. فهو يقبل مهام المنتجات والإعلانات والمحتوى من طابور بصيغة JSON Lines؛ يختار نموذجًا؛ يحدّ من التوازي؛ يعيد المحاولة عند الإخفاقات المؤقتة؛ يخزّن النتائج سواء كانت قائمة على URL أو base64؛ ويسجّل استهلاك كل مهمة وتكلفتها التقديرية. يحافظ المثال على هذه الأساسيات الإنتاجية ضمن كتلة مدمجة واحدة بحيث يمكنك اختبار سير العمل دون تحويل المقال إلى مرجع برمجي.

كيف يُبسّط واجه برمجة صور موحّد التوليد الدفعي

بنهاية الشرح، سيبدو سير العمل هكذا:

jobs.jsonl → bounded worker pool → CometAPI /v1/images/generations → object storage → manifest.jsonl

يبقى الطابور وطبقة التخزين ملكًا لك. تغيير نماذج الصور يغيّر قيمة model وليس نظام المصادقة أو مسار الطلب الرئيسي. هذه هي الميزة العملية لواجهة صور موحّدة: يصبح اختيار النموذج قرار توجيه داخل خط أنبوبي واحد بدلًا من تكامل منفصل لكل مزود.

ما الذي تحتاجه لأتمتة توليد الصور؟

تحتاج إلى Python 3.10 أو أحدث، حزمة requests، مفتاح CometAPI، موقع إخراج قابل للكتابة، وعلى الأقل معرّف نموذج صور حالي واحد.

ثبت الاعتماد الوحيد:

pip install requests

عيّن مفتاحك على الخادم، ليس في كود المتصفح أو المستودع:

export COMETAPI_KEY="your-key"

عنوان الأساس هو https://api.cometapi.com/v1، وتستخدم مهام النص إلى صورة المتوافقة المسار POST /images/generations. قبل النشر، تحقّق من كل نموذج في كتالوج النماذج المباشر؛ يُرجع الكتالوج المعرّف الحالي، والمسار المدعوم، والميزات، وبيانات التسعير دون الحاجة إلى ترويسة تفويض.

اعتبارًا من 20 أغسطس 2026، سرد الكتالوج المباشر هذين المسارين المفيدين:

الجدول نقطة بداية، وليس ادعاءً بأن لدى النماذج قدرات متطابقة. فالحجم والجودة والصيغة ودعم صور مرجعية وسلوك الاستجابة تبقى خاصة بكل نموذج. راجع سجل النموذج ووثائقه المرتبطة قبل تمرير المعلمات الاختيارية.

كيفية بناء سير عمل توليد صور دفعي في Python

1. امنح كل مهمة معرفًا دائمًا

استخدم كائن JSON واحدًا لكل سطر بحيث يمكن لطابور أو تصدير قاعدة بيانات أو وظيفة جدول بيانات تغذية العامل نفسه:

{"id":"sku-1001","kind":"product","prompt":"Studio product photo of a ceramic coffee dripper on a warm neutral background"}
{"id":"campaign-204","kind":"ad","prompt":"Editorial summer travel image, vivid natural light, wide composition, no text"}
{"id":"blog-088","kind":"content","prompt":"Minimal illustration of a developer automating a creative workflow, no text"}

يصبح المعرّف اسم ملف الإخراج ومفتاح البيان. في الإنتاج، استخدمه كمفتاح idempotency وتجاوز المعرّفات المعلّمة مسبقًا بأنها ناجحة قبل إعادة معالجة الطابور.

2. وجّه حسب نوع المهمة، ثم تحقّق مقابل الكتالوج المباشر

يرسم المثال أعمال المنتجات إلى gpt-image-2 وأعمال الإعلانات أو المحتوى إلى doubao-seedream-4-5-251128. قد تتجاوز مهمة ما هذا الاختيار بحقل model الخاص بها. عند بدء التشغيل، ينزّل العامل الكتالوج العام ويرفض أي معرّف لم يعد مُدرجًا.

هذا أكثر أمانًا من ترميز SDK خاص بمزود عبر التطبيق. يمكنك تغيير مسار في رسم واحد بعد تقييم الجودة والزمن والتكلفة لمدخلاتك الخاصة.

3. قيّد التوازي بدلًا من إطلاق الدفعة كاملة

يبدأ العامل بأربع طلبات متزامنة. هذا الرقم إعداد تطبيقي تحفظي، وليس حد خدمة عالمي. قِس زمن الاستجابة وردود 429 لحسابك، ثم ارفع أو اخفض MAX_WORKERS عن قصد.

يُعاد المحاولة فقط لردود 408 و429 و5xx مع تزايد أُسي وتشتت. أخطاء المصادقة، ومعرّفات النماذج غير الصالحة، والمعلمات غير المدعومة تفشل فورًا لأن إعادة إرسال الطلب السيئ نفسه لا تضيف سوى تأخير.

4. طبّع النتيجة قبل التخزين

لا تُعيد نماذج الصور دائمًا الحاوية نفسها. تحتوي استجابة GPT Image الموثقة على data[0].b64_json؛ وقد تُعيد نماذج متوافقة أخرى data[0].url. يتعامل العامل مع كلاهما، يكتب الصورة إلى ملف مؤقت، ويعيد تسميته فقط بعد نجاح التنزيل أو فك الترميز.

في الإنتاج، استبدل دليل output/ المحلي بـ S3 أو R2 أو GCS أو أي مخزن كائنات آخر. لا تتعامل مع URL مستضاف لدى مزود باعتباره تخزينًا دائمًا إلا إذا نصّت سياسة الاحتفاظ على ذلك صراحة.

5. سجّل الاستخدام والمحاولات والتكلفة التقديرية

تصبح كل نتيجة سطر بيان مدمجًا يحتوي على معرّف المهمة، والنموذج، والمسار المحفوظ، والحالة، والتكلفة التقديرية بالدولار الأمريكي عندما يوفر الكتالوج المباشر بيانات تسعير كافية. تحتفظ المهام الفاشلة بالخطأ بدلًا من الاختفاء من الدفعة.

سكربت Python كامل لتوليد الصور دفعيًا

احفظ ما يلي باسم batch_image_pipeline.py، وضع الطابور بجواره باسم jobs.jsonl، ثم شغّل python3 batch_image_pipeline.py.

import base64, json, os, random, time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
import requests

BASE_URL = "https://api.cometapi.com/v1"
KEY = os.environ["COMETAPI_KEY"]
WORKERS = int(os.getenv("MAX_WORKERS", "4"))
OUT = Path("output")
ROUTES = {
    "product": "gpt-image-2",
    "ad": "doubao-seedream-4-5-251128",
    "content": "doubao-seedream-4-5-251128",
}

catalog = requests.get("https://api.cometapi.com/api/models", timeout=30)
catalog.raise_for_status()
CATALOG = {model["id"]: model for model in catalog.json()["data"]}

def generate(job):
    model = job.get("model", ROUTES[job["kind"]])
    if model not in CATALOG:
        raise ValueError(f"Unknown model: {model}")
    payload = {"model": model, "prompt": job["prompt"], "n": 1}
    if model == "gpt-image-2":
        payload.update(quality="low", size="1024x1024", output_format="jpeg")

    for attempt in range(4):
        response = requests.post(
            f"{BASE_URL}/images/generations",
            headers={"Authorization": f"Bearer {KEY}"},
            json=payload,
            timeout=180,
        )
        if response.status_code not in {408, 429} and response.status_code < 500:
            break
        time.sleep(2**attempt + random.random())
    response.raise_for_status()
    body = response.json()
    item = body["data"][0]

    if item.get("b64_json"):
        data = base64.b64decode(item["b64_json"])
        extension = body.get("output_format", "png")
    else:
        download = requests.get(item["url"], timeout=120)
        download.raise_for_status()
        data = download.content
        extension = {"image/png": "png", "image/webp": "webp"}.get(
            download.headers.get("content-type"), "jpg"
        )
    path = OUT / f"{job['id']}.{extension}"
    path.write_bytes(data)

    price, usage = CATALOG[model].get("pricing") or {}, body.get("usage", {})
    cost = price.get("per_request")
    if cost is None and price.get("input") is not None:
        cost = (usage.get("input_tokens", 0) * price["input"] +
                usage.get("output_tokens", 0) * price["output"]) / 1_000_000
    return {"id": job["id"], "model": model, "path": str(path),
            "estimated_usd": cost * price.get("ratio", 1) if cost is not None else None}

def safe_generate(job):
    try:
        return {"status": "success", **generate(job)}
    except Exception as error:
        return {"id": job["id"], "status": "failed", "error": str(error)}

OUT.mkdir(exist_ok=True)
jobs = [json.loads(line) for line in Path("jobs.jsonl").read_text().splitlines() if line]
with ThreadPoolExecutor(max_workers=WORKERS) as pool:
    results = list(pool.map(safe_generate, jobs))
with (OUT / "manifest.jsonl").open("w") as manifest:
    manifest.writelines(json.dumps(result) + "\n" for result in results)

يستخدم السكربت الكتالوج الحالي وقت التشغيل، بينما تعد خريطتا التوجيه الاحتياطيتان أمثلة تم التحقق منها في 20 أغسطس 2026. أعد التحقق منهما قبل النشر أو نشر الكود في تاريخ آخر.

كيفية اختبار سير عمل توليد الصور الدفعي

ابدأ بمهمة واحدة وعامل واحد:

MAX_WORKERS=1 python3 batch_image_pipeline.py

تتبع استجابة GPT Image الناجحة هذا الهيكل:

{
  "created": 1776841943,
  "output_format": "jpeg",
  "quality": "low",
  "size": "1024x1024",
  "usage": {
    "input_tokens": 16,
    "output_tokens": 208,
    "total_tokens": 224
  },
  "data": [{"b64_json": "<base64-image-data>"}]
}

يفك العامل ترميز الصورة، ويكتب output/<job-id>.jpeg، ويضيف صف نجاح إلى output/manifest.jsonl. إذا أعاد نموذج URL بدلًا من ذلك، يقوم العامل بتنزيله ويخزّن المسار المحلي بالصيغة نفسها في البيان.

تم فحص الكود نحويًا محليًا. لا تزال مكالمة التوليد الحية تتطلب مفتاح CometAPI الخاص بك، لذا شغّل اختبار دخاني بمهمة واحدة قبل زيادة التوازي.

كم تكلفة توليد الصور دفعيًا؟

يجب ختم التسعير بطابع زمني لأن أسعار النماذج تتغير. اعتبارًا من 20 أغسطس 2026، أعاد كتالوج نماذج CometAPI المباشر حقول الأسعار الأساسية التالية ونسبة فوتره 0.8:

• gpt-image-2: 5 دولارات لكل مليون رمز إدخال و30 دولارًا لكل مليون رمز إخراج؛ بتطبيق النسبة المدرجة تصبح المعدلات الفعلية 4 و24 دولارًا لكل مليون رمز.
• doubao-seedream-4-5-251128: 0.04 دولار لكل طلب؛ بتطبيق النسبة المدرجة تصبح 0.032 دولار لكل طلب.

يشرح دليل تسعير CometAPI الفوترة القائمة على الرموز للنماذج ذات التسعير الرسمي والفوترة القائمة على الطلب للنماذج المسعّرة لكل طلب. يقرأ السكربت الكتالوج عند تشغيله ويستخدم القاعدة نفسها:

token cost = ratio × (input tokens × input rate + output tokens × output rate) / 1,000,000

request cost = ratio × per-request price

على سبيل المثال، تُبلغ استجابة GPT Image الموثقة أعلاه عن 16 رمز إدخال و208 رموز إخراج. باستخدام قيم كتالوج 20 أغسطس، يُقدَّر هذا الناتج التوضيحي بحوالي 0.005056 دولار. يتغير الإجمالي الحقيقي وفق النموذج والجودة والحجم والمَحفِّز والمحاولات وسجل الاستخدام في الاستجابة. اعتبر استجابة API ولوحة استخدام الحساب سجل الفوترة، وليس افتراضًا ثابتًا لكل صورة.

ضع ميزانية للعمل غير الناجح أيضًا. قد ينتج عن إعادة المحاولة بعد مهلة غير مؤكدة نتيجة ثانية قابلة للفوترة، كما أن صورة ناجحة تقنيًا فشلت في المراجعة ما تزال تستهلك الميزانية. تتبّع تكلفة API ومعدل القبول معًا:

effective cost per accepted image = total batch spend / approved images

أخطاء شائعة في واجهة توليد الصور وكيفية إصلاحها

لا تعِد المحاولة لكل خطأ. سيبقى طلب 400 دائمًا غير صالح، بينما يمكن أن يحوّل حلقة إعادة محاولات 429 غير المحدودة ذروة الزيارات إلى تكدّس.

أفضل الممارسات لتوليد الصور على نطاق إنتاجي

انتقل من JSON Lines إلى طابور متين عند استخدام عدة عمّال. عيّن مهلة رؤية أطول من أقصى زمن توليد، وأقِرّ بالمهمة فقط بعد تخزين الصورة والبيان، وأرسل المهام المستنفدة إلى طابور أخطاء للمراجعة.

احتفظ بالتحكمات الاختيارية في ضبط خاص لكل نموذج. يجب أن يحتوي الحمولة المشتركة فقط على الحقول المشتركة مثل model وprompt وn: 1؛ أضف quality أو size أو output_format فقط بعد تأكيد وثائق النموذج المحدد لها. إذا أضفت مسار توجيه احتياطيًا، اختر نموذجًا يدعم المهمة نفسها وأعد بناء الحمولة لذلك النموذج بدلًا من إعادة إرسال خيارات خاصة بمزود آخر دون وعي.

اخزن مفتاح API في مدير أسرار، وقيّد إدخال المَحفِّز، وافحص الأصول المُولَّدة وفق سياساتك، وأبقِ روابط المزود خارج سجلات المنتجات طويلة الأمد. سجّل معرّف المهمة، ومعرّف النموذج، والزمن، والمحاولات، والاستخدام، ومسار التخزين، ونتيجة المراجعة، وتاريخ لقطة الكتالوج. تتيح لك هذه الحقول مقارنة النماذج بتكلفة الصورة المقبولة بدلًا من السعر المُعلن فقط.

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

الأسئلة الشائعة حول أتمتة توليد الصور على نطاق واسع

ما أسهل طريقة لأتمتة توليد الصور على نطاق واسع دون إدارة عدة واجهات؟

استخدم طابورًا واحدًا وسير عمل تخزين واحدًا، ثم أرسل الطلبات المتوافقة عبر مفتاح CometAPI واحد وhttps://api.cometapi.com/v1/images/generations. غيّر معرّف النموذج في طبقة التوجيه بدلًا من صيانة مصادقة وSDK منفصلين لكل مزود.

هل يمكنني إرسال طلب واحد وطلب من عدة نماذج صور التوليد في وقت واحد؟

يرسل المثال نموذجًا واحدًا لكل مهمة. التفرّع عملية تطبيقية: انسخ مهمة مع معرّفات ونماذج مميزة، ثم قارن المخرجات المخزنة. هذا يحفظ التكلفة وحالة المراجعة منسوبة إلى كل نموذج.

ما درجة التوازي المناسبة؟

لا يوجد رقم عالمي لكل حساب ونموذج. ابدأ بمجموعة صغيرة محدودة مثل أربعة عمّال، راقب الزمن وردود 429، واضبط بناءً على الأدلة.

هل يجب أن أخزّن URL المُعاد أم الصورة نفسها؟

اخزّن الصورة في مخزن كائنات خاص بك. قد يكون URL المُعاد مؤقتًا، بينما قد تعيد نماذج GPT Image محتوى base64 بدلًا من URL.

كيف أختار النموذج الأرخص؟

احسب تكلفة الصورة المقبولة، وليس فقط سعر المكالمة. ضمّن رسوم الرموز أو الطلب، والمحاولات، والتنزيلات الفاشلة، والأصول المرفوضة، والمعالجة اللاحقة، والمراجعة البشرية. أعد فحص كتالوج النماذج المباشر في اليوم الذي تنشر فيه أو تنشر الكود.

أين ينبغي أن أتحقق من نقطة النهاية وصيغة الاستجابة؟

استخدم البدء السريع لـ CometAPI، ووثائق كتالوج النماذج، ومرجع توليد الصور، ودليل التسعير.