要把模型接进科研脚本,先要分清仓库交给你的是哪一种调用形态,因为调用形态决定了脚本里怎么发请求、在哪设置超时、返回值要不要自己解析。很多接入失败并不是 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 和示例里没有出现的接口、参数名,不要凭印象写进脚本。未确认的字段一旦写错,后面的输入输出对齐就建在猜测上,排查成本比多花十分钟看示例高得多。
把脚本输入整理成结构化字段
自然语言拼接的问题在于同一条脚本每次拼出的句子不一样,模型返回也跟着漂。把输入固定成字典,至少包含三块,脚本里所有调用都走同一个构造函数:
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
脚本里对应位置加两行输出即可比对:
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,连同输入快照和原始返回写入独立文件,流水线继续处理下一条。降级分支不做内容猜测和补全,只保留现场,事后补齐或人工重跑。
阈值取多少需要结合环境确认:本地服务冷启动和长文本推理的耗时差别明显,建议先拿固定输入跑几次,记录耗时分布,再把超时设到明显高于常见值的位置;重试次数控制在两次以内,避免一条卡住的样本占住整条流水线。