From f36eb66c186d142c864b09ae21edc10d0cf0e4ed Mon Sep 17 00:00:00 2001 From: iomgaa Date: Tue, 14 Jul 2026 17:07:55 -0400 Subject: [PATCH] fix: tolerate backfill under-delivery, fix predict None-fill, drop dead session_id --- app/question_gen/adversarial_filter.py | 23 ++++++++++++---- .../test_adversarial_filter_e2e.py | 24 +++++++++++++++++ tests/unit/test_adversarial_iteration.py | 26 +++++++++++++++++++ tools/generate_questions.py | 1 - 4 files changed, 68 insertions(+), 6 deletions(-) diff --git a/app/question_gen/adversarial_filter.py b/app/question_gen/adversarial_filter.py index a7739d6..fd919ce 100644 --- a/app/question_gen/adversarial_filter.py +++ b/app/question_gen/adversarial_filter.py @@ -756,9 +756,18 @@ async def run_adversarial_rounds( if deficit <= 0 or round_no + 1 >= config.adversarial_max_rounds: break new_qs = await backfill(deficit, round_no, all_questions) - assert len(new_qs) == deficit, ( - f"backfill 应产出 {deficit} 题，实得 {len(new_qs)}（缺额驱动契约）" + # 真实 backfill 经 run_pipeline_v2 只返回过门的题：部分题被 gate 拒 = 常态欠产。 + # 容忍 len(new_qs) <= deficit——欠产先并入，deficit 下轮重算继续补，max_rounds 兜底 + # 防死循环。多产（> deficit）才是契约被破坏的 bug，保留 sanity 断言。 + assert len(new_qs) <= deficit, ( + f"backfill 多产: 需 {deficit} 实得 {len(new_qs)}（超额 = 契约被破坏）" ) + if len(new_qs) < deficit: + logger.warning( + "backfill 欠产: 需 {} 实得 {}，本轮先并入，后续轮次继续补", + deficit, + len(new_qs), + ) for q in new_qs: all_questions[q.question_id] = q pending = new_qs # 只对新补的题重新过滤（已判题走续跑） @@ -843,7 +852,12 @@ class _RealAgentRunner: rows = await RunLogImpl(self._db_path).get_predictions( run_id, question_ids=[q.question_id for q in questions] ) - return {r["question_id"]: r["prediction"] for r in rows} + # 契约：每个入参 qid 都必须有键。先全填 None，再用回读行覆盖——回读缺行 + # （agent 未落库/异常）的 qid 保持 None，不静默丢键。 + preds: dict[str, str | None] = {q.question_id: None for q in questions} + for r in rows: + preds[r["question_id"]] = r["prediction"] + return preds # --------------------------------------------------------------------------- @@ -873,7 +887,6 @@ def build_backfill( store: QuestionGenStore, pipeline_config: PipelineConfig, filter_task_types: tuple[str, ...], - session_id: str, ) -> BackfillFn: """组装真实 backfill 回调 — 缺额驱动 run_pipeline_v2，返回新增 AR 题。 @@ -883,6 +896,7 @@ def build_backfill( - initial_embed_pool = 已有题 embedding（跨 run 去重）； - seq_offset = 已有题最大 seq（新题续编，防撞 question_id）。 仅补 filter_task_types（当前锁定为 AR 单类），返回 PipelineResult.accepted。 + session/run_id 由 run_pipeline_v2 内部自生成，故本层不接受 session_id。 参数: trees: video_id → 三层树索引（生成素材来源）。 @@ -892,7 +906,6 @@ def build_backfill( store: 出题持久化。 pipeline_config: ar30 原配置（除 per_type 外全部继承）。 filter_task_types: 补生成的题型（仅这些）。 - session_id: 遥测会话 ID（透传给管线子调用）。 返回: BackfillFn 闭包。 diff --git a/tests/integration/test_adversarial_filter_e2e.py b/tests/integration/test_adversarial_filter_e2e.py index 93ad4ca..751f301 100644 --- a/tests/integration/test_adversarial_filter_e2e.py +++ b/tests/integration/test_adversarial_filter_e2e.py @@ -275,3 +275,27 @@ async def test_real_agent_runner_predict_roundtrips_predictions(tmp_path: Path) "smoke_r0", question_ids=["smoke"] ) assert rows and rows[0]["prediction"] == "B" + + +@pytest.mark.asyncio +async def test_real_agent_runner_predict_fills_none_for_missing( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """predict 对 predictions 表缺失的 qid 返回 None（契约：每个入参 qid 都有键）。""" + + async def _noop_run_inference(*args: Any, **kwargs: Any) -> None: + """空跑：不写 predictions 表，模拟回读缺行。""" + return None + + monkeypatch.setattr("app.harness.inference.run_inference", _noop_run_inference) + runner = _RealAgentRunner( + llm=_MockLLM(answer="B"), + tool_dispatch_fn=lambda *a, **k: None, + prompt_builder=lambda q: ("s", "u"), + db_path=str(tmp_path / "harness.db"), + concurrency=1, + skill_mode="none", + model="mock", + ) + preds = await runner.predict([_smoke_q("missing")], max_steps=1, run_id="r_none") + assert preds == {"missing": None} # 缺回读行 → 该 qid 键存在且为 None diff --git a/tests/unit/test_adversarial_iteration.py b/tests/unit/test_adversarial_iteration.py index 443e64a..fd2815b 100644 --- a/tests/unit/test_adversarial_iteration.py +++ b/tests/unit/test_adversarial_iteration.py @@ -163,3 +163,29 @@ async def test_rounds_backfill_then_stop_at_max(tmp_path): ids = {d["question_id"] for d in json.loads(final_path.read_text(encoding="utf-8"))} assert "q0" in ids and any(x.startswith("bf0_") for x in ids) store.close() + + +@pytest.mark.asyncio +async def test_rounds_tolerate_backfill_under_delivery(tmp_path): + """backfill 欠产（实得 < deficit，被 gate 拒常态）→ 不崩，并入实得题，达上限停。""" + store = QuestionGenStore(str(tmp_path / "q.db")) + agent = _FakeAgent(pred="B") + # 每轮只产出 deficit-1 道（模拟部分题被 gate 拒的真实欠产） + backfill = _CountingBackfill(lambda d, r: [_q(f"bf{r}_{i}") for i in range(max(d - 1, 0))]) + final_path = tmp_path / "accepted_questions_final.json" + await run_adversarial_rounds( + [_q("q0")], + agent=agent, + vlm=object(), + store=store, + trees={}, + config=AdversarialFilterConfig(adversarial_max_rounds=3), + final_path=final_path, + target=99, # 永远达不到 → 靠 max_rounds 终止，不因欠产崩溃或死循环 + backfill=backfill, + session_id="s", + ) + assert backfill.calls == 2 # round0/round1 各补一次；round2 达上限 break + ids = {d["question_id"] for d in json.loads(final_path.read_text(encoding="utf-8"))} + assert "q0" in ids and any(x.startswith("bf0_") for x in ids) # 实得欠产题已并入 + store.close() diff --git a/tools/generate_questions.py b/tools/generate_questions.py index 8796a23..91e9976 100644 --- a/tools/generate_questions.py +++ b/tools/generate_questions.py @@ -1318,7 +1318,6 @@ async def _run_adversarial_filter(args: argparse.Namespace) -> None: store=store, pipeline_config=pipeline_config, filter_task_types=filter_config.filter_task_types, - session_id=args.session_id, ) # Phase 6: 运行对抗过滤