จำแนกข้อความหลายพันรายการด้วย AsyncOpenAI ใน Python

ในการจำแนกข้อความหลายพันรายการ ให้ใช้ AsyncOpenAI ใส่ asyncio.Semaphore ล้อมรอบทุกการเรียก เพื่อให้มีคำขอที่กำลังดำเนินการเพียง N รายการ ขอป้ายกำกับ JSON บรรทัดเดียว และรันซ้ำเฉพาะรายการที่ล้มเหลว สคริปต์ด้านล่างทำแบบนั้นกับ https://api.cheaptokens.dev/v1 และพิมพ์จำนวนโทเค็นรวมจากฟิลด์ usage ในตอนท้าย

เมื่อวันที่ 31 ตุลาคม 2026 Andrej Karpathy (ผู้ร่วมก่อตั้ง OpenAI อดีตผู้อำนวยการฝ่าย AI ของ Tesla) ได้อธิบายถึง 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 ตาม README ของ openai-python (ตรวจสอบเมื่อ 1 พฤศจิกายน 2026) ข้อผิดพลาดการเชื่อมต่อ, 408, 409, 429 และการตอบกลับ 5xx จะถูกลองใหม่สองครั้งโดยค่าเริ่มต้นพร้อมการหน่วงแบบเอ็กซ์โพเนนเชียลสั้นๆ max_retries=2 ในสคริปต์เพียงทำให้มองเห็นได้ชัดเจน; ตั้งให้สูงขึ้นด้วย client.with_options(max_retries=5) หรือเป็น 0 เพื่อปิดการทำงาน ไทม์เอาต์เริ่มต้นคือ 10 นาที และไทม์เอาต์จะทำให้เกิด APITimeoutError ซึ่งเป็นเหตุผลที่สคริปต์ตั้ง timeout=60 สำหรับป้ายกำกับสั้นๆ รายละเอียดอยู่ใน ไทม์เอาต์และการลองใหม่ของ OpenAI Python SDK และแนวคิดการหน่วงเบื้องหลังใน 429 Too Many Requests จาก 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 ดู รับ 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: คีย์, base URL หรือรหัสโมเดลผิด ข้อผิดพลาด API 401 หรือ 404 อธิบายแต่ละกรณี
• json.JSONDecodeError ในหลายรายการ: คำตอบไม่มี {...} มักเป็นเพราะ max_tokens=100 ตัดมัน พิมพ์ r.choices[0].finish_reason; length หมายถึงถึงขีดจำกัด ดังนั้นเพิ่ม max_tokens
• 429 จำนวนมากและการรันช้า: ลด concurrency

รายละเอียดโมเดลทั่วไปและเอนด์พอยต์อยู่ใน เอกสาร CheapTokens และบน หน้าราคา