使用 Python 中的 AsyncOpenAI 分类数千条文本

要分类数千条文本，请使用 AsyncOpenAI，在每次调用周围放置一个 asyncio.Semaphore，以便只有 N 个请求处于进行中，要求返回单行 JSON 标签，并且只重新运行失败的项目。下面的脚本正是针对 https://api.cheaptokens.dev/v1 执行此操作，并在最后打印来自 usage 字段的总 token 数。

2026 年 10 月 31 日，Andrej Karpathy（OpenAI 联合创始人、前特斯拉 AI 总监）描述了 HackerTube：他大约用了两个小时构建它，并使用 Sonnet 将 Hacker News 评论缩小到约 40K 条推荐 YouTube 视频的评论。他的帖子没有给出提示词、批处理大小或并发数，因此以下内容是此类任务的标准模式，而不是他的配方。

脚本#

假设你有 5,000 张支持工单，并希望每张都标记为 bug、question 或 praise。以下是完整内容。使用 pip install openai 安装 SDK（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（从未产生有效标签的索引，当一切顺利时为空列表）以及 token 总数。模型 id 是完整的 deepseek-ai/DeepSeek-V4-Flash-0731；其他模型的 id 请从 CheapTokens 文档 获取，因为不同提供商之间的短名称不同。

将流水线按顺序排列，以检查你脑中是否已形成整体结构：

为什么使用信号量而不是一次性发出 5,000 个请求#

asyncio.gather 会愉快地启动你交给它的一切。不加限制地交给它数千个协程，它们会同时启动，这种请求突发是快速导致 429 错误的一种方式。信号量是一个旋转门：concurrency=5 表示在任何时刻最多有五个调用位于 async with sem 内，其余调用等待轮到自己。Python 仍然会创建所有任务，但只有五个会与 API 通信。

concurrency 应该设为多少？从低开始，比如上面的 5，并在错误不出现时提高它。将其视为按任务调整的设置，而不是要复制的常量。

重试：两层，两项工作#

第一层内置于 SDK 中。根据 openai-python README（2026 年 11 月 1 日检查），连接错误、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。

任务成本#

Token 是唯一的计量单位，因此脚本会汇总 usage，成本随之而来。以常见情况为例：短文本、短标签。假设 10,000 条文本，每条约 300 个输入 token（文本加指令），JSON 标签约 20 个输出 token：

10,000 × 300 = 3,000,000 个输入 token；10,000 × 20 = 200,000 个输出 token。

在 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 和 token 总数。如果 20 条文本使用约 6,000 个输入 token，那么 5,000 条将使用约 1,500,000 个，你可以在提交前先乘一下。

如果它不工作#

• KeyError: 'CHEAPTOKENS_API_KEY'：该变量未在此终端中设置。导出它并重新启动 Python。
• 401 或 404：密钥、基础 URL 或模型 id 错误。API 401 或 404 错误 会逐一说明。
• 许多项目上出现 json.JSONDecodeError：答案中没有 {...}，通常是因为 max_tokens=100 截断了它。打印 r.choices[0].finish_reason；length 表示达到了限制，因此提高 max_tokens。
• 许多 429 且运行缓慢：降低 concurrency。

通用模型和端点详细信息见 CheapTokens 文档 和 定价页面。