跑完 LoHoSearch 只看最终答案够不够——中间步骤的错怎么定位?

文章导读
只看最终答案不够,但也不等于要把每条任务的中间步骤从头翻一遍。LoHoSearch 的结果文件通常按任务嵌套保存,最终判定在最外层,步骤记录在数组里;人在这种结构里翻找,很容易停在第一个看起来可疑的步骤上,却不清楚它是不是真正导致失败的那一步。比较稳的做法是先用一个小脚本把嵌套结果拉平成「一行一个步骤」的长表,再对每条失败任务按顺序取第一个不通过步骤,最后按步骤序号做计数分布,看失败集中在起步阶段
📋 目录
  1. 壹 用脚本把结果文件展开成步骤级表格
  2. 贰 给每条失败任务算出首个失分步骤
  3. 叁 把首个失分步骤聚合成分布
  4. 肆 对着该步输入判断是材料缺还是判断错
A A

只看最终答案不够,但也不等于要把每条任务的中间步骤从头翻一遍。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 过滤。

给每条失败任务算出首个失分步骤

人工翻找的问题是不可复算:换个人、换一天,停在哪一步可能不一样。把它固化成规则之后,同一个结果文件每次跑出来的定位都一致。判定顺序建议从前往后遍历步骤,取第一条判定为不通过的步骤作为该任务的首个失分步骤,后面的步骤一律不看,因为后续步骤很可能是在错误上下文上继续叠加的。

跑完 LoHoSearch 只看最终答案够不够——中间步骤的错怎么定位?
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 里该步骤的输出读一遍,确认它确实是第一个出问题的步骤,而不是被一个偏严格的步骤判定提前截断。如果某个任务的步骤判定字段整体缺失,只能退回用最终答案失败加上最后一个有输出的步骤来标记,并在输出里显式标注为待确认,避免把它和真正的首个失分步骤混在一起统计。

把首个失分步骤聚合成分布

单条定位只说明「这条错在哪」,聚合之后才看得出「错法有没有共性」。按步骤序号计数即可,输出不需要花哨。

跑完 LoHoSearch 只看最终答案够不够——中间步骤的错怎么定位?
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")

读这些样本时,按两条标准分拣。第一条是材料缺失:该步需要的上下文、检索结果或上游输出在这一步的输入里根本不存在,或者明显为空、被截断、字段对不上,这类问题的修法在链路上游,比如补齐检索召回、放宽过滤条件、修上游字段传递。第二条是判断错误:材料在该步输入里是齐的,但该步给出的结论与材料不符,比如材料里已写明却被判为不存在、该收束却继续发散,这类问题才轮到改提示词、改判定规则或换更强的模型。分不清时优先按材料缺失怀疑,因为判断错误里有一部分其实是输入不全导致的连带结果,把材料补齐后再看同一批样本是否仍然判错,比直接调提示词更省事。