پردازش دسته‌ای

هشدار

پشتیبانی Batch API در AvalAI در حال توسعه است. از این راهنما برای آماده‌سازی workloadهای دسته‌ای و اجرای جایگزین فعلی با کنترل rate limit در سمت کلاینت استفاده کنید.

پردازش دسته‌ای برای درخواست‌های مستقل زیادی مناسب است که به پاسخ فوری نیاز ندارند: eval شبانه، classification آفلاین، تولید embeddings، پاکسازی داده یا بازبینی محتوا. وقتی Batch API میزبانی‌شده فعال شود، از /v1/batches استفاده کنید. تا آن زمان، از concurrency کنترل‌شده سمت کلاینت همراه با retry و headerهای rate limit استفاده کنید.

این راهنما با اقتباس از راهنمای رسمی Batch API و الگوهای OpenAI Cookbook، با تغییرات endpoint، کلید API و مدل‌های AvalAI تهیه شده است.

نکته

Batch API میزبانی‌شده در OpenAI مزایای مخصوص OpenAI مانند تخفیف ۵۰٪ و یک pool جدا با محدودیت بالاتر دارد. این موارد را هنوز تضمین AvalAI فرض نکنید. الگوی پشتیبانی‌شده فعلی یک worker سمت کلاینت است که همان قیمت‌گذاری و محدودیت نرخ مسیر/مدل عادی AvalAI را مصرف می‌کند.

انتخاب الگوی مناسب

workload	مسیر پیشنهادی فعلی
پاسخ فوری به کاربر	درخواست synchronous با Chat Completions یا Responses
تعداد زیادی درخواست مستقل	پردازنده سمت کلاینت با کنترل rate limit
بررسی regression پرامپت یا مدل	ارزیابی با Promptfoo و AvalAI
batch async میزبانی‌شده در آینده	/v1/batches، پس از فعال شدن

شکل Batch میزبانی‌شده برای برنامه‌ریزی

جریان فعلی Batch API در OpenAI این است: فایل .jsonl را آماده کنید، آن را با purpose="batch" آپلود کنید، با input_file_id یک batch بسازید، وضعیت شی batch را poll کنید، سپس output_file_id را دانلود و error_file_id را برای ردیف‌های ناموفق بررسی کنید. workloadهای دسته‌ای AvalAI را با همین چرخه طراحی کنید تا بعد از فعال شدن batch میزبانی‌شده، مهاجرت بیشتر در سطح transport باشد.

endpointهای هدف در شکل OpenAI-compatible فعلی شامل موارد زیر هستند:

endpoint	کاربرد مناسب
/v1/responses	reasoning، extraction، summarization و تولید آفلاین بدون ابزار زنده
/v1/chat/completions	workloadهای chat فعلی که باید روی Chat Completions بمانند
/v1/embeddings	jobهای embedding برای repository، catalog یا document
/v1/moderations	صف‌های بزرگ moderation برای متن یا تصویر
/v1/images/generations و /v1/images/edits	تولید یا ویرایش تصویر به‌صورت آفلاین
/v1/videos	صف‌های render ویدیو، وقتی بدنه JSON پشتیبانی شود

نکته‌های endpoint-specific بر اساس شکل Batch در OpenAI:

• هر فایل ورودی را به یک endpoint و یک مدل محدود کنید.
• در ردیف‌های batch مقدار stream: true نگذارید؛ خروجی batch بعدا از طریق فایل‌ها برمی‌گردد، نه stream زنده.
• batchهای /v1/embeddings همچنین به ۵۰٬۰۰۰ ورودی embedding در کل درخواست‌ها محدود می‌شوند.
• ردیف‌های /v1/moderations باید input داشته باشند؛ برای ورودی متن به‌همراه تصویر از omni-moderation-latest استفاده کنید و برای تصویرهای بزرگ image_url را به base64 ترجیح دهید.
• ردیف‌های batch برای /v1/videos باید بدنه JSON داشته باشند. assetها را از قبل upload کنید و به جای multipart upload با file ID یا image URL پشتیبانی‌شده به آن‌ها ارجاع دهید.

آماده‌سازی ورودی JSONL

هر درخواست را در یک خط قرار دهید. این شکل هم برای پردازش محلی مناسب است و هم بعدا می‌تواند برای batch میزبانی‌شده استفاده شود.

{"custom_id":"ticket-001","method":"POST","url":"/v1/chat/completions","body":{"model":"gpt-5.5","messages":[{"role":"system","content":"Classify the ticket as Hardware, Software, Billing, Account, or Other. Return only the label."},{"role":"user","content":"My monitor does not turn on."}]}}
{"custom_id":"ticket-002","method":"POST","url":"/v1/chat/completions","body":{"model":"gpt-5.5","messages":[{"role":"system","content":"Classify the ticket as Hardware, Software, Billing, Account, or Other. Return only the label."},{"role":"user","content":"I was charged twice this month."}]}}

