iomgaa
8958eee11b
config/train_videomme.yaml 同时收录待入库的实验配置变更(run_id v2 / concurrency 32 / batch_size 40)。tests/integration/test_v3_contract_e2e.py 的 run_id 断言按 Task 5 显式契约同步修正(原断言依赖旧隐式实例注入)。
368 lines
13 KiB
Python
368 lines
13 KiB
Python
"""训练循环诊断注入：_run_diagnosis 按 batch question_ids 加载真实树注入 run_diagnosis。
|
||
|
||
覆盖诊断 tree_data 断链修复的训练循环侧（Task 4）：验证 runner._run_diagnosis 会把
|
||
batch 涉及 video 的真实树作为 tree_data 传给 core.run_diagnosis，而非空 dict。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from pathlib import Path # noqa: TC003 — 运行时 tmp_path 标注使用
|
||
from unittest.mock import AsyncMock, MagicMock, patch
|
||
|
||
import pytest
|
||
|
||
from app.harness.config import RunConfig
|
||
from app.harness.runner import Runner
|
||
from core.evolution.types import DiagnosisResult
|
||
|
||
# 真实样本：question 604-2 属于 video 0RxMZBLeqRI（111 节点树）。
|
||
_REAL_QUESTION_ID = "604-2"
|
||
_REAL_VIDEO_ID = "0RxMZBLeqRI"
|
||
|
||
|
||
def _empty_diagnosis_result(run_id: str = "infer_adhoc") -> DiagnosisResult:
|
||
"""构造仅含 run_id 的空诊断结果（其余字段走 dataclass 默认值）。"""
|
||
return DiagnosisResult(run_id=run_id)
|
||
|
||
|
||
def _base_config(workspace_dir: Path, store_dir: Path) -> RunConfig:
|
||
"""构造 diagnose 模式 RunConfig，所有必填字段给测试默认值。
|
||
|
||
参数:
|
||
workspace_dir: 已写入 manifest 的 workspace 根目录。
|
||
store_dir: 真实 store 根目录（含 questions/ 与 videos/）。
|
||
"""
|
||
return RunConfig(
|
||
workspace_dir=workspace_dir,
|
||
store_dir=store_dir,
|
||
mode="diagnose",
|
||
concurrency=1,
|
||
max_steps=5,
|
||
skill_mode="none",
|
||
n_samples=0,
|
||
questions="benchmarks/Video-MME",
|
||
skills_version="v1",
|
||
prompts_version="v1",
|
||
epochs=1,
|
||
diag_size=10,
|
||
diag_correct_ratio=0.5,
|
||
val_size=24,
|
||
val_correct_ratio=0.5,
|
||
edit_budget_start=5,
|
||
edit_budget_end=2,
|
||
batch_size=5,
|
||
min_class_per_batch=2,
|
||
eval_min_per_class=2,
|
||
trainable_min_units=8,
|
||
early_stop_patience=3,
|
||
test_size=10,
|
||
use_slow_momentum=False,
|
||
gate_e_confirm=20.0,
|
||
gate_e_provisional=3.0,
|
||
gate_w_net_min=2,
|
||
gate_delta_min=0.02,
|
||
gate_lambda_dir=-0.642,
|
||
gate_e_rollback=10.0,
|
||
gate_n_max=40,
|
||
gate_p_low=0.05,
|
||
gate_p_high=0.95,
|
||
gate_probe_quota=0.2,
|
||
gate_gamma_decay=0.9,
|
||
gate_cooldown_steps=2,
|
||
gate_guard_err=0.10,
|
||
skill_update_mode="patch",
|
||
appendix_consolidate_threshold=6,
|
||
run_id="infer_adhoc",
|
||
)
|
||
|
||
|
||
@pytest.fixture
|
||
def runner_with_real_store(tmp_path: Path) -> Runner:
|
||
"""构造 diagnose 模式 runner，questions/videos 指向真实 store。
|
||
|
||
manifest.store 写真实 store 绝对路径，使 resolve_paths 的 questions_dir
|
||
命中含 604-2 的题库、store_dir 命中 0RxMZBLeqRI 的真实 tree.json。
|
||
"""
|
||
store_dir = Path(__file__).resolve().parents[2] / "store"
|
||
ws = tmp_path / "ws"
|
||
ws.mkdir()
|
||
(ws / "skills" / "v1").mkdir(parents=True)
|
||
manifest = {
|
||
"name": "ws",
|
||
"created_at": "",
|
||
"store": store_dir.as_posix(),
|
||
"current": {
|
||
"videos": "videos",
|
||
"questions": "questions/benchmarks/Video-MME",
|
||
"skills": "skills/v1",
|
||
"prompts": "prompts/v1",
|
||
},
|
||
"history": [],
|
||
}
|
||
(ws / "manifest.json").write_text(json.dumps(manifest))
|
||
|
||
return Runner(
|
||
_base_config(ws, store_dir),
|
||
llm=MagicMock(),
|
||
evolve_llm=MagicMock(),
|
||
vlm=MagicMock(),
|
||
telemetry=MagicMock(),
|
||
)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_run_diagnosis_injects_tree_data(runner_with_real_store: Runner) -> None:
|
||
"""_run_diagnosis 把 batch question_ids 对应 video 的真实树注入 run_diagnosis。"""
|
||
captured: dict[str, object] = {}
|
||
|
||
async def _fake_run_diagnosis(**kwargs: object) -> DiagnosisResult:
|
||
captured["tree_data"] = kwargs["tree_data"]
|
||
return _empty_diagnosis_result()
|
||
|
||
with patch(
|
||
"core.evolution.diagnose.run_diagnosis",
|
||
new=AsyncMock(side_effect=_fake_run_diagnosis),
|
||
):
|
||
await runner_with_real_store._run_diagnosis("infer_adhoc", question_ids=[_REAL_QUESTION_ID])
|
||
|
||
tree_data = captured["tree_data"]
|
||
assert _REAL_VIDEO_ID in tree_data
|
||
assert tree_data[_REAL_VIDEO_ID]["nodes"]
|
||
|
||
|
||
def _fake_question(question_id: str, video_id: str) -> object:
|
||
"""构造仅设置 question_id/video_id 的 GeneratedQuestion（其余字段给占位默认）。"""
|
||
from core.types import GeneratedQuestion
|
||
|
||
return GeneratedQuestion(
|
||
question_id=question_id,
|
||
video_id=video_id,
|
||
task_type="Action Reasoning",
|
||
question="问题",
|
||
options=("A", "B", "C", "D"),
|
||
answer="A",
|
||
source_nodes=(),
|
||
difficulty="medium",
|
||
)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_diagnosis_reads_traces_from_steps_json(
|
||
runner_with_real_store: Runner,
|
||
) -> None:
|
||
"""traces 表为空但 predictions.steps_json 有轨迹时，诊断仍拿到非空 traces。
|
||
|
||
构造一条只写 steps_json、不写 traces 表的 predictions 行；patch run_diagnosis
|
||
捕获传入的 run_log，直接 await 其 get_traces 断言经 StepsJsonRunLog 从
|
||
steps_json 重建出非空轨迹（算法 #7 恢复）。
|
||
"""
|
||
from app.harness.inference import PREDICTIONS_SCHEMA
|
||
from app.harness.log import HarnessLog
|
||
|
||
steps_json = json.dumps(
|
||
[
|
||
{
|
||
"thought": "先看整体",
|
||
"tool_call": {"tool": "search_tree", "args": {"query": "开场"}},
|
||
"tool_output": "命中 L2 节点 A",
|
||
}
|
||
],
|
||
ensure_ascii=False,
|
||
)
|
||
with HarnessLog(str(runner_with_real_store._paths.db_path), "infer_adhoc") as log:
|
||
log.create_table("predictions", PREDICTIONS_SCHEMA)
|
||
log.create_table("traces", {"video_id": "TEXT", "question_id": "TEXT", "step": "INTEGER"})
|
||
log.insert(
|
||
"predictions",
|
||
{
|
||
"video_id": _REAL_VIDEO_ID,
|
||
"question_id": _REAL_QUESTION_ID,
|
||
"task_type": "Action Reasoning",
|
||
"prediction": "A",
|
||
"answer": "B",
|
||
"evidence": "",
|
||
"reasoning": "",
|
||
"steps_used": 1,
|
||
"prompt_tokens": 0,
|
||
"completion_tokens": 0,
|
||
"stop_reason": "finished",
|
||
"steps_json": steps_json,
|
||
},
|
||
)
|
||
|
||
captured: dict[str, object] = {}
|
||
|
||
async def _fake_run_diagnosis(**kwargs: object) -> DiagnosisResult:
|
||
captured["run_log"] = kwargs["run_log"]
|
||
return _empty_diagnosis_result()
|
||
|
||
with patch(
|
||
"core.evolution.diagnose.run_diagnosis",
|
||
new=AsyncMock(side_effect=_fake_run_diagnosis),
|
||
):
|
||
await runner_with_real_store._run_diagnosis(
|
||
"infer_adhoc", question_ids=[_REAL_QUESTION_ID]
|
||
)
|
||
|
||
run_log = captured["run_log"]
|
||
traces = await run_log.get_traces("infer_adhoc", question_ids=[_REAL_QUESTION_ID])
|
||
assert traces, "traces 表空时应从 steps_json 重建出非空轨迹"
|
||
assert traces[0]["tool_name"] == "search_tree"
|
||
|
||
|
||
def _fake_inference_result(accuracy: float) -> object:
|
||
"""构造 InferenceResult 供 holdout 去重测试（per_task_type 空、token 归零）。"""
|
||
from app.harness.inference import InferenceResult
|
||
|
||
return InferenceResult(
|
||
run_id="x",
|
||
accuracy=accuracy,
|
||
total=10,
|
||
correct=int(accuracy * 10),
|
||
per_task_type={},
|
||
steps_mean=1.0,
|
||
token_usage={"prompt_tokens": 0, "completion_tokens": 0},
|
||
stop_reason_counts={},
|
||
)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_holdout_dedup_skips_reevaluated_versions(
|
||
runner_with_real_store: Runner,
|
||
) -> None:
|
||
"""baseline 不跑推理（从基线预测推导）；best_hard==final 时不重复评估。"""
|
||
from unittest.mock import AsyncMock
|
||
|
||
runner = runner_with_real_store
|
||
eval_calls: list[str] = []
|
||
|
||
async def _fake_eval(sv, pv, questions, run_id, context): # noqa: ANN001
|
||
eval_calls.append(run_id)
|
||
return _fake_inference_result(0.5)
|
||
|
||
runner._eval_version_on_pool = AsyncMock(side_effect=_fake_eval)
|
||
# best_mixed 赢家取 final 版本（必落在 memo，0 推理）
|
||
runner._pick_mixed_best = AsyncMock(return_value=("skills_final", "prompts_final"))
|
||
# baseline 推导：patch 为 0 推理的假结果（不经 _eval_version_on_pool）
|
||
runner._derive_baseline_test_result = MagicMock(return_value=_fake_inference_result(0.4))
|
||
|
||
state = MagicMock()
|
||
state.holdout_memo = {}
|
||
state.baseline_skills_version = "skills_base"
|
||
state.baseline_prompts_version = "prompts_base"
|
||
# best_hard 版本 == final 版本 → 去重
|
||
state.best_skills_version = "skills_final"
|
||
state.best_prompts_version = "prompts_final"
|
||
|
||
pools = MagicMock()
|
||
pools.baseline_run_id = "infer_adhoc"
|
||
pools.test = [_fake_question("q1", "vA")]
|
||
|
||
await runner._holdout_four_way(
|
||
1, pools, state, eval_skills_version="skills_final", eval_prompts_version="prompts_final"
|
||
)
|
||
|
||
# baseline=0（推导）、best_hard=1（真评并存 memo）、final/best_mixed 引用 memo（0）
|
||
assert len(eval_calls) == 1
|
||
runner._derive_baseline_test_result.assert_called_once()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_run_step_deletes_stale_rows_before_rerun(
|
||
runner_with_real_store: Runner, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""同 run_id 重跑前先清 predictions/traces，避免重复行双计（幂等）。"""
|
||
from unittest.mock import AsyncMock
|
||
|
||
from app.harness.inference import PREDICTIONS_SCHEMA, TRACES_SCHEMA
|
||
from app.harness.log import HarnessLog
|
||
|
||
run_id = "infer_adhoc_e1_s0"
|
||
# 预置该 step run_id 的旧 predictions/traces 行
|
||
with HarnessLog(str(runner_with_real_store._paths.db_path), run_id) as log:
|
||
log.create_table("predictions", PREDICTIONS_SCHEMA)
|
||
log.create_table("traces", TRACES_SCHEMA)
|
||
log.insert("predictions", {"video_id": "vA", "question_id": "q1", "prediction": "A"})
|
||
log.insert("traces", {"video_id": "vA", "question_id": "q1", "step": 0})
|
||
|
||
batch = [_fake_question("q1", "vA")]
|
||
runner_with_real_store._rollout_batch = AsyncMock()
|
||
monkeypatch.setattr("app.harness.runner._apply_batch_correctness", lambda *a, **k: None)
|
||
runner_with_real_store._run_diagnosis = AsyncMock(return_value=DiagnosisResult(run_id=run_id))
|
||
runner_with_real_store._gate_batch_skills = AsyncMock()
|
||
|
||
state = MagicMock()
|
||
state.correctness = {"q1": True}
|
||
state.gate_cooldown = {}
|
||
pools = MagicMock()
|
||
pools.baseline_run_id = "infer_adhoc"
|
||
|
||
await runner_with_real_store._run_step(1, 0, 10, batch, pools, state)
|
||
|
||
with HarnessLog(
|
||
str(runner_with_real_store._paths.db_path), run_id, register_run=False
|
||
) as log:
|
||
preds = log.query("SELECT * FROM predictions WHERE run_id=?", (run_id,))
|
||
traces = log.query("SELECT * FROM traces WHERE run_id=?", (run_id,))
|
||
assert preds == []
|
||
assert traces == []
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_run_step_aborts_on_high_degrade_rate(
|
||
runner_with_real_store: Runner, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""诊断降级占比 > 50% 时 _run_step 中止（疑似 judge 基础设施故障）。"""
|
||
from unittest.mock import AsyncMock
|
||
|
||
batch = [_fake_question("q1", "vA"), _fake_question("q2", "vA")]
|
||
|
||
runner_with_real_store._rollout_batch = AsyncMock()
|
||
monkeypatch.setattr("app.harness.runner._apply_batch_correctness", lambda *a, **k: None)
|
||
runner_with_real_store._run_diagnosis = AsyncMock(
|
||
return_value=DiagnosisResult(run_id="r", degraded_count=2)
|
||
)
|
||
runner_with_real_store._gate_batch_skills = AsyncMock()
|
||
|
||
state = MagicMock()
|
||
state.correctness = {"q1": False, "q2": False}
|
||
state.gate_cooldown = {}
|
||
pools = MagicMock()
|
||
pools.baseline_run_id = "infer_adhoc"
|
||
|
||
with pytest.raises(RuntimeError, match="降级占比"):
|
||
await runner_with_real_store._run_step(1, 0, 10, batch, pools, state)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_run_diagnosis_full_scan_loads_all_video_trees(
|
||
runner_with_real_store: Runner,
|
||
) -> None:
|
||
"""question_ids=None（全量诊断）路径：加载全部 questions 涉及 video 的树。
|
||
|
||
轻量验证：patch load_benchmark 返回两个不同 video 的假题，patch
|
||
load_tree_data_for_videos 捕获 video_ids，断言全量路径覆盖全部 video。
|
||
"""
|
||
fake_qs = [
|
||
_fake_question("q-a", "vA"),
|
||
_fake_question("q-b", "vB"),
|
||
]
|
||
captured: dict[str, object] = {}
|
||
|
||
def _cap(store_dir: Path, video_ids: list[str]) -> dict[str, object]:
|
||
captured["video_ids"] = list(video_ids)
|
||
return {v: {"nodes": {}} for v in video_ids}
|
||
|
||
with (
|
||
patch("app.question_gen.load_benchmark", return_value=fake_qs),
|
||
patch("app.harness.tree_nodes.load_tree_data_for_videos", side_effect=_cap),
|
||
patch(
|
||
"core.evolution.diagnose.run_diagnosis",
|
||
new=AsyncMock(return_value=_empty_diagnosis_result()),
|
||
),
|
||
):
|
||
await runner_with_real_store._run_diagnosis("infer_adhoc", question_ids=None)
|
||
|
||
assert set(captured["video_ids"]) == {"vA", "vB"}
|