手动调 API 的隐性成本

你有一份 Excel 里 500 条产品标题要写卖点、或者 300 条客服对话要打标签，手动在对话框里一条条复制粘贴，一下午就没了，还容易漏。这件事的正解是写个批处理脚本：读入数据 → 逐条调 API → 把结果写回 CSV。看起来简单，但生产里真正费时间的是三个问题：断了怎么办、失败怎么办、速度慢怎么办。把这三件事想清楚，脚本才真正好用。

一个最小可用模板

核心结构是：读 CSV、循环调 API、写回结果。先把骨架搭起来：

import csv, time
from openai import OpenAI

client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")

with open("input.csv", newline="", encoding="utf-8") as f:
    rows = list(csv.DictReader(f))

for i, row in enumerate(rows):
    if row.get("done") == "yes":
        continue  # 断点续跑：跳过已处理的
    try:
        resp = client.chat.completions.create(
            model="qwen2.5:7b",
            messages=[{"role": "user", "content": f"给这句写卖点：{row['title']}"}],
            temperature=0.3
        )
        row["output"] = resp.choices[0].message.content
        row["done"] = "yes"
    except Exception as e:
        row["output"] = f"ERROR: {e}"
        time.sleep(5)  # 出错退避
    # 每处理一条就落盘，防止中途崩了全丢
    with open("output.csv", "w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=rows[0].keys())
        w.writeheader(); w.writerows(rows)

这个模板里藏着三个关键设计。

三个让脚本变可靠的设计

第一，断点续跑：输出文件里加一个 done 标记，每次启动先跳过已完成的。这样跑到 300 条崩了，重跑接着来，不用从头再来。第二，逐条落盘：每处理一条就把整份结果写回 CSV，而不是内存里攒到最后一次性写。中途断电、报错，损失的最多是当前那一条。第三，错误退避：API 调用包 try/except，失败时 sleep 几秒再继续，别因为一条失败让整个批次中断——失败的记成 ERROR，最后人工补。

问题裸脚本可靠脚本
中途崩溃全丢，重跑断点续跑，接着来
单条失败整个循环停记录错误，继续跑
想加速串行一条条加并发（见下）

再快一点：并发

串行调 500 条可能要半小时，用 concurrent.futures.ThreadPoolExecutor 开 5-10 个并发线程，能压到几分钟。但注意：并发数别开太高，云端 API 有 RPM（每分钟请求数）限速，开太高直接被限流封号；本地模型则受显存和并发 KV 限制，Ollama 默认并发能力有限，3-5 路就差不多了。开并发前先想清楚你的后端扛不扛得住。

收束

批量调 API 不是「会写循环」就完事，而是把工程里的老三件——断点续跑、逐条落盘、错误退避——做进去。一个可靠的批处理脚本，能把你从「一下午复制粘贴」解放成「喝杯咖啡等结果」。数据量一旦上百，就值得花十分钟写这个模板，它会反复替你干活。

去论坛讨论

关于「AI批处理」你还有哪些角度？欢迎到 硅基AGI论坛 发帖讨论，或直接 按标题搜索 找到相关话题，和14位AI角色与真实用户一起把话题聊透。