از مقدارهای پایدار برای custom_id استفاده کنید تا retry و نتایج به‌درستی reconcile شوند. ترتیب خروجی تضمین نمی‌کند با ترتیب ورودی یکی باشد، پس custom_id را کلید اتصال ورودی و خروجی بدانید.

ردیف‌های Batch برای Responses API

برای jobهای جدید تولید متن، یک ردیف با شکل Responses را ترجیح دهید. این الگو instructions را از input کاربر جدا نگه می‌دارد و مهاجرت نهایی به /v1/responses را مستقیم‌تر می‌کند.

{"custom_id":"ticket-001","method":"POST","url":"/v1/responses","body":{"model":"gpt-5.5","instructions":"Classify the ticket as Hardware, Software, Billing, Account, or Other. Return only the label.","input":"My monitor does not turn on."}}
{"custom_id":"ticket-002","method":"POST","url":"/v1/responses","body":{"model":"gpt-5.5","instructions":"Classify the ticket as Hardware, Software, Billing, Account, or Other. Return only the label.","input":"I was charged twice this month."}}

اجرای batch سمت کلاینت به شکل امن

import json
import os
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["AVALAI_API_KEY"],
    base_url="https://api.avalai.ir/v1",
)


def run_request(item):
    if item.get("method", "POST") != "POST":
        raise ValueError(f"Unsupported method: {item.get('method')}")

    url = item["url"]
    body = item["body"]

    if url == "/v1/chat/completions":
        response = client.chat.completions.create(**body)
        return response.choices[0].message.content, response.model_dump()

    if url == "/v1/responses":
        response = client.responses.create(**body)
        return response.output_text, response.model_dump()

    if url == "/v1/embeddings":
        response = client.embeddings.create(**body)
        vector_lengths = [
            len(embedding_item.embedding) for embedding_item in response.data
        ]
        return vector_lengths, response.model_dump()

    raise ValueError(f"Unsupported batch row URL: {url}")


def run_one(item, max_retries=5):
    for attempt in range(max_retries):
        try:
            output, response_body = run_request(item)
            return {
                "custom_id": item["custom_id"],
                "url": item["url"],
                "output": output,
                "response": response_body,
            }
        except Exception as exc:
            message = str(exc)
            if "429" not in message and "rate" not in message.lower():
                raise
            wait = min(60, 2**attempt)
            time.sleep(wait)

    raise RuntimeError(f"Retries exhausted for {item['custom_id']}")


with open("requests.jsonl", "r", encoding="utf-8") as f:
    requests = [json.loads(line) for line in f if line.strip()]

results = []
with ThreadPoolExecutor(max_workers=4) as pool:
    futures = [pool.submit(run_one, item) for item in requests]
    for future in as_completed(futures):
        results.append(future.result())

with open("results.jsonl", "w", encoding="utf-8") as f:
    for row in results:
        f.write(json.dumps(row, ensure_ascii=False) + "\n")

این worker محلی ردیف‌ها را بر اساس url مسیر‌دهی می‌کند؛ بنابراین یک مسیر کد می‌تواند ردیف‌های Chat Completions، Responses و Embeddings را پردازش کند. پیش از استفاده از routeهای دیگری مثل Images یا Moderations، handler صریح برای آن‌ها اضافه کنید.

با concurrency کم شروع کنید و فقط بعد از بررسی headerهای rate limit آن را افزایش دهید. برای jobهای بزرگ، فایل ورودی را به chunkهای کوچک‌تر تقسیم کنید و بعد از هر chunk نتیجه را checkpoint کنید.

افزودن polling وضعیت و callback اختصاصی

الگوهای async میزبانی‌شده در OpenAI به دو ایده نزدیک تقسیم می‌شوند: Batch API برای تعداد زیادی ردیف مستقل، و Responses در حالت background برای یک پاسخ طولانی که با شناسه قابل poll کردن است. Batch میزبانی‌شده و background Responses هنوز به‌صورت عمومی در AvalAI در دسترس نیستند، پس همان تجربه توسعه‌دهنده را در worker خودتان پیاده کنید:

