项目实录 · 12 分钟阅读

用 Python 跑批量 AI 任务：队列、重试、断点与成本

批量任务不是给 API 调用套一层 for 循环。这篇实录搭出让它跑得住、停得下、算得清的四件事：稳定任务 key、并发上限、断点文件、开跑前就能说清的价格。

先有任务清单，再有循环

一批任务的本质是一组带稳定身份的输入。把输入文本做哈希当 key，同一张工单永远落到同一个 key：重跑安全、重复可见，都靠它。花钱之前先数一遍总行数与去重后的 key 数。

import hashlib

def job_key(text):
    return hashlib.sha256(text.encode("utf-8")).hexdigest()[:12]

texts = ["first ticket", "second ticket", "first ticket"]
jobs = [{"key": job_key(t), "text": t} for t in texts]
print(len(jobs), len({job["key"] for job in jobs}))

key 让重复暴露出来

把并发压在配额之下

瓶颈不在你的 CPU，在服务端配额。给每次调用套一个信号量，限制同时在途的请求数，再用 gather 等全部结束。这个数字要写进配置：上午九点能扛住的并发，晚高峰通常扛不住。

import asyncio

async def call(text, gate):
    async with gate:
        await asyncio.sleep(0.01)   # 代替真实 API 调用
        return text.upper()

async def run(texts, limit=4):
    gate = asyncio.Semaphore(limit)
    return await asyncio.gather(*(call(t, gate) for t in texts))

print(asyncio.run(run(["a", "b", "c"])))

一个信号量管住整批

429 是让你慢一点，不是失败

限流的意思是等一会儿再来。每次失败把等待翻倍，再加一点随机抖动，避免所有 worker 同时重试；失败三四次就放弃。400、401 这类业务错误一次都不要重试——重试多少次结果都一样。

import asyncio, random

async def with_backoff(call, attempts=4):
    delay = 0.5
    for attempt in range(attempts):
        try:
            return await call()
        except RuntimeError:            # 代替 HTTP 429
            if attempt == attempts - 1:
                raise
            await asyncio.sleep(delay + random.uniform(0, delay / 2))
            delay *= 2

有上限的退避加抖动

断点续跑靠追加写

每完成一条就追加一行 JSON 并立即落盘。重启时先把这个文件读成「已完成 key 集合」，跳过其中的任务。追加写就是全部诀窍：崩溃只损失当前这一条，不会丢掉前面的五百条。

import json, pathlib

def load_done(path):
    p = pathlib.Path(path)
    if not p.exists():
        return set()
    rows = [json.loads(line) for line in p.read_text(encoding="utf-8").splitlines() if line.strip()]
    return {row["key"] for row in rows if row.get("ok")}

def save(path, key, text):
    with open(path, "a", encoding="utf-8") as fh:
        fh.write(json.dumps({"key": key, "ok": True, "text": text}, ensure_ascii=False) + "\n")

JSONL 扛得住进程被杀

落盘之前先校验

就算要求了 JSON 输出，返回也可能裹着代码围栏，或者在半个对象处被截断。剥掉围栏再解析，失败就降级成原文前 200 字，而不是在循环里抛异常——一条坏数据不该终结整批 800 条。

import json

def parse(text, fallback_chars=200):
    raw = text.strip()
    if raw.startswith("```"):
        raw = raw.split("\n", 1)[-1].rsplit("```", 1)[0]
    try:
        return json.loads(raw)
    except json.JSONDecodeError:
        return {"title": raw[:fallback_chars], "fallback": True}

print(parse('```json\n{"title": "ok"}\n```'))
print(parse("cut off mid")["fallback"])

解析失败要有兜底

先看清一次请求，再管一千次

前面所有做法都默认你知道单次调用长什么样：一个 POST、两个请求头、一个 JSON 体——模型名、消息列表、输出契约。response_format 只保证输出能被解析，上一节的 schema 校验仍然要跑。

import os, requests

resp = requests.post(
    "https://api.example.com/v1/chat/completions",
    headers={
        "Authorization": f"Bearer {os.environ['OPENAI_API_KEY']}",
        "Content-Type": "application/json",
    },
    json={
        "model": "gpt-4o-mini",
        "messages": [{"role": "user", "content": "Return the ticket title as JSON."}],
        "response_format": {"type": "json_object"},
        "max_tokens": 120,
    },
    timeout=(3.05, 60),
)
resp.raise_for_status()
data = resp.json()
print(data["choices"][0]["message"]["content"], data["usage"]["total_tokens"])

一次请求，四个决定

开跑之前先把价算出来

按字数估一遍输入 token，乘上单价打印出来；跑完再记实际用量做对照。差额通常来自你忘了计入的提示词模板，而它会在每次重试时再算一遍。

PRICE_IN = 0.15 / 1_000_000
PRICE_OUT = 0.60 / 1_000_000

def batch_cost(prompt_tokens, completion_tokens):
    return round(prompt_tokens * PRICE_IN + completion_tokens * PRICE_OUT, 4)

print(batch_cost(1_200_000, 300_000))

说得出数字的成本

你的练习清单

每条任务都有内容决定的 key，重跑不会重复付费。
并发数写在配置里，并且低于服务端配额。
429 用翻倍退避加抖动；4xx 业务错误不重试。
结果按 JSONL 逐行追加，写完即落盘。
结构化输出有明确的解析失败分支。
单次请求的四个要点写得出来：认证头、模型、输出契约、超时。
收尾打印任务数、失败数与实际花费。

常见问题

能不能直接用 ThreadPoolExecutor？

如果 SDK 是同步的、调用只是网络等待，用线程更简单也够用。需要单进程几百个并发等待时 asyncio 更省内存。选一种，别把两套混着嵌套。

结果文件会多大？

每行约 200 字节，一百万条约 200 MB。按天或按来源切文件；只需要「哪些做完了」的话，单独维护一份 key 索引即可。

重试几次合适？

三到四次，总等待时间控制在单条任务的超时预算之内。仍失败就写进 failed.jsonl 记下错误，继续下一条。

为什么用哈希而不是行号？

输入文件一变行号就全错位，也发现不了同一张工单来了两次。内容哈希两件事一起解决。

练一练重试与结构化输出

覆盖重试预算、超时、流式分片与 JSON 校验的题目。

打开练习

继续阅读：