صنّف آلاف النصوص باستخدام AsyncOpenAI في Python

لتصنيف آلاف النصوص، استخدم AsyncOpenAI، وضع asyncio.Semaphore حول كل استدعاء بحيث لا يكون هناك سوى N طلب قيد التنفيذ، واطلب تسمية JSON من سطر واحد، وأعد تشغيل العناصر التي فشلت فقط. يفعل السكربت أدناه ذلك بالضبط مقابل https://api.cheaptokens.dev/v1 ويطبع إجمالي الرموز من حقل usage في النهاية.

في 31 أكتوبر 2026، وصف Andrej Karpathy (أحد مؤسسي OpenAI، ومدير الذكاء الاصطناعي السابق في Tesla) HackerTube: بناه في حوالي ساعتين واستخدم Sonnet لتضييق تعليقات Hacker News إلى حوالي 40 ألف تعليق يوصي بفيديو YouTube. لا يذكر منشوره الـ prompt أو حجم الدفعة أو التزامن، لذا ما يلي هو النمط القياسي لهذا النوع من المهام، وليس وصفته.

السكربت#

لنفترض أن لديك 5,000 تذكرة دعم وتريد وسم كل واحدة بـ bug أو question أو praise. إليك كل شيء. ثبّت SDK باستخدام pip install openai (Python 3.10+؛ يُظهر python -c "import openai; print(openai.__version__)" ما لديك) وضع مفتاحك في متغير بيئة أولاً.

import asyncio
import json
import os

from openai import AsyncOpenAI

client = AsyncOpenAI(
    base_url='https://api.cheaptokens.dev/v1',
    api_key=os.environ['CHEAPTOKENS_API_KEY'],
    max_retries=2,
    timeout=60,
)
MODEL = 'deepseek-ai/DeepSeek-V4-Flash-0731'
LABELS = {'bug', 'question', 'praise'}
SYSTEM = (
    'Classify the text as bug, question or praise. '
    'Reply with JSON only: {"label": "<bug|question|praise>"}'
)


async def classify(text, sem, usage):
    async with sem:
        r = await client.chat.completions.create(
            model=MODEL,
            temperature=0,
            max_tokens=100,
            messages=[
                {'role': 'system', 'content': SYSTEM},
                {'role': 'user', 'content': text},
            ],
        )
    usage['in'] += r.usage.prompt_tokens
    usage['out'] += r.usage.completion_tokens
    raw = r.choices[0].message.content
    label = json.loads(raw[raw.find('{'):raw.rfind('}') + 1])['label']
    if label not in LABELS:
        raise ValueError(f'unexpected label: {label!r}')
    return label


async def run(texts, concurrency=5, rounds=3):
    sem = asyncio.Semaphore(concurrency)
    usage = {'in': 0, 'out': 0}
    labels, pending = {}, list(range(len(texts)))
    for _ in range(rounds):
        if not pending:
            break
        results = await asyncio.gather(
            *(classify(texts[i], sem, usage) for i in pending),
            return_exceptions=True,
        )
        failed = []
        for i, res in zip(pending, results):
            if isinstance(res, Exception):
                failed.append(i)
            else:
                labels[i] = res
        pending = failed
    return labels, pending, usage


texts = [
    'The export button crashes the app on Android 15.',
    'Does the Pro plan include API access?',
    'Love the new dark theme, thank you!',
]
labels, failed, usage = asyncio.run(run(texts))
print(labels)
print('failed:', failed)
print('tokens in:', usage['in'], 'out:', usage['out'])

ستحصل على labels (فهرس النص → التسمية)، وfailed (الفهارس التي لم تنتج تسمية صالحة أبداً، قائمة فارغة عندما نجح كل شيء) وإجماليات الرموز. معرّف النموذج هو deepseek-ai/DeepSeek-V4-Flash-0731 الكامل؛ خذ معرّفات النماذج الأخرى من وثائق CheapTokens، لأن الأسماء القصيرة تختلف بين المزودين.

رتّب خط الأنابيب للتحقق من أنك فهمت الشكل:

لماذا semaphore وليس 5,000 طلب دفعة واحدة#

asyncio.gather يبدأ كل ما تعطيه إياه بسعادة. أعطه آلاف coroutines دون حد وستبدأ جميعها دفعة واحدة، وهو انفجار من الطلبات يعد طريقة سريعة لأخطاء 429. الـ semaphore هو بوابة: concurrency=5 يعني أن خمسة استدعاءات على الأكثر داخل async with sem في أي لحظة، والباقي ينتظر دوره. لا يزال Python ينشئ جميع المهام، لكن خمسة فقط تتحدث إلى API.

