diff --git a/app/harness/steps_json_traces.py b/app/harness/steps_json_traces.py new file mode 100644 index 0000000..959a217 --- /dev/null +++ b/app/harness/steps_json_traces.py @@ -0,0 +1,51 @@ +"""把 predictions.steps_json 转成 RunLog.get_traces 的行形。 + +infer_adhoc 的 traces 表为空,轨迹存于 steps_json({thought, tool_call, tool_output})。 +诊断管线经 get_traces 消费轨迹,故需此确定性转换适配。 +""" + +from __future__ import annotations + +import json +from typing import Any + + +def steps_json_to_trace_rows( + video_id: str, question_id: str, steps_json: str +) -> list[dict[str, Any]]: + """将单题 steps_json 解析为 trace 行列表(step 从 0 递增)。 + + 参数: + video_id: 视频 ID。 + question_id: 题 ID。 + steps_json: predictions.steps_json 原文(JSON 数组字符串)。 + + 返回: + 行字典列表,字段对齐 traces 表 schema;空/空数组返回 []。 + + 关键实现细节: + - 工具名读 tool_call.tool(infer_adhoc 真实字段),对极少数历史 + 数据的 name 做 back-compat 回退。 + - steps_json 非 JSON 数组时直接报错,不做兜底掩盖。 + """ + if not steps_json or not steps_json.strip(): + return [] + steps = json.loads(steps_json) + if not isinstance(steps, list): + raise ValueError(f"steps_json 非数组: {question_id}") + rows: list[dict[str, Any]] = [] + for i, s in enumerate(steps): + call = s.get("tool_call") or {} + rows.append( + { + "video_id": video_id, + "question_id": question_id, + "step": i, + # infer_adhoc 用 "tool";back-compat 兼容极少数 "name" + "tool_name": call.get("tool", call.get("name")), + "tool_args": call.get("args", {}), + "tool_output": s.get("tool_output"), + "thought": s.get("thought"), + } + ) + return rows diff --git a/tests/unit/test_steps_json_traces.py b/tests/unit/test_steps_json_traces.py new file mode 100644 index 0000000..af750f0 --- /dev/null +++ b/tests/unit/test_steps_json_traces.py @@ -0,0 +1,30 @@ +import json + +from app.harness.steps_json_traces import steps_json_to_trace_rows + + +def test_parses_tool_call_into_name_and_args(): + # 真实 infer_adhoc steps_json 形态:tool_call={"tool":..., "args":...}(非 "name") + steps = [ + { + "thought": "看根节点", + "tool_call": {"tool": "view_node", "args": {"node_id": "v_L1_000"}}, + "tool_output": "o0", + }, + { + "thought": "搜索", + "tool_call": {"tool": "search_similar", "args": {"query": "gadget"}}, + "tool_output": "hit", + }, + ] + rows = steps_json_to_trace_rows("vid1", "q1", json.dumps(steps)) + assert [r["step"] for r in rows] == [0, 1] + assert rows[0]["tool_name"] == "view_node" + assert rows[0]["tool_args"] == {"node_id": "v_L1_000"} + assert rows[0]["video_id"] == "vid1" and rows[0]["question_id"] == "q1" + assert rows[1]["tool_output"] == "hit" + + +def test_empty_or_blank_steps_json_returns_empty(): + assert steps_json_to_trace_rows("v", "q", "") == [] + assert steps_json_to_trace_rows("v", "q", "[]") == []