只看最终答案不够,但也不等于要把每条任务的中间步骤从头翻一遍。LoHoSearch 的结果文件通常按任务嵌套保存,最终判定在最外层,步骤记录在数组里;人在这种结构里翻找,很容易停在第一个看起来可疑的步骤上,却不清楚它是不是真正导致失败的那一步。比较稳的做法是先用一个小脚本把嵌套结果拉平成「一行一个步骤」的长表,再对每条失败任务按顺序取第一个不通过步骤,最后按步骤序号做计数分布,看失败集中在起步阶段还是收尾阶段。脚本里的字段名是占位写法,需要按你环境里结果文件的实际键名替换。
适用场景:LoHoSearch 已跑完、只有任务级通过/不通过、需要定位中间步骤。操作动作:把结果文件展开成步骤级长表,按任务取首个不通过步骤,再按步骤序号聚合。验证方式:展开后的行数应等于各任务步骤数之和,人工抽查两三条确认首个失分步骤取值正确。边界:步骤判定字段缺失或语义不明时,只能标为待确认,不能直接当成结论。
用脚本把结果文件展开成步骤级表格
这一步的目的只有一个:把嵌套结构变成长表,让每一行对应「某个任务的某一步」,后续的筛选、计数、抽样都建立在这张长表上。下面是一个通用骨架,逐行读结果文件,打印任务标识、步骤序号、该步输出、该步判定四列,字段名用占位符,落地前先拿一条记录确认键名。
import json
RESULT_PATH = "results.jsonl" # 换成你环境里的结果文件路径
F = { # 字段名占位,按实际输出替换
"task_id": "task_id",
"steps": "steps",
"step_index": "step", # 没有序号就用数组下标兜底
"step_output": "output",
"step_judge": "judge", # 该步判定:pass / fail / 其他
}
n_task = 0
n_step = 0
with open(RESULT_PATH, encoding="utf-8") as fin:
print("task_id\tstep\tjudge\toutput")
for line in fin:
line = line.strip()
if not line:
continue
task = json.loads(line)
n_task += 1
steps = task.get(F["steps"]) or []
for i, s in enumerate(steps):
n_step += 1
idx = s.get(F["step_index"], i)
out = str(s.get(F["step_output"], ""))[:120].replace("\n", " ")
print(f'{task.get(F["task_id"])}\t{idx}\t{s.get(F["step_judge"])}\t{out}')
print(f"# tasks={n_task} steps={n_step}", flush=True)验证方式很直接:把 n_task 和结果文件的行数对上,把 n_step 和逐任务手工累加的步骤数对上;如果某个任务的步骤数组为空或字段名写错,n_step 会明显偏小,这时先修字段映射,不要急着看分布。输出建议重定向到文件,比如 python expand.py > steps.tsv,再用表格工具或 grep 过滤。
给每条失败任务算出首个失分步骤
人工翻找的问题是不可复算:换个人、换一天,停在哪一步可能不一样。把它固化成规则之后,同一个结果文件每次跑出来的定位都一致。判定顺序建议从前往后遍历步骤,取第一条判定为不通过的步骤作为该任务的首个失分步骤,后面的步骤一律不看,因为后续步骤很可能是在错误上下文上继续叠加的。
FAIL_VALUES = {"fail", "failed", "no", "false", "0"}
def first_fail(task, F):
for i, s in enumerate(task.get(F["steps"]) or []):
v = str(s.get(F["step_judge"], "")).strip().lower()
if v in FAIL_VALUES:
return s.get(F["step_index"], i), v
return None, None
# 用法:只处理任务级失败的任务,逐条打印定位结果
for line in open(RESULT_PATH, encoding="utf-8"):
line = line.strip()
if not line:
continue
task = json.loads(line)
idx, v = first_fail(task, F)
if idx is not None:
print(f'{task.get(F["task_id"])}\t{i}\t{v}') # task_id / 首个失分步骤 / 该步判定规则写完后不要直接信,人工抽两三条任务,把原始 JSON 里该步骤的输出读一遍,确认它确实是第一个出问题的步骤,而不是被一个偏严格的步骤判定提前截断。如果某个任务的步骤判定字段整体缺失,只能退回用最终答案失败加上最后一个有输出的步骤来标记,并在输出里显式标注为待确认,避免把它和真正的首个失分步骤混在一起统计。
把首个失分步骤聚合成分布
单条定位只说明「这条错在哪」,聚合之后才看得出「错法有没有共性」。按步骤序号计数即可,输出不需要花哨。
from collections import Counter
c = Counter()
for line in open(RESULT_PATH, encoding="utf-8"):
line = line.strip()
if not line:
continue
task = json.loads(line)
idx, _ = first_fail(task, F)
if idx is not None:
c[idx] += 1
for idx, n in sorted(c.items()):
print(f"step={idx}\tcount={n}")输出形如 step=0 count=…、step=3 count=…,一眼能看出失败是压在头几步(往往是理解任务、构造检索条件这类起步环节),还是散在尾段(多为汇总、收敛、输出格式这类收尾环节)。这里要克制:失败任务只有几条到十几条时,单个步骤的计数波动很大,只能当作线索,不能当作结论;先把样本量补上来,或者把同一类任务拆开分别看,再判断哪一步是主要失分点。
对着该步输入判断是材料缺还是判断错
分布只告诉你哪一步容易出问题,修法对不对要看这一步的输入。先把失败样本该步的输入抽出来存成小文件,逐条读,而不是在原文件里翻。
import json
WANT = {("task_001", 2), ("task_007", 2)} # (任务标识, 步骤序号) 来自上一步的定位结果
OUT = "fail_inputs.jsonl"
with open(OUT, "w", encoding="utf-8") as fout:
for line in open(RESULT_PATH, encoding="utf-8"):
line = line.strip()
if not line:
continue
task = json.loads(line)
tid = task.get(F["task_id"])
for i, s in enumerate(task.get(F["steps"]) or []):
if (tid, s.get(F["step_index"], i)) in WANT:
fout.write(json.dumps({
"task_id": tid,
"step": s.get(F["step_index"], i),
"input": s.get("input"), # 该步输入,键名按实际替换
"output": s.get(F["step_output"]),
}, ensure_ascii=False) + "\n")读这些样本时,按两条标准分拣。第一条是材料缺失:该步需要的上下文、检索结果或上游输出在这一步的输入里根本不存在,或者明显为空、被截断、字段对不上,这类问题的修法在链路上游,比如补齐检索召回、放宽过滤条件、修上游字段传递。第二条是判断错误:材料在该步输入里是齐的,但该步给出的结论与材料不符,比如材料里已写明却被判为不存在、该收束却继续发散,这类问题才轮到改提示词、改判定规则或换更强的模型。分不清时优先按材料缺失怀疑,因为判断错误里有一部分其实是输入不全导致的连带结果,把材料补齐后再看同一批样本是否仍然判错,比直接调提示词更省事。