From ae0a718f673212d034710811d5656474a34e26cf Mon Sep 17 00:00:00 2001 From: iomgaa Date: Tue, 14 Jul 2026 13:45:31 -0400 Subject: [PATCH] feat: thread and persist sub_pattern into accepted questions --- app/question_gen/pipeline_v2.py | 4 + core/types.py | 2 + .../test_generated_question_sub_pattern.py | 73 +++++++++++++++++++ tools/generate_questions.py | 44 +++++------ 4 files changed, 98 insertions(+), 25 deletions(-) create mode 100644 tests/unit/test_generated_question_sub_pattern.py diff --git a/app/question_gen/pipeline_v2.py b/app/question_gen/pipeline_v2.py index 8f95c48..16c1393 100644 --- a/app/question_gen/pipeline_v2.py +++ b/app/question_gen/pipeline_v2.py @@ -251,6 +251,7 @@ def _to_generated_question( family: str, options: tuple[str, ...] | None = None, answer: str | None = None, + sub_pattern: str | None = None, ) -> GeneratedQuestion: """将 CandidateQuestion 转换为 GeneratedQuestion。 @@ -259,6 +260,7 @@ def _to_generated_question( family: 问题家族名称（如 "RETRIEVAL"）。 options: 洗牌后的选项元组（若为 None 则使用 candidate 原始选项）。 answer: 重映射后的答案字母（若为 None 则使用 candidate 原始答案）。 + sub_pattern: 出题子模式标识（AR 特化策略使用，None 表示无）。 返回: GeneratedQuestion 实例（difficulty_steps 初始为 None）。 @@ -275,6 +277,7 @@ def _to_generated_question( family=family, skill_target=candidate.skill_target, difficulty_steps=None, + sub_pattern=sub_pattern, ) @@ -531,6 +534,7 @@ async def _process_one_slot( family=strategy.strategy_name, options=pp.options, answer=pp.answer, + sub_pattern=sub_pattern.name if sub_pattern else None, ) # 将题目 embedding 加入池（flatten 确保 1D） embed_pool.append(embed_fn(candidate.question).flatten()) diff --git a/core/types.py b/core/types.py index b3608b1..48f1ec0 100644 --- a/core/types.py +++ b/core/types.py @@ -48,6 +48,7 @@ class GeneratedQuestion: family: 问题家族名称（如 "RETRIEVAL"，v2 出题管线使用，None 表示未指定）。 skill_target: 目标技能标识（v2 出题管线使用，None 表示未指定）。 difficulty_steps: 推理步数估计（v2 出题管线使用，None 表示未指定）。 + sub_pattern: 出题子模式标识（AR 特化策略使用，None 表示无）。 """ question_id: str @@ -61,6 +62,7 @@ class GeneratedQuestion: family: str | None = field(default=None) skill_target: str | None = field(default=None) difficulty_steps: int | None = field(default=None) + sub_pattern: str | None = field(default=None) @dataclass(frozen=True) diff --git a/tests/unit/test_generated_question_sub_pattern.py b/tests/unit/test_generated_question_sub_pattern.py new file mode 100644 index 0000000..4298c58 --- /dev/null +++ b/tests/unit/test_generated_question_sub_pattern.py @@ -0,0 +1,73 @@ +"""GeneratedQuestion.sub_pattern 字段 + _to_generated_question 透传。""" + +from app.question_gen.generator_v2 import CandidateQuestion +from app.question_gen.pipeline_v2 import _to_generated_question +from core.types import GeneratedQuestion + + +def _candidate() -> CandidateQuestion: + return CandidateQuestion( + question_id="v1_Action Recognition_0001", + video_id="v1", + task_type="Action Recognition", + skill_target="M1_AR", + question="厨师最终采用了哪种烹饪方式？", + options=("A. 蒸", "B. 炒", "C. 煮", "D. 炸"), + answer="A", + source_nodes=("n1", "n2"), + difficulty="hard", + ) + + +def test_generated_question_has_sub_pattern_default_none(): + q = GeneratedQuestion( + question_id="q1", video_id="v1", task_type="Action Recognition", + question="?", options=("A. x",), answer="A", + source_nodes=("n1",), difficulty="easy", + ) + assert q.sub_pattern is None + + +def test_to_generated_question_threads_sub_pattern(): + q = _to_generated_question( + _candidate(), family="ACTION_RECOGNITION", + sub_pattern="premature_evidence_anchoring", + ) + assert q.sub_pattern == "premature_evidence_anchoring" + + +def test_to_generated_question_sub_pattern_defaults_none(): + q = _to_generated_question(_candidate(), family="RETRIEVAL") + assert q.sub_pattern is None + + +def test_question_to_entry_includes_sub_pattern(): + from tools.generate_questions import _question_to_entry + q = GeneratedQuestion( + question_id="v1_Action Recognition_0001", video_id="v1", + task_type="Action Recognition", question="?", + options=("A. 蒸", "B. 炒", "C. 煮", "D. 炸"), answer="A", + source_nodes=("n1",), difficulty="hard", + family="ACTION_RECOGNITION", skill_target="M1_AR", + sub_pattern="temporal_reasoning_failure", + ) + entry = _question_to_entry(q) + assert entry["sub_pattern"] == "temporal_reasoning_failure" + assert entry["question_id"] == "v1_Action Recognition_0001" + assert entry["options"] == ["A. 蒸", "B. 炒", "C. 煮", "D. 炸"] + + +def test_append_to_json_writes_sub_pattern(tmp_path): + from tools.generate_questions import _append_to_json + q = GeneratedQuestion( + question_id="v1_Action Recognition_0001", video_id="v1", + task_type="Action Recognition", question="?", + options=("A. 蒸", "B. 炒", "C. 煮", "D. 炸"), answer="A", + source_nodes=("n1",), difficulty="hard", + family="ACTION_RECOGNITION", skill_target="M1_AR", + sub_pattern="temporal_reasoning_failure", + ) + _append_to_json(tmp_path, q) + import json + data = json.loads((tmp_path / "v1.json").read_text(encoding="utf-8")) + assert data[0]["sub_pattern"] == "temporal_reasoning_failure" diff --git a/tools/generate_questions.py b/tools/generate_questions.py index b739d1e..b1fd760 100644 --- a/tools/generate_questions.py +++ b/tools/generate_questions.py @@ -323,6 +323,23 @@ def _build_embed_provider(): # --------------------------------------------------------------------------- +def _question_to_entry(question: GeneratedQuestion) -> dict: + """将题目序列化为 JSON entry（_append_to_json 与 _on_accept 共用）。""" + return { + "question_id": question.question_id, + "video_id": question.video_id, + "task_type": question.task_type, + "question": question.question, + "options": list(question.options), + "answer": question.answer, + "source_nodes": list(question.source_nodes), + "difficulty": question.difficulty, + "family": question.family, + "skill_target": question.skill_target, + "sub_pattern": question.sub_pattern, + } + + def _append_to_json(output_dir: Path, question: GeneratedQuestion) -> None: """将生成的题目追加到对应 video_id 的 JSON 文件。 @@ -341,17 +358,7 @@ def _append_to_json(output_dir: Path, question: GeneratedQuestion) -> None: logger.warning("读取 {} 失败，覆盖写入", json_path) existing = [] - entry = { - "question_id": question.question_id, - "task_type": question.task_type, - "question": question.question, - "options": list(question.options), - "answer": question.answer, - "source_nodes": list(question.source_nodes), - "difficulty": question.difficulty, - "family": question.family, - "skill_target": question.skill_target, - } + entry = _question_to_entry(question) existing.append(entry) # 原子写入 @@ -1120,20 +1127,7 @@ async def _run_generate_v2(args: argparse.Namespace) -> None: except (json.JSONDecodeError, OSError): existing = [] - existing.append( - { - "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, - } - ) + existing.append(_question_to_entry(q)) tmp_path = output_path.with_suffix(".tmp") tmp_path.write_text(