批量调用豆包处理一批评论（情感分类），做好三件事

批量调用豆包处理一批评论（情感分类），做好三件事：限制并发数、遇到 429 或 5xx 时按 Retry-After 退避重试、逐条记录 tokens 用量。费用只有在你自己提供单价（PRICE_IN / PRICE_OUT，单位是每百万 tokens）时才计算，脚本里不写死价格。

先看清成品

• 测试： 用模拟限流服务器测试（前 4 个请求返回 429），6 条评论全部处理完。
• 重试情况： 程序记录到 4 次重试，服务器一共收到 10 个请求，两边对得上。
• 分类错误： 本地 qwen2.5 1.5B 把“一般般吧，和图片差不多”分成了差评（应该是中评）。

没有在 box 上跑的部分： box 上没有方舟 API Key，没有调用过豆包 / 火山方舟的真实接口。代码按官方文档的 Chat API（OpenAI 兼容，base_url https://ark.cn-beijing.volces.com/api/v3）编写，测试时把 ARK_BASE_URL 指向两种替身：

• 本地 Ollama。
• 我自己写的模拟服务器：它只照着文档检查了请求格式，返回的内容是固定的假数据。

合规提醒：方舟 API Key（ARK_API_KEY）只放在环境变量里，不要写进网页前端。商品图、用户评论和聊天记录都可能含有个人信息，上传前要确认有权使用、做好脱敏。AI 生成的商品描述和客服回答必须经过人工审核才能发布，不能写夸大或无法证实的功效（《广告法》）。对公众提供生成式 AI 服务，需要按《生成式人工智能服务管理暂行办法》《人工智能生成合成内容标识办法》等规定履行相应义务。价格以方舟控制台的“模型定价”为准，本文不列价格。

做完之后你会得到什么？

需要准备哪些环境？

文件结构

43-doubao-C-src/
├── ark_batch.py
├── reviews.csv
└── fake_ratelimit.py

第 1 步：代码

ark_batch.py：

"""批量处理（例：商品评论情感分类），带并发上限、429/5xx 退避重试、逐条记录 tokens 用量。
费用只在你提供单价时计算：环境变量 PRICE_IN / PRICE_OUT = 每百万 tokens 的价格（以方舟控制台“模型定价”为准，本脚本不内置价格）。
环境变量：ARK_API_KEY、ARK_BASE_URL（默认 https://ark.cn-beijing.volces.com/api/v3）、ARK_MODEL（默认 doubao-seed-2-1-lite-260915）、CONCURRENCY（默认 4）
用法：python ark_batch.py reviews.csv 结果.csv"""
import csv, json, os, random, sys, time
from concurrent.futures import ThreadPoolExecutor
import openai
from openai import OpenAI

BASE = os.environ.get("ARK_BASE_URL", "https://ark.cn-beijing.volces.com/api/v3")
# max_retries=0：关掉 SDK 自带的静默重试，由下面的代码统一重试并记录，日志里才看得到真实的 429 次数
client = OpenAI(base_url=BASE, api_key=os.environ.get("ARK_API_KEY") or "missing-key", max_retries=0, timeout=120)
MODEL = os.environ.get("ARK_MODEL", "doubao-seed-2-1-lite-260915")
LABELS = ["好评", "中评", "差评"]

def call(text, tries=5):
    extra = {} if "11434" in BASE else {"thinking": {"type": "disabled"}}
    for i in range(tries):
        try:
            r = client.chat.completions.create(model=MODEL, temperature=0, max_tokens=50, extra_body=extra, messages=[
                {"role": "system", "content": f"判断商品评论的情感，只输出 {'/'.join(LABELS)} 中的一个词。"},
                {"role": "user", "content": text}])
            return r.choices[0].message.content.strip(), r.usage, i
        except (openai.RateLimitError, openai.InternalServerError, openai.APIConnectionError) as e:
            ra = getattr(getattr(e, "response", None), "headers", {}).get("retry-after")
            wait = float(ra) if ra else min(2 ** i, 30) + random.random()     # 优先听服务端的 Retry-After
            print(f"  {type(e).__name__}，{wait:.1f}s 后重试（第 {i+1} 次）", file=sys.stderr); time.sleep(wait)
    raise RuntimeError("重试次数用完")

