书生-A1.5 接进自己的科研脚本——输入输出格式从哪开始对齐?

文章导读
要把模型接进科研脚本,先要分清仓库交给你的是哪一种调用形态,因为调用形态决定了脚本里怎么发请求、在哪设置超时、返回值要不要自己解析。很多接入失败并不是 prompt 写得不好,而是输入端把一次结构化任务拼成了一段自然语言,输出端又指望模型自己给出可 json.loads 的字符串。建议的顺序是:先确认调用方式,再把输入写成字段,再把输出约束并解析,最后用一条固定输入把链路跑通,确认稳定之后才谈批量
📋 目录
  1. A 先确认仓库提供的调用方式是进程内、本地服务还是 SDK
  2. B 把脚本输入整理成结构化字段
  3. C 把输出约束成可解析格式
  4. D 用一条固定输入跑通全链路
  5. E 把超时、重试和半截返回写进脚本
A A

要把模型接进科研脚本,先要分清仓库交给你的是哪一种调用形态,因为调用形态决定了脚本里怎么发请求、在哪设置超时、返回值要不要自己解析。很多接入失败并不是 prompt 写得不好,而是输入端把一次结构化任务拼成了一段自然语言,输出端又指望模型自己给出可 json.loads 的字符串。建议的顺序是:先确认调用方式,再把输入写成字段,再把输出约束并解析,最后用一条固定输入把链路跑通,确认稳定之后才谈批量与效果。

对齐可以拆成四步:先用仓库 README 与示例目录确认调用形态(进程内、本地 HTTP 服务或 SDK),再把脚本输入写成包含任务类型、数据字段、输出格式要求的结构化字典,然后把返回按不可信字符串解析并在失败时保留原始返回,最后用固定一条输入同时打印原始返回与解析字段逐项比对。判断链路是否跑通看的是原始返回与解析字段能否对上,不是输出内容是否好看。超时与重试阈值要写在脚本侧显式位置,半截返回走降级而不是重试到底。

先确认仓库提供的调用方式是进程内、本地服务还是 SDK

仓库通常在三个地方暴露真实调用方式:README 的快速开始、examples 或 demo 目录下的脚本、依赖文件里的包名。三处说法不一致时,以能完整跑通的那条示例为准。判断依据可以按下面三类分:

  • 进程内:示例里出现 from 某模块 import 某类 后直接构造并调用,说明脚本要和模型装在同一环境。需要额外确认模型加载耗时与显存占用,这类调用在循环里重复加载会很慢,通常应该在脚本入口加载一次并复用。
  • 本地服务:示例里出现 uvicorn、FastAPI 启动命令或 curl 127.0.0.1:端口,脚本按普通 HTTP 客户端处理。端口、接口路径、请求体字段名都要写进配置,不要散落在代码里。
  • SDK:示例里出现客户端对象初始化并带 base_url、api_key 一类参数,此时按 SDK 的方式构造消息,模型名或版本字段要显式写入参数。

动手写请求代码之前,先把仓库翻一遍:

sed -n '1,80p' README.md
ls examples demo 2>/dev/null
grep -rnE 'chat|generate|FastAPI|uvicorn|base_url' `--include`='*.py' . | head -n 40

README 和示例里没有出现的接口、参数名,不要凭印象写进脚本。未确认的字段一旦写错,后面的输入输出对齐就建在猜测上,排查成本比多花十分钟看示例高得多。

把脚本输入整理成结构化字段

自然语言拼接的问题在于同一条脚本每次拼出的句子不一样,模型返回也跟着漂。把输入固定成字典,至少包含三块,脚本里所有调用都走同一个构造函数:

书生-A1.5 接进自己的科研脚本——输入输出格式从哪开始对齐?
payload = {
    'task_type': 'extract_metrics',
    'data': {
        'text': paragraph,
        'fields_hint': ['指标名', '数值', '单位'],
    },
    'output_format': {
        'type': 'json_object',
        'fields': ['metric', 'value', 'unit'],
        'no_extra_text': True,
    },
}
  • 任务类型:用有限枚举,脚本维护一份可选值列表,传入前先校验,不要在调用处临时拼一句任务描述。
  • 数据字段:原始文本、文件路径或已抽取的中间结果,保持每次调用结构一致。超出模型上下文长度时在脚本侧截断或分片,不要交给模型自己判断该丢掉哪一段。
  • 输出格式要求:把字段名和期望类型写清楚。仓库的模板或参数支持指定 JSON 时优先用参数约束;不支持时把格式要求作为同一段输入的一部分写进去,并接受它是软约束这一事实。