ما الرقم الذي يجب أن يكون عليه concurrency؟ ابدأ منخفضاً، مثل 5 أعلاه، وارفعه طالما بقيت الأخطاء بعيدة. اعتبره إعداداً تضبطه لكل مهمة، وليس ثابتاً تنسخه.

إعادة المحاولة: طبقتان، مهمتان#

الطبقة الأولى مدمجة في SDK. وفقاً لـ openai-python README (تم التحقق في 1 نوفمبر 2026)، تُعاد محاولة أخطاء الاتصال و408 و409 و429 واستجابات 5xx مرتين افتراضياً مع تراجع أسي قصير. max_retries=2 في السكربت يجعل ذلك مرئياً فقط؛ اضبطه أعلى باستخدام client.with_options(max_retries=5) أو إلى 0 لإيقافه. المهلة الافتراضية 10 دقائق والمهلة ترفع APITimeoutError، ولهذا يضبط السكربت timeout=60 للتسميات القصيرة. التفاصيل في مهلة وإعادة محاولات OpenAI Python SDK، وفكرة التراجع وراءها في 429 طلبات كثيرة جداً من AI API.

الطبقة الثانية لك، وتغطي ما لا يستطيع SDK رؤيته: نجح الاستدعاء، لكن النموذج أجاب بنثر دون JSON على الإطلاق (I can't classify this) أو اخترع تسمية تسمى complaint. النص حول {...} لا يهم، لأن الكود يقتطع JSON من الإجابة. ثم يرفع json.loads أو فحص LABELS، وgather(..., return_exceptions=True) يحتفظ بالخطأ كقيمة بدلاً من قتل الدفعة، وينتقل العنصر إلى الجولة التالية. بعد rounds محاولات، ما يبقى في failed هو لك لفحصه يدوياً. إجابة سيئة واحدة من 5,000 تكلفك طلباً إضافياً واحداً، وليس المهمة بأكملها.

يطلب السكربت JSON في الـ prompt ويتحقق منه في الكود. إذا أردت دفع التنسيق بقوة أكبر باستخدام response_format، راجع احصل على JSON صالح من LLM API.

تكلفة المهمة#

الرموز هي المقياس الوحيد، لذا يجمع السكربت usage وتتبع التكلفة. خذ الحالة الشائعة: نصوص قصيرة، تسميات قصيرة. افترض 10,000 نص بحوالي 300 رمز إدخال لكل منها (النص بالإضافة إلى التعليمات) وحوالي 20 رمز إخراج لتسمية JSON:

10,000 × 300 = 3,000,000 رمز إدخال؛ 10,000 × 20 = 200,000 رمز إخراج.

على DeepSeek-V4-Flash تبلغ هذه المهمة $0.065، وعلى GLM-5.3-Flash تبلغ $0.068، وعلى MiniMax-M2.7 تبلغ $0.18. يعرض الجدول نفس المهمة لكل نموذج، بجانب OpenRouter؛ مع احتساب الإدخال والإخراج معاً يخرج أرخص من OpenRouter إجمالاً.

حرّك المنزلقات في الحاسبة إلى أرقامك الخاصة. حقول in وout لطلب واحد، وruns هو عدد الطلبات التي ترسلها.

جانب الإدخال يهيمن هنا: النص يدخل، كلمة واحدة تخرج. لهذا يناسب نموذج بإدخال رخيص ونافذة سياق 380K، مثل DeepSeek-V4-Flash بسعر واحد في أي ساعة، الفلترة بالجملة؛ للتسميات القصيرة، GLM-5.3-Flash وMiniMax-M2.7 بدائل لنفس السكربت، تغير فقط MODEL. لست متأكداً كيف تنمو فاتورة كهذه؟ ما يفوتره عليك AI API يوضح الصورة.

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

شغّل 20 نصاً أولاً وانظر إلى failed وإجماليات الرموز. إذا استخدمت 20 نصاً حوالي 6,000 رمز إدخال، فستستخدم 5,000 حوالي 1,500,000، ويمكنك الضرب قبل الالتزام.

إذا لم يعمل#

• KeyError: 'CHEAPTOKENS_API_KEY': المتغير غير مضبوط في هذا الطرفية. صدّره وابدأ Python مرة أخرى.
• 401 أو 404: مفتاح أو عنوان URL أساسي أو معرّف نموذج خاطئ. خطأ API 401 أو 404 يشرح كل واحد.
• json.JSONDecodeError على العديد من العناصر: الإجابة لا تحتوي على {...}، غالباً لأن max_tokens=100 قطعها. اطبع r.choices[0].finish_reason؛ length يعني أن الحد وصل، لذا ارفع max_tokens.
• العديد من 429 وتشغيل بطيء: اخفض concurrency.

تفاصيل النموذج العام ونقطة النهاية موجودة في وثائق CheapTokens وعلى صفحة الأسعار.