用 Together AI Batch API 处理 JSONL 批量请求：校验与结果对账

大量不着急的请求（比如几千条用户反馈的摘要）适合用 Batch API：上传一个 JSONL 文件，异步处理，完成后下载结果文件。 这篇做 batch_tool.py 三个子命令：prepare 把 CSV 变成符合官方格式的 JSONL，check 在上传前查出重复 id、超长 id、超大行和坏 JSON，reconcile 按 custom_id 把输出文件和错误文件对回原始顺序，生成 CSV。

先看清成品

• prepare 和 check 在 box 上真实运行；故意做坏的文件被查出 2 个问题。
• 用无效 Key 真实调用官方上传接口，返回 401。
• reconcile 用我手写的模拟输出/错误文件（标明了是模拟的，格式按官方示例）演示：只看输出文件会漏掉 1 条失败。5 项 pytest 通过。

合规提醒：只用 Together AI 官方 API 和你自己的 API Key，Key 只放环境变量或服务端，不共享账号。按 token 计费，价格以 together.ai/pricing 为准，本文不写价格。每个开源模型有自己的许可证，商用前看模型页面；处理用户消息时注意个人信息保护，不把别人的隐私数据发给第三方服务而不告知。

运行说明：我没有 Together API Key，没有对 Together 发出过成功的请求（也没产生费用）。真实跑过的是：用无效 Key 调用官方接口（2026-10-21 21:44–21:46，UTC+8，返回原样贴在文中），以及全部本地代码。Batch 的输出和错误文件是我按官方示例格式手写的模拟数据，放在 mock/ 目录并有说明。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

86-together-ai-B-src/
├── batch_tool.py
├── submit.sh
├── feedback.csv
├── batch.jsonl          # prepare 生成，测试会用到
├── mock/
│   ├── README.md
│   ├── output.jsonl
│   └── errors.jsonl
└── test_batch_tool.py

第 1 步：示例 CSV

id,text
1,物流太慢了，等了一周才到
2,客服回复很快，问题当场解决
3,包装还行，东西和描述一致
4,第二次买了，质量稳定

第 2 步：工具脚本

"""Together Batch API 的本地三件套：prepare（生成 JSONL）、check（提交前校验）、reconcile（按 custom_id 对账）。
规则来自 docs.together.ai 的 Batch 文档：custom_id 唯一且 ≤64 字符；每行 < 10MB（超大行在校验阶段不会被发现，
会在处理时报 "error reading input file"）；输出顺序与输入不同，必须按 custom_id 对；COMPLETED 也可能有失败行，
要同时下载 output_file_id 和 error_file_id。"""
import argparse, csv, json, sys

MAX_LINE = 10 * 1024 * 1024


def prepare(csv_path, out_path, model, prompt_tpl):
    n = 0
    with open(csv_path, encoding="utf-8") as f, open(out_path, "w", encoding="utf-8") as o:
        for row in csv.DictReader(f):
            body = {"model": model, "max_tokens": 200, "temperature": 0,
                    "messages": [{"role": "user", "content": prompt_tpl.format(**row)}]}
            o.write(json.dumps({"custom_id": f"row-{row['id']}", "body": body}, ensure_ascii=False) + "\n")
            n += 1
    return n


def check(path):
    errs, seen = [], set()
    with open(path, "rb") as f:
        for i, raw in enumerate(f, 1):
            if len(raw) >= MAX_LINE:
                errs.append(f"第{i}行 {len(raw)} 字节，超过 10MB 上限"); continue
            try:
                obj = json.loads(raw)
            except json.JSONDecodeError as e:
                errs.append(f"第{i}行不是合法 JSON：{e.msg}"); continue
            cid = obj.get("custom_id")
            if not isinstance(cid, str) or not cid:
                errs.append(f"第{i}行缺 custom_id")
            elif len(cid) > 64:
                errs.append(f"第{i}行 custom_id 超过 64 字符")
            elif cid in seen:
                errs.append(f"第{i}行 custom_id 重复：{cid}")
            else:
                seen.add(cid)
            body = obj.get("body")
            if not isinstance(body, dict) or "model" not in body or not body.get("messages"):
                errs.append(f"第{i}行 body 缺 model 或 messages")
    return len(seen), errs


