Python में AsyncOpenAI के साथ हजारों टेक्स्ट का वर्गीकरण करें

हजारों टेक्स्ट को वर्गीकृत करने के लिए, AsyncOpenAI का उपयोग करें, हर कॉल के चारों ओर एक asyncio.Semaphore रखें ताकि केवल N अनुरोध ही प्रगति पर हों, एक-पंक्ति JSON लेबल मांगें, और केवल उन्हीं आइटम को फिर से चलाएं जो विफल हुए। नीचे दिया गया स्क्रिप्ट https://api.cheaptokens.dev/v1 के विरुद्ध ठीक यही करता है और अंत में usage फ़ील्ड से कुल टोकन प्रिंट करता है।

31 अक्टूबर 2026 को आंद्रेज कारपैथी (OpenAI के सह-संस्थापक, टेस्ला के पूर्व 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 टाइमआउट और पुनः प्रयास में हैं, और इसके पीछे का बैकऑफ़ विचार AI API से 429 Too Many Requests में है।

दूसरी परत आपकी है, और यह वह कवर करती है जो SDK नहीं देख सकता: कॉल सफल हुआ, लेकिन मॉडल ने गद्य में उत्तर दिया और कोई JSON नहीं (I can't classify this) या complaint नामक एक आविष्कृत लेबल दिया। {...} के आसपास का टेक्स्ट मायने नहीं रखता, क्योंकि कोड उत्तर से JSON काटता है। फिर json.loads या LABELS जाँच उठाती है, gather(..., return_exceptions=True) त्रुटि को मान के रूप में रखता है बजाय बैच को मारने के, और आइटम अगले दौर में चला जाता है। rounds प्रयासों के बाद, failed में जो बचा है वह आपके हाथ से निरीक्षण के लिए है। 5,000 में से एक खराब उत्तर आपको एक अतिरिक्त अनुरोध का खर्च देता है, पूरे कार्य का नहीं।

स्क्रिप्ट प्रॉम्प्ट में JSON मांगती है और कोड में इसे मान्य करती है। यदि आप response_format के साथ प्रारूप को और कठोर बनाना चाहते हैं, तो LLM API से वैध JSON प्राप्त करें देखें।

कार्य की लागत#

टोकन ही एकमात्र मीटर हैं, इसलिए स्क्रिप्ट 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 बदलते हैं। निश्चित नहीं कि इस तरह का बिल कैसे बढ़ता है? 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 दस्तावेज़ और मूल्य निर्धारण पृष्ठ पर हैं।