iomgaa
7abe92eb1c
- check_verbatim now computes n-gram overlap for BOTH question_text and correct_option vs source texts, returning max(question_ratio, option_ratio). Extracted _ngram_overlap_ratio helper for reuse. - Added 4 missing blacklist patterns: 'this segment', 'this frame', 'the current frame', 'frame summary'. - Added 5 new test cases covering the above changes. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
382 lines
13 KiB
Python
382 lines
13 KiB
Python
"""确定性后处理层 — 零 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"this\s+segment", re.IGNORECASE), "this segment"),
|
|
(re.compile(r"this\s+frame", re.IGNORECASE), "this frame"),
|
|
(re.compile(r"the\s+current\s+frame", re.IGNORECASE), "the current frame"),
|
|
(re.compile(r"frame\s+summary", re.IGNORECASE), "frame summary"),
|
|
(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 _ngram_overlap_ratio(
|
|
text_words: list[str],
|
|
source_ngrams: set[tuple[str, ...]],
|
|
window: int,
|
|
) -> float:
|
|
"""计算单段文本与来源 n-gram 集合的重叠率。
|
|
|
|
参数:
|
|
text_words: 待比较文本的分词列表（已小写化）。
|
|
source_ngrams: 来源素材的 n-gram 集合。
|
|
window: n-gram 窗口大小。
|
|
|
|
返回:
|
|
重叠率 [0.0, 1.0]，词数不足窗口大小时返回 0.0。
|
|
"""
|
|
if len(text_words) < window:
|
|
return 0.0
|
|
|
|
text_ngrams: set[tuple[str, ...]] = {
|
|
tuple(text_words[i : i + window]) for i in range(len(text_words) - window + 1)
|
|
}
|
|
|
|
if not text_ngrams:
|
|
return 0.0
|
|
|
|
overlap = text_ngrams & source_ngrams
|
|
return len(overlap) / len(text_ngrams)
|
|
|
|
|
|
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: 构造来源文本的 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 2: 计算题目文本的重叠率
|
|
question_words = question_text.lower().split()
|
|
question_ratio = _ngram_overlap_ratio(question_words, source_ngrams, window)
|
|
|
|
# Phase 3: 提取选项纯文本（去掉可能的 "X. " 前缀）
|
|
option_text = correct_option
|
|
if len(option_text) >= 3 and option_text[1] == "." and option_text[2] == " ":
|
|
option_text = option_text[3:]
|
|
|
|
option_words = option_text.lower().split()
|
|
option_ratio = _ngram_overlap_ratio(option_words, source_ngrams, window)
|
|
|
|
# Phase 4: 返回两者中较大的重叠率
|
|
return max(question_ratio, option_ratio)
|
|
|
|
|
|
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 _match_any(text: str, patterns: list[re.Pattern[str]]) -> bool:
|
|
"""判断文本是否匹配任一正则模式。
|
|
|
|
参数:
|
|
text: 待检测文本。
|
|
patterns: 预编译正则列表。
|
|
|
|
返回:
|
|
True 表示至少命中一条模式。
|
|
"""
|
|
return any(p.search(text) for p in patterns)
|
|
|
|
|
|
# 素材禁区规则表：(模式列表, 违规描述)
|
|
_FORBIDDEN_MATERIAL_RULES: list[tuple[list[re.Pattern[str]], str]] = [
|
|
(_T1_INSTANT_ACTION_PATTERNS, "T1 违规: 素材包含瞬时动作描述，不适合出题"),
|
|
(_T1_SCOREBOARD_PATTERNS, "T1 违规: 素材包含记分牌时序信息，不适合出题"),
|
|
(_T1_NO_DIALOGUE_CAUSAL_PATTERNS, "T1 违规: 素材缺乏对白因果线索，不适合出题"),
|
|
(_T7_OPTION_REPETITION_PATTERNS, "T7 违规: 素材暗示可能产生重复选项"),
|
|
(_T7_COUNTING_AMBIGUITY_PATTERNS, "T7 违规: 素材包含计数边界含糊描述"),
|
|
]
|
|
|
|
|
|
def check_forbidden_material(source_nodes_text: str, task_type: str) -> list[str]:
|
|
"""出题禁区：检测 T1 类素材和 T7 噪声模式。
|
|
|
|
T1 类素材（不适合出题的内容）:
|
|
- 瞬时动作：画面一闪而过，无法稳定观察
|
|
- 记分牌时序：依赖数字变化的时序信息
|
|
- 无对白因果：缺乏语言线索的因果推理
|
|
|
|
T7 噪声模式（选项质量问题）:
|
|
- 选项重复：多个选项表述相同
|
|
- 计数边界口径含糊：数量描述不确定
|
|
|
|
参数:
|
|
source_nodes_text: 来源节点的拼接文本。
|
|
task_type: 题型名称（用于上下文感知检测）。
|
|
|
|
返回:
|
|
违规描述列表，空列表表示通过。
|
|
"""
|
|
return [
|
|
message
|
|
for patterns, message in _FORBIDDEN_MATERIAL_RULES
|
|
if _match_any(source_nodes_text, patterns)
|
|
]
|
|
|
|
|
|
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,
|
|
)
|