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

chore: snapshot in-progress question-gen work before preflight fixes

Browse Source
This commit is contained in:
iomgaa
2026-07-16 04:12:21 -04:00
parent 11a5545f57
commit a4c429b247
39 changed files with 738 additions and 283 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit/test_search_summarizer.py
+35 -54
View File
@@ -219,11 +219,7 @@ class TestCheckAnchors:
def test_no_info_statement_not_counted(self) -> None:
"""声明句"未包含…相关…信息"不计入 n_assertions。"""
anchor_map = {"s1": "行1"}
summary = (
"[相关信息]\n"
"- 该节点未包含与问题直接相关的信息\n"
"- 关键发现(s1)"
)
summary = "[相关信息]\n- 该节点未包含与问题直接相关的信息\n- 关键发现(s1)"
_, stats = check_anchors(summary, anchor_map)
assert stats["n_assertions"] == 1 # 声明句不计
assert stats["n_anchored"] == 1
@@ -271,9 +267,7 @@ class TestAssembleAnchoredOutput:
"""ids_expand 模式：保留行号 + 附加引文段。"""
anchor_map = {"s1": "第一行内容", "s2": "第二行内容"}
summary = "关键发现(s1,s2)"
result, stats = assemble_anchored_output(
summary, anchor_map, "ids_expand"
)
result, stats = assemble_anchored_output(summary, anchor_map, "ids_expand")
assert "(s1,s2)" in result
assert "[引文]" in result
assert 's1: "第一行内容"' in result
@@ -284,9 +278,7 @@ class TestAssembleAnchoredOutput:
"""expand_only 模式：剥除行号 + 附加引文段。"""
anchor_map = {"s1": "第一行内容"}
summary = "关键发现(s1)"
result, stats = assemble_anchored_output(
summary, anchor_map, "expand_only"
)
result, stats = assemble_anchored_output(summary, anchor_map, "expand_only")
assert "(s1)" not in result
assert "[引文]" in result
assert 's1: "第一行内容"' in result
@@ -297,21 +289,15 @@ class TestAssembleAnchoredOutput:
anchor_map = {f"s{i}": f"行{i}" for i in range(1, 10)}
refs = ",".join(f"s{i}" for i in range(1, 10))
summary = f"发现({refs})"
result, stats = assemble_anchored_output(
summary, anchor_map, "ids_expand"
)
result, stats = assemble_anchored_output(summary, anchor_map, "ids_expand")
assert stats["n_expanded"] == 5
def test_max_chars_cap(self) -> None:
"""总字符超过 800 时截断。"""
anchor_map = {
f"s{i}": "A" * 300 for i in range(1, 6)
}
anchor_map = {f"s{i}": "A" * 300 for i in range(1, 6)}
refs = ",".join(f"s{i}" for i in range(1, 6))
summary = f"发现({refs})"
result, stats = assemble_anchored_output(
summary, anchor_map, "ids_expand"
)
result, stats = assemble_anchored_output(summary, anchor_map, "ids_expand")
# 300 字符原文 + 前缀 ≈ 310+ 每条，800 / 310 ≈ 2 条
assert stats["n_expanded"] < 5
@@ -319,9 +305,7 @@ class TestAssembleAnchoredOutput:
"""单行超 200 字符截断并标记 n_trunc。"""
anchor_map = {"s1": "A" * 250}
summary = "发现(s1)"
result, stats = assemble_anchored_output(
summary, anchor_map, "ids_expand"
)
result, stats = assemble_anchored_output(summary, anchor_map, "ids_expand")
assert stats["n_trunc"] == 1
assert "…" in result
@@ -388,10 +372,12 @@ class TestSummarizeNode:
async def test_anchor_mode(self, prompts_dir: Path) -> None:
"""锚模式：check_anchors + assemble。"""
anchor_map = {"s1": "第一行", "s2": "第二行"}
llm = FakeLLMProvider([
"[相关信息]\n- 关键发现(s1)\n- 补充(s2)",
"核实通过",
])
llm = FakeLLMProvider(
[
"[相关信息]\n- 关键发现(s1)\n- 补充(s2)",
"核实通过",
]
)
result = await summarize_node(
llm,
"带行号的内容",
@@ -409,10 +395,12 @@ class TestSummarizeNode:
"""锚模式 stats_sink 回调接收完整统计。"""
anchor_map = {"s1": "第一行"}
collected: list[dict] = []
llm = FakeLLMProvider([
"[相关信息]\n- 关键发现(s1)",
"核实通过",
])
llm = FakeLLMProvider(
[
"[相关信息]\n- 关键发现(s1)",
"核实通过",
]
)
await summarize_node(
llm,
"内容",
@@ -471,9 +459,7 @@ class TestSummarizeChildren:
{"id": "n2", "time_range": (30.0, 60.0), "summary": "中间"},
]
llm = FakeLLMProvider(["相关性标注结果", "核实通过"])
result = await summarize_children(
llm, children_info, "问题", prompts_dir
)
result = await summarize_children(llm, children_info, "问题", prompts_dir)
assert "相关性标注结果" in result
assert "[核实] 核实通过" in result
@@ -484,25 +470,19 @@ class TestSummarizeChildren:
{"id": "n1", "time_range": (0.0, 30.0), "summary": "开头"},
]
llm = FailingLLMProvider("网络错误")
result = await summarize_children(
llm, children_info, "问题", prompts_dir
)
result = await summarize_children(llm, children_info, "问题", prompts_dir)
assert "n1" in result
assert "0-30s" in result
assert "开头" in result
@pytest.mark.asyncio()
async def test_verify_failure_returns_extract_only(
self, prompts_dir: Path
) -> None:
async def test_verify_failure_returns_extract_only(self, prompts_dir: Path) -> None:
"""核实轮失败仍返回提取结果。"""
children_info = [
{"id": "n1", "time_range": (0.0, 30.0), "summary": "开头"},
]
llm = FailOnNthLLMProvider(["标注结果"], fail_on=2)
result = await summarize_children(
llm, children_info, "问题", prompts_dir
)
result = await summarize_children(llm, children_info, "问题", prompts_dir)
assert "标注结果" in result
@@ -513,19 +493,22 @@ class TestSummarizeNodesBatch:
async def test_batch_normal(self, prompts_dir: Path) -> None:
"""并发三个节点，结果顺序与输入一致。"""
# 每个节点需要 2 轮 LLM 调用（提取 + 核实）
llm = FakeLLMProvider([
"摘要A", "核实A",
"摘要B", "核实B",
"摘要C", "核实C",
])
llm = FakeLLMProvider(
[
"摘要A",
"核实A",
"摘要B",
"核实B",
"摘要C",
"核实C",
]
)
items = [
("n1", "内容1", "extra1"),
("n2", "内容2", "extra2"),
("n3", "内容3", "extra3"),
]
results = await summarize_nodes_batch(
llm, items, "问题", prompts_dir
)
results = await summarize_nodes_batch(llm, items, "问题", prompts_dir)
assert len(results) == 3
assert results[0][0] == "n1"
assert results[1][0] == "n2"
@@ -538,7 +521,5 @@ class TestSummarizeNodesBatch:
async def test_batch_empty(self, prompts_dir: Path) -> None:
"""空列表返回空结果。"""
llm = FakeLLMProvider([])
results = await summarize_nodes_batch(
llm, [], "问题", prompts_dir
)
results = await summarize_nodes_batch(llm, [], "问题", prompts_dir)
assert results == []