diff --git a/kb_workflow/srt/semantic_engine.py b/kb_workflow/srt/semantic_engine.py index ee5aef8..5b86f26 100644 --- a/kb_workflow/srt/semantic_engine.py +++ b/kb_workflow/srt/semantic_engine.py @@ -7,6 +7,7 @@ - 语义边界检测 - 内容完整性保护 - 主题标题生成 +- 可扩展的分段策略架构 """ from __future__ import annotations @@ -34,6 +35,288 @@ class AIClientProtocol(Protocol): ... +# ============================================================================ +# Segmentation Strategy Pattern (Plan 13-06) +# ============================================================================ + +class SegmentationStrategy(Protocol): + """分段策略接口 + + 定义了分段策略的通用接口，支持未来扩展不同的分段策略。 + + Examples: + >>> strategy = UnifiedSegmentationStrategy() + >>> segments = strategy.segment(subtitle_segments, metadata) + + Note: + 遵守用户决策：Phase 13 实现统一策略，不区分视频类型。 + """ + + def segment( + self, + segments: List, + metadata: dict + ) -> List[str]: + """对字幕分段进行语义分段 + + Args: + segments: 字幕分段列表（来自 parser.py 的 Segment 对象） + metadata: 视频元数据（title, duration 等） + + Returns: + 段落文本列表 + """ + ... + + +class UnifiedSegmentationStrategy: + """统一分段策略 + + 适用于所有视频类型（知识类、故事类、访谈类等）。 + 视频类型仅用于 Prompt 上下文，不改变分段策略。 + + 遵守用户决策：无类型区分，使用统一 Prompt。 + + Attributes: + ai_client: AI 客户端（实现 AIClientProtocol） + fallback_segmenter: 规则分段器（用于降级） + """ + + def __init__( + self, + ai_client: Optional[AIClientProtocol] = None, + fallback_segmenter: Optional['SemanticSegmenter'] = None + ): + """初始化统一分段策略 + + Args: + ai_client: AI 客户端实例，如果为 None 则尝试创建 SimpleAIClient + fallback_segmenter: 降级时使用的规则分段器 + """ + self.ai_client = ai_client + self.fallback_segmenter = fallback_segmenter + + # Lazy AI Client 初始化（与 ASRCorrector 相同模式） + if self.ai_client is None: + try: + from kb_workflow.deep_analyzer import SimpleAIClient + self.ai_client = SimpleAIClient() + if not self.ai_client.api_key: + logger.info("AI API key not configured") + self.ai_client = None + except Exception as e: + logger.warning(f"Failed to initialize AI client: {e}") + self.ai_client = None + + # 初始化降级分段器 + if self.fallback_segmenter is None: + from kb_workflow.srt.segmenter import SemanticSegmenter + self.fallback_segmenter = SemanticSegmenter() + + def segment( + self, + segments: List, + metadata: dict + ) -> List[str]: + """对字幕分段进行语义分段 + + 使用统一的 LLM Prompt 进行语义分段，不区分视频类型。 + 视频类型仅用于 Prompt 上下文。 + + Args: + segments: 字幕分段列表（来自 parser.py 的 Segment 对象） + metadata: 视频元数据（title, duration 等） + + Returns: + 段落文本列表 + """ + # AI 不可用，直接使用规则分段 + if self.ai_client is None: + logger.info("AI client unavailable, using fallback segmenter") + return self.fallback_segmenter.segment(segments) + + # 合并字幕文本 + content = " ".join(seg.text for seg in segments) + + # 构建 Prompt + prompt = self._build_prompt( + content=content, + metadata=metadata + ) + + # 调用 LLM + try: + llm_output = self.ai_client.generate( + prompt, + max_tokens=8000, + disable_reasoning=True # 分段不需要思考模式 + ) + + # 解析和验证 + video_segments = self._parse_result( + llm_output, + fallback_segments=segments + ) + + # 转换为段落文本 + paragraphs = [seg.content for seg in video_segments] + + logger.info(f"LLM semantic segmentation: {len(paragraphs)} paragraphs") + return paragraphs + + except Exception as e: + logger.error(f"LLM segmentation failed: {e}, using fallback") + return self.fallback_segmenter.segment(segments) + + def _build_prompt(self, content: str, metadata: dict) -> str: + """构建语义分段 Prompt + + Args: + content: 字幕内容 + metadata: 视频元数据 + + Returns: + 构建好的 Prompt 字符串 + """ + # 推断视频类型 + video_type = self._infer_video_type(metadata) + title = metadata.get("title", "") or "" + + # 构建 Prompt + total_chars = len(content) + prompt = f"""你是一个专业的视频内容分析师。请将以下字幕内容按语义分段。 + +【视频信息】 +类型：{video_type} +标题：{title} +总字数：{total_chars}字 + +【字幕内容】 +{content} + +【分段要求】 +1. 建议将内容分为 5-8 个段落（可根据语义完整性灵活调整） +2. 每个段落约 1000-2000 字（为保持语义完整性可以突破此范围） +3. 在话题转换处创建新分段，不要混入不同话题的内容 +4. **重要**：保持句子完整性，分段边界应放在自然停顿处（句号。、问号？、感叹号！、省略号……等标点符号后） + - 基于语义停顿判断，不仅依赖标点符号 + - 保持内容的连贯性和可读性 +5. 保持论点完整性，不要在前提-结论结构中间截断 +6. 为主题/章节类型的分段生成简洁标题（5-15字），论点/观点类型分段不加标题 + +【输出格式】 +必须返回以下 JSON 格式（不要添加任何额外文本）： +```json +{{ + "video_type": "{video_type}", + "segments": [ + {{ + "title": "分段标题（如果此分段不需要标题则设为 null）", + "content": "分段内容", + "start_index": 0, + "end_index": 123 + }} + ], + "actual_segment_count": 6 +}} +``` + +【注意事项】 +- 优先保证语义完整性，不要为了凑数而强行截断 +- 标题应简洁准确，如「机器学习概述」「梯度下降原理」 +- 如果某个分段无法生成合适的标题（如过渡性内容），将 title 设为 null""" + return prompt + + def _infer_video_type(self, metadata: dict) -> str: + """从元数据推断视频类型 + + Args: + metadata: 视频元数据，包含 title, category, tags 等字段 + + Returns: + 推断出的视频类型，默认为"知识类" + + Examples: + >>> strategy = UnifiedSegmentationStrategy() + >>> strategy._infer_video_type({"title": "Python教程"}) + '知识类' + >>> strategy._infer_video_type({"category": "影视"}) + '故事类' + """ + if not metadata: + return DEFAULT_VIDEO_TYPE + + # 提取所有文本字段 + title = metadata.get("title", "") or "" + category = metadata.get("category", "") or "" + tags = metadata.get("tags", []) or [] + description = metadata.get("description", "") or "" + + # 合并所有文本用于匹配 + all_text = " ".join([title, category, description] + list(tags)) + + # 如果没有有效文本，返回默认类型 + if not all_text or all_text.isspace(): + return DEFAULT_VIDEO_TYPE + + # 按优先级匹配类型关键词 + for video_type, keywords in VIDEO_TYPE_KEYWORDS.items(): + for keyword in keywords: + if keyword in all_text: + logger.debug( + f"Detected video type '{video_type}' " + f"from keyword '{keyword}'" + ) + return video_type + + # 未匹配到任何类型，返回默认 + logger.debug("No specific video type detected, using default") + return DEFAULT_VIDEO_TYPE + + def _parse_result( + self, + llm_output: str, + fallback_segments: List + ) -> List[VideoSegment]: + """解析 LLM 输出并验证 + + Args: + llm_output: LLM 原始输出 + fallback_segments: 降级时使用的原始分段 + + Returns: + VideoSegment 对象列表 + + Raises: + Exception: 解析或验证失败时 + """ + try: + # 提取 JSON + data = extract_json_from_llm_response(llm_output) + + # Pydantic 验证 + result = SegmentationResult(**data) + + logger.info( + f"LLM 分段成功: 类型={result.video_type}, " + f"分段数={result.actual_segment_count}" + ) + + return result.segments + + except (json.JSONDecodeError, ValidationError) as e: + logger.error(f"LLM 分段结果验证失败: {e}") + logger.error(f"原始 LLM 输出: {llm_output[:500]}...") + + # 降级到规则分段 + logger.info("降级到规则分段") + raise + + except Exception as e: + logger.error(f"未知错误: {e}") + raise + + # ============================================================================ # Pydantic Models # ============================================================================ @@ -207,161 +490,50 @@ DEFAULT_VIDEO_TYPE = "知识类" # ============================================================================ class SemanticSegmentationEngine: - """语义分段引擎 + """语义分段引擎（使用策略架构） - 使用 LLM 实现智能视频字幕语义分段。 - - 功能： - - 识别视频类型（从元数据推断） - - 构建包含类型信息的 Prompt - - 调用 LLM 进行语义分段 - - 验证和返回分段结果 + 使用策略模式实现可扩展的分段架构。 Attributes: - ai_client: AI 客户端（可选，暂未实现完整 LLM 调用） + strategy: 分段策略（实现 SegmentationStrategy 接口） + + Examples: + >>> # 使用默认统一策略 + >>> engine = SemanticSegmentationEngine() + >>> paragraphs = engine.segment(segments, metadata) + + >>> # 使用自定义策略（未来扩展） + >>> custom_strategy = KnowledgeTypeStrategy() + >>> engine = SemanticSegmentationEngine(strategy=custom_strategy) + >>> paragraphs = engine.segment(segments, metadata) + + Note: + Plan 13-06: 实现可扩展策略架构，Phase 13 使用统一策略。 """ def __init__( self, - ai_client: Optional[AIClientProtocol] = None, + strategy: Optional[SegmentationStrategy] = None, fallback_segmenter: Optional['SemanticSegmenter'] = None ): """初始化语义分段引擎 Args: - ai_client: AI 客户端实例，如果为 None 则尝试创建 SimpleAIClient - fallback_segmenter: 降级时使用的规则分段器 + strategy: 分段策略实例，如果为 None 则使用统一策略 + fallback_segmenter: 降级时使用的规则分段器（已弃用，保留兼容性） """ - self.ai_client = ai_client - self.fallback_segmenter = fallback_segmenter + # 默认使用统一策略 + if strategy is None: + self.strategy = UnifiedSegmentationStrategy( + fallback_segmenter=fallback_segmenter + ) + else: + self.strategy = strategy - # Lazy AI Client 初始化（与 ASRCorrector 相同模式） - if self.ai_client is None: - try: - from kb_workflow.deep_analyzer import SimpleAIClient - self.ai_client = SimpleAIClient() - if not self.ai_client.api_key: - logger.info("AI API key not configured") - self.ai_client = None - except Exception as e: - logger.warning(f"Failed to initialize AI client: {e}") - self.ai_client = None - - # 初始化降级分段器 - if self.fallback_segmenter is None: - from kb_workflow.srt.segmenter import SemanticSegmenter - self.fallback_segmenter = SemanticSegmenter() - - def _infer_video_type(self, metadata: dict) -> str: - """从元数据推断视频类型 - - Args: - metadata: 视频元数据，包含 title, category, tags 等字段 - - Returns: - 推断出的视频类型，默认为"知识类" - - Examples: - >>> engine = SemanticSegmentationEngine() - >>> engine._infer_video_type({"title": "Python教程"}) - '知识类' - >>> engine._infer_video_type({"category": "影视"}) - '故事类' - """ - if not metadata: - return DEFAULT_VIDEO_TYPE - - # 提取所有文本字段 - title = metadata.get("title", "") or "" - category = metadata.get("category", "") or "" - tags = metadata.get("tags", []) or [] - description = metadata.get("description", "") or "" - - # 合并所有文本用于匹配 - all_text = " ".join([title, category, description] + list(tags)) - - # 如果没有有效文本，返回默认类型 - if not all_text or all_text.isspace(): - return DEFAULT_VIDEO_TYPE - - # 按优先级匹配类型关键词 - for video_type, keywords in VIDEO_TYPE_KEYWORDS.items(): - for keyword in keywords: - if keyword in all_text: - logger.debug( - f"Detected video type '{video_type}' " - f"from keyword '{keyword}'" - ) - return video_type - - # 未匹配到任何类型，返回默认 - logger.debug("No specific video type detected, using default") - return DEFAULT_VIDEO_TYPE - - def _build_prompt(self, content: str, metadata: dict) -> str: - """构建语义分段 Prompt - - Args: - content: 字幕内容 - metadata: 视频元数据 - - Returns: - 构建好的 Prompt 字符串 - - Examples: - >>> engine = SemanticSegmentationEngine() - >>> prompt = engine._build_prompt("字幕内容", {"title": "教程"}) - >>> "视频类型" in prompt or "知识类" in prompt - True - """ - # 推断视频类型 - video_type = self._infer_video_type(metadata) - title = metadata.get("title", "") or "" - - # 构建 Prompt - total_chars = len(content) - prompt = f"""你是一个专业的视频内容分析师。请将以下字幕内容按语义分段。 - -【视频信息】 -类型：{video_type} -标题：{title} -总字数：{total_chars}字 - -【字幕内容】 -{content} - -【分段要求】 -1. 建议将内容分为 5-8 个段落（可根据语义完整性灵活调整） -2. 每个段落约 1000-2000 字（为保持语义完整性可以突破此范围） -3. 在话题转换处创建新分段，不要混入不同话题的内容 -4. **重要**：保持句子完整性，分段边界应放在自然停顿处（句号。、问号？、感叹号！、省略号……等标点符号后） - - 基于语义停顿判断，不仅依赖标点符号 - - 保持内容的连贯性和可读性 -5. 保持论点完整性，不要在前提-结论结构中间截断 -6. 为主题/章节类型的分段生成简洁标题（5-15字），论点/观点类型分段不加标题 - -【输出格式】 -必须返回以下 JSON 格式（不要添加任何额外文本）： -```json -{{ - "video_type": "{video_type}", - "segments": [ - {{ - "title": "分段标题（如果此分段不需要标题则设为 null）", - "content": "分段内容", - "start_index": 0, - "end_index": 123 - }} - ], - "actual_segment_count": 6 -}} -``` - -【注意事项】 -- 优先保证语义完整性，不要为了凑数而强行截断 -- 标题应简洁准确，如「机器学习概述」「梯度下降原理」 -- 如果某个分段无法生成合适的标题（如过渡性内容），将 title 设为 null""" - return prompt + logger.debug( + f"SemanticSegmentationEngine initialized with strategy: " + f"{self.strategy.__class__.__name__}" + ) def segment( self, @@ -370,6 +542,8 @@ class SemanticSegmentationEngine: ) -> List[str]: """对字幕分段进行语义分段 + 使用当前配置的策略进行分段。 + Args: segments: 字幕分段列表（来自 parser.py 的 Segment 对象） metadata: 视频元数据（title, duration 等） @@ -377,83 +551,4 @@ class SemanticSegmentationEngine: Returns: 段落文本列表 """ - # AI 不可用，直接使用规则分段 - if self.ai_client is None: - logger.info("AI client unavailable, using fallback segmenter") - return self.fallback_segmenter.segment(segments) - - # 合并字幕文本 - content = " ".join(seg.text for seg in segments) - - # 构建 Prompt - prompt = self._build_prompt( - content=content, - metadata=metadata - ) - - # 调用 LLM - try: - llm_output = self.ai_client.generate( - prompt, - max_tokens=8000, - disable_reasoning=True # 分段不需要思考模式 - ) - - # 解析和验证 - video_segments = self._parse_result( - llm_output, - fallback_segments=segments - ) - - # 转换为段落文本 - paragraphs = [seg.content for seg in video_segments] - - logger.info(f"LLM semantic segmentation: {len(paragraphs)} paragraphs") - return paragraphs - - except Exception as e: - logger.error(f"LLM segmentation failed: {e}, using fallback") - return self.fallback_segmenter.segment(segments) - - def _parse_result( - self, - llm_output: str, - fallback_segments: List - ) -> List[VideoSegment]: - """解析 LLM 输出并验证 - - Args: - llm_output: LLM 原始输出 - fallback_segments: 降级时使用的原始分段 - - Returns: - VideoSegment 对象列表 - - Raises: - Exception: 解析或验证失败时 - """ - try: - # 提取 JSON - data = extract_json_from_llm_response(llm_output) - - # Pydantic 验证 - result = SegmentationResult(**data) - - logger.info( - f"LLM 分段成功: 类型={result.video_type}, " - f"分段数={result.actual_segment_count}" - ) - - return result.segments - - except (json.JSONDecodeError, ValidationError) as e: - logger.error(f"LLM 分段结果验证失败: {e}") - logger.error(f"原始 LLM 输出: {llm_output[:500]}...") - - # 降级到规则分段 - logger.info("降级到规则分段") - raise - - except Exception as e: - logger.error(f"未知错误: {e}") - raise + return self.strategy.segment(segments, metadata) diff --git a/tests/test_semantic_engine.py b/tests/test_semantic_engine.py index 7390257..5478d89 100644 --- a/tests/test_semantic_engine.py +++ b/tests/test_semantic_engine.py @@ -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