feat(13-06): implement extensible strategy architecture for segmentation
- Define SegmentationStrategy Protocol interface (Plan 13-06) - Implement UnifiedSegmentationStrategy with unified approach for all video types - Update SemanticSegmentationEngine to use strategy pattern - Add comprehensive tests for strategy extensibility (STRAT-01, STRAT-02, STRAT-03) Key changes: - SegmentationStrategy Protocol defines the strategy interface - UnifiedSegmentationStrategy implements unified approach (per user decision) - SemanticSegmentationEngine now uses strategy injection pattern - Tests verify strategy protocol, unified strategy, custom strategy injection, and extensibility The architecture supports future extension (e.g., knowledge-type specific strategies) while Phase 13 uses the unified strategy approach. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -727,3 +727,266 @@ class TestParallelStructurePreservation:
|
||||
# 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
|
||||
|
||||
Reference in New Issue
Block a user