用 GPT-6.1 Terra 跑批量任务，先在脚本里把输入输出格式定死

批量任务输出时好时坏，通常不是模型能力不稳定，而是脚本里没定义“什么算一条合法结果”。可落地的做法是把格式约束前置到脚本层：输入用固定模板渲染，输出用固定字段清单断言，调用外面再包一层统一的重试外壳。校验不通过的结果不进写库环节，重试耗尽转人工队列。这样后面调提示词、调重试上限时，才有日志可以对照，而不是凭感觉改。

跑 GPT-6.1 Terra 这类批量任务，格式约束建议卡在脚本层，不要只写在提示词里：脚本负责输入模板渲染、输出字段断言和异常分类，提示词只承担内容质量。适用场景是任务要重复跑、结果要写库或进下游的批处理；边界在于字段清单只能约束结构，保证不了语义正确，语义仍需人工抽检。验证方法是干跑一批后，日志里每行都能看到校验结果与重试次数。

把一条任务拆成输入模板和输出字段清单

一条任务只对应一个模板，模板里只留占位符，不在脚本里拼字符串。占位符用双花括号标记，渲染时严格替换，缺一个就直接报错，而不是渲染出半条任务再发给模型。

INPUT_TEMPLATE = {
    "task_id": "{{task_id}}",          # 必填，唯一，用于日志、去重、失败回查
    "task_type": "{{task_type}}",      # 枚举：summary / classify / extract
    "schema_version": "v1",            # 与被校验的字段清单版本对齐
    "locale": "{{locale}}",            # 如 zh-CN
    "payload": "{{payload}}",          # 原始文本，超长在渲染前截断
}

def render(template, row):
    out = {}
    for k, v in template.items():
        if isinstance(v, str) and "{{" in v:
            key = v.strip("{}").strip()
            if key not in row or row[key] in (None, ""):
                raise ValueError(f"template_missing:{key}")
            out[k] = str(row[key])
        else:
            out[k] = v
    return out

输出字段清单是另一份独立定义，写进代码常量，同时以自然语言抄进请求提示词。两边不一致时以代码为准，提示词跟着改。

字段名	类型	是否必填	空值如何处理
task_id	string	必填	缺失或为空即判失败，不写库
label	string，白名单枚举	必填	空串视为失败，进重试分支
confidence	number，0 到 1	可选	缺失写 null，下游按“未知”处理，不补默认值
items	array<object>	条件必填（task_type=extract 时）	空数组合法；字段缺失判失败
reason	string	可选	缺失写空串，不参与检索
model	string	必填	由脚本回填配置值，不取模型自述

在脚本里写请求封装与重试外壳

业务代码只负责准备输入、消费合法结果；超时、重试、异常归类全部收进一个函数。这样换 SDK、换并发模型时，只改这一处。下面用的是通用调用形式，client.generate(payload, timeout=...) 需要替换成你实际 SDK 的调用方式，参数名按所用 SDK 文档填。

import time

class TransientError(Exception): pass   # 值得重试
class PermanentError(Exception): pass   # 重试也没用

def call_model(client, payload, timeout=60, max_retry=2):
    last = None
    for attempt in range(max_retry + 1):
        try:
            resp = client.generate(payload, timeout=timeout)
            text = getattr(resp, "text", None)
            if not text:
                raise TransientError("empty_response")
            return text, attempt
        except TimeoutError as e:
            last = TransientError(f"timeout:{e}")
        except Exception as e:
            msg = str(e).lower()
            if any(k in msg for k in ("rate limit", "429", "502", "503", "unavailable")):
                last = TransientError(msg)
            else:
                last = PermanentError(msg)
                break
        time.sleep(min(2 ** attempt, 8))
    raise last

异常分类的判据建议写在一处常量里，方便后面按日志调整：超时、限流、服务端 5xx、空响应归为可重试；鉴权失败、请求体超长、参数非法归为不可重试，直接落到永久失败分支。重试次数与超时都做成配置项，先给保守值，例如超时 60 秒、重试 2 次，再按干跑结果决定是否调整。

用字段断言校验返回结果

校验放在解析之后、写库之前。先容错解析 JSON，再用字段清单断言结构，任一断言失败就带着错误列表返回，而不是抛一个笼统异常。

import json

REQUIRED = {"task_id": str, "label": str, "model": str}
LABELS = {"退款", "物流", "其他"}

def parse_and_validate(raw, task_type):
    try:
        obj = json.loads(raw)
    except json.JSONDecodeError:
        return None, ["parse_error"]
    errs = []
    for k, t in REQUIRED.items():
        v = obj.get(k)
        if not isinstance(v, t) or v == "":
            errs.append(f"missing:{k}")
    if obj.get("label") not in LABELS:
        errs.append("enum:label")
    c = obj.get("confidence")
    if c is not None and not (isinstance(c, (int, float)) and 0 <= c <= 1):
        errs.append("range:confidence")
    if task_type == "extract" and not isinstance(obj.get("items"), list):
        errs.append("type:items")
    return obj, errs

失败后走三条分支，触发条件要能一眼判断：

丢弃：错误来自输入侧，比如模板占位符缺失、payload 为空、task_type 不在枚举内。这类问题重试结果不会变，记 fail_fast 后跳过该条。
重试：错误集中在输出结构，比如 parse_error、missing:label、enum:label，且本轮 attempt 未达上限。重试时把错误列表作为反馈拼进提示词，例如“上次返回缺少 label 字段，只输出 JSON 对象”。
转人工：重试次数耗尽仍不合法，或返回被内容策略拒绝。落到 review 队列，记录 task_id、原始输入、最后一次原始返回和错误列表，不做任何自动修补。

小批量干跑并统计失败分布

不要一上来全量跑。先从待处理集里抽一小批，例如 20 到 50 条，覆盖最短文本、最长文本、含换行与引号的文本、以及历史上容易出问题的类型，固定随机种子便于复现。干跑阶段关闭写库，只写本地日志文件，每行带 task_id、attempt、validate_errs、elapsed_ms。

失败类型	日志标记	判断依据	处理方向
输入越界	fail_fast / template_missing	请求发出前就被拦下	修输入或加预清洗，不重试
JSON 解析失败	parse_error	返回体不是合法 JSON	提示词强调只输出 JSON；SDK 支持结构化输出时优先启用
必填缺失	missing:xxx	校验函数 errs 列表	字段清单与示例补进提示词
枚举越界	enum:label	label 不在白名单	白名单原文写进提示词，避免模型自造标签
超时或限流	timeout / 429	异常分类为可重试	看分布决定降并发还是加超时
重试耗尽	exhausted	循环结束仍未通过校验	进人工队列，人工看过再改规则

干跑结束后用日志看分布，而不是看整体感受：

grep -o 'fail_type=[a-z_:]*' dryrun.log | sort | uniq -c | sort -rn
awk -F'attempt=' '{print $2}' dryrun.log | cut -d' ' -f1 | sort | uniq -c

如果失败集中在 parse_error 和 missing，先改提示词里的字段说明与示例；如果集中在 timeout，先降并发或放宽超时，而不是无脑加重试次数。重试上限调高之后必须重跑同一批样本对比失败分布，否则无法判断是规则变好还是样本变了。校验逻辑、字段清单、提示词三者的版本号要一起记进日志，出问题时才能定位是哪次改动引入的。