starmen1265
ef4a660faa
- 添加 enable_semantic_segmentation 参数到 SRTProcessor.__init__ - 启用时使用 SemanticSegmentationEngine，禁用时使用 SemanticSegmenter - AI 不可用时自动降级到规则分段 - 更新 _generate_markdown() 支持带标题的 VideoSegment 输出 - 调整步骤顺序：先加载元数据，再进行语义分段 - 更新 UnifiedSegmentationStrategy.segment() 返回 List[VideoSegment] - 添加代理方法到 SemanticSegmentationEngine 以保持向后兼容 - 创建全面的集成测试（9 个测试用例） - 更新现有语义引擎测试以适配新 API Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
1003 lines
37 KiB
Python
1003 lines
37 KiB
Python
"""测试语义分段引擎"""
|
||
from typing import Protocol
|
||
|
||
import pytest
|
||
import json
|
||
from unittest.mock import Mock
|
||
from pydantic import ValidationError
|
||
|
||
from kb_workflow.srt.parser import Segment
|
||
from kb_workflow.srt.semantic_engine import (
|
||
extract_json_from_llm_response,
|
||
VideoSegment,
|
||
SegmentationResult,
|
||
SemanticSegmentationEngine,
|
||
UnifiedSegmentationStrategy
|
||
)
|
||
|
||
|
||
class TestVideoTypeDetection:
|
||
"""测试视频类型识别功能"""
|
||
|
||
def test_video_type_detection_from_title(self):
|
||
"""测试从标题推断视频类型"""
|
||
# 知识类 - 从标题推断
|
||
metadata_edu = {"title": "Python编程教程", "category": "教育"}
|
||
engine = SemanticSegmentationEngine()
|
||
video_type = engine._infer_video_type(metadata_edu)
|
||
assert video_type == "知识类"
|
||
|
||
# 知识类 - 包含"课程"
|
||
metadata_course = {"title": "机器学习课程", "category": ""}
|
||
video_type = engine._infer_video_type(metadata_course)
|
||
assert video_type == "知识类"
|
||
|
||
def test_video_type_detection_from_category(self):
|
||
"""测试从分类推断视频类型"""
|
||
# 故事类 - 从分类推断
|
||
metadata_movie = {"title": "星际穿越", "category": "影视"}
|
||
engine = SemanticSegmentationEngine()
|
||
video_type = engine._infer_video_type(metadata_movie)
|
||
assert video_type == "故事类"
|
||
|
||
# 访谈类 - 从标签推断
|
||
metadata_interview = {"title": "名人访谈", "tags": ["访谈", "对话"]}
|
||
video_type = engine._infer_video_type(metadata_interview)
|
||
assert video_type == "访谈类"
|
||
|
||
def test_video_type_fallback_to_default(self):
|
||
"""测试元数据缺失时使用默认类型"""
|
||
# 空元数据
|
||
engine = SemanticSegmentationEngine()
|
||
video_type = engine._infer_video_type({})
|
||
assert video_type == "知识类" # 默认类型
|
||
|
||
# 无效元数据
|
||
video_type = engine._infer_video_type({"title": "", "category": ""})
|
||
assert video_type == "知识类"
|
||
|
||
|
||
class TestVideoTypeInPrompt:
|
||
"""测试视频类型在 Prompt 中的使用"""
|
||
|
||
def test_video_type_in_prompt(self):
|
||
"""测试 Prompt 包含视频类型信息"""
|
||
engine = SemanticSegmentationEngine()
|
||
metadata = {"title": "机器学习教程", "category": "教育"}
|
||
|
||
prompt = engine._build_prompt("测试内容", metadata)
|
||
|
||
assert "视频类型" in prompt or "知识类" in prompt
|
||
assert "机器学习教程" in prompt
|
||
|
||
def test_prompt_format(self):
|
||
"""测试 Prompt 格式正确"""
|
||
engine = SemanticSegmentationEngine()
|
||
content = "这是测试内容"
|
||
metadata = {"title": "测试视频"}
|
||
|
||
prompt = engine._build_prompt(content, metadata)
|
||
|
||
assert "请分析以下视频字幕" in prompt or "字幕内容" in prompt
|
||
assert content in prompt
|
||
|
||
|
||
class TestPydanticModels:
|
||
"""测试 Pydantic 模型验证"""
|
||
|
||
def test_video_segment_model(self):
|
||
"""测试 VideoSegment 模型"""
|
||
# 创建完整分段
|
||
segment = VideoSegment(
|
||
title="机器学习概述",
|
||
content="机器学习是人工智能的一个分支",
|
||
start_index=1,
|
||
end_index=5
|
||
)
|
||
assert segment.title == "机器学习概述"
|
||
assert segment.content == "机器学习是人工智能的一个分支"
|
||
assert segment.start_index == 1
|
||
assert segment.end_index == 5
|
||
|
||
# 创建无标题分段（title 可选）
|
||
segment_no_title = VideoSegment(
|
||
content="测试内容",
|
||
start_index=1,
|
||
end_index=3
|
||
)
|
||
assert segment_no_title.title is None
|
||
|
||
def test_segmentation_result_model(self):
|
||
"""测试 SegmentationResult 模型"""
|
||
segments = [
|
||
VideoSegment(title="第一章", content="内容1", start_index=1, end_index=5),
|
||
VideoSegment(title="第二章", content="内容2", start_index=6, end_index=10)
|
||
]
|
||
|
||
result = SegmentationResult(
|
||
video_type="知识类",
|
||
segments=segments,
|
||
actual_segment_count=2
|
||
)
|
||
|
||
assert result.video_type == "知识类"
|
||
assert len(result.segments) == 2
|
||
assert result.actual_segment_count == 2
|
||
|
||
def test_model_validation(self):
|
||
"""测试模型字段验证"""
|
||
# 缺少必填字段应该抛出异常
|
||
with pytest.raises(Exception): # Pydantic ValidationError
|
||
VideoSegment(
|
||
title="测试",
|
||
# 缺少 content
|
||
start_index=1,
|
||
end_index=3
|
||
)
|
||
|
||
def test_title_length_validation_fails(self):
|
||
"""测试标题长度验证（不超过 15 字）"""
|
||
with pytest.raises(ValidationError) as exc_info:
|
||
VideoSegment(
|
||
title="这是一个超过十五个字符限制的非常非常长的标题",
|
||
content="内容...",
|
||
start_index=1,
|
||
end_index=100
|
||
)
|
||
assert "标题长度" in str(exc_info.value) or "15" in str(exc_info.value)
|
||
|
||
def test_title_exactly_15_chars_passes(self):
|
||
"""测试标题正好 15 字时应该通过验证"""
|
||
# 使用明确的15字字符串
|
||
title_15 = "一二三四五六七八九十十一二十三"
|
||
segment = VideoSegment(
|
||
title=title_15,
|
||
content="内容...",
|
||
start_index=1,
|
||
end_index=100
|
||
)
|
||
assert len(segment.title) == 15
|
||
assert segment.title == title_15
|
||
|
||
|
||
class TestChineseSentenceBoundary:
|
||
"""测试中文句边界检测功能（Plan 13-05）"""
|
||
|
||
def test_chinese_sentence_boundary_punctuations(self):
|
||
"""Test 1: Prompt 列举所有中文句末标点（。？！……）"""
|
||
engine = SemanticSegmentationEngine()
|
||
|
||
# 构建一个测试 prompt
|
||
content = "测试内容"
|
||
metadata = {"title": "测试视频", "video_type": "知识类"}
|
||
prompt = engine._build_prompt(content, metadata)
|
||
|
||
# 验证 Prompt 包含所有中文句末标点
|
||
assert "。" in prompt or "句号" in prompt
|
||
assert "？" in prompt or "问号" in prompt
|
||
assert "！" in prompt or "感叹号" in prompt
|
||
assert "……" in prompt or "省略号" in prompt
|
||
|
||
def test_prompt_emphasizes_semantic_pause(self):
|
||
"""Test 2: Prompt 强调「自然停顿处」和「语义停顿」"""
|
||
engine = SemanticSegmentationEngine()
|
||
|
||
content = "测试内容"
|
||
metadata = {"title": "测试视频", "video_type": "知识类"}
|
||
prompt = engine._build_prompt(content, metadata)
|
||
|
||
# 验证 Prompt 强调语义停顿
|
||
assert "自然停顿" in prompt or "语义停顿" in prompt
|
||
|
||
def test_prompt_not_only_punctuation(self):
|
||
"""Test 3: Prompt 说明「不仅依赖标点符号」"""
|
||
engine = SemanticSegmentationEngine()
|
||
|
||
content = "测试内容"
|
||
metadata = {"title": "测试视频", "video_type": "知识类"}
|
||
prompt = engine._build_prompt(content, metadata)
|
||
|
||
# 验证 Prompt 说明不仅依赖标点
|
||
assert "不仅" in prompt or "不仅依赖" in prompt or "不仅限于" in prompt
|
||
|
||
def test_sentence_boundary_in_segmentation(self):
|
||
"""Test 4: 分段边界在标点符号后"""
|
||
# 创建 mock AI 客户端
|
||
mock_client = Mock()
|
||
mock_client.api_key = "test_key"
|
||
|
||
# 模拟包含正确句边界的分段响应
|
||
mock_response = json.dumps({
|
||
"video_type": "知识类",
|
||
"segments": [
|
||
{
|
||
"title": "第一段",
|
||
"content": "这是第一句话。这是第二句话？",
|
||
"start_index": 0,
|
||
"end_index": 20
|
||
},
|
||
{
|
||
"title": "第二段",
|
||
"content": "这是第三句话！这是第四句话……",
|
||
"start_index": 20,
|
||
"end_index": 40
|
||
}
|
||
],
|
||
"actual_segment_count": 2
|
||
}, ensure_ascii=False)
|
||
|
||
mock_client.generate.return_value = mock_response
|
||
|
||
# 创建测试数据
|
||
from kb_workflow.srt.parser import Segment
|
||
segments = [
|
||
Segment(1, 0.0, 5.0, "这是第一句话。"),
|
||
Segment(2, 5.0, 10.0, "这是第二句话？"),
|
||
Segment(3, 10.0, 15.0, "这是第三句话！"),
|
||
Segment(4, 15.0, 20.0, "这是第四句话……")
|
||
]
|
||
|
||
# 使用新的策略模式 API
|
||
from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy
|
||
strategy = UnifiedSegmentationStrategy(ai_client=mock_client)
|
||
result = strategy.segment(segments, {"title": "测试", "video_type": "知识类"})
|
||
|
||
# 验证分段边界在标点后
|
||
assert len(result) == 2
|
||
# 第一个分段在问号后结束
|
||
assert result[0].content.strip().endswith("？")
|
||
# 第二个分段在省略号后结束
|
||
assert result[1].content.strip().endswith("……")
|
||
|
||
|
||
# ============================================================================
|
||
# 计划 13-02: 语义边界检测测试
|
||
# ============================================================================
|
||
|
||
class AIClientProtocol(Protocol):
|
||
"""AI 客户端协议"""
|
||
|
||
def generate(self, prompt: str, max_tokens: int = 16000, disable_reasoning: bool = False) -> str:
|
||
"""生成文本"""
|
||
...
|
||
|
||
|
||
class MockAIClient:
|
||
"""模拟 AI 客户端"""
|
||
|
||
def __init__(self, response: str):
|
||
self.response = response
|
||
self.api_key = "mock_key"
|
||
|
||
def generate(self, prompt: str, max_tokens: int = 16000, disable_reasoning: bool = False) -> str:
|
||
"""返回预设的响应"""
|
||
return self.response
|
||
|
||
|
||
class TestTopicBoundaryDetection:
|
||
"""测试话题转换点检测"""
|
||
|
||
def test_topic_boundary_detection(self):
|
||
"""测试 LLM 正确识别话题转换点
|
||
|
||
场景：一段包含两个不同主题的字幕
|
||
- 主题1：机器学习介绍（0-500字）
|
||
- 主题2：深度学习介绍（500-1000字）
|
||
|
||
预期：LLM 在话题转换处创建分段边界
|
||
"""
|
||
# 准备测试数据
|
||
segments = self._create_multi_topic_segments()
|
||
|
||
# 模拟 LLM 响应：正确识别两个主题
|
||
llm_response = json.dumps({
|
||
"video_type": "知识类",
|
||
"segments": [
|
||
{
|
||
"title": "机器学习概述",
|
||
"content": "机器学习是人工智能的一个分支。它使计算机能够从数据中学习。",
|
||
"start_index": 0,
|
||
"end_index": 35
|
||
},
|
||
{
|
||
"title": "深度学习基础",
|
||
"content": "深度学习是机器学习的子领域，使用神经网络模拟人脑。",
|
||
"start_index": 36,
|
||
"end_index": 71
|
||
}
|
||
],
|
||
"actual_segment_count": 2
|
||
}, ensure_ascii=False)
|
||
|
||
mock_client = MockAIClient(llm_response)
|
||
strategy = UnifiedSegmentationStrategy(ai_client=mock_client)
|
||
engine = SemanticSegmentationEngine(strategy=strategy)
|
||
|
||
metadata = {
|
||
"title": "AI技术介绍",
|
||
"video_type": "知识类"
|
||
}
|
||
|
||
paragraphs = engine.segment(segments, metadata)
|
||
|
||
# 验证：结果包含两个段落
|
||
assert len(paragraphs) == 2, f"Expected 2 paragraphs, got {len(paragraphs)}"
|
||
assert "机器学习是人工智能" in paragraphs[0].content, "First paragraph should mention 机器学习"
|
||
assert "深度学习是机器学习" in paragraphs[1].content, "Second paragraph should mention 深度学习"
|
||
|
||
def test_semantic_completeness(self):
|
||
"""测试单个话题内容完整，不混入其他话题
|
||
|
||
场景：LLM 应该保持每个话题的完整性
|
||
- 不应该在话题中间截断
|
||
- 不应该混入不同话题的内容
|
||
"""
|
||
# 准备测试数据
|
||
segments = self._create_single_topic_segments()
|
||
|
||
# 模拟 LLM 响应：单个完整话题
|
||
llm_response = json.dumps({
|
||
"video_type": "知识类",
|
||
"segments": [
|
||
{
|
||
"title": "梯度下降算法",
|
||
"content": "梯度下降是一种优化算法。它通过迭代调整参数来最小化损失函数。梯度下降有三种变体：批量梯度下降、随机梯度下降和小批量梯度下降。",
|
||
"start_index": 0,
|
||
"end_index": 105
|
||
}
|
||
],
|
||
"actual_segment_count": 1
|
||
}, ensure_ascii=False)
|
||
|
||
mock_client = MockAIClient(llm_response)
|
||
strategy = UnifiedSegmentationStrategy(ai_client=mock_client)
|
||
engine = SemanticSegmentationEngine(strategy=strategy)
|
||
|
||
metadata = {
|
||
"title": "优化算法介绍",
|
||
"video_type": "知识类"
|
||
}
|
||
|
||
paragraphs = engine.segment(segments, metadata)
|
||
|
||
# 验证：单个段落包含完整内容
|
||
assert len(paragraphs) == 1, f"Expected 1 paragraph, got {len(paragraphs)}"
|
||
assert "梯度下降是一种优化算法" in paragraphs[0].content
|
||
assert "批量梯度下降" in paragraphs[0].content
|
||
assert "随机梯度下降" in paragraphs[0].content
|
||
assert "小批量梯度下降" in paragraphs[0].content
|
||
|
||
def test_no_mixed_topics(self):
|
||
"""测试分段不混入不同话题的内容
|
||
|
||
验证每个分段只包含一个话题的内容
|
||
"""
|
||
segments = self._create_multi_topic_segments()
|
||
|
||
# 模拟 LLM 响应：确保分段不重叠
|
||
llm_response = json.dumps({
|
||
"video_type": "知识类",
|
||
"segments": [
|
||
{
|
||
"title": "机器学习概述",
|
||
"content": "机器学习是人工智能的一个分支。",
|
||
"start_index": 0,
|
||
"end_index": 20
|
||
},
|
||
{
|
||
"title": "深度学习基础",
|
||
"content": "深度学习是机器学习的子领域。",
|
||
"start_index": 21,
|
||
"end_index": 40
|
||
}
|
||
],
|
||
"actual_segment_count": 2
|
||
}, ensure_ascii=False)
|
||
|
||
mock_client = MockAIClient(llm_response)
|
||
strategy = UnifiedSegmentationStrategy(ai_client=mock_client)
|
||
engine = SemanticSegmentationEngine(strategy=strategy)
|
||
|
||
metadata = {
|
||
"title": "AI技术介绍",
|
||
"video_type": "知识类"
|
||
}
|
||
|
||
paragraphs = engine.segment(segments, metadata)
|
||
|
||
# 验证：分段内容不混入其他话题
|
||
assert len(paragraphs) == 2
|
||
# 第一个段落不应该包含深度学习的内容（除了可能的引用）
|
||
assert "深度学习" not in paragraphs[0].content or "子领域" in paragraphs[0].content
|
||
# 第二个段落可以提到"机器学习"因为它是"深度学习的超类"
|
||
|
||
def _create_multi_topic_segments(self) -> list[Segment]:
|
||
"""创建包含多个话题的字幕分段"""
|
||
return [
|
||
Segment(1, 0.0, 2.0, "机器学习是人工智能的一个分支。"),
|
||
Segment(2, 2.0, 4.0, "它使计算机能够从数据中学习。"),
|
||
Segment(3, 4.0, 6.0, "深度学习是机器学习的子领域，"),
|
||
Segment(4, 6.0, 8.0, "使用神经网络模拟人脑。"),
|
||
]
|
||
|
||
def _create_single_topic_segments(self) -> list[Segment]:
|
||
"""创建包含单个话题的字幕分段"""
|
||
return [
|
||
Segment(1, 0.0, 2.0, "梯度下降是一种优化算法。"),
|
||
Segment(2, 2.0, 4.0, "它通过迭代调整参数来最小化损失函数。"),
|
||
Segment(3, 4.0, 6.0, "梯度下降有三种变体："),
|
||
Segment(4, 6.0, 8.0, "批量梯度下降、随机梯度下降和小批量梯度下降。"),
|
||
]
|
||
|
||
|
||
class TestJSONExtraction:
|
||
"""测试 JSON 提取功能"""
|
||
|
||
def test_extract_json_from_code_block(self):
|
||
"""测试从 ```json ``` 代码块中提取 JSON"""
|
||
response = '''这是分析结果：
|
||
|
||
```json
|
||
{
|
||
"video_type": "知识类",
|
||
"segments": []
|
||
}
|
||
```
|
||
|
||
希望这个结果有帮助。'''
|
||
|
||
result = extract_json_from_llm_response(response)
|
||
assert result["video_type"] == "知识类"
|
||
assert "segments" in result
|
||
|
||
def test_extract_json_from_plain_text(self):
|
||
"""测试从纯 JSON 文本中提取"""
|
||
response = '{"video_type": "知识类", "segments": []}'
|
||
|
||
result = extract_json_from_llm_response(response)
|
||
assert result["video_type"] == "知识类"
|
||
|
||
def test_extract_json_from_mixed_text(self):
|
||
"""测试从混合文本中提取 JSON"""
|
||
response = '分析如下：{"video_type": "知识类", "segments": []} 以上是结果。'
|
||
|
||
result = extract_json_from_llm_response(response)
|
||
assert result["video_type"] == "知识类"
|
||
|
||
def test_extract_json_invalid_response(self):
|
||
"""测试无效响应抛出异常"""
|
||
with pytest.raises(ValueError, match="未找到 JSON 内容|无法从 LLM 响应中提取有效 JSON"):
|
||
extract_json_from_llm_response("这是一段普通文本，没有 JSON")
|
||
|
||
|
||
class TestPydanticValidationExtended:
|
||
"""测试 Pydantic 验证功能（扩展）"""
|
||
|
||
def test_validate_segment_overlap(self):
|
||
"""测试分段不重叠"""
|
||
# 不重叠的分段
|
||
data = {
|
||
"video_type": "知识类",
|
||
"segments": [
|
||
{
|
||
"title": "第一段",
|
||
"content": "第一段内容",
|
||
"start_index": 0,
|
||
"end_index": 10
|
||
},
|
||
{
|
||
"title": "第二段",
|
||
"content": "第二段内容",
|
||
"start_index": 10,
|
||
"end_index": 20
|
||
}
|
||
],
|
||
"actual_segment_count": 2
|
||
}
|
||
|
||
result = SegmentationResult(**data)
|
||
assert len(result.segments) == 2
|
||
# 验证不重叠
|
||
for i in range(len(result.segments) - 1):
|
||
assert result.segments[i].end_index <= result.segments[i + 1].start_index, \
|
||
f"Segment {i} overlaps with segment {i+1}"
|
||
|
||
def test_allow_null_title(self):
|
||
"""测试允许标题为 null"""
|
||
data = {
|
||
"video_type": "知识类",
|
||
"segments": [
|
||
{
|
||
"title": None,
|
||
"content": "这是一段过渡性内容，不需要标题",
|
||
"start_index": 0,
|
||
"end_index": 20
|
||
}
|
||
],
|
||
"actual_segment_count": 1
|
||
}
|
||
|
||
result = SegmentationResult(**data)
|
||
assert result.segments[0].title is None
|
||
|
||
|
||
# =============================================================================
|
||
# Test Content Completeness Protection (COMP-01, COMP-02, COMP-03)
|
||
# =============================================================================
|
||
|
||
class TestSentenceBoundaryProtection:
|
||
"""测试句边界保护功能 - COMP-01: 不在句子中间截断"""
|
||
|
||
def test_sentence_boundary_protection(self):
|
||
"""Test: 分段边界应在标点符号后，不在句子中间截断"""
|
||
from kb_workflow.srt.parser import Segment
|
||
|
||
# Sample segments with clear sentence boundaries
|
||
segments = [
|
||
Segment(id=1, start_time=0.0, end_time=3.0, text="机器学习是人工智能的一个分支。"),
|
||
Segment(id=2, start_time=3.0, end_time=6.0, text="它使用算法来解析数据、从中学习。"),
|
||
Segment(id=3, start_time=6.0, end_time=9.0, text="机器学习的目标是让计算机能够自动学习！"),
|
||
]
|
||
|
||
# Mock AI client
|
||
mock_client = Mock()
|
||
mock_client.api_key = "test-key"
|
||
mock_response = """```json
|
||
{
|
||
"video_type": "知识类",
|
||
"segments": [
|
||
{
|
||
"title": "机器学习概述",
|
||
"content": "机器学习是人工智能的一个分支。它使用算法来解析数据、从中学习。",
|
||
"start_index": 0,
|
||
"end_index": 35
|
||
},
|
||
{
|
||
"title": null,
|
||
"content": "机器学习的目标是让计算机能够自动学习！",
|
||
"start_index": 36,
|
||
"end_index": 55
|
||
}
|
||
],
|
||
"actual_segment_count": 2
|
||
}
|
||
```"""
|
||
mock_client.generate.return_value = mock_response
|
||
|
||
# Create engine and segment
|
||
strategy = UnifiedSegmentationStrategy(ai_client=mock_client)
|
||
engine = SemanticSegmentationEngine(strategy=strategy)
|
||
metadata = {"title": "机器学习介绍", "duration": 9.0}
|
||
paragraphs = engine.segment(segments, metadata)
|
||
|
||
# Verify segmentation occurred
|
||
assert len(paragraphs) == 2
|
||
|
||
# Verify sentence boundaries - each paragraph should end with punctuation
|
||
import re
|
||
sentence_endings = r'[。？！……]$'
|
||
|
||
for para in paragraphs:
|
||
para_content = para.content if hasattr(para, 'content') else para
|
||
para_stripped = para_content.strip()
|
||
assert re.search(sentence_endings, para_stripped), \
|
||
f"段落应以标点符号结尾: {para_content[:50]}..."
|
||
|
||
def test_sentence_boundary_in_prompt(self):
|
||
"""Test: SEGMENTATION_PROMPT 包含句边界保护指令"""
|
||
engine = SemanticSegmentationEngine()
|
||
|
||
# Get the prompt template
|
||
content = "测试内容"
|
||
metadata = {"title": "测试", "duration": 10.0, "video_type": "知识类"}
|
||
prompt = engine._build_prompt(content, metadata)
|
||
|
||
# Verify prompt contains sentence boundary instructions
|
||
assert "句号" in prompt or "。" in prompt
|
||
assert "问号" in prompt or "？" in prompt
|
||
assert "感叹号" in prompt or "！" in prompt
|
||
assert "省略号" in prompt or "……" in prompt
|
||
assert "不要在句子中间截断" in prompt or "保持句子完整性" in prompt
|
||
|
||
|
||
class TestArgumentCompleteness:
|
||
"""测试论点完整性保护 - COMP-02: 不破坏论点完整性"""
|
||
|
||
def test_argument_completeness(self):
|
||
"""Test: 论点结构应保持完整，前提-结论不被拆分"""
|
||
from kb_workflow.srt.parser import Segment
|
||
|
||
# Sample segments with premise-conclusion structures
|
||
segments = [
|
||
Segment(id=1, start_time=0.0, end_time=4.0, text="首先，让我们理解什么是深度学习。"),
|
||
Segment(id=2, start_time=4.0, end_time=8.0, text="深度学习是机器学习的一个子集。"),
|
||
Segment(id=3, start_time=8.0, end_time=12.0, text="因此，深度学习能够处理更复杂的数据模式。"),
|
||
]
|
||
|
||
# Mock AI client
|
||
mock_client = Mock()
|
||
mock_client.api_key = "test-key"
|
||
mock_response = """```json
|
||
{
|
||
"video_type": "知识类",
|
||
"segments": [
|
||
{
|
||
"title": "深度学习简介",
|
||
"content": "首先，让我们理解什么是深度学习。深度学习是机器学习的一个子集。因此，深度学习能够处理更复杂的数据模式。",
|
||
"start_index": 0,
|
||
"end_index": 50
|
||
}
|
||
],
|
||
"actual_segment_count": 1
|
||
}
|
||
```"""
|
||
mock_client.generate.return_value = mock_response
|
||
|
||
# Create engine and segment
|
||
strategy = UnifiedSegmentationStrategy(ai_client=mock_client)
|
||
engine = SemanticSegmentationEngine(strategy=strategy)
|
||
metadata = {"title": "深度学习介绍", "duration": 12.0}
|
||
paragraphs = engine.segment(segments, metadata)
|
||
|
||
# Verify segmentation occurred
|
||
assert len(paragraphs) == 1
|
||
|
||
# Verify argument completeness - premise and conclusion together
|
||
first_para = paragraphs[0].content
|
||
assert "首先" in first_para
|
||
assert "因此" in first_para
|
||
|
||
def test_argument_completeness_in_prompt(self):
|
||
"""Test: SEGMENTATION_PROMPT 包含论点完整性指令"""
|
||
engine = SemanticSegmentationEngine()
|
||
|
||
# Get the prompt template
|
||
content = "测试内容"
|
||
metadata = {"title": "测试", "duration": 10.0, "video_type": "知识类"}
|
||
prompt = engine._build_prompt(content, metadata)
|
||
|
||
# Verify prompt contains argument completeness instructions
|
||
assert "论点完整性" in prompt or "保持论点" in prompt
|
||
assert "前提" in prompt or "结论" in prompt or "不要截断" in prompt
|
||
|
||
|
||
class TestParallelStructurePreservation:
|
||
"""测试并列结构保护 - COMP-03: 并列结构不被错误拆分"""
|
||
|
||
def test_parallel_structure_preserved(self):
|
||
"""Test: 并列结构（首先/其次/最后）应保持完整"""
|
||
from kb_workflow.srt.parser import Segment
|
||
|
||
# Sample segments with parallel structures
|
||
segments = [
|
||
Segment(id=1, start_time=0.0, end_time=4.0, text="首先，我们需要收集数据。"),
|
||
Segment(id=2, start_time=4.0, end_time=8.0, text="其次，我们清洗和预处理数据。"),
|
||
Segment(id=3, start_time=8.0, end_time=12.0, text="最后，我们训练模型并进行评估。"),
|
||
]
|
||
|
||
# Mock AI client
|
||
mock_client = Mock()
|
||
mock_client.api_key = "test-key"
|
||
mock_response = """```json
|
||
{
|
||
"video_type": "教程类",
|
||
"segments": [
|
||
{
|
||
"title": "数据收集",
|
||
"content": "首先，我们需要收集数据。",
|
||
"start_index": 0,
|
||
"end_index": 14
|
||
},
|
||
{
|
||
"title": "数据预处理",
|
||
"content": "其次，我们清洗和预处理数据。",
|
||
"start_index": 15,
|
||
"end_index": 30
|
||
},
|
||
{
|
||
"title": "模型训练",
|
||
"content": "最后，我们训练模型并进行评估。",
|
||
"start_index": 31,
|
||
"end_index": 45
|
||
}
|
||
],
|
||
"actual_segment_count": 3
|
||
}
|
||
```"""
|
||
mock_client.generate.return_value = mock_response
|
||
|
||
# Create engine and segment
|
||
strategy = UnifiedSegmentationStrategy(ai_client=mock_client)
|
||
engine = SemanticSegmentationEngine(strategy=strategy)
|
||
metadata = {"title": "数据处理流程", "duration": 12.0}
|
||
paragraphs = engine.segment(segments, metadata)
|
||
|
||
# Verify segmentation occurred
|
||
assert len(paragraphs) == 3
|
||
|
||
# Verify each parallel point is complete in its segment
|
||
assert "首先" in paragraphs[0].content
|
||
assert "其次" in paragraphs[1].content
|
||
assert "最后" in paragraphs[2].content
|
||
|
||
def test_no_special_handling_for_parallel_in_prompt(self):
|
||
"""Test: SEGMENTATION_PROMPT 不包含并列结构特殊处理
|
||
|
||
遵守用户决策：让 LLM 自然判断
|
||
"""
|
||
engine = SemanticSegmentationEngine()
|
||
|
||
# Get the prompt template
|
||
content = "测试内容"
|
||
metadata = {"title": "测试", "duration": 10.0, "video_type": "知识类"}
|
||
prompt = engine._build_prompt(content, metadata)
|
||
|
||
# Verify prompt does NOT have specific instructions for parallel structures
|
||
# (per user decision to let LLM naturally judge)
|
||
# The prompt should not have special handling like "保持并列结构完整"
|
||
assert "保持并列结构完整" not in prompt
|
||
assert "并列结构不拆分" not in prompt
|
||
|
||
|
||
# =============================================================================
|
||
# Test Strategy Pattern Architecture (Plan 13-06)
|
||
# =============================================================================
|
||
|
||
class TestStrategyPatternArchitecture:
|
||
"""测试策略模式架构 - STRAT-01, STRAT-02, STRAT-03"""
|
||
|
||
def test_segmentation_strategy_protocol_exists(self):
|
||
"""Test: SegmentationStrategy Protocol 接口定义存在"""
|
||
from kb_workflow.srt.semantic_engine import SegmentationStrategy
|
||
|
||
# 验证 SegmentationStrategy 是 Protocol
|
||
assert hasattr(SegmentationStrategy, '__protocol_attrs__')
|
||
# 验证包含 segment 方法
|
||
assert 'segment' in SegmentationStrategy.__protocol_attrs__
|
||
|
||
def test_unified_segmentation_strategy_exists(self):
|
||
"""Test: UnifiedSegmentationStrategy 统一策略实现存在"""
|
||
from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy
|
||
|
||
# 验证 UnifiedSegmentationStrategy 可以实例化
|
||
strategy = UnifiedSegmentationStrategy()
|
||
assert strategy is not None
|
||
# 验证实现 segment 方法
|
||
assert hasattr(strategy, 'segment')
|
||
assert callable(strategy.segment)
|
||
|
||
def test_semantic_engine_uses_strategy(self):
|
||
"""Test: SemanticSegmentationEngine 使用策略接口"""
|
||
from kb_workflow.srt.semantic_engine import (
|
||
SemanticSegmentationEngine,
|
||
UnifiedSegmentationStrategy,
|
||
SegmentationStrategy
|
||
)
|
||
|
||
# 验证默认使用 UnifiedSegmentationStrategy
|
||
engine = SemanticSegmentationEngine()
|
||
assert isinstance(engine.strategy, UnifiedSegmentationStrategy)
|
||
# 验证策略实现 SegmentationStrategy 接口
|
||
# Protocol 类型检查在运行时是隐式的，我们检查方法签名
|
||
assert hasattr(engine.strategy, 'segment')
|
||
|
||
def test_custom_strategy_can_be_injected(self):
|
||
"""Test: 可以注入自定义策略（验证可扩展性）"""
|
||
from kb_workflow.srt.semantic_engine import (
|
||
SemanticSegmentationEngine,
|
||
SegmentationStrategy
|
||
)
|
||
|
||
# 创建自定义策略
|
||
class CustomSegmentationStrategy:
|
||
"""自定义策略（示例）"""
|
||
|
||
def __init__(self):
|
||
self.call_count = 0
|
||
|
||
def segment(self, segments: list, metadata: dict) -> list:
|
||
"""简单的自定义分段逻辑"""
|
||
self.call_count += 1
|
||
# 简单地每个字幕分段作为一个段落
|
||
return [seg.text for seg in segments]
|
||
|
||
# 注入自定义策略
|
||
custom_strategy = CustomSegmentationStrategy()
|
||
engine = SemanticSegmentationEngine(strategy=custom_strategy)
|
||
|
||
# 验证使用了自定义策略
|
||
assert engine.strategy is custom_strategy
|
||
assert isinstance(engine.strategy, CustomSegmentationStrategy)
|
||
|
||
# 验证调用自定义策略
|
||
test_segments = [
|
||
Segment(1, 0.0, 1.0, "第一句"),
|
||
Segment(2, 1.0, 2.0, "第二句")
|
||
]
|
||
result = engine.segment(test_segments, {"title": "测试"})
|
||
|
||
# 验证自定义策略被调用
|
||
assert custom_strategy.call_count == 1
|
||
assert len(result) == 2
|
||
assert result[0] == "第一句"
|
||
assert result[1] == "第二句"
|
||
|
||
def test_unified_strategy_applies_to_all_video_types(self):
|
||
"""Test: 统一策略适用于所有视频类型（遵守用户决策）"""
|
||
from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy
|
||
|
||
# 创建 mock AI 客户端
|
||
mock_client = Mock()
|
||
mock_client.api_key = "test_key"
|
||
mock_response = json.dumps({
|
||
"video_type": "知识类",
|
||
"segments": [
|
||
{
|
||
"title": "测试段落",
|
||
"content": "测试内容",
|
||
"start_index": 0,
|
||
"end_index": 10
|
||
}
|
||
],
|
||
"actual_segment_count": 1
|
||
}, ensure_ascii=False)
|
||
mock_client.generate.return_value = mock_response
|
||
|
||
# 创建统一策略
|
||
strategy = UnifiedSegmentationStrategy(ai_client=mock_client)
|
||
|
||
# 测试不同视频类型都使用相同的统一策略
|
||
test_segments = [Segment(1, 0.0, 1.0, "测试")]
|
||
|
||
# 知识类
|
||
result_edu = strategy.segment(test_segments, {"title": "Python教程"})
|
||
assert len(result_edu) > 0
|
||
|
||
# 故事类
|
||
result_story = strategy.segment(test_segments, {"title": "电影解说"})
|
||
assert len(result_story) > 0
|
||
|
||
# 访谈类
|
||
result_interview = strategy.segment(test_segments, {"title": "名人访谈"})
|
||
assert len(result_interview) > 0
|
||
|
||
# 验证 AI 客户端被调用相同次数（策略不因类型而变化）
|
||
assert mock_client.generate.call_count == 3
|
||
|
||
def test_strategy_switching_does_not_affect_functionality(self):
|
||
"""Test: 策略切换不影响功能"""
|
||
from kb_workflow.srt.semantic_engine import (
|
||
SemanticSegmentationEngine,
|
||
UnifiedSegmentationStrategy
|
||
)
|
||
|
||
# 创建 mock AI 客户端
|
||
mock_client = Mock()
|
||
mock_client.api_key = "test_key"
|
||
mock_response = json.dumps({
|
||
"video_type": "知识类",
|
||
"segments": [
|
||
{
|
||
"title": "段落1",
|
||
"content": "内容1",
|
||
"start_index": 0,
|
||
"end_index": 5
|
||
},
|
||
{
|
||
"title": "段落2",
|
||
"content": "内容2",
|
||
"start_index": 5,
|
||
"end_index": 10
|
||
}
|
||
],
|
||
"actual_segment_count": 2
|
||
}, ensure_ascii=False)
|
||
mock_client.generate.return_value = mock_response
|
||
|
||
# 测试数据
|
||
test_segments = [
|
||
Segment(1, 0.0, 1.0, "内容1"),
|
||
Segment(2, 1.0, 2.0, "内容2")
|
||
]
|
||
metadata = {"title": "测试"}
|
||
|
||
# 使用显式的 UnifiedSegmentationStrategy（两个都使用 mock）
|
||
strategy1 = UnifiedSegmentationStrategy(ai_client=mock_client)
|
||
engine1 = SemanticSegmentationEngine(strategy=strategy1)
|
||
result1 = engine1.segment(test_segments, metadata)
|
||
|
||
strategy2 = UnifiedSegmentationStrategy(ai_client=mock_client)
|
||
engine2 = SemanticSegmentationEngine(strategy=strategy2)
|
||
result2 = engine2.segment(test_segments, metadata)
|
||
|
||
# 验证两种方式结果一致
|
||
assert len(result1) == len(result2)
|
||
assert result1 == result2
|
||
|
||
def test_strategy_extensibility_for_future_types(self):
|
||
"""Test: 策略架构支持未来扩展（STRAT-03）"""
|
||
from kb_workflow.srt.semantic_engine import (
|
||
SemanticSegmentationEngine,
|
||
SegmentationStrategy
|
||
)
|
||
|
||
# 模拟未来的特定类型策略（如知识类专用策略）
|
||
class FutureKnowledgeTypeStrategy:
|
||
"""未来可能的知识类专用策略（示例）"""
|
||
|
||
def segment(self, segments: list, metadata: dict) -> list:
|
||
"""针对知识类视频的特殊分段逻辑"""
|
||
# 这里可以实现知识类特定的分段逻辑
|
||
# 例如：更细粒度地识别技术术语、概念定义等
|
||
return [seg.text for seg in segments]
|
||
|
||
# 验证可以轻松创建新策略
|
||
future_strategy = FutureKnowledgeTypeStrategy()
|
||
engine = SemanticSegmentationEngine(strategy=future_strategy)
|
||
|
||
# 验证新策略可以正常工作
|
||
test_segments = [Segment(1, 0.0, 1.0, "测试")]
|
||
result = engine.segment(test_segments, {"title": "测试"})
|
||
|
||
assert len(result) > 0
|
||
assert isinstance(engine.strategy, FutureKnowledgeTypeStrategy)
|
||
|
||
# 验证策略接口一致性
|
||
assert hasattr(future_strategy, 'segment')
|
||
assert callable(future_strategy.segment)
|
||
|
||
|
||
class TestStrategyPatternIntegration:
|
||
"""测试策略模式与现有功能的集成"""
|
||
|
||
def test_strategy_preserves_video_type_detection(self):
|
||
"""Test: 策略架构保留视频类型识别功能"""
|
||
from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy
|
||
|
||
strategy = UnifiedSegmentationStrategy()
|
||
|
||
# 验证视频类型识别仍然有效
|
||
video_type = strategy._infer_video_type({"title": "Python教程"})
|
||
assert video_type == "知识类"
|
||
|
||
video_type = strategy._infer_video_type({"title": "电影解说"})
|
||
assert video_type == "故事类"
|
||
|
||
def test_strategy_preserves_prompt_building(self):
|
||
"""Test: 策略架构保留 Prompt 构建功能"""
|
||
from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy
|
||
|
||
strategy = UnifiedSegmentationStrategy()
|
||
|
||
# 验证 Prompt 构建仍然有效
|
||
prompt = strategy._build_prompt(
|
||
content="测试内容",
|
||
metadata={"title": "测试视频"}
|
||
)
|
||
|
||
# 验证 Prompt 包含必要元素
|
||
assert "视频类型" in prompt or "类型：" in prompt
|
||
assert "测试视频" in prompt
|
||
assert "测试内容" in prompt
|
||
assert "5-8" in prompt # 分段数量建议
|
||
|
||
def test_strategy_preserves_fallback_mechanism(self):
|
||
"""Test: 策略架构保留降级机制"""
|
||
from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy
|
||
|
||
# 创建没有 AI 客户端的策略（触发降级）
|
||
strategy = UnifiedSegmentationStrategy(ai_client=None)
|
||
|
||
# 验证降级分段器存在
|
||
assert strategy.fallback_segmenter is not None
|
||
|
||
# 验证降级机制工作
|
||
test_segments = [
|
||
Segment(1, 0.0, 1.0, "第一句"),
|
||
Segment(2, 1.0, 2.0, "第二句")
|
||
]
|
||
result = strategy.segment(test_segments, {"title": "测试"})
|
||
|
||
# 降级到规则分段，应该有结果
|
||
assert len(result) > 0
|