Python-এ AsyncOpenAI দিয়ে হাজার হাজার টেক্সট শ্রেণীবদ্ধ করুন

হাজার হাজার টেক্সট শ্রেণীবদ্ধ করতে, AsyncOpenAI ব্যবহার করুন, প্রতিটি কলের চারপাশে একটি asyncio.Semaphore রাখুন যাতে শুধুমাত্র N অনুরোধ একসাথে চলে, এক-লাইনের JSON লেবেল চান, এবং শুধুমাত্র ব্যর্থ আইটেমগুলো আবার চালান। নিচের স্ক্রিপ্টটি ঠিক এটাই করে https://api.cheaptokens.dev/v1 এর বিরুদ্ধে এবং শেষে usage ফিল্ড থেকে মোট টোকেন প্রিন্ট করে।

৩১ অক্টোবর ২০২৬-এ 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 অনুসারে (১ নভেম্বর ২০২৬-এ যাচাই করা), সংযোগ ত্রুটি, 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 ডকস-এ এবং প্রাইসিং পেজ-এ আছে।