替换项只有三处:task_type 的枚举值、data 里的业务字段、output_format 的字段列表。这三处定了,解析代码才有稳定的目标结构。

把输出约束成可解析格式

无论提示怎么写,都应按不可信字符串处理模型返回:先原样保留,再尝试解析,解析结果里带上原始文本。下面这段骨架区分了三种失败,替换 OUTPUT_FIELDS 即可复用到别的任务:

import json

OUTPUT_FIELDS = ('metric', 'value', 'unit')

def parse_result(raw):
    obj = None
    try:
        obj = json.loads(raw)
    except json.JSONDecodeError:
        i, j = raw.find('{'), raw.rfind('}')
        if i != -1 and j > i:
            try:
                obj = json.loads(raw[i:j + 1])
            except json.JSONDecodeError:
                obj = None
    if obj is None:
        return {'ok': False, 'reason': 'not_json', 'raw': raw}
    missing = [f for f in OUTPUT_FIELDS if f not in obj]
    if missing:
        return {'ok': False, 'reason': 'missing_fields', 'missing': missing, 'raw': raw}
    if not isinstance(obj['value'], (int, float)):
        return {'ok': False, 'reason': 'type_mismatch', 'raw': raw}
    return {'ok': True, 'data': obj, 'raw': raw}

截取首尾大括号再解析,能救回模型在 JSON 前后多说一句话的情况,但救不回字段缺失。字段类型也不要在脚本里静默转换,例如 value 期望数字却拿到带单位的字符串时,标记出来交给人工确认,比强行 float() 后再出错更容易定位。

用一条固定输入跑通全链路

准备一条最小但真实的输入,存成 fixtures/one_case.json,跑的时候把原始返回和解析结果同时打出来,日志留存:

python run_agent.py `--input` fixtures/one_case.json `--debug` 2>&1 | tee logs/one_case.log

脚本里对应位置加两行输出即可比对:

书生-A1.5 接进自己的科研脚本——输入输出格式从哪开始对齐?
print('RAW >>>', repr(raw)[:800])
print('PARSED >>>', parsed)

逐项比对四个点:一是返回里是否混有解释性文字,二是字段名是否与 OUTPUT_FIELDS 完全一致,三是字段类型是否匹配,四是同一段输入连续跑两次结果结构是否稳定。四项都对上,说明输入输出这一层是通的;对不上时,先回到输入 schema 检查字段是否表达清楚,再考虑调整提示措辞。链路没跑通之前不建议批量执行,否则会把格式问题扩散到整批结果里。

把超时、重试和半截返回写进脚本

长文本推理耗时不稳定,超时阈值应写在请求构造处,重试次数写在调用外层,两处分开配置,不要混成一个常量:

TIMEOUT_S = 60     # 单次请求超时,写在 client 调用处
MAX_RETRY = 2      # 重试次数上限,写在调用外层

def call_with_guard(payload):
    last_raw = ''
    for attempt in range(MAX_RETRY + 1):
        try:
            last_raw = client.send(payload, timeout=TIMEOUT_S)
        except TimeoutError:
            continue
        except Exception:
            continue
        res = parse_result(last_raw)
        if res['ok']:
            return res
        if res['reason'] == 'not_json':
            continue
        return res
    return {'ok': False, 'reason': 'timeout_or_incomplete', 'raw': last_raw}

半截返回的典型表现是 JSON 少右括号、字段写到一半断掉,这类情况按失败处理,重试到上限仍失败时走降级分支:把该条目标记为 pending_review,连同输入快照和原始返回写入独立文件,流水线继续处理下一条。降级分支不做内容猜测和补全,只保留现场,事后补齐或人工重跑。

阈值取多少需要结合环境确认:本地服务冷启动和长文本推理的耗时差别明显,建议先拿固定输入跑几次,记录耗时分布,再把超时设到明显高于常见值的位置;重试次数控制在两次以内,避免一条卡住的样本占住整条流水线。