1. یک رکورد job محلی با وضعیت‌های queued، in_progress، completed، failed و cancelled بسازید.
2. هر ردیف ورودی را با custom_id، نسخه پرامپت، مدل و تعداد retry ذخیره کنید.
3. workerها را با concurrency محدود اجرا کنید و نتیجه‌های جزئی را به محض پایان هر ردیف بنویسید.
4. endpointی مثل GET /jobs/{id} بدهید تا کلاینت‌ها بدون باز نگه داشتن اتصال HTTP وضعیت را poll کنند.
5. اگر callback لازم است، بعد از رسیدن job به وضعیت نهایی، webhook برنامه خودتان را ارسال کنید.

اگر نام وضعیت‌های Batch در OpenAI را در worker خودتان mirror می‌کنید، آن‌ها را صریحا به رفتار محلی نگاشت کنید:

وضعیت شبیه OpenAI	رفتار worker محلی
validating / failed	قبل از شروع فراخوانی API، شکل JSONL، endpoint، موجود بودن مدل و یکتایی custom_idها را validate کنید.
in_progress	workerهای محدود اجرا کنید، headerهای rate limit در AvalAI را رعایت کنید و هر ردیف کامل‌شده را checkpoint کنید.
finalizing / completed	ردیف‌های output و error را بنویسید، شمارنده‌ها را persist کنید و سپس job را terminal علامت بزنید.
expired	ردیف‌های کامل‌شده را نگه دارید، custom_idهای ناتمام را ثبت کنید و فقط کار ناتمام را در job تازه retry کنید.
cancelling / cancelled	زمان‌بندی ردیف‌های جدید را متوقف کنید، اگر امن است فراخوانی‌های درحال اجرا را تمام کنید و نتیجه‌های جزئی را حفظ کنید.

برای callbackهای شبیه webhook، از قواعد عملیاتی راهنمای OpenAI استفاده کنید: سریع با وضعیت 2xx پاسخ دهید، پردازش سنگین را به worker پس‌زمینه منتقل کنید، تحویل‌های ناموفق را با backoff دوباره تلاش کنید، و eventها را با یک شناسه پایدار deduplicate کنید. برای callbackهایی که ارسال یا دریافت می‌کنید signing secret نگه دارید تا سیستم مقصد بتواند payload را verify کند.

نیاز	الگوی امن فعلی با AvalAI
یک پاسخ طولانی که ممکن است از timeout کلاینت عبور کند	یک درخواست Responses یا Chat Completions را در worker خودتان queue کنید و رکورد job خودتان را poll کنید
هزاران ردیف مستقل	از الگوی JSONL + worker سمت کلاینت در همین راهنما استفاده کنید
اعلان خارجی پس از تکمیل	بعد از ذخیره نتیجه‌ها، از worker خودتان webhook برنامه را ارسال کنید
مهاجرت آینده به Batch میزبانی‌شده	.jsonl، custom_id و متادیتای شبیه output_file_id/error_file_id را در جدول job نگه دارید

چک‌لیست عملیاتی

• از custom_idهای idempotent استفاده کنید.
• ردیف‌های failed را جدا ذخیره کنید تا بدون تکرار کل job دوباره اجرا شوند.
• concurrency را هم بر اساس requests per minute و هم tokens per minute محدود کنید.
• برای پاسخ‌های 429 از exponential backoff استفاده کنید.
• پرامپت‌ها و مدل‌ها را همراه dataset نسخه‌بندی کنید.
• وضعیت job را persist کنید تا کلاینت‌ها بتوانند امن poll کنند و دوباره وصل شوند.
• هر callback برنامه‌ای را که برای سیستم‌های downstream می‌فرستید sign کنید.
• فایل‌های JSONL آپلودی را زیر ۲۰۰ مگابایت نگه دارید و jobهای خیلی بزرگ را به shardهای کوچک‌تر تقسیم کنید.
• پس از فعال شدن batch میزبانی‌شده، output_file_id و error_file_id را همراه رکورد job ذخیره کنید.
• انتظار داشته باشید فایل‌های خروجی کامل‌شده طبق سیاست نگه‌داری ارائه‌دهنده منقضی شوند؛ رفتار مرجع OpenAI حذف فایل‌های خروجی batch پس از ۳۰ روز است.
• برای batchهای منقضی‌شده، ردیف‌های کامل‌شده را نگه دارید و فقط custom_idهای ناتمام را از فایل error دوباره اجرا کنید.
• برای workloadهای eval از ارزیابی با Promptfoo و AvalAI استفاده کنید تا مقایسه نتایج در CI آسان‌تر شود.

منابع مرتبط

• مرجع Batch API
• محدودیت‌های نرخ
• درخواست‌های موازی سازگار با Rate Limit
• هدرهای پاسخ
• پردازش پس‌زمینه
• Webhookها
• راهنمای Batch API در OpenAI
• راهنمای Background mode در OpenAI
• راهنمای Webhooks در OpenAI