Phân loại hàng nghìn văn bản với AsyncOpenAI trong Python

Để phân loại hàng nghìn văn bản, hãy dùng AsyncOpenAI, đặt một asyncio.Semaphore quanh mỗi lệnh gọi để chỉ có N yêu cầu đang xử lý, yêu cầu nhãn JSON một dòng, và chỉ chạy lại những mục bị lỗi. Đoạn script dưới đây làm chính xác điều đó với https://api.cheaptokens.dev/v1 và in tổng số token từ trường usage ở cuối.

Vào ngày 31 tháng 10 năm 2026, Andrej Karpathy (đồng sáng lập OpenAI, cựu giám đốc AI của Tesla) đã mô tả HackerTube: ông xây dựng nó trong khoảng hai giờ và dùng Sonnet để thu hẹp bình luận Hacker News xuống còn khoảng 40K bình luận giới thiệu một video YouTube. Bài đăng của ông không đưa ra prompt, kích thước lô hay mức độ đồng thời, nên những gì sau đây là mẫu chuẩn cho loại công việc này, không phải công thức của ông.

Đoạn script#

Giả sử bạn có 5.000 phiếu hỗ trợ và muốn gắn thẻ mỗi phiếu là bug, question hoặc praise. Đây là toàn bộ. Cài SDK bằng pip install openai (Python 3.10+; python -c "import openai; print(openai.__version__)" cho biết bạn đang có gì) và đặt khóa của bạn vào một biến môi trường trước.

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'])

Bạn nhận lại labels (chỉ số văn bản → nhãn), failed (các chỉ số không bao giờ tạo ra nhãn hợp lệ, danh sách rỗng khi mọi thứ đều thành công) và tổng số token. Id mô hình là deepseek-ai/DeepSeek-V4-Flash-0731 đầy đủ; lấy id cho các mô hình khác từ tài liệu CheapTokens, vì tên ngắn khác nhau giữa các nhà cung cấp.

Sắp xếp pipeline theo thứ tự để kiểm tra bạn đã nắm được hình dạng trong đầu:

Tại sao dùng semaphore mà không phải 5.000 yêu cầu cùng lúc#

asyncio.gather vui vẻ khởi động mọi thứ bạn đưa cho nó. Đưa cho nó hàng nghìn coroutine mà không có giới hạn và tất cả sẽ khởi động cùng lúc, một đợt bùng nổ yêu cầu là cách nhanh chóng để gặp lỗi 429. Semaphore là một cổng quay: concurrency=5 nghĩa là tối đa năm lệnh gọi nằm trong async with sem tại bất kỳ thời điểm nào, và phần còn lại chờ lượt. Python vẫn tạo tất cả các task, nhưng chỉ năm task nói chuyện với API.

concurrency nên là bao nhiêu? Bắt đầu thấp, như 5 ở trên, và tăng lên trong khi lỗi vẫn không xuất hiện. Hãy coi nó như một cài đặt bạn tinh chỉnh cho từng công việc, không phải một hằng số để sao chép.

Thử lại: hai lớp, hai nhiệm vụ#

Lớp đầu tiên được tích hợp trong SDK. Theo README openai-python (kiểm tra ngày 1 tháng 11 năm 2026), lỗi kết nối, 408, 409, 429 và phản hồi 5xx được thử lại hai lần theo mặc định với backoff lũy thừa ngắn. max_retries=2 trong script chỉ làm cho điều đó hiển thị; đặt cao hơn với client.with_options(max_retries=5) hoặc thành 0 để tắt. Thời gian chờ mặc định là 10 phút và thời gian chờ sẽ ném APITimeoutError, đó là lý do script đặt timeout=60 cho nhãn ngắn. Chi tiết có trong OpenAI Python SDK timeout và retries, và ý tưởng backoff đằng sau nó trong 429 Too Many Requests từ một AI API.

Lớp thứ hai là của bạn, và nó bao gồm những gì SDK không thể thấy: lệnh gọi thành công, nhưng mô hình trả lời bằng văn xuôi và không có JSON nào cả (I can't classify this) hoặc bịa ra một nhãn gọi là complaint. Văn bản xung quanh một {...} không quan trọng, vì mã cắt JSON ra khỏi câu trả lời. Sau đó json.loads hoặc kiểm tra LABELS ném lỗi, gather(..., return_exceptions=True) giữ lỗi như một giá trị thay vì giết cả lô, và mục đó chuyển sang vòng tiếp theo. Sau rounds lần thử, bất cứ thứ gì còn trong failed là của bạn để kiểm tra bằng tay. Một câu trả lời xấu trong 5.000 chỉ tốn một yêu cầu thêm, không phải toàn bộ công việc.

Script yêu cầu JSON trong prompt và xác thực nó trong mã. Nếu bạn muốn đẩy định dạng mạnh hơn với response_format, xem Lấy JSON hợp lệ từ một LLM API.

Công việc tốn bao nhiêu#

Token là đồng hồ đo duy nhất, vì vậy script tính tổng usage và chi phí theo đó. Lấy trường hợp phổ biến: văn bản ngắn, nhãn ngắn. Giả sử 10.000 văn bản với khoảng 300 token đầu vào mỗi văn bản (văn bản cộng hướng dẫn) và khoảng 20 token đầu ra cho nhãn JSON:

10.000 × 300 = 3.000.000 token đầu vào; 10.000 × 20 = 200.000 token đầu ra.

Trên DeepSeek-V4-Flash công việc đó tốn $0.065, trên GLM-5.3-Flash là $0.068, trên MiniMax-M2.7 là $0.18. Bảng hiển thị cùng công việc cho mọi mô hình, bên cạnh OpenRouter; với đầu vào và đầu ra được tính cùng nhau, tổng cộng rẻ hơn OpenRouter.

Di chuyển thanh trượt trong máy tính đến số của bạn. Các trường in và out dành cho một yêu cầu, và runs là số yêu cầu bạn gửi.

Phía đầu vào chiếm ưu thế ở đây: văn bản đi vào, một từ đi ra. Đó là lý do một mô hình có đầu vào rẻ và cửa sổ ngữ cảnh 380K, như DeepSeek-V4-Flash với một mức giá bất kể giờ nào, phù hợp để lọc hàng loạt; cho nhãn ngắn, GLM-5.3-Flash và MiniMax-M2.7 là các lựa chọn thay thế cho cùng một script, bạn chỉ thay đổi MODEL. Không chắc hóa đơn như thế này tăng lên thế nào? AI API tính phí bạn những gì có bức tranh.

Trước lần chạy lớn#

Chạy 20 văn bản trước và xem failed và tổng số token. Nếu 20 văn bản dùng khoảng 6.000 token đầu vào, 5.000 sẽ dùng khoảng 1.500.000, và bạn có thể nhân trước khi cam kết.

Nếu nó không hoạt động#

• KeyError: 'CHEAPTOKENS_API_KEY': biến chưa được đặt trong terminal này. Xuất nó và khởi động lại Python.
• 401 hoặc 404: sai khóa, URL cơ sở hoặc id mô hình. Lỗi API 401 hoặc 404 hướng dẫn từng cái.
• json.JSONDecodeError trên nhiều mục: câu trả lời không có {...}, thường vì max_tokens=100 cắt nó. In r.choices[0].finish_reason; length nghĩa là đã chạm giới hạn, vì vậy tăng max_tokens.
• Nhiều 429 và lần chạy chậm: giảm concurrency.

Chi tiết chung về mô hình và endpoint nằm trong tài liệu CheapTokens và trên trang giá.