批量跑几千条时输出格式不统一，问题通常不在模型本身，而在三处没被约束：输入文件没有稳定主键、提示里没把字段清单写成硬约束、队列层没对每条输出做校验和失败留档。把 Ling 3.0 Flash 接进批处理队列，按“一行一请求 → 提示写死字段 → 并发受控 → 逐条校验 → 失败回灌”的顺序搭起来，下游脚本拿到的就是固定结构，出问题也能定位到具体哪一行。

这套做法适用于“一条输入产出一条结构化结果”的批量场景，不适合需要多轮对话或自由文本的任务。操作顺序是：先固定输入主键和字段清单，再在队列里加并发上限、重试和失败写入磁盘，最后用校验函数按错误码统计不合规条目并回灌提示。验证方式是失败文件里每条都能追溯到输入 id 和具体错误码；边界是缺值占位、字段长度这类规则必须提前定死，否则校验规则会随批次漂移，统计结果也就失去比较意义。

把待处理数据整理成一行一请求的输入文件

输入建议用 JSONL：一行一条请求，跑完后行号、id、原始提示、模型输出能逐条对齐。CSV 也能用，但提示文本里常带逗号和换行，转义会额外制造一类难查的错误。把变量在生成输入文件时就拼好，队列里只做透传，避免执行阶段二次拼接导致同一条数据出现两个版本。

一行自造样例（非真实业务数据）：

{"id": "row-000001", "prompt": "把下面这句话归为 投诉/咨询/其他：快递三天没到", "tag": "batch-01"}

id 建议带行号前缀，例如 row-000001，这样失败样本一出来就能反查缓存或重跑单条。tag 不参与模型调用，只在你后面做分批次统计时有用。

在提示里写死字段清单和缺值占位写法

提示模板骨架，system 负责格式约束，user 只放单条任务文本：

system:
你是批量任务中的单条处理器。只输出一个 JSON 对象，不要输出解释、前后缀或 Markdown 代码围栏。
字段清单（必须全部出现，顺序不限）：
- id: string，原样回填输入里的 id
- label: string，只能是 [投诉, 咨询, 其他] 之一
- reason: string，不超过 60 个字符
- confidence: number，0 到 1 之间
缺值写法：无法判断时，字符串字段写 'unknown'，数值字段写 null。
不要省略字段，不要写 '-'、空白或 'N/A'。

user:
{{prompt}}

缺值占位的取舍：字符串字段统一写 unknown 而不是空串，空串在下游 if 判断里容易被当成假值而误判成“有结果”；数值字段写 null 比写 0 安全，0 会被当成真实评分参与排序。代价是校验函数必须显式允许 null，否则类型检查会把合法的缺值判死。

如果接入接口支持 JSON 模式或 response_format 之类的参数，可以同时打开，但不要只依赖它：这类参数保证的是“能解析成 JSON”，不保证“字段齐全、类型正确”。字段清单和缺值写法的约束仍然要写在提示里。

给队列加并发上限与失败重投

下面的骨架用信号量控制并发、用简单重试加退避、失败样本统一写入磁盘。client 调用是占位，endpoint、model 名和鉴权头按你所在环境的接口文档替换。

import asyncio, json, time

CONCURRENCY = 8        # 按接口额度和本机资源调，先小后大
MAX_RETRY = 2
FAIL_PATH = 'failures.jsonl'

sem = asyncio.Semaphore(CONCURRENCY)
write_lock = asyncio.Lock()

async def call_model(prompt):
    async with sem:
        resp = await client.complete(prompt)   # 占位：替换为实际接入
    return resp

async def dump_failure(row_id, raw, errs):
    async with write_lock:
        with open(FAIL_PATH, 'a', encoding='utf-8') as f:
            f.write(json.dumps({
                'id': row_id,
                'errors': errs,
                'raw': (raw or '')[:2000],
                'ts': int(time.time()),
            }, ensure_ascii=False) + '\n')

async def handle(row):
    for attempt in range(MAX_RETRY + 1):
        try:
            raw = await call_model(row['prompt'])
            obj, errs = validate(raw, row['id'])
            if obj is not None:
                return obj
            last = errs
        except Exception as e:
            last = ['E_EXC|%s|%r' % (row['id'], e)]
        await asyncio.sleep(min(2 ** attempt, 8))
    await dump_failure(row['id'], locals().get('raw'), last)
    return None

