feat: add pure decision core for adversarial filter (hash/fingerprint/canonical/flip)

This commit is contained in:
2026-07-14 15:51:28 -04:00
parent 334fbbc94d
commit c109f2257a
2 changed files with 155 additions and 0 deletions
+92
View File
@@ -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:
"""题 payloadquestion+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
@@ -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