多模态批处理结果如何稳定对账：custom_id、JSONL 顺序与失败重试边界

多模态批处理最容易让人误判的地方，不是请求有没有完成，而是完成后的结果能不能和原始任务一一对上。可靠的做法是把 custom_id 当作业务主键：输入 JSONL 先建立索引，输出 JSONL 不按行号匹配，而是按主键回填；成功、失败和缺失三类记录分开统计，只有确认失败可重试时才进入下一批。

不要依赖输出文件顺序对账。用唯一的 custom_id 连接输入、成功结果和错误结果，再用“总数 = 成功 + 可重试失败 + 不可重试失败 + 未返回”验收批次。

批处理完成，不等于业务结果齐全

批处理接口通常返回一个批次状态和若干文件标识。状态进入完成阶段，只说明平台已经处理完这个批次；真正的业务核对还要读取成功输出文件，必要时再读取错误文件。官方 Batch API 的请求对象要求每行带唯一 custom_id，返回对象也用它把输出匹配回输入，不能把 JSONL 的物理顺序当成契约。

多模态请求还多了一层风险：图片、文本和参数往往由任务表拼装而来。若用数组下标匹配，某一行失败或输出顺序变化后，后面的图片就可能被写进错误的商品、工单或素材记录。

先用 custom_id 对齐输入和结果

下面的示例只依赖 JSONL 文件，适合放在批处理完成后的对账脚本中。输入行包含 custom_id 和业务字段，输出行包含同名主键以及 response 或 error。build_index 只建立索引，不改写原始任务。

import json
from pathlib import Path

def read_jsonl(path):
    with Path(path).open(encoding="utf-8") as f:
        for line_no, line in enumerate(f, 1):
            if line.strip():
                yield line_no, json.loads(line)

def build_index(input_path):
    index = {}
    for line_no, row in read_jsonl(input_path):
        key = row["custom_id"]
        if key in index:
            raise ValueError(f"duplicate custom_id at line {line_no}: {key}")
        index[key] = {"input_line": line_no, "request": row, "state": "pending"}
    return index

def reconcile(output_path, index):
    for line_no, row in read_jsonl(output_path):
        key = row.get("custom_id")
        if key not in index:
            raise ValueError(f"unknown custom_id at line {line_no}: {key}")
        if row.get("error") is not None:
            index[key].update(state="failed", error=row["error"])
        else:
            index[key].update(state="succeeded", response=row.get("response"))
    return index

核对点有三个：输入侧不能有重复主键，输出侧不能出现陌生主键，同一个输出文件重复读取时不能把已经确认的成功记录改成 pending。这里的 output.jsonl 是待解析的成功结果文件，reconcile 负责按 custom_id 回填。真实项目里可以把输入索引和最终状态落到数据库，示例先把边界写清楚。

失败行进入可重试队列

成功和失败不是同一种空值。建议先把 error.jsonl 并入同一个状态表，再按错误类型决定是否重试。临时额度不足、短暂网络错误这类 retryable 状态可以进入 retry_queue；参数不合法、图片不可读等确定性错误应保留在 final_report，不要无限循环。

RETRYABLE = {"rate_limit", "timeout", "temporary_unavailable"}

def classify_errors(index):
    retry_queue = []
    for key, item in index.items():
        if item["state"] != "failed":
            continue
        error_type = item["error"].get("type")
        if error_type in RETRYABLE:
            item["state"] = "retryable"
            retry_queue.append({
                "custom_id": f"retry-{key}",
                "source_custom_id": key,
                "request": item["request"],
            })
        else:
            item["state"] = "permanent_failure"
    return retry_queue

def final_report(index):
    counts = {}
    for item in index.values():
        counts[item["state"]] = counts.get(item["state"], 0) + 1
    return counts

重试任务使用新的批次内主键，例如 retry-，同时保存 source_custom_id。这样第二批结果可以独立对账，最终再回填原任务；如果重试脚本意外执行两次，也能通过原主键和批次号做幂等判断。

用四项数量验收批次

报表不要只打印“处理成功”。至少要同时给出输入总数、成功数、可重试失败数、不可重试失败数和未返回数。未返回数不应该被静默当成失败，因为它可能意味着读取错了文件、分页未读完，或对账脚本中途退出。

def verify_counts(index):
    total = len(index)
    succeeded = sum(x["state"] == "succeeded" for x in index.values())
    retryable = sum(x["state"] == "retryable" for x in index.values())
    permanent = sum(x["state"] == "permanent_failure" for x in index.values())
    pending = sum(x["state"] == "pending" for x in index.values())
    if total != succeeded + retryable + permanent + pending:
        raise AssertionError("reconcile count mismatch")
    return {"total": total, "succeeded": succeeded,
            "retryable": retryable, "permanent": permanent,
            "pending": pending}

这里的 pending 就是流程要重点追查的异常：如果批次声称已完成但仍有 pending，先确认成功文件和错误文件是否都读取完，再检查 custom_id 是否在中间转换时被截断。

常见误区和回滚边界

把输出第 N 行配给输入第 N 行

这是最危险的捷径。任何失败、重试或服务端排序变化都会让整批错位。只允许按 custom_id 关联，行号最多用于诊断。

看到批次完成就把空响应记成成功

成功行必须有可解析的响应对象；错误行应保存原始错误类型和批次标识。解析异常要停在对账阶段，不能用空字符串覆盖原记录。

重试时复用原 custom_id

重试批次使用新的批次内主键，同时保留 source_custom_id。最终写回时按源主键做幂等更新，避免第二次运行重复产生业务结果。

相关问题

为什么不直接依赖 output_file_id？

output_file_id 只告诉你成功输出文件在哪里，不能代替每条任务的业务身份。错误文件、重试批次和数据库回填仍然需要 custom_id。

什么时候可以关闭批次？

当成功、可重试失败、不可重试失败和 pending 的数量闭合，且重试队列已经独立落盘后，才可以把原批次标记为已对账。若还有 pending，应保留处理中状态。

把对账结果留成可复查记录

一份合格的最终记录至少包含批次标识、输入文件摘要、成功文件摘要、错误文件摘要、数量核对结果和重试批次关系。这样下游拿到的是可解释的结果，而不是一张无法追溯来源的“成功列表”。