From c109f2257ad2bf42ad8e6604a86c4b04f002cf5a Mon Sep 17 00:00:00 2001 From: iomgaa Date: Tue, 14 Jul 2026 15:51:28 -0400 Subject: [PATCH] feat: add pure decision core for adversarial filter (hash/fingerprint/canonical/flip) --- app/question_gen/adversarial_filter.py | 92 ++++++++++++++++++++++ tests/unit/test_adversarial_filter_core.py | 63 +++++++++++++++ 2 files changed, 155 insertions(+) create mode 100644 app/question_gen/adversarial_filter.py create mode 100644 tests/unit/test_adversarial_filter_core.py diff --git a/app/question_gen/adversarial_filter.py b/app/question_gen/adversarial_filter.py new file mode 100644 index 0000000..143b695 --- /dev/null +++ b/app/question_gen/adversarial_filter.py @@ -0,0 +1,92 @@ +"""Phase B 独立后置对抗过滤层 — 作弊者门 + 配对翻转门。 + +在 Phase A 产物 accepted_questions.json 之上，用完整 inference agent 揪残余 +shortcut：作弊门（agent 秒杀=太简单，剔除）+ 翻转门（agent 答案须随问题翻转）。 +不改 Phase A 状态机；过滤进度存独立 adversarial_verdicts 表。 + +设计: research-wiki/designs/2026-07-14-adversarial-question-gen-phaseB-design.md +""" + +from __future__ import annotations + +import enum +import hashlib +import json +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from core.types import GeneratedQuestion + + +class FlipDecision(enum.Enum): + """翻转门判定结果。""" + + PASSED = "passed" + FILTERED_NO_FLIP = "filtered_no_flip" + FLIP_SKIPPED = "flip_skipped" + + +def question_hash(question: GeneratedQuestion) -> str: + """题 payload（question+options+answer）的稳定 hash，防 JSON 变动误用旧 verdict。 + + 参数: + question: 题目。 + + 返回: + 16 位十六进制摘要。 + """ + payload = json.dumps( + { + "question": question.question, + "options": list(question.options), + "answer": question.answer, + }, + ensure_ascii=False, + sort_keys=True, + ) + return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:16] + + +def agent_config_fingerprint(*, skill_mode: str, max_steps: int, model: str) -> str: + """agent 配置指纹（skill_mode/max_steps/model），变化则该题 verdict 作废。""" + raw = f"{skill_mode}|{max_steps}|{model}" + return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:16] + + +def canonical_answer_text(question: GeneratedQuestion, letter: str | None) -> str | None: + """把 agent 预测的选项字母映射为选项规范化文本；非法/越界返回 None。 + + 镜像题选项会重洗牌，字母无语义，必须按选项文本比较。 + + 参数: + question: 题目（提供 options）。 + letter: agent 预测字母（大小写不敏感），None/空/越界视为无效。 + + 返回: + 去掉 "X. " 前缀的选项文本；无效时 None。 + """ + if not letter or not isinstance(letter, str): + return None + idx = ord(letter.strip().upper()) - ord("A") + if not 0 <= idx < len(question.options): + return None + opt = question.options[idx] + prefix = f"{letter.strip().upper()}. " + return opt[len(prefix):] if opt.startswith(prefix) else opt + + +def judge_flip(*, p_text: str | None, q_text: str | None) -> FlipDecision: + """按 canonical 文本判翻转：任一无效→skipped；不同→passed；相同→filtered。 + + 参数: + p_text: 原题 P 的 agent 所选 canonical 文本。 + q_text: 镜像题 Q 的 agent 所选 canonical 文本。 + + 返回: + FlipDecision。 + """ + if p_text is None or q_text is None: + return FlipDecision.FLIP_SKIPPED + if p_text.strip() != q_text.strip(): + return FlipDecision.PASSED + return FlipDecision.FILTERED_NO_FLIP diff --git a/tests/unit/test_adversarial_filter_core.py b/tests/unit/test_adversarial_filter_core.py new file mode 100644 index 0000000..0c126ce --- /dev/null +++ b/tests/unit/test_adversarial_filter_core.py @@ -0,0 +1,63 @@ +"""adversarial_filter 纯判定：hash / 指纹 / canonical / 翻转判定。""" + +from app.question_gen.adversarial_filter import ( + FlipDecision, + agent_config_fingerprint, + canonical_answer_text, + judge_flip, + question_hash, +) +from core.types import GeneratedQuestion + + +def _q(qid="q1", options=("A. 蒸", "B. 炒", "C. 煮", "D. 炸"), answer="A"): + return GeneratedQuestion( + question_id=qid, video_id="v1", task_type="Action Recognition", + question="?", options=options, answer=answer, + source_nodes=("n1",), difficulty="hard", + sub_pattern="temporal_reasoning_failure", + ) + + +def test_question_hash_stable_and_payload_sensitive(): + h1 = question_hash(_q()) + h2 = question_hash(_q()) + assert h1 == h2 + h3 = question_hash(_q(answer="B")) # answer 变 → hash 变 + assert h1 != h3 + h4 = question_hash(_q(options=("A. 蒸", "B. 炒", "C. 煮", "D. 烤"))) # option 变 → 变 + assert h1 != h4 + + +def test_agent_config_fingerprint_changes_with_inputs(): + a = agent_config_fingerprint(skill_mode="auto", max_steps=40, model="m1") + b = agent_config_fingerprint(skill_mode="auto", max_steps=41, model="m1") + c = agent_config_fingerprint(skill_mode="manual", max_steps=40, model="m1") + assert a != b and a != c + + +def test_canonical_answer_text_maps_letter_to_option_text(): + assert canonical_answer_text(_q(), "C") == "煮" + assert canonical_answer_text(_q(), "c") == "煮" + + +def test_canonical_answer_text_invalid_returns_none(): + assert canonical_answer_text(_q(), "Z") is None + assert canonical_answer_text(_q(), "") is None + assert canonical_answer_text(_q(), None) is None + + +def test_judge_flip_different_answers_passed(): + # P 选"蒸"，Q（镜像）选"炒"→ 语义不同 → passed + d = judge_flip(p_text="蒸", q_text="炒") + assert d is FlipDecision.PASSED + + +def test_judge_flip_same_answer_filtered(): + d = judge_flip(p_text="蒸", q_text="蒸") + assert d is FlipDecision.FILTERED_NO_FLIP + + +def test_judge_flip_invalid_answer_skipped(): + assert judge_flip(p_text=None, q_text="炒") is FlipDecision.FLIP_SKIPPED + assert judge_flip(p_text="蒸", q_text=None) is FlipDecision.FLIP_SKIPPED