برای توسعه‌دهندگانی که با مدل‌های زبانی بزرگ (LLM) کار می‌کنند، هزینه و تأخیر (Latency) اغلب گلوگاه‌های اصلی هنگام پردازش مجموعه‌های داده بزرگ هستند. چه در حال تولید توضیحات محصولات برای یک سایت تجارت الکترونیک با میلیون‌ها قلم کالا باشید، چه خلاصه‌سازی هزاران بلیت پشتیبانی، یا انجام تحلیل احساسات روی لاگ‌های تاریخی، انجام فراخوانی‌های API فردی برای هر قلم، ناکارآمد و پرهزینه است.
در اینجا API دسته‌ای OpenAI وارد می‌شود. این قابلیت به شما امکان می‌دهد حجم بالایی از درخواست‌های غیرفوری را با تخفیف قابل توجهی به OpenAI ارسال کنید. با بسته‌بندی درخواست‌ها در یک شغل (Job) واحد که به صورت ناهمگام (Asynchronous) پردازش می‌شود، می‌توانید هزینه توکن‌های ورودی و خروجی خود را تا ۵۰٪ در مقایسه با فراخوانی‌های API بلادرنگ کاهش دهید. در این راهنما، بررسی خواهیم کرد که چگونه API دسته‌ای را به طور مؤثر برای پردازش مقیاس بزرگ داده پیاده‌سازی کنیم.
چرا از API دسته‌ای استفاده کنیم؟
مزیت اصلی API دسته‌ای، کارایی از نظر هزینه است. OpenAI یک تخفیف ۵۰ درصدی برای تمام توکن‌های استفاده شده در درخواست‌های دسته‌ای ارائه می‌دهد. با این حال، این مزیت با برخی مصالحه‌ها همراه است:
- پردازش ناهمگام: نمی‌توانید انتظار پاسخ فوری داشته باشید. شغل‌ها معمولاً در عرض ۲۴ ساعت تکمیل می‌شوند.
- عدم تعامل بلادرنگ: این روش برای چت‌بات‌ها یا کاربردهایی که نیاز به بازخورد فوری کاربر دارند، مناسب نیست.
- اندازه ثابت دسته: باید تمام درخواست‌ها را قبل از ارسال از قبل آماده کنید.
اگر کاربرد شما شامل تحلیل آفلاین، تولید محتوا یا برچسب‌گذاری داده است، API دسته‌ای یک انتخاب ایده‌آل است.
گام ۱: آماده‌سازی فایل JSONL
اولین گام در هر شغل دسته‌ای، ایجاد یک فایل JSON Lines (JSONL) است. هر خط در این فایل نماینده یک درخواست API واحد است. ساختار باید با قالب استاندارد درخواست API OpenAI مطابقت داشته باشد، از جمله یک `custom_id` منحصربه‌فرد برای ردیابی.
import json
def create_batch_file(data_points, output_filename="batch_input.jsonl"):
"""
یک فایل JSONL برای API دسته‌ای OpenAI تولید می‌کند.
Args:
data_points: لیستی از رشته‌هایی که باید پردازش شوند.
output_filename: نام فایل JSONL خروجی.
"""
with open(output_filename, 'w') as f:
for i, text in enumerate(data_points):
request = {
"custom_id": f"request-{i}",
"method": "POST",
"url": "/v1/chat/completions",
"body": {
"model": "gpt-4o",
"messages": [
{
"role": "user",
"content": f"Summarize the following text in one sentence: {text}"
}
],
"max_tokens": 100
}
}
f.write(json.dumps(request) + '\n')
print(f"Batch file '{output_filename}' created successfully.")
# مثال استفاده
sample_data = [
"The quick brown fox jumps over the lazy dog.",
"Python is a high-level programming language known for its readability.",
"Machine learning is a subset of artificial intelligence."
]
create_batch_file(sample_data)گام ۲: بارگذاری فایل و ایجاد شغل دسته‌ای
پس از آماده شدن فایل JSONL، باید آن را به OpenAI بارگذاری کرده و با استفاده از SDK پایتون، شغل دسته‌ای را آغاز کنید.
from openai import OpenAI
import time
client = OpenAI() # کلید API از متغیر محیطی خوانده می‌شود
# ۱. بارگذاری فایل
file_response = client.files.create(
file=open("batch_input.jsonl", "rb"),
purpose="batch"
)
file_id = file_response.id
print(f"File uploaded with ID: {file_id}")
# ۲. ایجاد شغل دسته‌ای
batch = client.batches.create(
input_file_id=file_id,
endpoint="/v1/chat/completions",
completion_window="24h"
)
batch_id = batch.id
print(f"Batch job created with ID: {batch_id}")گام ۳: پایش تکمیل
از آنجا که شغل دسته‌ای ناهمگام است، باید وضعیت را پایش کنید تا زمانی که تکمیل شود. بسته به اندازه شغل، این کار ممکن است چند ساعت طول بکشد. برای سیستم‌های تولیدی (Production)، توصیه می‌شود به جای مسدود کردن در یک اسکریپت، از یک شغل cron یا سیستم اعلان مبتنی بر وب‌هوک استفاده کنید.
def poll_batch_status(batch_id, client, interval=60):
"""
وضعیت شغل دسته‌ای را تا زمانی که تکمیل یا شکست بخورد پایش می‌کند.
"""
while True:
batch = client.batches.retrieve(batch_id=batch_id)
status = batch.status
print(f"Current status: {status}")
if status in ["completed", "failed", "expired", "cancelled"]:
return batch
time.sleep(interval)
# انتظار برای تکمیل
final_batch = poll_batch_status(batch_id, client, interval=300) # بررسی هر ۵ دقیقهگام ۴: دریافت و پردازش نتایج
پس از آنکه دسته به عنوان `completed` علامت‌گذاری شد، می‌توانید فایل خروجی را دانلود کنید. این فایل پاسخ‌های متناظر با هر `custom_id` از فایل ورودی را شامل می‌شود.
def download_and_process_results(batch_id, client):
"""
فایل خروجی دسته را دانلود کرده و نتایج را تجزیه می‌کند.
"""
# دریافت شناسه فایل خروجی
output_file_id = client.batches.retrieve(batch_id).output_file_id
if output_file_id is None:
raise ValueError("No output file found. Job may have failed.")
# دریافت محتوای فایل
file_content = client.files.content(output_file_id)
# تجزیه محتوای JSONL
results = []
for line in file_content.splitlines():
if line.strip():
data = json.loads(line)
# استخراج محتوا از پاسخ
response_content = data.get('response', {}).get('body', {}).get('choices', [{}])[0].get('message', {}).get('content', '')
results.append({
'id': data.get('custom_id'),
'summary': response_content
})
return results
# پردازش نتایج
results = download_and_process_results(batch_id, client)
for r in results:
print(f"{r['id']}: {r['summary']}")بهترین روش‌ها برای پردازش دسته‌ای
- مدیریت مناسب خطاها: برخی درخواست‌ها ممکن است به دلیل نقض سیاست‌های محتوا یا مشکلات قالب‌بندی شکست بخورند. همیشه فیلد `error` در خطوط JSONL خروجی را بررسی کنید.
- تکه‌تکه کردن مجموعه‌های داده بزرگ: اگرچه محدودیت سخت‌گیرانه‌ای وجود ندارد، اما فایل‌های بسیار بزرگ می‌توانند عیب‌یابی دشواری داشته باشند. در نظر بگیرید که میلیون‌ها ردیف را به تکه‌هایی از ۱۰,۰۰۰ تا ۵۰,۰۰۰ ردیف تقسیم کنید تا مدیریت آسان‌تری داشته باشید.
- کنترل نسخه برای پرامپت‌ها: از آنجا که شغل‌های دسته‌ای زمان‌بر هستند، اطمینان حاصل کنید که قالب پرامپت شما پایدار است. اگر نیاز به تغییر پرامپت دارید، باید یک فایل جدید و یک شغل دسته‌ای جدید ایجاد کنید.
- پایش هزینه‌ها: از داشبورد OpenAI برای ردیابی استفاده از توکن و هزینه‌های هر شغل دسته‌ای استفاده کنید تا اطمینان حاصل کنید که حداکثر صرفه‌جویی را دارید.
نتیجه‌گیری
API دسته‌ای OpenAI یک ابزار قدرتمند برای توسعه‌دهندگانی است که به دنبال مقیاس‌پذیر کردن کاربردهای هوش مصنوعی خود هستند، بدون اینکه هزینه‌هایشان به صورت خطی افزایش یابد. با بهره‌گیری از پردازش ناهمگام، می‌توانید چالش‌های داده‌ای مقیاس بزرگ را حل کنید که با فراخوانی‌های API بلادرنگ، بسیار پرهزینه یا کند خواهند بود. با ادامه تکامل کاربردهای LLM، تسلط بر پردازش دسته‌ای یک مهارت حیاتی برای ساخت سیستم‌های هوش مصنوعی کارآمد و مقرون‌به‌صرفه خواهد بود.
با یک مجموعه داده آزمایشی کوچک شروع کنید، نتایج خود را تأیید کنید و سپس مقیاس را افزایش دهید. تخفیف ۵۰ درصدی، API دسته‌ای را به یک بخش ضروری در هر استک توسعه LLM جدی تبدیل کرده است.