手动调 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 路就差不多了。开并发前先想清楚你的后端扛不扛得住。
...