def work(row):
    t0 = time.time()
    try:
        label, u, retries = call(row["review"])
        if label not in LABELS: label = f"无法识别：{label}"
        return {**row, "label": label, "in_tokens": u.prompt_tokens, "out_tokens": u.completion_tokens, "retries": retries, "seconds": round(time.time() - t0, 2)}
    except Exception as e:
        return {**row, "label": f"失败：{e}", "in_tokens": 0, "out_tokens": 0, "retries": "-", "seconds": round(time.time() - t0, 2)}

if __name__ == "__main__":
    rows = list(csv.DictReader(open(sys.argv[1], encoding="utf-8-sig")))
    with ThreadPoolExecutor(int(os.environ.get("CONCURRENCY", 4))) as ex: out = list(ex.map(work, rows))
    with open(sys.argv[2], "w", newline="", encoding="utf-8-sig") as f:
        w = csv.DictWriter(f, fieldnames=list(out[0])); w.writeheader(); w.writerows(out)
    for o in out: print(f"{o['id']:>3} {o['label']:6} 重试 {o['retries']}  {o['review']}")
    tin, tout = sum(o["in_tokens"] for o in out), sum(o["out_tokens"] for o in out)
    print(f"共 {len(out)} 条，输入 {tin} tokens，输出 {tout} tokens，429/5xx 重试 {sum(o['retries'] for o in out if o['retries'] != '-')} 次")
    if os.environ.get("PRICE_IN") and os.environ.get("PRICE_OUT"):
        print(f"按你提供的单价估算费用：{tin/1e6*float(os.environ['PRICE_IN']) + tout/1e6*float(os.environ['PRICE_OUT']):.4f}")

fake_ratelimit.py（仅用于离线测试）：

"""模拟限流的服务：前 N 个请求返回 429（带 Retry-After: 1），之后转发给本地 Ollama。统计收到的请求总数。
用法：FAIL_FIRST=4 python fake_ratelimit.py   （监听 127.0.0.1:18201）"""
import os, threading, requests
from flask import Flask, jsonify, request
app, lock, n = Flask(__name__), threading.Lock(), {"total": 0}
FAIL = int(os.environ.get("FAIL_FIRST", 4))

@app.post("/v1/chat/completions")
def chat():
    with lock: n["total"] += 1; k = n["total"]
    if k <= FAIL: return jsonify(error={"message": "fake rate limit"}), 429, {"Retry-After": "1"}
    b = request.get_json(); b["model"] = os.environ.get("FAKE_MODEL", "qwen2.5-local:1.5b")
    return requests.post("http://127.0.0.1:11434/v1/chat/completions", json=b, timeout=300).json()

@app.get("/stats")
def stats(): return jsonify(n)

if __name__ == "__main__":
    app.run(port=18201, threaded=True)

第 2 步：运行

$ FAIL_FIRST=4 python fake_ratelimit.py &
$ ARK_BASE_URL=http://127.0.0.1:18201/v1 ARK_MODEL=qwen2.5-local:1.5b python ark_batch.py reviews.csv 结果.csv
  RateLimitError，1.0s 后重试（第 1 次）
  （同样的提示共 4 行）
  1 好评     重试 1  物流很快，包装也很好，下次还来
  2 差评     重试 1  杯子有裂缝，客服态度还行但是退货麻烦
  3 差评     重试 1  一般般吧，和图片差不多
  4 差评     重试 1  颜色比图片暗很多，失望
  5 好评     重试 0  性价比很高，推荐
  6 差评     重试 0  发错货了，催了三天才补发
共 6 条，输入 254 tokens，输出 18 tokens，429/5xx 重试 4 次
$ curl 127.0.0.1:18201/stats
{"total":10}

检查点：6 条 + 4 次重试 = 10 个请求，和服务器的统计一致。真实使用时，设置 PRICE_IN / PRICE_OUT 后会多输出一行估算费用。

调试示例：重试“消失”了

第一版使用 SDK 的默认设置。跑完之后日志里显示 0 次重试，但服务器同样收到了 10 个请求。原因是 openai SDK 默认会自动重试 429（max_retries 默认为 2），重试发生在 SDK 内部，程序完全不知道。这样一来，统计不准，也没法控制退避策略。改成 max_retries=0，由程序自己按 Retry-After 退避并计数之后，两边的数字对上了。

常见问题

模板规则

• 重试由自己控制，并且要计数。
• 用量逐条记录，单价由用户提供。
• 批量结果抽样核对后再使用。

项目资源

批量任务能不能放心交出去，看的是重试和用量这两本账记得清不清楚。