Python میں AsyncOpenAI کے ساتھ ہزاروں متنوں کی درجہ بندی کریں

ہزاروں متنوں کی درجہ بندی کرنے کے لیے، AsyncOpenAI استعمال کریں، ہر کال کے گرد ایک asyncio.Semaphore رکھیں تاکہ صرف N درخواستیں زیرِ عمل رہیں، ایک سطری JSON لیبل طلب کریں، اور صرف ناکام ہونے والے آئٹمز کو دوبارہ چلائیں۔ نیچے دیا گیا اسکرپٹ بالکل یہی کرتا ہے https://api.cheaptokens.dev/v1 کے خلاف اور آخر میں usage فیلڈ سے کل ٹوکنز پرنٹ کرتا ہے۔

31 اکتوبر 2026 کو Andrej Karpathy (OpenAI کے شریک بانی، Tesla AI کے سابق ڈائریکٹر) نے HackerTube کی وضاحت کی: انہوں نے اسے تقریباً دو گھنٹے میں بنایا اور Sonnet کا استعمال کرتے ہوئے Hacker News کے تبصروں کو تقریباً 40K تک محدود کیا جو YouTube ویڈیو تجویز کرتے ہیں۔ ان کی پوسٹ میں پرامپٹ، بیچ سائز یا کنکرنسی نہیں دی گئی، لہٰذا جو آگے آتا ہے وہ اس قسم کے کام کے لیے معیاری پیٹرن ہے، ان کی ترکیب نہیں۔

اسکرپٹ#

فرض کریں آپ کے پاس 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 ڈاکس سے لیں، کیونکہ مختصر نام فراہم کنندگان کے درمیان مختلف ہوتے ہیں۔

پائپ لائن کو ترتیب دیں تاکہ آپ جانچ سکیں کہ آپ کے ذہن میں شکل درست ہے:

سیمفور کیوں اور ایک وقت میں 5,000 درخواستیں کیوں نہیں#

asyncio.gather خوشی سے ہر وہ چیز شروع کر دیتا ہے جو آپ اسے دیتے ہیں۔ اسے بغیر کسی حد کے ہزاروں کوروٹینز دیں اور وہ سب ایک ساتھ شروع ہو جاتے ہیں، درخواستوں کا ایک دھماکہ جو 429 غلطیوں کا تیز راستہ ہے۔ سیمفور ایک ٹرن اسٹائل ہے: 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 timeout and retries میں ہیں، اور اس کے پیچھے بیک آف کا خیال 429 Too Many Requests from an AI API میں ہے۔

دوسری پرت آپ کی ہے، اور یہ وہ چیز ڈھانپتی ہے جو SDK نہیں دیکھ سکتا: کال کامیاب ہوئی، لیکن ماڈل نے نثر میں جواب دیا اور کوئی JSON نہیں (I can't classify this) یا complaint نام کا لیبل ایجاد کیا۔ {...} کے ارد گرد متن سے کوئی فرق نہیں پڑتا، کیونکہ کوڈ جواب سے JSON کاٹ لیتا ہے۔ پھر json.loads یا LABELS چیک استثنا پیدا کرتا ہے، gather(..., return_exceptions=True) غلطی کو بطور ویلیو رکھتا ہے بجائے بیچ کو مارنے کے، اور آئٹم اگلے راؤنڈ میں چلا جاتا ہے۔ rounds کوششوں کے بعد، جو کچھ failed میں باقی ہے وہ آپ کے ہاتھ سے معائنہ کرنے کے لیے ہے۔ 5,000 میں سے ایک خراب جواب آپ کو ایک اضافی درخواست کا خرچ دیتا ہے، پورے کام کا نہیں۔

اسکرپٹ پرامپٹ میں JSON طلب کرتا ہے اور کوڈ میں اس کی توثیق کرتا ہے۔ اگر آپ response_format کے ساتھ فارمیٹ کو مزید سخت کرنا چاہتے ہیں، تو Get valid JSON from an LLM API دیکھیں۔

کام کا خرچ#

ٹوکنز واحد پیمانہ ہیں، لہٰذا اسکرپٹ usage کا مجموعہ کرتا ہے اور لاگت اس کے مطابق ہوتی ہے۔ عام صورت لیں: مختصر متن، مختصر لیبل۔ فرض کریں 10,000 متن ہر ایک تقریباً 300 ان پٹ ٹوکنز (متن جمع ہدایت) اور JSON لیبل کے لیے تقریباً 20 آؤٹ پٹ ٹوکنز:

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 تبدیل کرتے ہیں۔ یقین نہیں کہ اس طرح کا بل کیسے بڑھتا ہے؟ What an AI API bills you for میں تصویر موجود ہے۔

بڑے رن سے پہلے#

پہلے 20 متن چلائیں اور failed اور ٹوکن کے مجموعے دیکھیں۔ اگر 20 متن تقریباً 6,000 ان پٹ ٹوکنز استعمال کرتے ہیں، تو 5,000 تقریباً 1,500,000 استعمال کریں گے، اور آپ پابندی سے پہلے ضرب لگا سکتے ہیں۔

اگر یہ کام نہ کرے#

• KeyError: 'CHEAPTOKENS_API_KEY': متغیر اس ٹرمینل میں سیٹ نہیں ہے۔ اسے ایکسپورٹ کریں اور Python دوبارہ شروع کریں۔
• 401 یا 404: غلط کلید، بیس URL یا ماڈل آئی ڈی۔ API 401 or 404 error ہر ایک سے گزرتا ہے۔
• بہت سے آئٹمز پر json.JSONDecodeError: جواب میں {...} نہیں ہے، اکثر اس لیے کہ max_tokens=100 نے اسے کاٹ دیا۔ r.choices[0].finish_reason پرنٹ کریں؛ length کا مطلب ہے حد لگ گئی، لہٰذا max_tokens بڑھائیں۔
• بہت سے 429 اور سست رن: concurrency کم کریں۔

عام ماڈل اور اینڈ پوائنٹ کی تفصیلات CheapTokens ڈاکس اور قیمتوں کے صفحے پر موجود ہیں۔