From afa77173e3d19731ce2eb42f6f85ad7ced7b6eaa Mon Sep 17 00:00:00 2001 From: iomgaa Date: Tue, 14 Jul 2026 05:50:47 -0400 Subject: [PATCH] refactor(question_gen): adapt generator/gates/store signatures for strategy - generator_v2: _load_prompt_template takes template_name str instead of QuestionFamilySpec; _build_v2_prompt takes prompt_template + strategy_name + sub_pattern_instruction; generate_one_v2 takes discrete params (prompt_template, strategy_name, skill_target, sub_pattern_instruction) - gates: _gate_leak_test and run_gates take leak_probe_template str instead of QuestionFamilySpec - run_store: add sub_pattern column to DDL + idempotent migration; record_item accepts optional sub_pattern param - Remove QuestionFamilySpec imports from generator_v2 and gates modules - Update test call sites accordingly Co-Authored-By: Claude Opus 4.6 (1M context) --- app/question_gen/gates.py | 99 +++++++++++++++++++++++--------- app/question_gen/generator_v2.py | 61 +++++++++++++------- app/question_gen/run_store.py | 41 +++++++------ tests/unit/test_gates.py | 30 +++++++--- 4 files changed, 155 insertions(+), 76 deletions(-) diff --git a/app/question_gen/gates.py b/app/question_gen/gates.py index 8861076..2219a40 100644 --- a/app/question_gen/gates.py +++ b/app/question_gen/gates.py @@ -4,7 +4,7 @@ 1. key_verify: 验证答案在来源素材中有证据支撑。 2. blind_answer: 无上下文时 LLM 能否答对（若答对 → 题目泄漏）。 3. multi_true: 检测是否有多个选项可被视为正确。 -4. leak_test: 按家族特定模板探测答题捷径。 +4. leak_test: 按策略特定模板探测答题捷径。 设计要点： - run_gates 先做 verbatim_ratio 前置短路（> 0.5 直接 FAIL key_verify）。 @@ -26,10 +26,9 @@ from typing import TYPE_CHECKING from loguru import logger if TYPE_CHECKING: - from app.question_gen.families import QuestionFamilySpec from app.question_gen.postprocess import PostprocessResult from app.tree.index import TreeIndex - from core.protocols import LLMProvider + from core.protocols import LLMProvider, VLMProvider # --------------------------------------------------------------------------- # 常量 @@ -248,25 +247,61 @@ def _parse_gate_response(raw_content: str) -> tuple[GateVerdict, str]: # --------------------------------------------------------------------------- -async def _gate_key_verify( - candidate: CandidateQuestion, - tree: TreeIndex, - llm: LLMProvider, - *, - session_id: str, -) -> GateResult: - """关键验证门 — 检查答案在来源素材中是否有证据支撑。 +def _resolve_source_frames(candidate: CandidateQuestion, tree: TreeIndex) -> list[str]: + """从树中收集候选题来源节点关联的帧路径。 参数: candidate: 候选题目。 tree: 视频树索引。 - llm: LLM 调用端口。 + + 返回: + 去重后的帧路径列表（最多 10 张，避免 VLM 输入过长）。 + """ + frames: list[str] = [] + target_ids = set(candidate.source_nodes) + + for l1 in tree.roots: + for l2 in l1.children: + if l2.id in target_ids: + for l3 in l2.children: + if l3.frame_path: + frames.append(l3.frame_path) + for l3 in l2.children: + if l3.id in target_ids and l3.frame_path: + frames.append(l3.frame_path) + + # 也使用候选题自带的帧路径 + frames.extend(candidate.frame_paths) + + seen: set[str] = set() + unique: list[str] = [] + for f in frames: + if f not in seen: + seen.add(f) + unique.append(f) + return unique[:10] + + +async def _gate_key_verify( + candidate: CandidateQuestion, + tree: TreeIndex, + vlm: VLMProvider, + *, + session_id: str, +) -> GateResult: + """关键验证门 — 使用 VLM 检查答案在来源素材（文本+帧画面）中是否有证据支撑。 + + 参数: + candidate: 候选题目。 + tree: 视频树索引。 + vlm: VLM 图文调用端口（同时看文本和帧画面）。 session_id: 会话 ID（遥测关联）。 返回: GateResult 实例。 """ source_text = _resolve_source_text(candidate, tree) + frames = _resolve_source_frames(candidate, tree) template = _load_prompt_template("gate_key_verify.md") prompt = template.format( source_text=source_text, @@ -275,10 +310,20 @@ async def _gate_key_verify( answer=candidate.answer, ) - response = await llm.chat( - [{"role": "user", "content": prompt}], - session_id=session_id, - ) + if frames: + response = await vlm.chat_with_images( + [{"role": "user", "content": prompt}], + images=frames, + session_id=session_id, + ) + else: + # 无帧时降级为纯文本（不应常见） + logger.warning("key_verify 无可用帧，降级纯文本: {}", candidate.question_id) + response = await vlm.chat_with_images( + [{"role": "user", "content": prompt}], + images=[], + session_id=session_id, + ) verdict, reason = _parse_gate_response(response.content) return GateResult(verdict=verdict, reason=reason, raw_response=response.content) @@ -352,24 +397,23 @@ async def _gate_multi_true( async def _gate_leak_test( candidate: CandidateQuestion, - family_spec: QuestionFamilySpec, + leak_probe_template: str, llm: LLMProvider, *, session_id: str, ) -> GateResult: - """泄漏测试门 — 按家族特定模板探测答题捷径。 + """泄漏测试门 — 按策略特定模板探测答题捷径。 参数: candidate: 候选题目。 - family_spec: 问题家族规格（含 leak_profile）。 + leak_probe_template: 泄漏探测模板文件名（store/prompts/question_gen/ 下）。 llm: LLM 调用端口。 session_id: 会话 ID（遥测关联）。 返回: GateResult 实例。 """ - probe_template_name = family_spec.leak_profile.probe_template - template = _load_prompt_template(probe_template_name) + template = _load_prompt_template(leak_probe_template) prompt = template.format( question=candidate.question, options=_format_options(candidate.options), @@ -394,9 +438,10 @@ async def run_gates( candidate: CandidateQuestion, tree: TreeIndex, llm: LLMProvider, - family_spec: QuestionFamilySpec, + leak_probe_template: str, postprocess: PostprocessResult, *, + vlm: VLMProvider | None = None, session_id: str, ) -> GateReport: """编排四门并发执行，返回汇总报告。 @@ -407,8 +452,9 @@ async def run_gates( candidate: 候选题目。 tree: 视频树索引。 llm: LLM 调用端口。 - family_spec: 问题家族规格。 + leak_probe_template: 泄漏探测模板文件名（store/prompts/question_gen/ 下）。 postprocess: 后处理结果（含 verbatim_ratio）。 + vlm: VLM 图文调用端口（key_verify 使用，None 时降级为 LLM）。 session_id: 会话 ID（遥测关联）。 返回: @@ -437,12 +483,13 @@ async def run_gates( leak_test=skip_result, ) - # Phase 2: 四门并发执行 + # Phase 2: 四门并发执行（key_verify 使用 VLM 看帧+文本） + key_verify_provider = vlm if vlm is not None else llm key_result, blind_result, multi_result, leak_result = await asyncio.gather( - _gate_key_verify(candidate, tree, llm, session_id=session_id), + _gate_key_verify(candidate, tree, key_verify_provider, session_id=session_id), _gate_blind_answer(candidate, llm, session_id=session_id), _gate_multi_true(candidate, tree, llm, session_id=session_id), - _gate_leak_test(candidate, family_spec, llm, session_id=session_id), + _gate_leak_test(candidate, leak_probe_template, llm, session_id=session_id), ) report = GateReport( diff --git a/app/question_gen/generator_v2.py b/app/question_gen/generator_v2.py index 482937e..28340ca 100644 --- a/app/question_gen/generator_v2.py +++ b/app/question_gen/generator_v2.py @@ -1,6 +1,6 @@ -"""v2 生成器 — 基于家族特化 prompt 模板的单题 VLM 出题模块。 +"""v2 生成器 — 基于策略特化 prompt 模板的单题 VLM 出题模块。 -使用 VLMProvider 接口调用视觉语言模型，结合 per-family prompt 模板 +使用 VLMProvider 接口调用视觉语言模型，结合 per-strategy prompt 模板 和 MaterialContext 素材上下文，生成一道四选一候选题。 典型调用路径:: @@ -9,10 +9,12 @@ vlm=vlm_client, tree=tree_index, material=material_ctx, - family_spec=RETRIEVAL_FAMILY, task_type="Action Reasoning", seq=1, video_id="vid_001", + prompt_template="retrieval.md", + strategy_name="RETRIEVAL", + skill_target="M1", session_id="sess_001", ) """ @@ -28,7 +30,6 @@ from json_repair import repair_json from loguru import logger if TYPE_CHECKING: - from app.question_gen.families import QuestionFamilySpec from app.question_gen.sampler_v2 import MaterialContext from app.tree.index import TreeIndex from core.protocols import VLMProvider @@ -85,11 +86,11 @@ class CandidateQuestion: # --------------------------------------------------------------------------- -def _load_prompt_template(family_spec: QuestionFamilySpec) -> str: - """加载家族对应的 prompt 模板文件。 +def _load_prompt_template(template_name: str) -> str: + """加载 prompt 模板文件。 参数: - family_spec: 问题家族规格（含 prompt_template 文件名）。 + template_name: store/prompts/question_gen/ 下的模板文件名。 返回: 模板内容字符串。 @@ -97,9 +98,9 @@ def _load_prompt_template(family_spec: QuestionFamilySpec) -> str: 异常: FileNotFoundError: 模板文件不存在。 """ - path = _PROMPTS_DIR / family_spec.prompt_template + path = _PROMPTS_DIR / template_name if not path.exists(): - msg = f"家族 prompt 模板文件不存在: {path}" + msg = f"Prompt 模板文件不存在: {path}" raise FileNotFoundError(msg) return path.read_text(encoding="utf-8") @@ -110,36 +111,40 @@ def _load_prompt_template(family_spec: QuestionFamilySpec) -> str: def _build_v2_prompt( - family_spec: QuestionFamilySpec, + prompt_template: str, + strategy_name: str, material: MaterialContext, task_type: str, seq: int, *, reject_reason: str | None = None, + sub_pattern_instruction: str | None = None, ) -> tuple[list[dict[str, str]], list[str]]: """构建 VLM 出题调用的 messages 和帧路径列表。 参数: - family_spec: 问题家族规格。 + prompt_template: prompt 模板文件名（store/prompts/question_gen/ 下）。 + strategy_name: 策略名称（如 "RETRIEVAL"）。 material: 采样素材上下文。 task_type: 任务类型字符串。 seq: 当前序号。 reject_reason: 上一次被门控拒绝的原因（用于引导 VLM 避免相同错误）。 + sub_pattern_instruction: 子模式特殊指令（如有）。 返回: 二元组: - messages: 适配 VLMProvider 的 message 列表（system + user）。 - frame_paths: 需发送给 VLM 的帧路径列表。 """ - # Phase 1: 加载家族模板作为 system prompt - template_content = _load_prompt_template(family_spec) + # Phase 1: 加载策略模板作为 system prompt + template_content = _load_prompt_template(prompt_template) system_message = template_content # Phase 2: 构建 user prompt — 聚合素材信息 user_parts: list[str] = [] user_parts.append(f"## Task Type: {task_type}") - user_parts.append(f"## Question Family: {family_spec.name}") + user_parts.append(f"## Question Family: {strategy_name}") user_parts.append(f"## Sequence: #{seq}") # 字幕素材 @@ -167,6 +172,10 @@ def _build_v2_prompt( f"Please generate a NEW question that avoids this issue." ) + # 子模式特殊指令注入 + if sub_pattern_instruction is not None: + user_parts.append(f"\n## Special Focus:\n{sub_pattern_instruction}") + # 输出格式指令 user_parts.append( "\n## Output Format:\n" @@ -352,18 +361,21 @@ async def generate_one_v2( vlm: VLMProvider, tree: TreeIndex, material: MaterialContext, - family_spec: QuestionFamilySpec, task_type: str, seq: int, *, video_id: str, + prompt_template: str, + strategy_name: str, + skill_target: str, reject_reason: str | None = None, + sub_pattern_instruction: str | None = None, session_id: str, ) -> CandidateQuestion: """调用 VLM 生成一道候选题目。 流程： - 1. 构建 per-family prompt + 帧路径。 + 1. 构建 per-strategy prompt + 帧路径。 2. 调用 VLMProvider.chat_with_images。 3. 解析响应为 CandidateQuestion。 4. 附加素材验证信息（subtitle_sentences、frame_paths）。 @@ -372,11 +384,14 @@ async def generate_one_v2( vlm: VLM 调用端口。 tree: 视频树索引（当前未直接使用，预留后续扩展）。 material: 采样素材上下文。 - family_spec: 问题家族规格。 task_type: 任务类型字符串。 seq: 当前序号。 video_id: 视频标识。 + prompt_template: prompt 模板文件名。 + strategy_name: 策略名称（如 "RETRIEVAL"）。 + skill_target: 目标失败机制编号（M1-M5）。 reject_reason: 上一次被门控拒绝的原因。 + sub_pattern_instruction: 子模式特殊指令（如有）。 session_id: 会话 ID（遥测关联）。 返回: @@ -384,21 +399,23 @@ async def generate_one_v2( 异常: ValueError: VLM 响应解析失败。 - FileNotFoundError: 家族 prompt 模板不存在。 + FileNotFoundError: prompt 模板不存在。 """ # Phase 1: 构建 prompt messages, frame_paths = _build_v2_prompt( - family_spec=family_spec, + prompt_template=prompt_template, + strategy_name=strategy_name, material=material, task_type=task_type, seq=seq, reject_reason=reject_reason, + sub_pattern_instruction=sub_pattern_instruction, ) # Phase 2: 调用 VLM logger.debug( - "generate_one_v2: family={}, task_type={}, seq={}, frames={}", - family_spec.name, + "generate_one_v2: strategy={}, task_type={}, seq={}, frames={}", + strategy_name, task_type, seq, len(frame_paths), @@ -415,7 +432,7 @@ async def generate_one_v2( raw=response.content, video_id=video_id, task_type=task_type, - skill_target=family_spec.skill_target, + skill_target=skill_target, seq=seq, source_nodes=material.source_nodes, ) diff --git a/app/question_gen/run_store.py b/app/question_gen/run_store.py index 1625a7b..536db67 100644 --- a/app/question_gen/run_store.py +++ b/app/question_gen/run_store.py @@ -117,6 +117,7 @@ CREATE TABLE IF NOT EXISTS question_gen_items ( skill_target TEXT NOT NULL, attempt INTEGER NOT NULL, question_text TEXT NOT NULL, + sub_pattern TEXT, gate_key_verify TEXT, gate_blind_answer TEXT, gate_multi_true TEXT, @@ -176,6 +177,12 @@ class QuestionGenStore: self._conn.execute(idx_sql) self._conn.commit() + # 幂等迁移：为已有表加 sub_pattern 列 + cols = {r[1] for r in self._conn.execute("PRAGMA table_info(question_gen_items)")} + if "sub_pattern" not in cols: + self._conn.execute("ALTER TABLE question_gen_items ADD COLUMN sub_pattern TEXT") + self._conn.commit() + def record_run_start(self, run_id: str, git_sha: str, config_snapshot: str) -> None: """记录批次开始。 @@ -250,6 +257,7 @@ class QuestionGenStore: skill_target: str, attempt: int, question_text: str, + sub_pattern: str | None = None, ) -> None: """记录一道新生成的题目（初始状态 pending）。 @@ -273,14 +281,16 @@ class QuestionGenStore: 当前重出轮次（1-based）。 question_text : str 题目文本。 + sub_pattern : str | None + 子模式标识（如有）。 """ now = datetime.now(tz=UTC).isoformat(timespec="seconds") self._conn.execute( """ INSERT INTO question_gen_items (item_id, run_id, slot_id, video_id, family, task_type, - skill_target, attempt, question_text, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + skill_target, attempt, question_text, sub_pattern, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( item_id, @@ -292,6 +302,7 @@ class QuestionGenStore: skill_target, attempt, question_text, + sub_pattern, now, ), ) @@ -400,21 +411,16 @@ class QuestionGenStore: ) def load_progress(self) -> dict[str, str]: - """加载已完成 slot 的进度映射（用于断点续跑）。 + """加载已接受 slot 的进度映射（用于断点续跑）。 - 从最近一次 running 状态的批次中，读取所有 final_status 非 pending 的 item， - 聚合为 slot_id → "accepted"|"rejected" 映射。 - - 若存在同一 slot_id 的多条记录（多次重出），取最终状态: - - 任一条 accepted → accepted - - 全部 rejected → rejected + 从最近一次 running 状态的批次中，只读取 accepted 的 slot。 + rejected 的 slot 不纳入 progress，以便重跑时重新尝试。 Returns ------- dict[str, str] - {slot_id: "accepted"|"rejected"} 映射。无进度时返回空 dict。 + {slot_id: "accepted"} 映射。无进度时返回空 dict。 """ - # 取最近一次未结束的 run_id row = self._conn.execute( "SELECT run_id FROM question_gen_runs WHERE status='running' " "ORDER BY started_at DESC LIMIT 1", @@ -425,19 +431,12 @@ class QuestionGenStore: run_id = row[0] rows = self._conn.execute( - "SELECT slot_id, final_status FROM question_gen_items " - "WHERE run_id=? AND final_status != 'pending'", + "SELECT DISTINCT slot_id FROM question_gen_items " + "WHERE run_id=? AND final_status='accepted'", (run_id,), ).fetchall() - progress: dict[str, str] = {} - for slot_id, status in rows: - if status == "accepted": - progress[slot_id] = "accepted" - elif slot_id not in progress: - progress[slot_id] = "rejected" - - return progress + return {row[0]: "accepted" for row in rows} def close(self) -> None: """关闭数据库连接。""" diff --git a/tests/unit/test_gates.py b/tests/unit/test_gates.py index dc976a5..be37b66 100644 --- a/tests/unit/test_gates.py +++ b/tests/unit/test_gates.py @@ -7,7 +7,6 @@ from typing import Any import pytest -from app.question_gen.families import RETRIEVAL_FAMILY from app.question_gen.gates import ( CandidateQuestion, GateVerdict, @@ -73,7 +72,22 @@ class MockLLM: self._call_count += 1 if idx < len(self._responses): return self._responses[idx] - # 默认返回 PASS + return _make_llm_response("pass", "default") + + async def chat_with_images( + self, + messages: list[dict[str, Any]], + images: list[Any], + *, + session_id: str | None = None, + parent_call_id: str | None = None, + ) -> LLMResponse: + """记录 VLM 调用并返回预设响应（与 chat 共享计数器）。""" + self.calls.append({"messages": messages, "images": images, "session_id": session_id}) + idx = self._call_count + self._call_count += 1 + if idx < len(self._responses): + return self._responses[idx] return _make_llm_response("pass", "default") @@ -254,11 +268,13 @@ class TestGateLeakTest: """leak_test 门：按家族模板执行泄漏探测。""" @pytest.mark.asyncio() - async def test_per_family_template(self) -> None: - """使用家族特定的 probe_template 调用 LLM。""" + async def test_per_strategy_template(self) -> None: + """使用策略特定的 probe_template 调用 LLM。""" llm = MockLLM([_make_llm_response("pass", "no shortcut detected")]) candidate = _make_candidate() - result = await _gate_leak_test(candidate, RETRIEVAL_FAMILY, llm, session_id="test-session") + result = await _gate_leak_test( + candidate, "gate_leak_retrieval.md", llm, session_id="test-session" + ) assert result.verdict == GateVerdict.PASS # 验证 session_id 被正确传递 assert llm.calls[0]["session_id"] == "test-session" @@ -284,7 +300,7 @@ class TestRunGates: candidate=candidate, tree=tree, llm=llm, - family_spec=RETRIEVAL_FAMILY, + leak_probe_template="gate_leak_retrieval.md", postprocess=postprocess, session_id="test-session", ) @@ -303,7 +319,7 @@ class TestRunGates: candidate=candidate, tree=tree, llm=llm, - family_spec=RETRIEVAL_FAMILY, + leak_probe_template="gate_leak_retrieval.md", postprocess=postprocess, session_id="test-session", )