diff --git a/app/question_gen/adversarial_filter.py b/app/question_gen/adversarial_filter.py index 133b725..62f70b1 100644 --- a/app/question_gen/adversarial_filter.py +++ b/app/question_gen/adversarial_filter.py @@ -13,7 +13,7 @@ import enum import hashlib import json from pathlib import Path -from typing import TYPE_CHECKING, Protocol +from typing import TYPE_CHECKING, Protocol, overload from json_repair import repair_json from loguru import logger @@ -38,17 +38,39 @@ class FlipDecision(enum.Enum): FLIP_SKIPPED = "flip_skipped" -def question_hash(question: str, options: tuple[str, ...], answer: str) -> str: +@overload +def question_hash(question: GeneratedQuestion) -> str: ... + + +@overload +def question_hash(question: str, options: tuple[str, ...], answer: str) -> str: ... + + +def question_hash( + question: GeneratedQuestion | str, + options: tuple[str, ...] | None = None, + answer: str | None = None, +) -> str: """题 payload（question+options+answer）的稳定 hash，防 JSON 变动误用旧 verdict。 + 两种等价调用形态：整题 `question_hash(q)` 或散参 `question_hash(题面, 选项, 答案)`； + 前者按题面/选项/答案拆解后走同一路径，保证与散参形态哈希一致。 + 参数: - question: 题目文本。 - options: 选项元组。 - answer: 正确答案字母。 + question: 整条 GeneratedQuestion，或题目文本字符串。 + options: 选项元组（散参形态必传）。 + answer: 正确答案字母（散参形态必传）。 返回: 16 位十六进制摘要。 + + 异常: + TypeError: 传入题面字符串却缺 options / answer（散参形态参数不全）。 """ + if isinstance(question, GeneratedQuestion): + return question_hash(question.question, question.options, question.answer) + if options is None or answer is None: + raise TypeError("散参形态 question_hash 需同时传入 (question, options, answer)") payload = json.dumps( { "question": question, @@ -90,7 +112,7 @@ def canonical_answer_text(options: tuple[str, ...], letter: str | None) -> str | return None opt = options[idx] prefix = f"{s}. " - return opt[len(prefix):] if opt.startswith(prefix) else opt + return opt[len(prefix) :] if opt.startswith(prefix) else opt def judge_flip(*, p_text: str | None, q_text: str | None) -> FlipDecision: @@ -131,8 +153,8 @@ class AgentRunner(Protocol): def _cheat_hash(question: GeneratedQuestion) -> str: - """按散参数签名计算作弊门题面 hash（question/options/answer）。""" - return question_hash(question.question, question.options, question.answer) + """计算作弊门题面 hash（question/options/answer），供两门与终判统一续跑主键。""" + return question_hash(question) def _recover_survivors( @@ -208,15 +230,24 @@ async def run_cheater_gate( correct = pred is not None and pred.strip().upper() == q.answer.strip().upper() verdict = "filtered_too_easy" if correct else "passed" store.record_verdict( - question_id=q.question_id, question_hash=_cheat_hash(q), stage="cheat", - round=round_no, agent_prediction=pred, agent_correct=correct, - verdict=verdict, pair_id=None, agent_config=cfg_fp, + question_id=q.question_id, + question_hash=_cheat_hash(q), + stage="cheat", + round=round_no, + agent_prediction=pred, + agent_correct=correct, + verdict=verdict, + pair_id=None, + agent_config=cfg_fp, ) if not correct: survivors.append(q) logger.info( "作弊门: {} 题（续跑复用 {}，新判 {}）→ 存活 {}", - len(questions), len(completed), len(todo), len(survivors), + len(questions), + len(completed), + len(todo), + len(survivors), ) return survivors @@ -293,9 +324,7 @@ def _existing_frames(frame_paths: list[str]) -> list[str]: return [p for p in frame_paths if Path(p).exists()] -def _build_mirror_question( - question: GeneratedQuestion, mirror: dict -) -> GeneratedQuestion | None: +def _build_mirror_question(question: GeneratedQuestion, mirror: dict) -> GeneratedQuestion | None: """从解析出的 mirror dict 构造镜像题；字段缺失/类型错误 → None。""" try: options = tuple(str(o) for o in mirror["options"]) @@ -371,3 +400,221 @@ async def generate_mirror_question( ) return None return mirror_q + + +def _read_cheat_prediction( + store: QuestionGenStore, q: GeneratedQuestion, cfg_fp: str +) -> str | None: + """从 adversarial_verdicts 读作弊门落库的 P 预测字母（C2：绝不重跑 agent）。 + + 按 (question_id, 当前题面 hash, stage='cheat', 当前 agent_config) 定位那条 + 由 `run_cheater_gate` 写入的预测；无匹配行返回 None。 + """ + row = store._conn.execute( + "SELECT agent_prediction FROM adversarial_verdicts " + "WHERE question_id=? AND question_hash=? AND stage='cheat' AND agent_config=?", + (q.question_id, _cheat_hash(q), cfg_fp), + ).fetchone() + return row[0] if row else None + + +def _persist_flip( + store: QuestionGenStore, + q: GeneratedQuestion, + decision: FlipDecision, + mirror_pred: str | None, + round_no: int, + cfg_fp: str, + pair_id: str, +) -> None: + """写 flip_original + flip_mirror 两条 verdict，并按判定改写原题 cheat 行。 + + flip_original 复用 P 的作弊门预测；flip_mirror 记镜像预测；两者同 pair_id 关联。 + FILTERED_NO_FLIP 时把 cheat 行 verdict 改判 filtered_no_flip（与终判排除双保险）； + passed / flip_skipped 时 cheat 行保持 passed。 + """ + h = _cheat_hash(q) + p_pred = _read_cheat_prediction(store, q, cfg_fp) + store.record_verdict( + question_id=q.question_id, + question_hash=h, + stage="flip_original", + round=round_no, + agent_prediction=p_pred, + agent_correct=None, + verdict=decision.value, + pair_id=pair_id, + agent_config=cfg_fp, + ) + store.record_verdict( + question_id=q.question_id, + question_hash=h, + stage="flip_mirror", + round=round_no, + agent_prediction=mirror_pred, + agent_correct=None, + verdict=decision.value, + pair_id=pair_id, + agent_config=cfg_fp, + ) + if decision is FlipDecision.FILTERED_NO_FLIP: + store.record_verdict( + question_id=q.question_id, + question_hash=h, + stage="cheat", + round=round_no, + agent_prediction=p_pred, + agent_correct=False, + verdict="filtered_no_flip", + pair_id=None, + agent_config=cfg_fp, + ) + + +async def _judge_one_flip( + q: GeneratedQuestion, + flip_axis: str | None, + *, + agent: AgentRunner, + vlm: VLMProvider, + trees: dict[str, TreeIndex], + store: QuestionGenStore, + cfg_fp: str, + config: AdversarialFilterConfig, + run_id: str, + session_id: str, +) -> tuple[FlipDecision, str | None]: + """跑单题翻转判定，返回 (decision, 镜像预测字母)。 + + 原题 P 预测**只从 adversarial_verdicts 表读作弊门落的行**（不重跑 agent）， + 故 `store` 与 `cfg_fp` 必传（C2：按 (question_id, question_hash, stage='cheat', + agent_config) 定位那条预测）。镜像造不出 / 素材缺失 → FLIP_SKIPPED（不跑 agent）。 + """ + tree = trees.get(q.video_id) + if tree is None or flip_axis is None: + return FlipDecision.FLIP_SKIPPED, None + material = _rebuild_material(tree, q.source_nodes) + mirror = await generate_mirror_question( + q, flip_axis=flip_axis, vlm=vlm, material=material, session_id=session_id + ) + if mirror is None: + return FlipDecision.FLIP_SKIPPED, None + preds = await agent.predict( + [mirror], max_steps=config.adversarial_agent_max_steps, run_id=f"{run_id}_mirror" + ) + q_pred = preds.get(mirror.question_id) + p_pred = _read_cheat_prediction(store, q, cfg_fp) # 复用作弊门 P 预测（不重跑） + p_text = canonical_answer_text(q.options, p_pred) + q_text = canonical_answer_text(mirror.options, q_pred) + return judge_flip(p_text=p_text, q_text=q_text), q_pred + + +async def run_flip_gate( + survivors: list[GeneratedQuestion], + *, + agent: AgentRunner, + vlm: VLMProvider, + store: QuestionGenStore, + trees: dict[str, TreeIndex], + config: AdversarialFilterConfig, + round_no: int, + run_id: str, + session_id: str, +) -> list[GeneratedQuestion]: + """翻转门：不支持 flip 的终判 passed；支持的按 canonical 翻转判定。 + + P 预测复用作弊门落表结果（不重跑）；仅新跑镜像 Q。任一无效 / 镜像失败 → + flip_skipped（保留题，只经作弊门，不误杀）。镜像题只用于判定，不进题库。 + + 参数: + survivors: 作弊门存活（agent 答错）的题列表。 + agent: 完整 agent 试答端口（仅对镜像 Q 调用）。 + vlm: 镜像题生成 VLM 端口。 + store: verdict 持久化（含作弊门 P 预测来源）。 + trees: video_id → 三层树索引（重建镜像素材用）。 + config: 过滤配置（提供 max_steps）。 + round_no: 当前轮次（构造 pair_id）。 + run_id: agent 推理 run 标识。 + session_id: VLM 遥测会话 ID。 + + 返回: + 终判 verdict∈{passed, flip_skipped} 的题（filtered_no_flip 被剔除）。 + """ + from app.question_gen.strategy_action_recognition import _AR_PATTERN_BY_NAME + + cfg_fp = agent_config_fingerprint( + skill_mode=agent.skill_mode, + max_steps=config.adversarial_agent_max_steps, + model=agent.model, + ) + kept: list[GeneratedQuestion] = [] + for q in survivors: + sp = _AR_PATTERN_BY_NAME.get(q.sub_pattern or "") + if sp is None or not sp.supports_flip: + kept.append(q) # cheat 已记 passed，无需改写 + continue + decision, mirror_pred = await _judge_one_flip( + q, + sp.flip_axis, + agent=agent, + vlm=vlm, + trees=trees, + store=store, + cfg_fp=cfg_fp, + config=config, + run_id=run_id, + session_id=session_id, + ) + pair_id = f"{q.question_id}::{round_no}" + _persist_flip(store, q, decision, mirror_pred, round_no, cfg_fp, pair_id) + if decision is not FlipDecision.FILTERED_NO_FLIP: + kept.append(q) # passed 或 flip_skipped 都保留 + logger.info("翻转门: {} 存活 → 保留 {}", len(survivors), len(kept)) + return kept + + +def _question_to_record(q: GeneratedQuestion) -> dict: + """把题目序列化为最终题库 JSON 记录（字段与 loader.load_benchmark 读回口径一致）。""" + return { + "question_id": q.question_id, + "video_id": q.video_id, + "task_type": q.task_type, + "question": q.question, + "options": list(q.options), + "answer": q.answer, + "source_nodes": list(q.source_nodes), + "difficulty": q.difficulty, + "family": q.family, + "skill_target": q.skill_target, + "difficulty_steps": q.difficulty_steps, + "sub_pattern": q.sub_pattern, + } + + +def write_final_bank( + out_path: Path, + store: QuestionGenStore, + questions_by_id: dict[str, GeneratedQuestion], + cfg_fp: str, +) -> list[dict]: + """按终判 passed 全量重建最终题库 JSON（镜像题绝不入库）。 + + 终判集合来自 `final_passed_question_ids`（当前 hash+config 下 cheat=passed 且 + 无 filtered_no_flip 行）。镜像题不在 questions_by_id 中，天然被排除。 + + 参数: + out_path: 输出 JSON 路径。 + store: verdict 来源。 + questions_by_id: question_id → 原题（仅 Phase A 产物，不含镜像）。 + cfg_fp: 当前 agent 配置指纹。 + + 返回: + 写入的记录列表（保持 questions_by_id 的插入顺序）。 + """ + hash_by_qid = {qid: _cheat_hash(q) for qid, q in questions_by_id.items()} + passed = store.final_passed_question_ids(hash_by_qid, cfg_fp) + records = [ + _question_to_record(questions_by_id[qid]) for qid in questions_by_id if qid in passed + ] + out_path.write_text(json.dumps(records, ensure_ascii=False, indent=2), encoding="utf-8") + return records diff --git a/tests/unit/test_adversarial_flip_gate.py b/tests/unit/test_adversarial_flip_gate.py new file mode 100644 index 0000000..bb8baaa --- /dev/null +++ b/tests/unit/test_adversarial_flip_gate.py @@ -0,0 +1,242 @@ +"""翻转门四路径：passed / filtered_no_flip / flip_skipped / 镜像不入库。""" + +import json + +import pytest + +from app.question_gen.adversarial_config import AdversarialFilterConfig +from app.question_gen.adversarial_filter import ( + agent_config_fingerprint, + question_hash, + run_flip_gate, + write_final_bank, +) +from app.question_gen.run_store import QuestionGenStore +from core.types import GeneratedQuestion, LLMResponse + + +class _FakeAgent: + """复用 Task 6 语义；带 skill_mode 属性（AgentRunner Protocol 要求）。""" + + def __init__(self, preds, model="m1", skill_mode="auto"): + self._preds = preds + self.model = model + self.skill_mode = skill_mode + self.calls: list[str] = [] + + async def predict(self, questions, *, max_steps, run_id): + self.calls.extend(q.question_id for q in questions) + return {q.question_id: self._preds.get(q.question_id) for q in questions} + + +class _FakeVLM: + def __init__(self, content): + self._content = content + + async def chat_with_images(self, messages, images, *, session_id=None, parent_call_id=None): + return LLMResponse( + content=self._content, + thinking="", + model="fake", + provider="fake", + prompt_tokens=0, + completion_tokens=0, + latency_ms=0, + ttft_ms=None, + max_inter_token_ms=None, + cache_hit=False, + call_id="c", + ) + + +class _FakeMaterial: + subtitle_sentences = ["先炒后蒸"] + frame_paths = ["/f1.jpg"] + + +def _q(qid, sub="temporal_reasoning_failure"): + return GeneratedQuestion( + question_id=qid, + video_id="v1", + task_type="Action Recognition", + question="X 之前做了什么？", + options=("A. 蒸", "B. 炒", "C. 煮", "D. 炸"), + answer="A", + source_nodes=("n1",), + difficulty="hard", + sub_pattern=sub, + ) + + +def _fp(): + return agent_config_fingerprint(skill_mode="auto", max_steps=40, model="m1") + + +@pytest.fixture(autouse=True) +def _stub_material(monkeypatch): + """隔离建树素材重建，直接给镜像生成喂假素材。""" + monkeypatch.setattr( + "app.question_gen.adversarial_filter._rebuild_material", + lambda tree, source_nodes: _FakeMaterial(), + ) + + +def _preset_cheat(store, q, pred="A"): + """预置作弊门 P 预测行（翻转门须复用它，不重跑 agent）。""" + store.record_verdict( + question_id=q.question_id, + question_hash=question_hash(q), + stage="cheat", + round=0, + agent_prediction=pred, + agent_correct=False, + verdict="passed", + pair_id=None, + agent_config=_fp(), + ) + + +@pytest.mark.asyncio +async def test_flip_gate_answer_flips_passed(tmp_path): + store = QuestionGenStore(str(tmp_path / "q.db")) + q = _q("hard") + _preset_cheat(store, q, pred="A") # P canonical="蒸" + agent = _FakeAgent({"hard_mirror": "A"}) # 镜像洗牌后 A=炒 → canonical≠蒸 + vlm = _FakeVLM( + json.dumps( + { + "mirror": { + "question": "X 之后？", + "options": ["A. 炒", "B. 蒸", "C. 煮", "D. 炸"], + "answer": "A", + } + }, + ensure_ascii=False, + ) + ) + kept = await run_flip_gate( + [q], + agent=agent, + vlm=vlm, + store=store, + trees={"v1": object()}, + config=AdversarialFilterConfig(), + round_no=0, + run_id="r0", + session_id="s", + ) + assert {x.question_id for x in kept} == {"hard"} + assert agent.calls == ["hard_mirror"] # C2: 原题 P 未被重跑，只跑镜像 + cheat = store._conn.execute( + "SELECT verdict FROM adversarial_verdicts WHERE question_id='hard' AND stage='cheat'" + ).fetchone()[0] + assert cheat == "passed" + mrow = store._conn.execute( + "SELECT verdict, pair_id FROM adversarial_verdicts WHERE stage='flip_mirror'" + ).fetchone() + assert mrow[0] == "passed" and mrow[1] # 镜像独立行 + pair_id 非空 + store.close() + + +@pytest.mark.asyncio +async def test_flip_gate_same_answer_filtered(tmp_path): + store = QuestionGenStore(str(tmp_path / "q.db")) + q = _q("stick") + _preset_cheat(store, q, pred="A") # P canonical="蒸" + agent = _FakeAgent({"stick_mirror": "A"}) # 镜像 A=蒸 → canonical 与 P 相同 + vlm = _FakeVLM( + json.dumps( + { + "mirror": { + "question": "X 之后？", + "options": ["A. 蒸", "B. 炒", "C. 煮", "D. 炸"], + "answer": "B", + } + }, + ensure_ascii=False, + ) + ) + kept = await run_flip_gate( + [q], + agent=agent, + vlm=vlm, + store=store, + trees={"v1": object()}, + config=AdversarialFilterConfig(), + round_no=0, + run_id="r0", + session_id="s", + ) + assert kept == [] # 未随问题翻转 → 剔除 + cheat = store._conn.execute( + "SELECT verdict FROM adversarial_verdicts WHERE question_id='stick' AND stage='cheat'" + ).fetchone()[0] + assert cheat == "filtered_no_flip" # cheat 行被改写 → final 不含它 + store.close() + + +@pytest.mark.asyncio +async def test_flip_gate_invalid_mirror_skipped_but_kept(tmp_path): + store = QuestionGenStore(str(tmp_path / "q.db")) + q = _q("murky") + _preset_cheat(store, q, pred="A") + agent = _FakeAgent({}) # 镜像造不出 → agent 不该被调用 + vlm = _FakeVLM('{"mirror": null}') + kept = await run_flip_gate( + [q], + agent=agent, + vlm=vlm, + store=store, + trees={"v1": object()}, + config=AdversarialFilterConfig(), + round_no=0, + run_id="r0", + session_id="s", + ) + assert {x.question_id for x in kept} == {"murky"} # 退回只经作弊门，保留不误杀 + assert agent.calls == [] # 镜像 None → 未跑 agent + cheat = store._conn.execute( + "SELECT verdict FROM adversarial_verdicts WHERE question_id='murky' AND stage='cheat'" + ).fetchone()[0] + assert cheat == "passed" + store.close() + + +@pytest.mark.asyncio +async def test_flip_gate_mirror_excluded_and_unsupported_passes(tmp_path): + store = QuestionGenStore(str(tmp_path / "q.db")) + q = _q("hard") # 支持 flip + npq = _q("plain", sub="premature_evidence_anchoring") # 不支持 flip + _preset_cheat(store, q, pred="A") + _preset_cheat(store, npq, pred="B") + agent = _FakeAgent({"hard_mirror": "A"}) + vlm = _FakeVLM( + json.dumps( + { + "mirror": { + "question": "X 之后？", + "options": ["A. 炒", "B. 蒸", "C. 煮", "D. 炸"], + "answer": "A", + } + }, + ensure_ascii=False, + ) + ) + kept = await run_flip_gate( + [q, npq], + agent=agent, + vlm=vlm, + store=store, + trees={"v1": object()}, + config=AdversarialFilterConfig(), + round_no=0, + run_id="r0", + session_id="s", + ) + assert {x.question_id for x in kept} == {"hard", "plain"} # 不支持 flip 直接 passed + assert agent.calls == ["hard_mirror"] # 不支持 flip 的题不跑 agent/VLM + out = tmp_path / "final.json" + write_final_bank(out, store, {"hard": q, "plain": npq}, _fp()) + ids = [d["question_id"] for d in json.loads(out.read_text(encoding="utf-8"))] + assert "hard_mirror" not in ids and set(ids) == {"hard", "plain"} # 镜像不入题库 + store.close()