def reconcile(input_path, output_path, error_path=None):
    wanted = [json.loads(l)["custom_id"] for l in open(input_path, encoding="utf-8") if l.strip()]
    ok, failed = {}, {}
    for l in open(output_path, encoding="utf-8"):
        if l.strip():
            o = json.loads(l)
            ok[o["custom_id"]] = o["response"]["body"]["choices"][0]["message"]["content"]
    if error_path:
        for l in open(error_path, encoding="utf-8"):
            if l.strip():
                o = json.loads(l)
                failed[o["custom_id"]] = f'{o["error"].get("code")}: {o["error"].get("message")}'
    missing = [c for c in wanted if c not in ok and c not in failed]
    return [(c, ok.get(c), failed.get(c)) for c in wanted], missing


def main(argv=None):
    ap = argparse.ArgumentParser(); sub = ap.add_subparsers(dest="cmd", required=True)
    p = sub.add_parser("prepare"); p.add_argument("csv"); p.add_argument("out")
    p.add_argument("--model", default="Qwen/Qwen3.5-9B")
    p.add_argument("--prompt", default="用一句话总结这条用户反馈，并判断情绪（正面/负面/中性）：{text}")
    c = sub.add_parser("check"); c.add_argument("jsonl")
    r = sub.add_parser("reconcile"); r.add_argument("input"); r.add_argument("output"); r.add_argument("--errors")
    r.add_argument("--csv", default="results.csv")
    a = ap.parse_args(argv)
    if a.cmd == "prepare":
        print(f"写入 {prepare(a.csv, a.out, a.model, a.prompt)} 行 → {a.out}"); return 0
    if a.cmd == "check":
        n, errs = check(a.jsonl)
        for e in errs: print("✗", e)
        print(f"{n} 个有效 custom_id，{len(errs)} 个问题"); return 1 if errs else 0
    rows, missing = reconcile(a.input, a.output, a.errors)
    with open(a.csv, "w", encoding="utf-8-sig", newline="") as f:
        w = csv.writer(f); w.writerow(["custom_id", "result", "error"]); w.writerows(rows)
    nok = sum(1 for _, o, _ in rows if o); nerr = sum(1 for _, _, e in rows if e)
    print(f"成功 {nok}，失败 {nerr}，缺失 {len(missing)} → {a.csv}")
    for m in missing: print("  缺失：", m)
    return 1 if (nerr or missing) else 0


if __name__ == "__main__":
    sys.exit(main())

第 3 步：上传脚本

#!/usr/bin/env bash
# 先本地 check 通过再上传；字段按 docs.together.ai Batch 教程的 cURL 示例（REST 直传需带 file_name）
set -euo pipefail
F=${1:-batch.jsonl}
: "${TOGETHER_API_KEY:?先 export TOGETHER_API_KEY}"
python batch_tool.py check "$F"
curl -sS -X POST https://api.together.ai/v1/files/upload \
  -H "Authorization: Bearer $TOGETHER_API_KEY" \
  -F purpose=batch-api -F "file_name=$(basename "$F")" -F "file=@$F"
echo
echo "拿到 file id 后：curl -X POST https://api.together.ai/v1/batches -H 'Authorization: Bearer …' -H 'Content-Type: application/json' -d '{\"input_file_id\":\"<id>\",\"endpoint\":\"/v1/chat/completions\"}'"

第 4 步：模拟输出（我写的）

{"id":"mock-1","custom_id":"row-3","response":{"status_code":200,"body":{"choices":[{"message":{"role":"assistant","content":"包装尚可、与描述一致；中性"}}]}}}
{"id":"mock-2","custom_id":"row-1","response":{"status_code":200,"body":{"choices":[{"message":{"role":"assistant","content":"抱怨物流慢；负面"}}]}}}
{"id":"mock-3","custom_id":"row-2","response":{"status_code":200,"body":{"choices":[{"message":{"role":"assistant","content":"客服快速解决问题；正面"}}]}}}{"custom_id":"row-4","error":{"message":"Invalid model specified","code":"invalid_model"}}

第 5 步：测试

import json, batch_tool as bt

def w(p, lines): p.write_text("".join(l + "\n" for l in lines), encoding="utf-8"); return str(p)

def test_prepare_and_check(tmp_path):
    c = tmp_path / "a.csv"; c.write_text("id,text\n1,好\n2,差\n", encoding="utf-8")
    out = tmp_path / "b.jsonl"
    assert bt.prepare(str(c), str(out), "m", "{text}") == 2
    n, errs = bt.check(str(out)); assert n == 2 and errs == []
    first = json.loads(out.read_text(encoding="utf-8").splitlines()[0])
    assert first["custom_id"] == "row-1" and first["body"]["messages"][0]["content"] == "好"

