想让模型逐行处理一张几百行的表，起步时先不要急着写循环。比较稳的顺序是：先量清这张表有多少行、多少列、单行大约多少字符，再用几行连同表头做一次小样本试跑，确认返回的列名、列序和空值处理符合预期，最后才把读表、分批、写回、失败记录四段固定进脚本。单次请求能带多少行，取决于单行体积和批处理脚本的分批大小参数，通常要按单行实际长度估一个保守行数，再通过试跑日志确认没有超限或截断。

批量处理表格的起步判断：先统计行数、列数和单行平均字符，再取少量行连同表头试跑，确认返回格式可解析；随后固定每批行数，用脚本分四段执行（读表、分批、写回、失败记录）。适用场景是几百行以内的逐行结构化处理；边界在于单次请求体量需要结合所用接入方式的实际限制确认，不能只看行数不看单行长度。

量一张表的行数、列数和单行体积

先把表的体量写进日志。行数决定要切几批，列数和单行平均字符决定单次请求能带多少行，两者缺一个都容易在第一批就超限。下面是一段可以放进脚本或单独运行的骨架，字段名按实际表头替换。

python3 - <<'PY'
import csv, json

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

header = list(rows[0].keys()) if rows else []
row_count = len(rows)
col_count = len(header)
char_total = 0
for r in rows:
    char_total += sum(len(str(v or "")) for v in r.values())
avg_chars = char_total / row_count if row_count else 0

info = {
    "file": path,
    "row_count": row_count,
    "col_count": col_count,
    "header": header,
    "avg_chars_per_row": round(avg_chars, 1),
    "max_chars_per_row": max((sum(len(str(v or "")) for v in r.values()) for r in rows), default=0),
}
print(json.dumps(info, ensure_ascii=False, indent=2))
PY

日志里至少保留这些字段：文件名、行数、列数、表头列表、单行平均字符、单行最大字符。平均值决定常规批次能装多少行，最大值提示是否要单独处理个别长行。估算时把表头也计入每次请求的体积，因为通常每批都会带表头，这部分不能忽略。

拿几行做小样本试跑

小样本试跑的目的不是看模型回答得好不好，而是确认三件事：返回的列名是否与表头一致、列序是否稳定、空值是否有一致的表示。建议选 3 到 5 行，覆盖普通值、空字段和特别长的字段，连同表头一起发出去。

你收到的是一张表的片段，表头为：
[列A, 列B, 列C]

请对每一行输出一个 JSON 对象，键名与表头完全一致，列序不变；
空字段输出空字符串，不要补写内容。
输出整体为一个 JSON 数组，不要加解释文字。

数据行：
[这里粘贴选出的3到5行]

拿到返回后逐项核对：键名是否逐字匹配表头，有没有被改写成更口语的名字；数组长度是否等于发送的行数；空字段是否被填成了猜测值。如果返回不可直接解析，先在提示词里收紧输出格式，再决定是否扩大批量；这一步没过就写全量脚本，后面排查会非常费劲。

写批处理脚本骨架：读表、分批、写回

手工试跑通过后，把流程固化成脚本。骨架里要有一个明确的分批大小参数，先在试跑里跑小值，再按单行体积调整。请求部分只给通用调用位置，具体参数、鉴权字段和返回结构要按实际接入文档核对，不要照搬。

python3 - <<'PY'
import csv, json, time

SRC = "input.csv"
OUT = "output.csv"
FAIL = "failed_batches.jsonl"
BATCH_SIZE = 20  # 先按试跑结果设小值，确认稳定后再调整

def call_model(header, batch_rows):
    """通用调用骨架：按实际接入文档填写请求体、鉴权与返回解析。"""
    payload = {
        "header": header,
        "rows": batch_rows,
        "instruction": "逐行输出JSON，键名与表头一致，空值为空字符串",
    }
    # resp = client.invoke(payload)
    # return parse(resp)
    raise NotImplementedError("按实际接入文档补齐调用与解析")

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

with open(OUT, "w", newline="", encoding="utf-8") as out:
    writer = csv.DictWriter(out, fieldnames=header)
    writer.writeheader()
    for start in range(0, len(rows), BATCH_SIZE):
        end = min(start + BATCH_SIZE, len(rows))
        batch = rows[start:end]
        try:
            result = call_model(header, batch)
            writer.writerows(result)
        except Exception as e:
            print(f"batch {start}-{end} failed: {e}")
            raise
PY

写回的位置单独开一个新文件，不要原地覆盖原始表。这样即使中途失败，原始数据仍在，输出表也能从零重跑。BATCH_SIZE 的取值方法通常是：用单次请求的字符预算除以单行平均字符，再留出表头和提示词的开销，得到的行数向下取整。这个值只作为起点，还要通过试跑日志确认没有超限或截断。

给失败批次加记录与重试位

整表重跑的时间和调用成本都不低，所以失败要能定位到具体区间。做法是每批在调用前写下起始行和结束行，失败时追加一条记录文件，重试时只读这个文件里的区间。

{"batch_id": 7, "start_row": 140, "end_row": 160, "error": "返回不是合法JSON", "retried": false}

失败记录文件建议用 JSON Lines，一行一条，字段包括批次号、起始行、结束行、错误信息、是否重试。重试脚本的逻辑是：读取失败记录，只取 retried 为 false 或重试次数未超上限的条目，按起始行和结束行从原始表切出对应行再调用一次；成功后把结果追加到输出表，并把记录里的 retried 改为 true。这样出错时你面对的是某一段行号，而不是整张表。

核对输出表与原始表的行数

最后一步不能省。先比总行数：原始表数据行数与输出表数据行数应当相等，不等就说明有批次漏写或重复写入。再比主键：如果表里有唯一列，用它在两侧各做一次去重计数，数量也应一致。

wc -l input.csv output.csv

python3 - <<'PY'
import csv

def key_set(path, key):
    with open(path, newline="", encoding="utf-8") as f:
        return [r[key] for r in csv.DictReader(f)]

keys = key_set("input.csv", "id")
out = key_set("output.csv", "id")
print("输入行数", len(keys), "输出行数", len(out))
print("缺少", len(set(keys) - set(out)), "重复", len(out) - len(set(out)))
PY

发现缺行时，不要凭印象补。用缺的起始行号去失败记录里查落在哪些批次区间内，先确认那一批是否调用失败或解析失败，再决定是重试该批还是单独取出对应行手工处理。行数、主键和失败记录三者对上，才有依据说这批处理完成。