From 24ed7ca32225abbd7b89994e237c3cd2e1f7a9b4 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Tue, 14 Jul 2026 15:55:36 -0400 Subject: [PATCH] fix: harden canonical_answer_text and align decision-core signatures to spec --- app/question_gen/adversarial_filter.py | 38 ++++++++++++---------- tests/unit/test_adversarial_filter_core.py | 36 ++++++++++---------- 2 files changed, 39 insertions(+), 35 deletions(-) diff --git a/app/question_gen/adversarial_filter.py b/app/question_gen/adversarial_filter.py index 143b695..334be91 100644 --- a/app/question_gen/adversarial_filter.py +++ b/app/question_gen/adversarial_filter.py @@ -12,10 +12,8 @@ from __future__ import annotations import enum import hashlib import json -from typing import TYPE_CHECKING -if TYPE_CHECKING: - from core.types import GeneratedQuestion +_VALID_LETTERS = ("A", "B", "C", "D") class FlipDecision(enum.Enum): @@ -26,20 +24,22 @@ class FlipDecision(enum.Enum): FLIP_SKIPPED = "flip_skipped" -def question_hash(question: GeneratedQuestion) -> str: +def question_hash(question: str, options: tuple[str, ...], answer: str) -> str: """题 payload（question+options+answer）的稳定 hash，防 JSON 变动误用旧 verdict。 参数: - question: 题目。 + question: 题目文本。 + options: 选项元组。 + answer: 正确答案字母。 返回: 16 位十六进制摘要。 """ payload = json.dumps( { - "question": question.question, - "options": list(question.options), - "answer": question.answer, + "question": question, + "options": list(options), + "answer": answer, }, ensure_ascii=False, sort_keys=True, @@ -53,25 +53,29 @@ def agent_config_fingerprint(*, skill_mode: str, max_steps: int, model: str) -> return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:16] -def canonical_answer_text(question: GeneratedQuestion, letter: str | None) -> str | None: +def canonical_answer_text(options: tuple[str, ...], letter: str | None) -> str | None: """把 agent 预测的选项字母映射为选项规范化文本；非法/越界返回 None。 - 镜像题选项会重洗牌，字母无语义，必须按选项文本比较。 + 镜像题选项会重洗牌，字母无语义，必须按选项文本比较。防御性：空串、多字符、 + 非 A-D 字母、越界一律返回 None，下游据此保守 flip_skipped，绝不崩溃、绝不误杀。 参数: - question: 题目（提供 options）。 - letter: agent 预测字母（大小写不敏感），None/空/越界视为无效。 + options: 选项元组。 + letter: agent 预测字母（大小写不敏感，须为单个 A-D），其余一律无效。 返回: 去掉 "X. " 前缀的选项文本；无效时 None。 """ - if not letter or not isinstance(letter, str): + if not isinstance(letter, str): return None - idx = ord(letter.strip().upper()) - ord("A") - if not 0 <= idx < len(question.options): + s = letter.strip().upper() + if s not in _VALID_LETTERS: return None - opt = question.options[idx] - prefix = f"{letter.strip().upper()}. " + idx = ord(s) - ord("A") + if idx >= len(options): + return None + opt = options[idx] + prefix = f"{s}. " return opt[len(prefix):] if opt.startswith(prefix) else opt diff --git a/tests/unit/test_adversarial_filter_core.py b/tests/unit/test_adversarial_filter_core.py index 0c126ce..7aac00f 100644 --- a/tests/unit/test_adversarial_filter_core.py +++ b/tests/unit/test_adversarial_filter_core.py @@ -7,26 +7,20 @@ from app.question_gen.adversarial_filter import ( 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", - ) +_OPTIONS = ("A. 蒸", "B. 炒", "C. 煮", "D. 炸") def test_question_hash_stable_and_payload_sensitive(): - h1 = question_hash(_q()) - h2 = question_hash(_q()) + h1 = question_hash("?", _OPTIONS, "A") + h2 = question_hash("?", _OPTIONS, "A") assert h1 == h2 - h3 = question_hash(_q(answer="B")) # answer 变 → hash 变 + h3 = question_hash("?", _OPTIONS, "B") # answer 变 → hash 变 assert h1 != h3 - h4 = question_hash(_q(options=("A. 蒸", "B. 炒", "C. 煮", "D. 烤"))) # option 变 → 变 + h4 = question_hash("?", ("A. 蒸", "B. 炒", "C. 煮", "D. 烤"), "A") # option 变 → 变 assert h1 != h4 + h5 = question_hash("!", _OPTIONS, "A") # question 变 → 变 + assert h1 != h5 def test_agent_config_fingerprint_changes_with_inputs(): @@ -37,14 +31,20 @@ def test_agent_config_fingerprint_changes_with_inputs(): def test_canonical_answer_text_maps_letter_to_option_text(): - assert canonical_answer_text(_q(), "C") == "煮" - assert canonical_answer_text(_q(), "c") == "煮" + assert canonical_answer_text(_OPTIONS, "C") == "煮" + assert canonical_answer_text(_OPTIONS, "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 + assert canonical_answer_text(_OPTIONS, "Z") is None + assert canonical_answer_text(_OPTIONS, "") is None + assert canonical_answer_text(_OPTIONS, None) is None + + +def test_canonical_answer_text_invalid_inputs_return_none(): + for bad in ["", "AB", "E", "1", " ", "AA"]: + assert canonical_answer_text(_OPTIONS, bad) is None, bad + assert canonical_answer_text(_OPTIONS, "b") == "炒" # 合法仍工作（去前缀、大小写不敏感） def test_judge_flip_different_answers_passed():