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

feat(question_gen): add deterministic postprocess layer

Browse Source 
Add app/question_gen/postprocess.py with zero-LLM deterministic
post-processing for generated questions:

- shuffle_options: deterministic option permutation with answer remapping
- check_referent_blacklist: detect self-referential language (this clip, etc.)
- check_verbatim: word-level n-gram overlap ratio measurement
- has_time_anchor: timestamp and temporal phrase detection
- check_forbidden_material: T1/T7 source material validation
- run_postprocess: orchestration returning PostprocessResult

Tests: 31 unit tests covering all functions.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
iomgaa
2026-07-11 23:14:43 -04:00
parent 75e6d8c550
commit c83d771923
2 changed files with 698 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/question_gen/postprocess.py
+362
View File
@@ -0,0 +1,362 @@
"""确定性后处理层 — 零 LLM 的选项洗牌、指代黑名单、逐字重复率、时间锚点、素材禁区检测。
所有函数均为纯函数（给定输入必定产出相同输出），
用于 pipeline 出题后、门控前的确定性质量检查。
"""
from __future__ import annotations
import re
from dataclasses import dataclass
from typing import TYPE_CHECKING
from loguru import logger
if TYPE_CHECKING:
import random
# ---------------------------------------------------------------------------
# 数据结构
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class PostprocessResult:
"""后处理结果汇总。
属性:
options: 洗牌后的选项元组。
answer: 重映射后的答案字母。
referent_violations: 指代黑名单违规描述列表。
verbatim_ratio: 正确选项与来源素材的逐字重复率 [0.0, 1.0]。
has_time_anchor: 题目是否包含时间锚点。
"""
options: tuple[str, ...]
answer: str
referent_violations: list[str]
verbatim_ratio: float
has_time_anchor: bool
# ---------------------------------------------------------------------------
# 指代黑名单（预编译正则）
# ---------------------------------------------------------------------------
_BLACKLIST_PATTERNS: list[tuple[re.Pattern[str], str]] = [
(re.compile(r"this\s+clip", re.IGNORECASE), "this clip"),
(re.compile(r"the\s+video", re.IGNORECASE), "the video"),
(re.compile(r"this\s+video", re.IGNORECASE), "this video"),
(re.compile(r"this\s+scene", re.IGNORECASE), "this scene"),
(re.compile(r"the\s+clip", re.IGNORECASE), "the clip"),
(re.compile(r"this\s+footage", re.IGNORECASE), "this footage"),
(re.compile(r"the\s+footage", re.IGNORECASE), "the footage"),
(re.compile(r"上面的片段", re.IGNORECASE), "上面的片段"),
(re.compile(r"这段视频", re.IGNORECASE), "这段视频"),
(re.compile(r"该视频", re.IGNORECASE), "该视频"),
(re.compile(r"这个片段", re.IGNORECASE), "这个片段"),
(re.compile(r"视频中", re.IGNORECASE), "视频中"),
]
# ---------------------------------------------------------------------------
# 时间锚点正则与短语
# ---------------------------------------------------------------------------
_TIMESTAMP_PATTERN = re.compile(r"\d{1,2}:\d{2}")
_TIME_ANCHOR_PHRASES: list[re.Pattern[str]] = [
re.compile(r"at\s+the\s+beginning", re.IGNORECASE),
re.compile(r"at\s+the\s+end", re.IGNORECASE),
re.compile(r"in\s+the\s+beginning", re.IGNORECASE),
re.compile(r"at\s+the\s+start", re.IGNORECASE),
re.compile(r"开头", re.IGNORECASE),
re.compile(r"结尾", re.IGNORECASE),
re.compile(r"末尾", re.IGNORECASE),
re.compile(r"片头", re.IGNORECASE),
re.compile(r"片尾", re.IGNORECASE),
]
# ---------------------------------------------------------------------------
# T1 / T7 素材禁区正则
# ---------------------------------------------------------------------------
# T1: 瞬时动作
_T1_INSTANT_ACTION_PATTERNS: list[re.Pattern[str]] = [
re.compile(r"瞬间|一闪而过|转瞬即逝|一瞬间|刹那", re.IGNORECASE),
re.compile(r"flash|instant|split\s*second|blink", re.IGNORECASE),
]
# T1: 记分牌时序
_T1_SCOREBOARD_PATTERNS: list[re.Pattern[str]] = [
re.compile(r"记分牌|比分.*变|比分.*更新|得分.*变化", re.IGNORECASE),
re.compile(r"scoreboard|score\s*(changed|updated|went)", re.IGNORECASE),
]
# T1: 无对白因果
_T1_NO_DIALOGUE_CAUSAL_PATTERNS: list[re.Pattern[str]] = [
re.compile(r"无对白.*因果|因果.*无法.*判断", re.IGNORECASE),
re.compile(r"no\s+dialogue.*caus|cannot.*determin.*caus", re.IGNORECASE),
]
# T7: 选项重复
_T7_OPTION_REPETITION_PATTERNS: list[re.Pattern[str]] = [
re.compile(r"重复选项|选项.*相同|选项.*重复|完全相同", re.IGNORECASE),
re.compile(r"duplicate\s+option|identical\s+option|same\s+option", re.IGNORECASE),
]
# T7: 计数边界口径含糊
_T7_COUNTING_AMBIGUITY_PATTERNS: list[re.Pattern[str]] = [
re.compile(r"大约.*数量|数量不确定|难以.*计数|不确定.*几|约.*个", re.IGNORECASE),
re.compile(r"approximate.*count|uncertain.*number|hard\s+to\s+count", re.IGNORECASE),
]
# ---------------------------------------------------------------------------
# 公开函数
# ---------------------------------------------------------------------------
def shuffle_options(
options: tuple[str, ...],
answer: str,
rng: random.Random,
) -> tuple[tuple[str, ...], str]:
"""确定性洗牌选项并重映射答案字母。
通过 index permutation 打乱选项顺序，为每个选项重新分配 A/B/C/D 前缀，
并将答案字母映射到正确选项的新位置。
参数:
options: 原始选项元组，格式为 ("A. text1", "B. text2", ...)。
answer: 原始答案字母（"A"/"B"/"C"/"D"）。
rng: 可控随机数生成器（保证确定性）。
返回:
(新选项元组, 新答案字母) — 选项文本不变，仅前缀和顺序改变。
"""
# Phase 1: 提取纯文本（去掉 "X. " 前缀）
texts = [opt[3:] for opt in options]
# Phase 2: 确定原正确选项的文本
correct_idx = ord(answer) - ord("A")
correct_text = texts[correct_idx]
# Phase 3: 生成随机排列
indices = list(range(len(texts)))
rng.shuffle(indices)
# Phase 4: 按排列重组，分配新前缀
new_options: list[str] = []
new_answer = ""
for new_pos, old_idx in enumerate(indices):
letter = chr(ord("A") + new_pos)
new_options.append(f"{letter}. {texts[old_idx]}")
if texts[old_idx] == correct_text:
new_answer = letter
return tuple(new_options), new_answer
def check_referent_blacklist(question_text: str) -> list[str]:
"""检测题目文本中的自指代语言。
在 benchmark 题目中，自指代（如 "this clip"、"这段视频"）会泄露视频上下文，
使题目脱离视频后无法独立理解。
参数:
question_text: 题目文本。
返回:
违规描述列表，空列表表示通过。
"""
violations: list[str] = []
for pattern, label in _BLACKLIST_PATTERNS:
if pattern.search(question_text):
violations.append(f"检测到自指代: '{label}'")
return violations
def check_verbatim(
question_text: str,
correct_option: str,
source_texts: list[str],
window: int = 6,
) -> float:
"""计算正确选项与来源素材的逐字重复率。
使用滑动窗口 n-gram 集合交集方法：从选项文本提取所有 n-gram，
与来源文本的 n-gram 集合求交集，计算重叠比例。
参数:
question_text: 题目文本（当前未使用，预留接口）。
correct_option: 正确选项文本（含 "X. " 前缀）。
source_texts: 来源素材文本列表。
window: n-gram 窗口大小。
返回:
重复率 [0.0, 1.0]。0.0 表示无重叠，1.0 表示完全复制。
"""
# Phase 1: 提取选项纯文本（去掉可能的 "X. " 前缀）
option_text = correct_option
if len(option_text) >= 3 and option_text[1] == "." and option_text[2] == " ":
option_text = option_text[3:]
# Phase 2: 分词（简单空格分词，转小写）
option_words = option_text.lower().split()
# Phase 3: 选项词数不足窗口大小则无法构成 n-gram
if len(option_words) < window:
return 0.0
# Phase 4: 构造选项的 n-gram 集合
option_ngrams: set[tuple[str, ...]] = set()
for i in range(len(option_words) - window + 1):
option_ngrams.add(tuple(option_words[i : i + window]))
if not option_ngrams:
return 0.0
# Phase 5: 构造来源文本的 n-gram 集合
source_ngrams: set[tuple[str, ...]] = set()
for source in source_texts:
words = source.lower().split()
for i in range(len(words) - window + 1):
source_ngrams.add(tuple(words[i : i + window]))
if not source_ngrams:
return 0.0
# Phase 6: 计算交集比例
overlap = option_ngrams & source_ngrams
return len(overlap) / len(option_ngrams)
def has_time_anchor(question_text: str) -> bool:
"""检测题目中是否包含时间锚点。
时间锚点包括：数字时间戳（如 "01:30"）和时间短语（如 "at the beginning"、"开头"）。
参数:
question_text: 题目文本。
返回:
True 表示包含时间锚点。
"""
# Phase 1: 检查数字时间戳
if _TIMESTAMP_PATTERN.search(question_text):
return True
# Phase 2: 检查时间短语
return any(pattern.search(question_text) for pattern in _TIME_ANCHOR_PHRASES)
def check_forbidden_material(source_nodes_text: str, task_type: str) -> list[str]:
"""出题禁区：检测 T1 类素材和 T7 噪声模式。
T1 类素材（不适合出题的内容）:
- 瞬时动作：画面一闪而过，无法稳定观察
- 记分牌时序：依赖数字变化的时序信息
- 无对白因果：缺乏语言线索的因果推理
T7 噪声模式（选项质量问题）:
- 选项重复：多个选项表述相同
- 计数边界口径含糊：数量描述不确定
参数:
source_nodes_text: 来源节点的拼接文本。
task_type: 题型名称（用于上下文感知检测）。
返回:
违规描述列表，空列表表示通过。
"""
violations: list[str] = []
# Phase 1: T1 瞬时动作检测
for pattern in _T1_INSTANT_ACTION_PATTERNS:
if pattern.search(source_nodes_text):
violations.append("T1 违规: 素材包含瞬时动作描述，不适合出题")
break
# Phase 2: T1 记分牌时序检测
for pattern in _T1_SCOREBOARD_PATTERNS:
if pattern.search(source_nodes_text):
violations.append("T1 违规: 素材包含记分牌时序信息，不适合出题")
break
# Phase 3: T1 无对白因果检测
for pattern in _T1_NO_DIALOGUE_CAUSAL_PATTERNS:
if pattern.search(source_nodes_text):
violations.append("T1 违规: 素材缺乏对白因果线索，不适合出题")
break
# Phase 4: T7 选项重复检测
for pattern in _T7_OPTION_REPETITION_PATTERNS:
if pattern.search(source_nodes_text):
violations.append("T7 违规: 素材暗示可能产生重复选项")
break
# Phase 5: T7 计数边界口径含糊检测
for pattern in _T7_COUNTING_AMBIGUITY_PATTERNS:
if pattern.search(source_nodes_text):
violations.append("T7 违规: 素材包含计数边界含糊描述")
break
return violations
def run_postprocess(
question_text: str,
options: tuple[str, ...],
answer: str,
source_texts: list[str],
rng: random.Random,
) -> PostprocessResult:
"""编排全部后处理检查，返回汇总结果。
执行流程:
1. 选项洗牌（确定性）
2. 指代黑名单检测
3. 逐字重复率计算（基于洗牌后的正确选项）
4. 时间锚点检测
参数:
question_text: 题目文本。
options: 原始选项元组（"A. text", "B. text", ...）。
answer: 原始答案字母。
source_texts: 来源素材文本列表。
rng: 可控随机数生成器。
返回:
PostprocessResult 汇总实例。
"""
# Phase 1: 选项洗牌
shuffled_options, new_answer = shuffle_options(options, answer, rng)
# Phase 2: 指代黑名单
referent_violations = check_referent_blacklist(question_text)
# Phase 3: 逐字重复率（用洗牌后正确选项的文本）
correct_idx = ord(new_answer) - ord("A")
correct_option_text = shuffled_options[correct_idx]
verbatim_ratio = check_verbatim(question_text, correct_option_text, source_texts)
# Phase 4: 时间锚点
time_anchor = has_time_anchor(question_text)
logger.debug(
"后处理完成: referent_violations={}, verbatim_ratio={:.3f}, has_time_anchor={}",
len(referent_violations),
verbatim_ratio,
time_anchor,
)
return PostprocessResult(
options=shuffled_options,
answer=new_answer,
referent_violations=referent_violations,
verbatim_ratio=verbatim_ratio,
has_time_anchor=time_anchor,
)
tests/unit/test_postprocess.py
+336
View File
@@ -0,0 +1,336 @@
"""postprocess 模块单元测试 — 选项洗牌、指代黑名单、逐字重复率、时间锚点、素材禁区、run_postprocess 编排。"""
from __future__ import annotations
import random
from app.question_gen.postprocess import (
PostprocessResult,
check_forbidden_material,
check_referent_blacklist,
check_verbatim,
has_time_anchor,
run_postprocess,
shuffle_options,
)
# ---------------------------------------------------------------------------
# TestShuffleOptions
# ---------------------------------------------------------------------------
class TestShuffleOptions:
"""选项洗牌确定性与答案重映射测试。"""
def test_deterministic_with_seed(self) -> None:
"""相同种子产生相同置换结果。"""
options = ("A. 苹果", "B. 香蕉", "C. 橙子", "D. 葡萄")
answer = "B"
rng1 = random.Random(42)
rng2 = random.Random(42)
result1 = shuffle_options(options, answer, rng1)
result2 = shuffle_options(options, answer, rng2)
assert result1 == result2
def test_answer_remapped_correctly(self) -> None:
"""洗牌后 answer 字母始终指向原正确文本。"""
options = ("A. 苹果", "B. 香蕉", "C. 橙子", "D. 葡萄")
answer = "B" # 正确文本是 "香蕉"
# 用多个不同种子确保覆盖不同排列
for seed in range(100):
rng = random.Random(seed)
new_options, new_answer = shuffle_options(options, answer, rng)
# 答案字母对应的选项文本必须包含 "香蕉"
answer_idx = ord(new_answer) - ord("A")
assert "香蕉" in new_options[answer_idx], (
f"seed={seed}: 新答案 {new_answer} 对应 '{new_options[answer_idx]}'，不含 '香蕉'"
)
def test_all_options_preserved(self) -> None:
"""洗牌后选项内容集合不变（仅字母前缀重映射）。"""
options = ("A. 苹果", "B. 香蕉", "C. 橙子", "D. 葡萄")
answer = "A"
rng = random.Random(7)
new_options, _ = shuffle_options(options, answer, rng)
# 提取纯文本（去除 "X. " 前缀）
original_texts = {opt[3:] for opt in options}
shuffled_texts = {opt[3:] for opt in new_options}
assert original_texts == shuffled_texts
def test_output_letters_sequential(self) -> None:
"""洗牌后选项前缀始终为 A/B/C/D 顺序。"""
options = ("A. one", "B. two", "C. three", "D. four")
answer = "C"
rng = random.Random(99)
new_options, _ = shuffle_options(options, answer, rng)
for i, opt in enumerate(new_options):
expected_prefix = f"{chr(ord('A') + i)}. "
assert opt.startswith(expected_prefix), (
f"选项 {i} 应以 '{expected_prefix}' 开头，实际为 '{opt[:3]}'"
)
# ---------------------------------------------------------------------------
# TestReferentBlacklist
# ---------------------------------------------------------------------------
class TestReferentBlacklist:
"""指代黑名单检测测试。"""
def test_clean_passes(self) -> None:
"""无自指代语言的正常题目返回空列表。"""
q = "What is the main character doing in the kitchen?"
violations = check_referent_blacklist(q)
assert violations == []
def test_this_clip_caught(self) -> None:
"""'this clip' 被检测为违规。"""
q = "What happens in this clip after the man enters?"
violations = check_referent_blacklist(q)
assert len(violations) >= 1
assert any("this clip" in v.lower() for v in violations)
def test_the_video_caught(self) -> None:
"""'the video' 被检测为违规。"""
q = "In the video, what color is the car?"
violations = check_referent_blacklist(q)
assert len(violations) >= 1
def test_chinese_referent_caught(self) -> None:
"""中文自指代 '这段视频' 被检测。"""
q = "这段视频中男人在做什么？"
violations = check_referent_blacklist(q)
assert len(violations) >= 1
def test_this_scene_caught(self) -> None:
"""'this scene' 被检测为违规。"""
q = "What is shown in this scene before the explosion?"
violations = check_referent_blacklist(q)
assert len(violations) >= 1
def test_case_insensitive(self) -> None:
"""大小写不影响检测。"""
q = "What does THIS CLIP show at the end?"
violations = check_referent_blacklist(q)
assert len(violations) >= 1
# ---------------------------------------------------------------------------
# TestVerbatim
# ---------------------------------------------------------------------------
class TestVerbatim:
"""逐字重复率检测测试。"""
def test_zero_overlap(self) -> None:
"""正确选项与来源文本完全无重叠时返回 0.0。"""
ratio = check_verbatim(
question_text="What color is the sky?",
correct_option="A. The sky is painted in bright orange",
source_texts=["A man walks into a store and buys groceries for dinner."],
window=6,
)
assert ratio == 0.0
def test_full_copy_returns_one(self) -> None:
"""正确选项完全来自来源文本时返回 1.0。"""
source = "The cat sat on the mat near the window looking outside"
option = "A. The cat sat on the mat near the window looking outside"
ratio = check_verbatim(
question_text="What did the animal do?",
correct_option=option,
source_texts=[source],
window=6,
)
assert ratio == 1.0
def test_partial(self) -> None:
"""部分重叠返回 0-1 之间的值。"""
source = "The large brown dog jumped over the lazy red fox quickly"
# 选项中 6 词窗口 "large brown dog jumped over the" 出现在 source 中
option = "A. A large brown dog jumped over the small fence"
ratio = check_verbatim(
question_text="What happened?",
correct_option=option,
source_texts=[source],
window=6,
)
assert 0.0 < ratio < 1.0
def test_empty_source_returns_zero(self) -> None:
"""来源文本为空列表时返回 0.0。"""
ratio = check_verbatim(
question_text="What is this?",
correct_option="A. Something interesting",
source_texts=[],
window=6,
)
assert ratio == 0.0
def test_short_option_below_window(self) -> None:
"""选项词数不足窗口大小时仍正常工作（返回 0.0）。"""
ratio = check_verbatim(
question_text="What?",
correct_option="A. Hello world",
source_texts=["Some very long source text that goes on and on"],
window=6,
)
assert ratio == 0.0
# ---------------------------------------------------------------------------
# TestTimeAnchor
# ---------------------------------------------------------------------------
class TestTimeAnchor:
"""时间锚点检测测试。"""
def test_timestamp(self) -> None:
"""包含 '01:30' 格式时间戳 → True。"""
assert has_time_anchor("What happens at 01:30 in the recording?") is True
def test_no_anchor(self) -> None:
"""无任何时间锚点 → False。"""
assert has_time_anchor("What is the man doing in the park?") is False
def test_beginning_phrase(self) -> None:
"""'at the beginning' → True。"""
assert has_time_anchor("At the beginning, what does the woman say?") is True
def test_chinese_beginning(self) -> None:
"""'开头' → True。"""
assert has_time_anchor("视频开头出现了什么？") is True
def test_chinese_ending(self) -> None:
"""'结尾' → True。"""
assert has_time_anchor("在视频结尾发生了什么？") is True
def test_minute_second_format(self) -> None:
"""'12:45' 格式 → True。"""
assert has_time_anchor("At 12:45, who appears on screen?") is True
# ---------------------------------------------------------------------------
# TestCheckForbiddenMaterial
# ---------------------------------------------------------------------------
class TestCheckForbiddenMaterial:
"""素材禁区检测测试。"""
def test_clean_material_passes(self) -> None:
"""正常素材无违规。"""
source = "一个男人走进厨房，开始准备晚餐，切菜并烧水。"
violations = check_forbidden_material(source, "Action Recognition")
assert violations == []
def test_t1_instant_action_detected(self) -> None:
"""T1 瞬时动作模式被检测。"""
source = "球员瞬间射门得分。一闪而过的画面。"
violations = check_forbidden_material(source, "Action Recognition")
assert len(violations) >= 1
def test_t1_scoreboard_temporal_detected(self) -> None:
"""T1 记分牌时序模式被检测。"""
source = "比分从2:1变为3:1，记分牌显示更新。"
violations = check_forbidden_material(source, "Counting Problem")
assert len(violations) >= 1
def test_t7_option_repetition_detected(self) -> None:
"""T7 选项重复模式被检测。"""
source = "选项A和选项B完全相同。重复选项无法区分。"
violations = check_forbidden_material(source, "Object Recognition")
assert len(violations) >= 1
def test_t7_counting_ambiguity_detected(self) -> None:
"""T7 计数边界口径含糊被检测。"""
source = "大约有三到五个人，数量不确定，难以精确计数。"
violations = check_forbidden_material(source, "Counting Problem")
assert len(violations) >= 1
# ---------------------------------------------------------------------------
# TestRunPostprocess
# ---------------------------------------------------------------------------
class TestRunPostprocess:
"""run_postprocess 编排测试。"""
def test_returns_postprocess_result(self) -> None:
"""返回类型正确。"""
result = run_postprocess(
question_text="What is the main character doing?",
options=("A. Running", "B. Swimming", "C. Reading", "D. Sleeping"),
answer="A",
source_texts=["The protagonist goes for a morning jog in the park."],
rng=random.Random(42),
)
assert isinstance(result, PostprocessResult)
def test_shuffle_applied(self) -> None:
"""选项顺序被打乱（至少在多次运行中出现过不同排列）。"""
options = ("A. Alpha", "B. Beta", "C. Gamma", "D. Delta")
answer = "A"
seen_orders: set[tuple[str, ...]] = set()
for seed in range(50):
result = run_postprocess(
question_text="Pick one",
options=options,
answer=answer,
source_texts=["Unrelated source text about something else entirely."],
rng=random.Random(seed),
)
seen_orders.add(result.options)
# 50 个不同种子应产生多于 1 种排列
assert len(seen_orders) > 1
def test_referent_violations_populated(self) -> None:
"""有指代违规时正确填入结果。"""
result = run_postprocess(
question_text="What happens in this clip?",
options=("A. One", "B. Two", "C. Three", "D. Four"),
answer="B",
source_texts=["Some text here."],
rng=random.Random(0),
)
assert len(result.referent_violations) >= 1
def test_verbatim_ratio_range(self) -> None:
"""verbatim_ratio 始终在 [0.0, 1.0] 范围内。"""
result = run_postprocess(
question_text="What is shown?",
options=(
"A. A cat sits on the mat",
"B. A dog runs",
"C. A bird flies",
"D. A fish swims",
),
answer="A",
source_texts=["A cat sits on the mat by the window."],
rng=random.Random(1),
)
assert 0.0 <= result.verbatim_ratio <= 1.0
def test_has_time_anchor_field(self) -> None:
"""时间锚点字段正确填写。"""
result = run_postprocess(
question_text="What happens at 02:30?",
options=("A. X", "B. Y", "C. Z", "D. W"),
answer="C",
source_texts=["Some source text."],
rng=random.Random(5),
)
assert result.has_time_anchor is True