def test_check_flags(tmp_path):
    good = '{"custom_id":"a","body":{"model":"m","messages":[{"role":"user","content":"x"}]}}'
    p = w(tmp_path / "x.jsonl", [good, good, '{"custom_id":"' + "z"*65 + '","body":{"model":"m","messages":[1]}}',
                                  '{"custom_id":"b","body":{}}', "{oops"])
    _, errs = bt.check(p); s = " ".join(errs)
    assert "重复" in s and "64" in s and "缺 model" in s and "合法 JSON" in s

def test_oversized_line(tmp_path):
    p = tmp_path / "big.jsonl"; p.write_bytes(b'{"custom_id":"a","body":"' + b"x" * bt.MAX_LINE + b'"}\n')
    _, errs = bt.check(str(p)); assert "10MB" in errs[0]

def test_reconcile_order_and_errors():
    rows, missing = bt.reconcile("batch.jsonl", "mock/output.jsonl", "mock/errors.jsonl")
    assert [r[0] for r in rows] == ["row-1", "row-2", "row-3", "row-4"]
    assert rows[0][1].endswith("负面") and rows[3][2].startswith("invalid_model")
    assert missing == []

def test_reconcile_without_error_file_reports_missing():
    _, missing = bt.reconcile("batch.jsonl", "mock/output.jsonl")
    assert missing == ["row-4"]

第 6 步：运行

$ python batch_tool.py prepare feedback.csv batch.jsonl
写入 4 行 → batch.jsonl
$ python batch_tool.py check batch.jsonl
4 个有效 custom_id，0 个问题

# 故意做一个坏文件：重复一行 + 截断的 JSON
$ python batch_tool.py check bad.jsonl
✗ 第2行 custom_id 重复：row-1
✗ 第3行不是合法 JSON：Expecting ',' delimiter
1 个有效 custom_id，2 个问题

# 真实上传，无效 Key（2026-10-21 21:46 UTC+8）
$ curl -s -H 'Authorization: Bearer invalid-demo' -F purpose=batch-api -F file=@batch.jsonl https://api.together.ai/v1/files/upload
{"error":{"message":"Invalid API key provided. You can find your API key at https://api.together.xyz/settings/api-keys.","type":"invalid_request_error"}}   ← HTTP 401

# 对账（模拟文件）
$ python batch_tool.py reconcile batch.jsonl mock/output.jsonl
成功 3，失败 0，缺失 1 → results.csv
  缺失： row-4
$ python batch_tool.py reconcile batch.jsonl mock/output.jsonl --errors mock/errors.jsonl
成功 3，失败 1，缺失 0 → results.csv
$ cat results.csv
custom_id,result,error
row-1,抱怨物流慢；负面,
row-2,客服快速解决问题；正面,
row-3,包装尚可、与描述一致；中性,
row-4,,invalid_model: Invalid model specified
$ python -m pytest -q
5 passed

怎么检查结果？

• results.csv 按原始输入顺序排列（模拟输出里是 3、1、2 的乱序）。
• 没给 --errors 时，失败的那条显示为“缺失”，退出码 1，提醒你还有一个文件没下载。
• 真实使用时，先用 5–10 行的小文件走一遍上传→创建→下载，再放大。

调试示例

1. 只看 output 文件会漏掉失败。 上面第一次 reconcile 没带 --errors：输出文件里只有 3 行，如果代码只遍历输出文件，得到的就是 3 条“成功”，row-4 的失败悄无声息。官方管理页明确说 COMPLETED 的批次也可能有失败行。所以 reconcile 以输入文件的 custom_id 为准逐个找，既不在输出也不在错误文件里的列为“缺失”，退出码为 1——运行结果里那行“缺失：row-4”就是它起的作用。

2. 两个接口的错误信息指向不同域名。 真实的 401：chat/completions 返回的提示是 api.together.ai/settings/api-keys，带 code: invalid_api_key；files/upload 返回的是 api.together.xyz/settings/api-keys，而且没有 code 字段。所以代码只看 HTTP 状态码，不依赖 code 字段判断错误类型。

3. 文档里的状态列表不一致。 管理页的状态表没有 CANCELLED，教程页有。轮询代码要把 CANCELLED 也当作终态，否则取消后会一直轮询。

常见问题

模板规则

• 上传前本地校验。
• custom_id 用业务 id 生成，保证唯一。
• 同时下载输出和错误文件。
• 按 custom_id 对账，不按行号。

项目资源

批处理的关键不在提交，在对账：每一条都要有着落。