iomgaa/Video-Tree-TRM5
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(question_gen): adapt generator/gates/store signatures for strategy

Browse Source 
- 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) <noreply@anthropic.com>
This commit is contained in:
iomgaa
2026-07-14 05:50:47 -04:00
parent c49d0ff12f
commit afa77173e3
4 changed files with 155 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/question_gen/gates.py
+73 -26
View File
@@ -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(
app/question_gen/generator_v2.py
+39 -22
View File
@@ -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,
)
app/question_gen/run_store.py
+20 -21
View File
@@ -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:
"""关闭数据库连接。"""