iomgaa
afa77173e3
- 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>
512 lines
16 KiB
Python
512 lines
16 KiB
Python
"""轻量四门质量检查 — 对生成题目执行 4 个独立的 LLM 质量门控。
|
||
|
||
四门：
|
||
1. key_verify: 验证答案在来源素材中有证据支撑。
|
||
2. blind_answer: 无上下文时 LLM 能否答对（若答对 → 题目泄漏）。
|
||
3. multi_true: 检测是否有多个选项可被视为正确。
|
||
4. leak_test: 按策略特定模板探测答题捷径。
|
||
|
||
设计要点：
|
||
- run_gates 先做 verbatim_ratio 前置短路（> 0.5 直接 FAIL key_verify）。
|
||
- 四门并发执行（asyncio.gather）。
|
||
- 每门加载 store/prompts/question_gen/ 下对应模板，构造 messages，调 LLM。
|
||
- LLM 返回 JSON {"verdict": "pass"|"fail", "reason": "..."}，解析失败视为 FAIL。
|
||
- session_id 必须透传至每次 LLM 调用（遥测关联）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
from dataclasses import dataclass, field
|
||
from enum import Enum
|
||
from pathlib import Path
|
||
from typing import TYPE_CHECKING
|
||
|
||
from loguru import logger
|
||
|
||
if TYPE_CHECKING:
|
||
from app.question_gen.postprocess import PostprocessResult
|
||
from app.tree.index import TreeIndex
|
||
from core.protocols import LLMProvider, VLMProvider
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 常量
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_PROMPTS_DIR = Path(__file__).resolve().parent.parent.parent / "store" / "prompts" / "question_gen"
|
||
|
||
_VERBATIM_THRESHOLD = 0.5
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# CandidateQuestion（Task 5 尚未实现，本地定义，后续迁移至 generator_v2.py）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class CandidateQuestion:
|
||
"""候选题目 — 出题管线生成、待门控审核的题目数据。
|
||
|
||
属性:
|
||
question_id: 题目唯一标识。
|
||
video_id: 所属视频标识。
|
||
task_type: 题型（如 "Action Reasoning"）。
|
||
skill_target: 目标失败机制编号（M1-M5）。
|
||
question: 题目文本。
|
||
options: 选项元组（如 ("A. ...", "B. ...", "C. ...", "D. ...")）。
|
||
answer: 正确答案字母（如 "A"）。
|
||
source_nodes: 来源节点 ID 元组。
|
||
difficulty: 难度等级。
|
||
subtitle_sentences: 字幕句子元组。
|
||
frame_paths: 帧图片路径元组。
|
||
"""
|
||
|
||
question_id: str
|
||
video_id: str
|
||
task_type: str
|
||
skill_target: str
|
||
question: str
|
||
options: tuple[str, ...]
|
||
answer: str
|
||
source_nodes: tuple[str, ...]
|
||
difficulty: str
|
||
subtitle_sentences: tuple[str, ...] = field(default_factory=tuple)
|
||
frame_paths: tuple[str, ...] = field(default_factory=tuple)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 门控结果类型
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class GateVerdict(Enum):
|
||
"""门控判决枚举。"""
|
||
|
||
PASS = "pass"
|
||
FAIL = "fail"
|
||
SKIP = "skip"
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class GateResult:
|
||
"""单门判决结果。
|
||
|
||
属性:
|
||
verdict: 门控判决（PASS / FAIL / SKIP）。
|
||
reason: 判决原因说明。
|
||
raw_response: LLM 原始返回内容。
|
||
"""
|
||
|
||
verdict: GateVerdict
|
||
reason: str
|
||
raw_response: str
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class GateReport:
|
||
"""四门判决汇总报告。
|
||
|
||
属性:
|
||
key_verify: 关键验证门结果。
|
||
blind_answer: 盲答门结果。
|
||
multi_true: 多正确门结果。
|
||
leak_test: 泄漏测试门结果。
|
||
"""
|
||
|
||
key_verify: GateResult
|
||
blind_answer: GateResult
|
||
multi_true: GateResult
|
||
leak_test: GateResult
|
||
|
||
@property
|
||
def passed(self) -> bool:
|
||
"""全门 PASS 或 SKIP 时视为通过。"""
|
||
for gate in (self.key_verify, self.blind_answer, self.multi_true, self.leak_test):
|
||
if gate.verdict == GateVerdict.FAIL:
|
||
return False
|
||
return True
|
||
|
||
@property
|
||
def reject_reason(self) -> str | None:
|
||
"""首个 FAIL 门的 reason，全部通过返回 None。"""
|
||
for gate in (self.key_verify, self.blind_answer, self.multi_true, self.leak_test):
|
||
if gate.verdict == GateVerdict.FAIL:
|
||
return gate.reason
|
||
return None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 内部工具函数
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _load_prompt_template(filename: str) -> str:
|
||
"""加载 store/prompts/question_gen/ 下的模板文件。
|
||
|
||
参数:
|
||
filename: 模板文件名（如 "gate_key_verify.md"）。
|
||
|
||
返回:
|
||
模板内容字符串。
|
||
|
||
异常:
|
||
FileNotFoundError: 模板文件不存在。
|
||
"""
|
||
path = _PROMPTS_DIR / filename
|
||
if not path.exists():
|
||
msg = f"门控模板文件不存在: {path}"
|
||
raise FileNotFoundError(msg)
|
||
return path.read_text(encoding="utf-8")
|
||
|
||
|
||
def _resolve_source_text(candidate: CandidateQuestion, tree: TreeIndex) -> str:
|
||
"""从树中解析候选题的来源节点文本。
|
||
|
||
遍历树的所有层级，匹配 source_nodes 中的 ID，提取对应描述文本。
|
||
|
||
参数:
|
||
candidate: 候选题目。
|
||
tree: 视频树索引。
|
||
|
||
返回:
|
||
拼接的来源节点描述文本。
|
||
"""
|
||
texts: list[str] = []
|
||
target_ids = set(candidate.source_nodes)
|
||
|
||
for l1 in tree.roots:
|
||
if l1.id in target_ids:
|
||
texts.append(f"[L1 {l1.id}] {l1.card.scene_summary}")
|
||
for l2 in l1.children:
|
||
if l2.id in target_ids:
|
||
texts.append(f"[L2 {l2.id}] {l2.card.event_description}")
|
||
for l3 in l2.children:
|
||
if l3.id in target_ids:
|
||
texts.append(f"[L3 {l3.id}] {l3.card.frame_summary}")
|
||
|
||
if not texts:
|
||
logger.warning(
|
||
"未找到来源节点: candidate={}, source_nodes={}",
|
||
candidate.question_id,
|
||
candidate.source_nodes,
|
||
)
|
||
return "(no source material found)"
|
||
|
||
return "\n".join(texts)
|
||
|
||
|
||
def _format_options(options: tuple[str, ...]) -> str:
|
||
"""将选项元组格式化为可读字符串。"""
|
||
return "\n".join(options)
|
||
|
||
|
||
def _parse_gate_response(raw_content: str) -> tuple[GateVerdict, str]:
|
||
"""解析 LLM 返回的门控 JSON 响应。
|
||
|
||
期望格式: {"verdict": "pass"|"fail", "reason": "..."}
|
||
解析失败时返回 FAIL + parse_error。
|
||
|
||
参数:
|
||
raw_content: LLM 返回的原始文本。
|
||
|
||
返回:
|
||
(verdict, reason) 元组。
|
||
"""
|
||
try:
|
||
# 尝试从文本中提取 JSON（可能被 markdown 代码块包裹）
|
||
content = raw_content.strip()
|
||
if "```" in content:
|
||
# 提取代码块中的内容
|
||
parts = content.split("```")
|
||
for part in parts:
|
||
stripped = part.strip()
|
||
if stripped.startswith("json"):
|
||
stripped = stripped[4:].strip()
|
||
if stripped.startswith("{"):
|
||
content = stripped
|
||
break
|
||
|
||
data = json.loads(content)
|
||
verdict_str = data.get("verdict", "").lower()
|
||
reason = data.get("reason", "")
|
||
|
||
if verdict_str == "pass":
|
||
return GateVerdict.PASS, reason
|
||
elif verdict_str == "fail":
|
||
return GateVerdict.FAIL, reason
|
||
else:
|
||
return GateVerdict.FAIL, f"parse_error: invalid verdict '{verdict_str}'"
|
||
|
||
except (json.JSONDecodeError, AttributeError, TypeError) as e:
|
||
return GateVerdict.FAIL, f"parse_error: {e}"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 四门实现
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _resolve_source_frames(candidate: CandidateQuestion, tree: TreeIndex) -> list[str]:
|
||
"""从树中收集候选题来源节点关联的帧路径。
|
||
|
||
参数:
|
||
candidate: 候选题目。
|
||
tree: 视频树索引。
|
||
|
||
返回:
|
||
去重后的帧路径列表（最多 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,
|
||
question=candidate.question,
|
||
options=_format_options(candidate.options),
|
||
answer=candidate.answer,
|
||
)
|
||
|
||
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)
|
||
|
||
|
||
async def _gate_blind_answer(
|
||
candidate: CandidateQuestion,
|
||
llm: LLMProvider,
|
||
*,
|
||
session_id: str,
|
||
) -> GateResult:
|
||
"""盲答门 — 无上下文时 LLM 能否答对（答对 → 题目泄漏）。
|
||
|
||
参数:
|
||
candidate: 候选题目。
|
||
llm: LLM 调用端口。
|
||
session_id: 会话 ID（遥测关联）。
|
||
|
||
返回:
|
||
GateResult 实例。
|
||
"""
|
||
template = _load_prompt_template("gate_blind_answer.md")
|
||
prompt = template.format(
|
||
question=candidate.question,
|
||
options=_format_options(candidate.options),
|
||
)
|
||
|
||
response = await llm.chat(
|
||
[{"role": "user", "content": prompt}],
|
||
session_id=session_id,
|
||
)
|
||
|
||
verdict, reason = _parse_gate_response(response.content)
|
||
return GateResult(verdict=verdict, reason=reason, raw_response=response.content)
|
||
|
||
|
||
async def _gate_multi_true(
|
||
candidate: CandidateQuestion,
|
||
tree: TreeIndex,
|
||
llm: LLMProvider,
|
||
*,
|
||
session_id: str,
|
||
) -> GateResult:
|
||
"""多正确门 — 检测是否有多个选项可被视为正确。
|
||
|
||
参数:
|
||
candidate: 候选题目。
|
||
tree: 视频树索引。
|
||
llm: LLM 调用端口。
|
||
session_id: 会话 ID（遥测关联）。
|
||
|
||
返回:
|
||
GateResult 实例。
|
||
"""
|
||
source_text = _resolve_source_text(candidate, tree)
|
||
template = _load_prompt_template("gate_multi_true.md")
|
||
prompt = template.format(
|
||
source_text=source_text,
|
||
question=candidate.question,
|
||
options=_format_options(candidate.options),
|
||
)
|
||
|
||
response = await llm.chat(
|
||
[{"role": "user", "content": prompt}],
|
||
session_id=session_id,
|
||
)
|
||
|
||
verdict, reason = _parse_gate_response(response.content)
|
||
return GateResult(verdict=verdict, reason=reason, raw_response=response.content)
|
||
|
||
|
||
async def _gate_leak_test(
|
||
candidate: CandidateQuestion,
|
||
leak_probe_template: str,
|
||
llm: LLMProvider,
|
||
*,
|
||
session_id: str,
|
||
) -> GateResult:
|
||
"""泄漏测试门 — 按策略特定模板探测答题捷径。
|
||
|
||
参数:
|
||
candidate: 候选题目。
|
||
leak_probe_template: 泄漏探测模板文件名（store/prompts/question_gen/ 下）。
|
||
llm: LLM 调用端口。
|
||
session_id: 会话 ID（遥测关联）。
|
||
|
||
返回:
|
||
GateResult 实例。
|
||
"""
|
||
template = _load_prompt_template(leak_probe_template)
|
||
prompt = template.format(
|
||
question=candidate.question,
|
||
options=_format_options(candidate.options),
|
||
answer=candidate.answer,
|
||
)
|
||
|
||
response = await llm.chat(
|
||
[{"role": "user", "content": prompt}],
|
||
session_id=session_id,
|
||
)
|
||
|
||
verdict, reason = _parse_gate_response(response.content)
|
||
return GateResult(verdict=verdict, reason=reason, raw_response=response.content)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 编排入口
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
async def run_gates(
|
||
candidate: CandidateQuestion,
|
||
tree: TreeIndex,
|
||
llm: LLMProvider,
|
||
leak_probe_template: str,
|
||
postprocess: PostprocessResult,
|
||
*,
|
||
vlm: VLMProvider | None = None,
|
||
session_id: str,
|
||
) -> GateReport:
|
||
"""编排四门并发执行，返回汇总报告。
|
||
|
||
前置规则：verbatim_ratio > 0.5 时直接短路 key_verify 为 FAIL（不调 LLM）。
|
||
|
||
参数:
|
||
candidate: 候选题目。
|
||
tree: 视频树索引。
|
||
llm: LLM 调用端口。
|
||
leak_probe_template: 泄漏探测模板文件名（store/prompts/question_gen/ 下）。
|
||
postprocess: 后处理结果（含 verbatim_ratio）。
|
||
vlm: VLM 图文调用端口（key_verify 使用，None 时降级为 LLM）。
|
||
session_id: 会话 ID（遥测关联）。
|
||
|
||
返回:
|
||
GateReport 四门汇总。
|
||
"""
|
||
# Phase 1: verbatim 前置短路 — 不调 LLM，直接返回
|
||
if postprocess.verbatim_ratio > _VERBATIM_THRESHOLD:
|
||
logger.info(
|
||
"verbatim_ratio={:.3f} > {:.1f}，key_verify 短路 FAIL，其余三门 SKIP",
|
||
postprocess.verbatim_ratio,
|
||
_VERBATIM_THRESHOLD,
|
||
)
|
||
skip_result = GateResult(
|
||
verdict=GateVerdict.SKIP,
|
||
reason="skipped due to verbatim short-circuit",
|
||
raw_response="",
|
||
)
|
||
return GateReport(
|
||
key_verify=GateResult(
|
||
verdict=GateVerdict.FAIL,
|
||
reason=f"verbatim_ratio={postprocess.verbatim_ratio:.3f} exceeds threshold {_VERBATIM_THRESHOLD}",
|
||
raw_response="",
|
||
),
|
||
blind_answer=skip_result,
|
||
multi_true=skip_result,
|
||
leak_test=skip_result,
|
||
)
|
||
|
||
# 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, 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, leak_probe_template, llm, session_id=session_id),
|
||
)
|
||
|
||
report = GateReport(
|
||
key_verify=key_result,
|
||
blind_answer=blind_result,
|
||
multi_true=multi_result,
|
||
leak_test=leak_result,
|
||
)
|
||
|
||
if report.passed:
|
||
logger.debug("四门全部通过: question_id={}", candidate.question_id)
|
||
else:
|
||
logger.info(
|
||
"门控拒绝: question_id={}, reason={}",
|
||
candidate.question_id,
|
||
report.reject_reason,
|
||
)
|
||
|
||
return report
|