feat: add steps_json-backed RunLog wrapper
This commit is contained in:
@@ -0,0 +1,71 @@
|
||||
"""RunLog 包装器:traces 表空时从 predictions.steps_json 重建轨迹。
|
||||
|
||||
用于对 infer_adhoc 这类 traces 未落表、轨迹在 steps_json 的历史 run 跑离线诊断。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from app.harness.steps_json_traces import steps_json_to_trace_rows
|
||||
|
||||
|
||||
class StepsJsonRunLog:
|
||||
"""委托内层 RunLog;get_traces 空表时回退 steps_json。
|
||||
|
||||
实现 core/evolution/protocols.py 的 RunLog Protocol(duck-typing)。
|
||||
对 traces 已落表的正常 run 完全透传;仅当底层 traces 为空时,
|
||||
才从 predictions.steps_json 经 steps_json_to_trace_rows 重建轨迹行。
|
||||
"""
|
||||
|
||||
def __init__(self, inner: Any) -> None:
|
||||
"""构造包装器。
|
||||
|
||||
参数:
|
||||
inner: 内层 RunLog 实现(如 app/harness/log.py::RunLogImpl),
|
||||
需提供 get_predictions / get_traces 两个 async 方法。
|
||||
"""
|
||||
self._inner = inner
|
||||
|
||||
async def get_predictions(
|
||||
self, run_id: str, *, question_ids: list[str] | None = None
|
||||
) -> list[dict[str, Any]]:
|
||||
"""透传内层预测查询。
|
||||
|
||||
参数:
|
||||
run_id: 运行标识。
|
||||
question_ids: 可选的题目 ID 过滤列表。
|
||||
|
||||
返回:
|
||||
内层返回的预测记录字典列表,原样透传。
|
||||
"""
|
||||
return await self._inner.get_predictions(run_id, question_ids=question_ids)
|
||||
|
||||
async def get_traces(
|
||||
self, run_id: str, *, question_ids: list[str] | None = None
|
||||
) -> list[dict[str, Any]]:
|
||||
"""查询轨迹;底层 traces 表空时从 steps_json 回退重建。
|
||||
|
||||
参数:
|
||||
run_id: 运行标识。
|
||||
question_ids: 可选的题目 ID 过滤列表。
|
||||
|
||||
返回:
|
||||
轨迹行字典列表。
|
||||
|
||||
关键实现细节:
|
||||
- 内层 traces 非空 → 原样返回,不触发回退(正常 run 路径)。
|
||||
- 内层 traces 为空 → 拉取同一过滤条件下的 predictions,
|
||||
逐题经 steps_json_to_trace_rows 展开为轨迹行并拼接。
|
||||
- steps_json 缺失时以空串传入,由下游确定性返回 []。
|
||||
"""
|
||||
inner_rows = await self._inner.get_traces(run_id, question_ids=question_ids)
|
||||
if inner_rows:
|
||||
return inner_rows
|
||||
preds = await self._inner.get_predictions(run_id, question_ids=question_ids)
|
||||
rows: list[dict[str, Any]] = []
|
||||
for p in preds:
|
||||
rows.extend(
|
||||
steps_json_to_trace_rows(p["video_id"], p["question_id"], p.get("steps_json") or "")
|
||||
)
|
||||
return rows
|
||||
@@ -0,0 +1,33 @@
|
||||
import json
|
||||
|
||||
import pytest
|
||||
|
||||
from app.harness.baseline_run_log import StepsJsonRunLog
|
||||
|
||||
|
||||
class _FakeInner:
|
||||
def __init__(self, preds, traces):
|
||||
self._preds, self._traces = preds, traces
|
||||
|
||||
async def get_predictions(self, run_id, *, question_ids=None):
|
||||
return [p for p in self._preds if not question_ids or p["question_id"] in question_ids]
|
||||
|
||||
async def get_traces(self, run_id, *, question_ids=None):
|
||||
return list(self._traces)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_traces_falls_back_to_steps_json_when_table_empty():
|
||||
steps = [{"thought": "t", "tool_call": {"tool": "view_node", "args": {}}, "tool_output": "o"}]
|
||||
preds = [{"video_id": "v1", "question_id": "q1", "steps_json": json.dumps(steps)}]
|
||||
log = StepsJsonRunLog(_FakeInner(preds, traces=[]))
|
||||
rows = await log.get_traces("r", question_ids=["q1"])
|
||||
assert rows[0]["tool_name"] == "view_node" and rows[0]["question_id"] == "q1"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_traces_prefers_nonempty_inner_table():
|
||||
inner_traces = [{"video_id": "v1", "question_id": "q1", "step": 0, "tool_name": "x"}]
|
||||
log = StepsJsonRunLog(_FakeInner([], inner_traces))
|
||||
rows = await log.get_traces("r")
|
||||
assert rows == inner_traces
|
||||
Reference in New Issue
Block a user