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

feat(tree): 质量校验 — 交叉验证 entities/visible_text

Browse Source 
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
iomgaa
2026-07-07 01:57:32 -04:00
parent fb6f9964d8
commit 12f20493c1
2 changed files with 447 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/tree/verify.py
+291
View File
@@ -0,0 +1,291 @@
"""质量校验模块：交叉验证树节点 Card 字段与子节点证据。
验证策略:
- L2 entities: 仅保留在子 L3 文本语料中模糊匹配到的实体。
- L2 visible_text: 仅保留在子 L3 visible_text 中出现的条目。
- L1 visible_text: 仅保留在后代 L2/L3 visible_text 中出现的条目。
- L1 key_entities: 仅保留在后代 L2/L3 文本语料中模糊匹配到的实体。
Card 为 frozen dataclass，无法原地修改——移除幻觉字段时
创建新 Card 实例并赋值给 node.card（Node 非 frozen）。
"""
from __future__ import annotations
import string
from dataclasses import dataclass
from loguru import logger
from app.tree.index import (
L1Card,
L1Node,
L2Card,
L2Node,
TreeIndex,
)
# ---------------------------------------------------------------------------
# 校验统计
# ---------------------------------------------------------------------------
@dataclass
class VerifyStats:
"""校验统计信息。"""
l2_entities_kept: int = 0
l2_entities_removed: int = 0
l2_visible_text_kept: int = 0
l2_visible_text_removed: int = 0
l1_visible_text_kept: int = 0
l1_visible_text_removed: int = 0
l1_key_entities_kept: int = 0
l1_key_entities_removed: int = 0
# ---------------------------------------------------------------------------
# 文本归一化 & 模糊匹配
# ---------------------------------------------------------------------------
def _normalize(text: str) -> str:
"""归一化文本：小写 + 去除标点。
参数:
text: 原始文本。
返回:
归一化后的纯小写无标点字符串。
"""
return text.lower().translate(str.maketrans("", "", string.punctuation))
def fuzzy_match(entity: str | None, corpus: str | None) -> bool:
"""模糊子串匹配：归一化后判断 entity 是否为 corpus 的子串。
参数:
entity: 待匹配的实体文本（None 视为不匹配）。
corpus: 证据语料文本（None 视为空）。
返回:
True 表示匹配成功。
"""
if not entity or not corpus:
return False
return _normalize(str(entity)) in _normalize(str(corpus))
# ---------------------------------------------------------------------------
# 语料收集
# ---------------------------------------------------------------------------
def _collect_l3_text(l2_node: L2Node) -> str:
"""收集 L2 节点所有子 L3 的文本语料。
从每个 L3 子节点的 card 和顶层字段中提取:
frame_summary、visible_text、subtitle。
参数:
l2_node: L2 节点。
返回:
拼接后的文本语料（用换行分隔）。
"""
parts: list[str] = []
for l3 in l2_node.children:
parts.append(l3.card.frame_summary)
parts.extend(l3.card.visible_text)
if l3.subtitle:
parts.append(l3.subtitle)
return "\n".join(parts)
def _collect_descendant_visible_text(l1_node: L1Node) -> str:
"""收集 L1 节点所有后代（L2/L3）的 visible_text。
参数:
l1_node: L1 节点。
返回:
所有后代 visible_text 拼接后的文本（用换行分隔）。
"""
parts: list[str] = []
for l2 in l1_node.children:
parts.extend(l2.card.visible_text)
for l3 in l2.children:
parts.extend(l3.card.visible_text)
return "\n".join(parts)
def _collect_descendant_text_corpus(l1_node: L1Node) -> str:
"""收集 L1 节点所有后代（L2/L3）的完整文本语料。
用于 L1 key_entities 的交叉验证，范围包括
L2/L3 的所有文本字段（frame_summary、visible_text、subtitle 等）。
参数:
l1_node: L1 节点。
返回:
所有后代文本语料拼接后的文本（用换行分隔）。
"""
parts: list[str] = []
for l2 in l1_node.children:
parts.append(l2.card.event_description)
parts.extend(l2.card.entities)
parts.extend(l2.card.visible_text)
for l3 in l2.children:
parts.append(l3.card.frame_summary)
parts.extend(l3.card.visible_text)
if l3.subtitle:
parts.append(l3.subtitle)
return "\n".join(parts)
# ---------------------------------------------------------------------------
# 主校验函数
# ---------------------------------------------------------------------------
def verify_tree(index: TreeIndex) -> VerifyStats:
"""交叉验证视频树的 Card 字段与子节点证据，原地替换不合格的 Card。
Cards 为 frozen dataclass，移除幻觉字段时创建新 Card 实例
并赋值给 node.card。
参数:
index: 树索引（会被原地修改）。
返回:
VerifyStats 校验统计。
"""
stats = VerifyStats()
for l1 in index.roots:
# Phase 1: L2 字段验证
for l2 in l1.children:
_verify_l2(l2, stats)
# Phase 2: L1 字段验证
_verify_l1(l1, stats)
logger.info(
"verify_tree: source={} "
"l2_ent_kept={} l2_ent_rm={} "
"l2_vt_kept={} l2_vt_rm={} "
"l1_vt_kept={} l1_vt_rm={} "
"l1_ke_kept={} l1_ke_rm={}",
index.metadata.source_path,
stats.l2_entities_kept,
stats.l2_entities_removed,
stats.l2_visible_text_kept,
stats.l2_visible_text_removed,
stats.l1_visible_text_kept,
stats.l1_visible_text_removed,
stats.l1_key_entities_kept,
stats.l1_key_entities_removed,
)
return stats
def _verify_l2(l2: L2Node, stats: VerifyStats) -> None:
"""校验单个 L2 节点的 entities 和 visible_text。
参数:
l2: L2 节点（card 可能被替换）。
stats: 统计对象（原地累加）。
"""
corpus = _collect_l3_text(l2)
old_card = l2.card
# entities: 模糊匹配过滤
kept_entities = [e for e in old_card.entities if fuzzy_match(e, corpus)]
stats.l2_entities_kept += len(kept_entities)
stats.l2_entities_removed += len(old_card.entities) - len(kept_entities)
# visible_text: 子 L3 visible_text 中必须存在
l3_visible = _collect_l3_visible_text_set(l2)
kept_vt = [vt for vt in old_card.visible_text if _text_in_set(vt, l3_visible)]
stats.l2_visible_text_kept += len(kept_vt)
stats.l2_visible_text_removed += len(old_card.visible_text) - len(kept_vt)
# 创建新 Card 替换（frozen dataclass）
l2.card = L2Card(
event_description=old_card.event_description,
entities=kept_entities,
actions=old_card.actions,
action_subjects=old_card.action_subjects,
visible_text=kept_vt,
spatial_relations=old_card.spatial_relations,
state_changes=old_card.state_changes,
)
def _verify_l1(l1: L1Node, stats: VerifyStats) -> None:
"""校验单个 L1 节点的 visible_text 和 key_entities。
参数:
l1: L1 节点（card 可能被替换）。
stats: 统计对象（原地累加）。
"""
old_card = l1.card
# visible_text: 必须出现在后代 L2/L3 visible_text 中
descendant_vt = _collect_descendant_visible_text(l1)
kept_vt = [vt for vt in old_card.visible_text if fuzzy_match(vt, descendant_vt)]
stats.l1_visible_text_kept += len(kept_vt)
stats.l1_visible_text_removed += len(old_card.visible_text) - len(kept_vt)
# key_entities: 交叉验证后代文本语料
descendant_corpus = _collect_descendant_text_corpus(l1)
kept_ke = [ke for ke in old_card.key_entities if fuzzy_match(ke, descendant_corpus)]
stats.l1_key_entities_kept += len(kept_ke)
stats.l1_key_entities_removed += len(old_card.key_entities) - len(kept_ke)
# 创建新 Card 替换（frozen dataclass）
l1.card = L1Card(
scene_summary=old_card.scene_summary,
main_setting=old_card.main_setting,
key_entities=kept_ke,
main_actions=old_card.main_actions,
topic_keywords=old_card.topic_keywords,
visible_text=kept_vt,
temporal_flow=old_card.temporal_flow,
)
# ---------------------------------------------------------------------------
# 辅助函数
# ---------------------------------------------------------------------------
def _collect_l3_visible_text_set(l2: L2Node) -> set[str]:
"""收集 L2 下所有 L3 子节点的 visible_text 归一化集合。
参数:
l2: L2 节点。
返回:
归一化后的 visible_text 集合。
"""
result: set[str] = set()
for l3 in l2.children:
for vt in l3.card.visible_text:
result.add(_normalize(vt))
return result
def _text_in_set(text: str, normalized_set: set[str]) -> bool:
"""检查文本归一化后是否存在于集合中。
参数:
text: 待检查文本。
normalized_set: 归一化后的文本集合。
返回:
True 表示匹配成功。
"""
return _normalize(text) in normalized_set
tests/unit/test_verify.py
+156
View File
@@ -0,0 +1,156 @@
"""质量校验模块单元测试。"""
from __future__ import annotations
from app.tree.index import (
IndexMeta,
L1Card,
L1Node,
L2Card,
L2Node,
L3Card,
L3Node,
TreeIndex,
)
from app.tree.verify import VerifyStats, _normalize, fuzzy_match, verify_tree
class TestNormalize:
def test_lowercase(self):
assert _normalize("Hello World") == "hello world"
def test_strip_punctuation(self):
assert _normalize("Hello, World!") == "hello world"
def test_empty(self):
assert _normalize("") == ""
class TestFuzzyMatch:
def test_exact_match(self):
assert fuzzy_match("hello", "hello world")
def test_case_insensitive(self):
assert fuzzy_match("Hello", "say hello world")
def test_no_match(self):
assert not fuzzy_match("xyz", "hello world")
def test_none_entity(self):
assert not fuzzy_match(None, "hello")
def test_none_corpus(self):
assert not fuzzy_match("hello", None)
class TestVerifyTree:
def _make_tree(self):
"""构造一棵树，L2 有混合实体（有出处/无出处）。"""
l3_0 = L3Node(
id="l1_0_l2_0_l3_0",
card=L3Card(
frame_summary="一个运动员在跑步",
visible_entities=["运动员", "跑道"],
ongoing_actions=["跑步"],
visible_text=["Nike", "2024"],
spatial_layout="居中",
visual_attributes={},
),
timestamp=1.0,
subtitle="the athlete is running fast",
)
l3_1 = L3Node(
id="l1_0_l2_0_l3_1",
card=L3Card(
frame_summary="观众在欢呼",
visible_entities=["观众"],
ongoing_actions=["欢呼"],
visible_text=["Stadium"],
spatial_layout="广角",
visual_attributes={},
),
timestamp=3.0,
)
l2 = L2Node(
id="l1_0_l2_0",
card=L2Card(
event_description="比赛片段",
entities=["运动员", "裁判", "幻觉实体"], # "裁判"和"幻觉实体"无 L3 出处
actions=["跑步"],
action_subjects=["运动员"],
visible_text=["Nike", "不存在的文字"], # "不存在的文字"无 L3 出处
spatial_relations="",
state_changes=None,
),
time_range=(0.0, 10.0),
children=[l3_0, l3_1],
)
l1 = L1Node(
id="l1_0",
card=L1Card(
scene_summary="体育比赛",
main_setting="体育场",
key_entities=["运动员", "不存在的人"], # "不存在的人"无出处
main_actions=["比赛"],
topic_keywords=["体育"],
visible_text=["Nike", "Ghost"], # "Ghost"无出处
temporal_flow="从左到右",
),
time_range=(0.0, 10.0),
children=[l2],
)
return TreeIndex(metadata=IndexMeta("/test.mp4", "video"), roots=[l1])
def test_removes_ungrounded_l2_entities(self):
index = self._make_tree()
stats = verify_tree(index)
l2 = index.roots[0].children[0]
assert "运动员" in l2.card.entities
assert "幻觉实体" not in l2.card.entities
assert stats.l2_entities_removed >= 1
def test_removes_ungrounded_l2_visible_text(self):
index = self._make_tree()
stats = verify_tree(index)
l2 = index.roots[0].children[0]
assert "Nike" in l2.card.visible_text
assert "不存在的文字" not in l2.card.visible_text
assert stats.l2_visible_text_removed >= 1
def test_removes_ungrounded_l1_visible_text(self):
index = self._make_tree()
stats = verify_tree(index)
l1 = index.roots[0]
assert "Nike" in l1.card.visible_text
assert "Ghost" not in l1.card.visible_text
assert stats.l1_visible_text_removed >= 1
def test_removes_ungrounded_l1_key_entities(self):
index = self._make_tree()
stats = verify_tree(index)
l1 = index.roots[0]
assert "运动员" in l1.card.key_entities
assert "不存在的人" not in l1.card.key_entities
assert stats.l1_key_entities_removed >= 1
def test_preserves_grounded_entities(self):
index = self._make_tree()
verify_tree(index)
l2 = index.roots[0].children[0]
assert "运动员" in l2.card.entities
def test_returns_verify_stats(self):
index = self._make_tree()
stats = verify_tree(index)
assert isinstance(stats, VerifyStats)
total_kept = stats.l2_entities_kept + stats.l1_key_entities_kept
assert total_kept > 0
def test_frozen_card_replaced(self):
"""验证 Card 被替换为新实例（frozen dataclass 不能原地修改）。"""
index = self._make_tree()
old_l2_card = index.roots[0].children[0].card
verify_tree(index)
new_l2_card = index.roots[0].children[0].card
# Card should be a different object if anything was removed
assert old_l2_card is not new_l2_card