并发上限是唯一需要现场调的参数，先给个保守值跑一小段，观察超时和限流错误再往上加；重试只处理网络或超时这类瞬时失败，校验不通过的输出重试一两次就该进入失败文件，否则会用相同提示反复触发同一个格式问题。写入失败文件需要加锁，否则并发下 JSONL 会写乱行。

逐条校验JSON字段并统计不合规比例

字段定义与校验规则表（这张表就是 validate 的实现依据，改字段时先改表）：

import json

ENUM_LABEL = {'投诉', '咨询', '其他'}
SPEC = {
    'id':         {'type': str,          'max_len': 64,   'nullable': False},
    'label':      {'type': str,          'max_len': 8,    'nullable': False},
    'reason':     {'type': str,          'max_len': 60,   'nullable': False},
    'confidence': {'type': (int, float), 'max_len': None, 'nullable': True},
}

def validate(text, row_id):
    try:
        obj = json.loads(text)
    except json.JSONDecodeError as e:
        return None, ['E_JSON|%s|%s@%d' % (row_id, e.msg, e.pos)]
    if not isinstance(obj, dict):
        return None, ['E_NOT_OBJECT|%s|%s' % (row_id, type(obj).__name__)]
    errs = []
    for name, spec in SPEC.items():
        if name not in obj:                                  # 齐全性
            errs.append('E_MISSING|%s|%s' % (row_id, name))
            continue
        val = obj[name]
        if val is None:
            if not spec['nullable']:
                errs.append('E_NULL|%s|%s' % (row_id, name))
            continue
        if isinstance(val, bool) or not isinstance(val, spec['type']):   # 类型
            errs.append('E_TYPE|%s|%s|%s' % (row_id, name, type(val).__name__))
            continue
        if spec['max_len'] and len(val) > spec['max_len']:               # 长度
            errs.append('E_LEN|%s|%s|%d' % (row_id, name, len(val)))
        if name == 'label' and val not in ENUM_LABEL:
            errs.append('E_ENUM|%s|%s' % (row_id, name))
        if name == 'confidence' and not (0 <= val <= 1):
            errs.append('E_RANGE|%s|%s' % (row_id, name))
    return (obj if not errs else None), errs

bool 要单独排除，因为它是 int 的子类，不排除的话 True 会被当成合法数值。统计时按错误码加字段两级聚合，别只报一个总数：

from collections import Counter
counter = Counter()
for line in open('failures.jsonl', encoding='utf-8'):
    for err in json.loads(line)['errors']:
        parts = err.split('|')
        counter[(parts[0], parts[2] if len(parts) > 2 else '')] += 1
for (code, field), n in counter.most_common():
    print('%-14s %-12s %d' % (code, field, n))

看结果时先区分两类：E_JSON 和 E_MISSING 属于提示约束不足，改提示有机会收敛；E_LEN 和 E_RANGE 往往说明任务本身定义不清，比如 reason 要求 60 字却经常需要更长。后者改提示收益有限，更实际的做法是放宽长度或拆字段。

把不合规样本回灌成新一轮提示补丁

回灌不是把提示写得更长，而是按错误码定点加约束，并且每轮只改一处，这样下一轮才分得清哪条约束起了作用。

1. 按错误码分组，每类抽几条到十几条原始输出，保留被截断的片段，不要只看第一条。
2. 判断成因：E_JSON 多为输出了代码围栏或前后解释句；E_TYPE 多为数值被写成字符串；E_MISSING 多为没强调“必须全部出现”；E_LEN 多为把整段分析塞进 reason。
3. 在 system 提示里追加一条针对性约束，例如“不要输出代码围栏，第一个字符必须是左花括号”。
4. 记录补丁：版本、触发错误码、样本 id、新增约束句、影响字段。
5. 只重跑失败样本加少量已通过样本做回归，确认没有把原来能过的样本改坏，再合并进主模板。

回灌的目标不是把错误清零，而是把错误码分布收敛到能由重试兜住的范围。如果某一类错误反复出现在同一字段，更稳的处理是把该字段从模型手里拿走：例如 id 一律由代码回填、label 用枚举列表在调用后做一次映射，只把真正需要模型判断的部分留在输出里。