docs(14): create phase 14 plans (5 plans + wave 0 tests)

2026-03-23 20:56:54 +08:00
parent 3c9859c30d
commit cefb79d311
6 changed files with 1939 additions and 0 deletions
@@ -0,0 +1,510 @@
+---
+phase: 14-robustness-cost
+plan: 00
+type: execute
+wave: 0
+depends_on: []
+files_modified:
+  - tests/test_semantic_robustness.py
+  - tests/conftest.py
+  - tests/test_srt_processor.py
+autonomous: true
+requirements:
+  - ROBUST-01
+  - ROBUST-02
+  - ROBUST-03
+  - ROBUST-04
+  - COST-01
+  - COST-02
+  - COST-03
+must_haves:
+  truths:
+    - "所有计划任务都有对应的自动化测试"
+    - "测试框架配置正确，可以运行测试"
+    - "Mock fixtures 可用，模拟 LLM 客户端行为"
+  artifacts:
+    - path: "tests/test_semantic_robustness.py"
+      provides: "鲁棒性和成本控制测试套件"
+      exports: ["test_*"]
+    - path: "tests/conftest.py"
+      provides: "共享测试 fixtures"
+      exports: ["mock_ai_client", "sample_segments", "sample_metadata"]
+    - path: "tests/test_srt_processor.py"
+      provides: "SRTProcessor 集成测试"
+      exports: ["test_semantic_segmentation_toggle"]
+  key_links:
+    - from: "test_semantic_robustness.py"
+      to: "conftest.py"
+      via: "pytest fixtures"
+      pattern: "@pytest.fixture"
+---
+
+<objective>
+创建阶段 14 的测试基础设施，覆盖所有鲁棒性和成本控制功能。
+
+Purpose: 确保所有计划任务都有对应的自动化测试，遵循 TDD 原则。
+Output: 完整的测试套件，包括鲁棒性测试、成本控制测试、集成测试。
+</objective>
+
+<execution_context>
+@/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md
+@/home/starmen1265/.claude/get-shit-done/templates/summary.md
+</execution_context>
+
+<context>
+@.planning/phases/14-robustness-cost/14-CONTEXT.md
+@.planning/phases/14-robustness-cost/14-RESEARCH.md
+@.planning/phases/14-robustness-cost/14-VALIDATION.md
+@.planning/ROADMAP.md
+@.planning/REQUIREMENTS.md
+@.planning/STATE.md
+
+@tests/conftest.py
+@tests/test_semantic_engine.py
+@kb_workflow/srt/semantic_engine.py
+</context>
+
+<interfaces>
+<!-- Key fixtures and test patterns from existing test files -->
+
+From tests/conftest.py (existing fixtures):
+```python
+import pytest
+from unittest.mock import Mock, MagicMock
+from kb_workflow.srt.parser import Segment
+
+@pytest.fixture
+def mock_ai_client():
+    """Mock AI client"""
+    client = Mock()
+    client.api_key = "test_key"
+    client.generate = Mock(return_value="Test response")
+    return client
+
+@pytest.fixture
+def sample_segments():
+    """示例字幕分段"""
+    return [
+        Segment(index=1, start_time=0.0, end_time=1.0, text="第一句"),
+        Segment(index=2, start_time=1.0, end_time=2.0, text="第二句"),
+    ]
+
+@pytest.fixture
+def sample_metadata():
+    """示例视频元数据"""
+    return {
+        "title": "测试视频",
+        "duration": 120,
+        "category": "知识"
+    }
+```
+
+From 14-VALIDATION.md (Wave 0 Requirements):
+```python
+# Wave 0 Requirements
+- [ ] tests/test_semantic_robustness.py — 覆盖 ROBUST-01, ROBUST-02, ROBUST-03, ROBUST-04, COST-01, COST-02
+- [ ] tests/test_srt_processor.py — 覆盖 COST-03（配置开关）
+- [ ] tests/conftest.py — 共享 fixtures（mock LLM client, 样本字幕数据）
+- [ ] 框架安装: 无需安装，pytest 已配置
+
+# Mock Strategy
+使用 unittest.mock.patch 模拟 SimpleAIClient.generate()，测试不同失败场景（网络错误、解析失败、超时等）
+```
+
+From 14-RESEARCH.md (Test Requirements):
+| Req ID | Behavior | Test Type | Automated Command |
+|--------|----------|-----------|-------------------|
+| ROBUST-01 | LLM 调用失败时自动降级 | unit | test_llm_failure_fallback |
+| ROBUST-02 | 解析失败时重试或降级 | unit | test_parse_failure_retry |
+| ROBUST-03 | AI 不可用时使用规则分段 | unit | test_ai_unavailable_fallback |
+| ROBUST-04 | 降级日志记录 | unit | test_fallback_logging |
+| COST-01 | 长文本分块处理 | unit | test_long_text_chunking |
+| COST-02 | 缓存命中避免重复调用 | unit | test_cache_hit |
+| COST-03 | 配置开关控制 | integration | test_semantic_segmentation_toggle |
+</interfaces>
+
+<tasks>
+
+<task type="auto">
+  <name>Task 1: 创建 test_semantic_robustness.py 测试文件</name>
+  <files>tests/test_semantic_robustness.py</files>
+  <action>
+    创建 tests/test_semantic_robustness.py，包含所有鲁棒性和成本控制测试:
+
+    1. 导入依赖:
+       ```python
+       import pytest
+       import json
+       from unittest.mock import Mock, patch, MagicMock
+       from pathlib import Path
+       from requests.exceptions import ConnectionError, Timeout, HTTPError
+
+       from kb_workflow.srt.semantic_engine import (
+           UnifiedSegmentationStrategy,
+           SemanticSegmentationEngine,
+           VideoSegment,
+           SegmentationResult
+       )
+       from kb_workflow.srt.cache import SemanticCache, compute_content_hash
+       ```
+
+    2. 创建测试类和方法（遵循 14-VALIDATION.md 的测试映射）:
+
+    **TestLLMFailureFallback (ROBUST-01)**:
+    - test_llm_failure_fallback: 模拟 LLM 调用失败（网络错误），验证降级到规则分段
+    - test_recoverable_error_retry: 模拟可恢复错误（ConnectionError），验证重试 1 次
+    - test_non_recoverable_error_no_retry: 模拟不可恢复错误（HTTPError 400），验证直接降级
+
+    **TestParseFailureRetry (ROBUST-02, ROBUST-04)**:
+    - test_parse_failure_retry: 模拟 JSON 解析失败（JSONDecodeError），验证重试后降级
+    - test_validation_error_no_retry: 模拟 Pydantic 验证失败（ValidationError），验证直接降级
+    - test_fallback_logging: 验证降级时记录 warning 级别日志，包含错误类型和原因
+
+    **TestAIAvailabilityFallback (ROBUST-03)**:
+    - test_ai_unavailable_fallback: 模拟 AI 客户端未配置（api_key 为空），验证使用规则分段
+    - test_ai_client_none: 模拟 AI 客户端初始化失败，验证使用规则分段
+
+    **TestLongTextChunking (COST-01)**:
+    - test_long_text_chunking: 使用超过 5000 字符的文本，验证分块处理
+    - test_chunk_sentence_boundary: 验证分块在句子边界（标点符号后）分割
+    - test_chunk_index_correction: 验证 VideoSegment.start_index/end_index 相对于原文
+
+    **TestCacheHit (COST-02)**:
+    - test_cache_hit: 第一次调用 LLM，第二次调用从缓存读取
+    - test_cache_key_hash: 验证缓存键使用 SHA256 哈希
+    - test_cache_miss_content_change: 内容变化时缓存未命中
+    - test_cache_file_format: 验证缓存文件格式（JSON + version 字段）
+
+    3. 使用 Mock 模拟 LLM 客户端:
+       ```python
+       @pytest.fixture
+       def mock_ai_client_with_failure():
+           """模拟失败的 AI 客户端"""
+           client = Mock()
+           client.api_key = "test_key"
+           client.generate = Mock(side_effect=ConnectionError("Network error"))
+           return client
+
+       @pytest.fixture
+       def mock_ai_client_with_parse_error():
+           """模拟解析失败的 AI 客户端"""
+           client = Mock()
+           client.api_key = "test_key"
+           # 第一次返回无效 JSON，第二次返回有效 JSON
+           client.generate = Mock(side_effect=[
+               "Invalid JSON",
+               '{"video_type": "知识类", "segments": [], "actual_segment_count": 0}'
+           ])
+           return client
+       ```
+
+    4. 示例测试实现:
+       ```python
+       def test_llm_failure_fallback(sample_segments, sample_metadata):
+           """测试 LLM 调用失败时降级到规则分段"""
+           strategy = UnifiedSegmentationStrategy(
+               ai_client=mock_ai_client_with_failure()
+           )
+
+           result = strategy.segment(sample_segments, sample_metadata)
+
+           # 验证降级到规则分段（返回 VideoSegment 列表）
+           assert isinstance(result, list)
+           assert all(isinstance(seg, VideoSegment) for seg in result)
+           # 规则分段没有标题
+           assert all(seg.title is None for seg in result)
+
+       def test_recoverable_error_retry(sample_segments, sample_metadata):
+           """测试可恢复错误重试 1 次"""
+           client = Mock()
+           client.api_key = "test_key"
+           # 第一次失败，第二次成功
+           client.generate = Mock(side_effect=[
+               ConnectionError("Network error"),
+               '{"video_type": "知识类", "segments": [], "actual_segment_count": 0}'
+           ])
+
+           strategy = UnifiedSegmentationStrategy(ai_client=client)
+           result = strategy.segment(sample_segments, sample_metadata)
+
+           # 验证重试了 2 次
+           assert client.generate.call_count == 2
+       ```
+
+    确保所有测试都是 RED 状态（功能尚未实现）。
+  </action>
+  <verify>
+    <automated>pytest tests/test_semantic_robustness.py -v --tb=no</automated>
+  </verify>
+  <done>
+    - test_semantic_robustness.py 文件创建，包含所有测试
+    - 所有测试都是 RED 状态（预期失败）
+    - 测试覆盖所有需求（ROBUST-01 到 COST-03）
+    - Mock fixtures 正确模拟 LLM 客户端行为
+  </done>
+</task>
+
+<task type="auto">
+  <name>Task 2: 更新 conftest.py 添加新 fixtures</name>
+  <files>tests/conftest.py</files>
+  <action>
+    更新 tests/conftest.py，添加阶段 14 需要的新 fixtures:
+
+    1. 添加长文本样本（用于分块测试）:
+       ```python
+       @pytest.fixture
+       def long_text_sample():
+           """长文本样本（>5000 字符，用于分块测试）"""
+           # 生成一个超过 5000 字符的文本，包含多个句子
+           sentences = ["这是一个测试句子。" * 10] * 15  # 约 6000 字符
+           return " ".join(sentences)
+       ```
+
+    2. 添加模拟失败的 AI 客户端 fixtures:
+       ```python
+       @pytest.fixture
+       def mock_ai_client_network_error():
+           """模拟网络错误的 AI 客户端"""
+           client = Mock()
+           client.api_key = "test_key"
+           client.generate = Mock(side_effect=ConnectionError("Network error"))
+           return client
+
+       @pytest.fixture
+       def mock_ai_client_timeout():
+           """模拟超时的 AI 客户端"""
+           client = Mock()
+           client.api_key = "test_key"
+           client.generate = Mock(side_effect=Timeout("Request timeout"))
+           return client
+
+       @pytest.fixture
+       def mock_ai_client_rate_limit():
+           """模拟限流的 AI 客户端"""
+           response = Mock()
+           response.status_code = 429
+           client = Mock()
+           client.api_key = "test_key"
+           client.generate = Mock(side_effect=HTTPError(response=response))
+           return client
+
+       @pytest.fixture
+       def mock_ai_client_api_error():
+           """模拟 API 错误的 AI 客户端（不可恢复）"""
+           response = Mock()
+           response.status_code = 400
+           client = Mock()
+           client.api_key = "test_key"
+           client.generate = Mock(side_effect=HTTPError(response=response))
+           return client
+
+       @pytest.fixture
+       def mock_ai_client_parse_error():
+           """模拟解析失败的 AI 客户端"""
+           client = Mock()
+           client.api_key = "test_key"
+           # 第一次返回无效 JSON
+           client.generate = Mock(return_value="Invalid JSON response")
+           return client
+
+       @pytest.fixture
+       def mock_ai_client_success():
+           """模拟成功的 AI 客户端"""
+           client = Mock()
+           client.api_key = "test_key"
+           client.generate = Mock(return_value='''{
+               "video_type": "知识类",
+               "segments": [
+                   {
+                       "title": "测试标题",
+                       "content": "测试内容",
+                       "start_index": 0,
+                       "end_index": 10
+                   }
+               ],
+               "actual_segment_count": 1
+           }''')
+           return client
+       ```
+
+    3. 添加缓存相关的 fixtures:
+       ```python
+       @pytest.fixture
+       def temp_cache_dir(tmp_path):
+           """临时缓存目录"""
+           cache_dir = tmp_path / "cache" / "semantic"
+           return cache_dir
+
+       @pytest.fixture
+       def semantic_cache(temp_cache_dir):
+           """语义分段缓存实例"""
+           from kb_workflow.srt.cache import SemanticCache
+           return SemanticCache(cache_dir=temp_cache_dir)
+       ```
+
+    4. 添加长字幕分段样本:
+       ```python
+       @pytest.fixture
+       def long_subtitle_segments():
+           """长字幕分段样本（>5000 字符）"""
+           segments = []
+           for i in range(100):
+               segments.append(Segment(
+                   index=i + 1,
+                   start_time=i * 1.0,
+                   end_time=(i + 1) * 1.0,
+                   text=f"这是第 {i + 1} 句字幕内容。" * 5
+               ))
+           return segments
+       ```
+  </action>
+  <verify>
+    <automated>pytest tests/conftest.py --collect-only</automated>
+  </verify>
+  <done>
+    - conftest.py 添加所有需要的 fixtures
+    - Fixtures 包括: 模拟失败的 AI 客户端、长文本样本、缓存目录
+    - 所有 fixtures 可以正确导入和使用
+  </done>
+</task>
+
+<task type="auto">
+  <name>Task 3: 在 test_srt_processor.py 中添加配置开关测试</name>
+  <files>tests/test_srt_processor.py</files>
+  <action>
+    在 tests/test_srt_processor.py 中添加配置开关测试（COST-03）:
+
+    1. 如果文件不存在，创建基础文件:
+       ```python
+       import pytest
+       from unittest.mock import Mock, patch
+       from pathlib import Path
+
+       from kb_workflow.srt.processor import SRTProcessor
+       from kb_workflow.srt.parser import SRTParser
+       ```
+
+    2. 添加 TestSemanticSegmentationToggle 测试类:
+       ```python
+       class TestSemanticSegmentationToggle:
+           """测试语义分段配置开关 (COST-03)"""
+
+           def test_semantic_segmentation_enabled(self, sample_srt_file):
+               """测试启用语义分段"""
+               with patch('kb_workflow.srt.semantic_engine.SimpleAIClient') as mock_client:
+                   mock_client.return_value = Mock(api_key="test_key")
+
+                   processor = SRTProcessor(
+                       config={},
+                       enable_semantic_segmentation=True
+                   )
+
+                   # 验证创建了 SemanticSegmentationEngine
+                   assert processor.semantic_engine is not None
+                   assert processor.segmenter is None
+
+           def test_semantic_segmentation_disabled(self, sample_srt_file):
+               """测试禁用语义分段（默认）"""
+               processor = SRTProcessor(
+                   config={},
+                   enable_semantic_segmentation=False
+               )
+
+               # 验证使用规则分段器
+               assert processor.semantic_engine is None
+               assert processor.segmenter is not None
+
+           def test_semantic_segmentation_with_cache(self, sample_srt_file):
+               """测试启用语义分段和缓存"""
+               with patch('kb_workflow.srt.semantic_engine.SimpleAIClient') as mock_client:
+                   mock_client.return_value = Mock(api_key="test_key")
+
+                   processor = SRTProcessor(
+                       config={},
+                       enable_semantic_segmentation=True,
+                       enable_cache=True
+                   )
+
+                   # 验证创建了缓存
+                   assert processor.semantic_engine is not None
+                   # 验证策略使用了缓存
+                   assert processor.semantic_engine.strategy.cache is not None
+
+           def test_semantic_segmentation_without_cache(self, sample_srt_file):
+               """测试启用语义分段但禁用缓存"""
+               with patch('kb_workflow.srt.semantic_engine.SimpleAIClient') as mock_client:
+                   mock_client.return_value = Mock(api_key="test_key")
+
+                   processor = SRTProcessor(
+                       config={},
+                       enable_semantic_segmentation=True,
+                       enable_cache=False
+                   )
+
+                   # 验证没有创建缓存
+                   assert processor.semantic_engine is not None
+                   assert processor.semantic_engine.strategy.cache is None
+
+           @pytest.fixture
+           def sample_srt_file(self, tmp_path):
+               """创建示例 SRT 文件"""
+               srt_content = """1
+       00:00:00,000 --> 00:00:01,000
+       第一句字幕
+
+       2
+       00:00:01,000 --> 00:00:02,000
+       第二句字幕
+       """
+               srt_file = tmp_path / "test.srt"
+               srt_file.write_text(srt_content, encoding='utf-8')
+               return srt_file
+       ```
+
+    3. 确保测试是 RED 状态（功能尚未实现）。
+  </action>
+  <verify>
+    <automated>pytest tests/test_srt_processor.py::test_semantic_segmentation_toggle -v --tb=no</automated>
+  </verify>
+  <done>
+    - test_srt_processor.py 添加配置开关测试
+    - 测试覆盖启用/禁用语义分段、启用/禁用缓存
+    - 所有测试都是 RED 状态（预期失败）
+  </done>
+</task>
+
+</tasks>
+
+<verification>
+## 整体验证标准
+
+1. **测试文件创建**: test_semantic_robustness.py 包含所有鲁棒性和成本控制测试
+2. **Fixtures 更新**: conftest.py 添加所有需要的 fixtures
+3. **集成测试**: test_srt_processor.py 添加配置开关测试
+4. **测试状态**: 所有测试都是 RED 状态（功能尚未实现）
+
+运行测试验证文件结构:
+```bash
+# 验证测试文件存在
+ls tests/test_semantic_robustness.py tests/conftest.py tests/test_srt_processor.py
+
+# 验证测试可以收集
+python3 -m pytest tests/test_semantic_robustness.py --collect-only
+python3 -m pytest tests/test_srt_processor.py::test_semantic_segmentation_toggle --collect-only
+
+# 验证测试是 RED 状态
+python3 -m pytest tests/test_semantic_robustness.py -v --tb=no
+```
+</verification>
+
+<success_criteria>
+1. test_semantic_robustness.py 文件创建，包含 6 个测试类
+2. conftest.py 添加 10+ 个新 fixtures
+3. test_srt_processor.py 添加 TestSemanticSegmentationToggle 测试类
+4. 所有测试都是 RED 状态（预期失败）
+5. 测试覆盖所有需求（ROBUST-01 到 COST-03）
+</success_criteria>
+
+<output>
+执行完成后创建 `.planning/phases/14-robustness-cost/14-00-SUMMARY.md`
+</output>
@@ -0,0 +1,184 @@
+---
+phase: 14-robustness-cost
+plan: 01
+type: execute
+wave: 1
+depends_on: []
+files_modified:
+  - kb_workflow/srt/semantic_engine.py
+autonomous: true
+requirements:
+  - ROBUST-01
+  - ROBUST-03
+must_haves:
+  truths:
+    - "LLM 调用失败时系统自动降级到规则分段，不影响整体流程"
+    - "AI 客户端不可用时（未配置 API key）使用规则分段"
+    - "LLM 响应解析失败时降级到规则分段"
+  artifacts:
+    - path: "kb_workflow/srt/semantic_engine.py"
+      provides: "UnifiedSegmentationStrategy.segment() 带重试和降级逻辑"
+      exports: ["segment", "_is_recoverable_error"]
+  key_links:
+    - from: "UnifiedSegmentationStrategy.segment()"
+      to: "规则分段器"
+      via: "fallback_segmenter.segment()"
+      pattern: "self.fallback_segmenter.segment"
+    - from: "UnifiedSegmentationStrategy.segment()"
+      to: "SimpleAIClient.generate()"
+      via: "LLM API 调用"
+      pattern: "self.ai_client.generate"
+---
+
+<objective>
+实现 LLM 调用的优雅降级策略，确保 AI 不可用时系统仍能正常运行。
+
+Purpose: 提高系统稳定性，避免 LLM 故障导致整个处理流程失败。
+Output: 带重试和降级逻辑的 UnifiedSegmentationStrategy.segment() 方法。
+</objective>
+
+<execution_context>
+@/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md
+@/home/starmen1265/.claude/get-shit-done/templates/summary.md
+</execution_context>
+
+<context>
+@.planning/phases/14-robustness-cost/14-CONTEXT.md
+@.planning/phases/14-robustness-cost/14-RESEARCH.md
+@.planning/phases/14-robustness-cost/14-VALIDATION.md
+@.planning/ROADMAP.md
+@.planning/REQUIREMENTS.md
+@.planning/STATE.md
+
+@kb_workflow/srt/semantic_engine.py
+@kb_workflow/srt/corrector.py
+@kb_workflow/deep_analyzer/__init__.py
+</context>
+
+<interfaces>
+<!-- Key types and contracts from semantic_engine.py -->
+
+From kb_workflow/srt/semantic_engine.py:
+```python
+class UnifiedSegmentationStrategy:
+    def __init__(
+        self,
+        ai_client: Optional[AIClientProtocol] = None,
+        fallback_segmenter: Optional['SemanticSegmenter'] = None
+    ):
+        # Lazy AI Client 初始化
+        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
+
+    def segment(
+        self,
+        segments: List,
+        metadata: dict
+    ) -> List['VideoSegment']:
+        # AI 不可用，直接使用规则分段
+        if self.ai_client is None:
+            logger.info("AI client unavailable, using fallback segmenter")
+            paragraphs = self.fallback_segmenter.segment(segments)
+            return [
+                VideoSegment(title=None, content=para, start_index=0, end_index=0)
+                for para in paragraphs
+            ]
+
+        # 调用 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)
+            return video_segments
+        except Exception as e:
+            logger.error(f"LLM segmentation failed: {e}, using fallback")
+            paragraphs = self.fallback_segmenter.segment(segments)
+            return [VideoSegment(title=None, content=para, start_index=0, end_index=0) for para in paragraphs]
+```
+
+From kb_workflow/deep_analyzer/__init__.py (SimpleAIClient 重试模式):
+- SimpleAIClient 已有内置重试逻辑 (max_retries=3)
+- 使用 requests 库进行 HTTP 调用
+- 可能抛出的异常: ConnectionError, Timeout, HTTPError
+</interfaces>
+
+<tasks>
+
+<task type="auto" tdd="true">
+  <name>Task 1: 实现可恢复错误检测和重试逻辑</name>
+  <files>kb_workflow/srt/semantic_engine.py</files>
+  <behavior>
+    - 添加 _is_recoverable_error() 方法判断错误类型
+    - 可恢复错误: ConnectionError, Timeout, HTTPError with status 429/500/502/503/504
+    - 不可恢复错误: 其他 HTTPError (4xx), JSONDecodeError, ValidationError
+    - 重试逻辑: 仅对可恢复错误重试 1 次
+    - 重试延迟: 立即重试（无延迟）
+    - 最后一次失败或不可恢复错误: 降级到规则分段
+  </behavior>
+  <action>
+    在 UnifiedSegmentationStrategy 类中添加:
+
+    1. _is_recoverable_error() 静态方法:
+       - 导入 requests.exceptions (ConnectionError, Timeout, HTTPError)
+       - 判断异常类型和状态码
+       - 返回 bool
+
+    2. 修改 segment() 方法的 LLM 调用部分:
+       - 添加 for 循环 (max_attempts = 2)
+       - 在 try-except 中捕获异常
+       - 检查是否可恢复且还有重试机会
+       - 可恢复: logger.info 重试, continue
+       - 不可恢复或最后一次: logger.warning 降级原因, break
+
+    3. 改进日志记录:
+       - 重试: info 级别, "LLM call failed (attempt 1/2): {error}, retrying..."
+       - 降级: warning 级别, "{error_type}: {error_message}, 降级到规则分段"
+
+    遵守用户决策: 重试 1 次, 仅可恢复错误重试, 立即重试。
+  </action>
+  <verify>
+    <automated>pytest tests/test_semantic_robustness.py::test_llm_failure_fallback -x</automated>
+  </verify>
+  <done>
+    - _is_recoverable_error() 方法正确识别可恢复/不可恢复错误
+    - LLM 调用失败时重试 1 次（可恢复错误）或直接降级（不可恢复错误）
+    - 降级时记录具体的错误类型和消息 (warning 级别)
+    - AI 不可用时 (ai_client is None) 直接使用规则分段，无重试
+  </done>
+</task>
+
+</tasks>
+
+<verification>
+## 整体验证标准
+
+1. **功能验证**: LLM 调用失败时系统仍能返回分段结果（使用规则分段）
+2. **重试验证**: 可恢复错误（网络错误、超时、限流）触发重试
+3. **降级验证**: 不可恢复错误（API 错误、解析失败）直接降级，不重试
+4. **日志验证**: 降级时记录具体原因，便于问题诊断
+
+运行完整测试套件:
+```bash
+python3 -m pytest tests/test_semantic_robustness.py::test_llm_failure_fallback -v -x
+```
+</verification>
+
+<success_criteria>
+1. LLM 调用失败时自动降级到规则分段，不中断处理流程
+2. 可恢复错误（网络、超时、限流）重试 1 次后降级
+3. 不可恢复错误（API 错误、解析失败）直接降级，不浪费配额
+4. AI 客户端未配置时直接使用规则分段，无 LLM 调用
+5. 降级时记录 warning 级别日志，包含错误类型和原因
+6. 所有测试通过
+</success_criteria>
+
+<output>
+执行完成后创建 `.planning/phases/14-robustness-cost/14-01-SUMMARY.md`
+</output>
@@ -0,0 +1,240 @@
+---
+phase: 14-robustness-cost
+plan: 02
+type: execute
+wave: 1
+depends_on: []
+files_modified:
+  - kb_workflow/srt/semantic_engine.py
+autonomous: true
+requirements:
+  - ROBUST-02
+  - ROBUST-04
+must_haves:
+  truths:
+    - "LLM 响应解析失败时自动重试或降级"
+    - "降级过程记录详细日志，包含错误类型和原因"
+    - "Pydantic 验证失败时直接降级（不可恢复错误）"
+  artifacts:
+    - path: "kb_workflow/srt/semantic_engine.py"
+      provides: "_parse_result() 带解析失败处理"
+      exports: ["_parse_result"]
+  key_links:
+    - from: "_parse_result()"
+      to: "extract_json_from_llm_response()"
+      via: "JSON 提取"
+      pattern: "extract_json_from_llm_response"
+    - from: "_parse_result()"
+      to: "SegmentationResult"
+      via: "Pydantic 验证"
+      pattern: "SegmentationResult(**data)"
+---
+
+<objective>
+增强 LLM 响应解析的错误处理，区分解析失败类型并采取不同策略。
+
+Purpose: 提高 LLM 输出格式不稳定时的容错能力，记录详细日志便于诊断。
+Output: 改进的 _parse_result() 方法，带详细的错误日志和重试逻辑。
+</objective>
+
+<execution_context>
+@/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md
+@/home/starmen1265/.claude/get-shit-done/templates/summary.md
+</execution_context>
+
+<context>
+@.planning/phases/14-robustness-cost/14-CONTEXT.md
+@.planning/phases/14-robustness-cost/14-RESEARCH.md
+@.planning/phases/14-robustness-cost/14-VALIDATION.md
+@.planning/ROADMAP.md
+@.planning/REQUIREMENTS.md
+@.planning/STATE.md
+
+@kb_workflow/srt/semantic_engine.py
+@.planning/phases/14-robustness-cost/14-01-PLAN.md
+</context>
+
+<interfaces>
+<!-- Key types and contracts from semantic_engine.py -->
+
+From kb_workflow/srt/semantic_engine.py:
+```python
+def extract_json_from_llm_response(response: str) -> dict:
+    """从 LLM 响应中提取 JSON
+
+    处理以下情况：
+    - 纯 JSON
+    - 包裹在 ```json ``` 中的 JSON
+    - 前后包含额外说明文本的 JSON
+
+    Raises:
+        ValueError: 无法提取有效 JSON
+    """
+    # 尝试直接解析
+    try:
+        return json.loads(response.strip())
+    except json.JSONDecodeError:
+        pass
+
+    # 尝试提取 ```json ``` 代码块
+    json_match = re.search(r'```json\s*(.*?)\s*```', response, re.DOTALL)
+    if json_match:
+        try:
+            return json.loads(json_match.group(1))
+        except json.JSONDecodeError:
+            pass
+
+    # 尝试提取任何 JSON 对象
+    json_match = re.search(r'\{.*\}', response, re.DOTALL)
+    if json_match:
+        try:
+            return json.loads(json_match.group())
+        except json.JSONDecodeError as e:
+            raise ValueError(f"无法从 LLM 响应中提取有效 JSON: {e}")
+
+    raise ValueError("LLM 响应中未找到 JSON 内容")
+
+class SegmentationResult(BaseModel):
+    video_type: str
+    segments: List[VideoSegment]
+    actual_segment_count: int
+
+def _parse_result(
+    self,
+    llm_output: str,
+    fallback_segments: List
+) -> List[VideoSegment]:
+    """解析 LLM 输出并验证
+
+    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
+```
+
+关键观察:
+- extract_json_from_llm_response() 可能抛出 ValueError (无法提取 JSON)
+- SegmentationResult(**data) 可能抛出 ValidationError (Pydantic 验证失败)
+- 当前实现: 所有解析失败都抛出异常，由 segment() 方法的 try-except 捕获并降级
+</interfaces>
+
+<tasks>
+
+<task type="auto" tdd="true">
+  <name>Task 1: 改进解析失败的错误处理和日志</name>
+  <files>kb_workflow/srt/semantic_engine.py</files>
+  <behavior>
+    - 区分 JSONDecodeError (无法提取 JSON) 和 ValidationError (Pydantic 验证失败)
+    - JSONDecodeError 是可恢复错误（可能 LLM 输出格式问题），可以重试
+    - ValidationError 是不可恢复错误（数据结构错误），直接降级
+    - 所有解析失败都记录详细日志:
+      - 错误类型 (JSONDecodeError/ValidationError/ValueError)
+      - 错误消息
+      - LLM 输出前 500 字符（用于诊断）
+  </behavior>
+  <action>
+    修改 UnifiedSegmentationStrategy._parse_result() 方法:
+
+    1. 拆分 except 块，区分不同的错误类型:
+
+    ```python
+    except json.JSONDecodeError as e:
+        # JSON 解析失败（可恢复）
+        logger.error(
+            f"JSON 解析失败: {e}\n"
+            f"LLM 输出片段: {llm_output[:500]}..."
+        )
+        # 抛出特定异常，由 segment() 决定是否重试
+        raise
+
+    except ValidationError as e:
+        # Pydantic 验证失败（不可恢复）
+        logger.error(
+            f"Pydantic 验证失败: {e}\n"
+            f"LLM 输出片段: {llm_output[:500]}..."
+        )
+        # 直接降级，不重试
+        raise
+
+    except ValueError as e:
+        # 无法提取 JSON（可恢复）
+        logger.error(
+            f"JSON 提取失败: {e}\n"
+            f"LLM 输出片段: {llm_output[:500]}..."
+        )
+        raise
+    ```
+
+    2. 在 segment() 方法中，修改 LLM 调用部分的重试逻辑:
+       - 对 JSONDecodeError 和 ValueError 进行重试
+       - 对 ValidationError 不重试，直接降级
+       - 使用 _is_recoverable_error() 的模式，创建 _is_recoverable_parse_error()
+
+    3. 统一日志格式:
+       - 重试: info 级别
+       - 降级: warning 级别，包含 "降级到规则分段"
+
+    遵守用户决策: 解析失败直接降级（不重试），但 JSONDecodeError 可以重试 1 次。
+  </action>
+  <verify>
+    <automated>pytest tests/test_semantic_robustness.py::test_parse_failure_retry -x</automated>
+  </verify>
+  <done>
+    - JSONDecodeError 和 ValueError 触发重试 1 次
+    - ValidationError 直接降级，不重试
+    - 所有解析失败都记录详细日志（错误类型 + LLM 输出片段）
+    - 测试覆盖各种解析失败场景
+  </done>
+</task>
+
+</tasks>
+
+<verification>
+## 整体验证标准
+
+1. **JSON 解析失败**: JSONDecodeError 触发重试 1 次，仍失败则降级
+2. **Pydantic 验证失败**: ValidationError 直接降级，不重试
+3. **JSON 提取失败**: ValueError 触发重试 1 次，仍失败则降级
+4. **日志完整性**: 所有解析失败都记录错误类型和 LLM 输出片段
+
+运行完整测试套件:
+```bash
+python3 -m pytest tests/test_semantic_robustness.py::test_parse_failure_retry -v -x
+```
+</verification>
+
+<success_criteria>
+1. JSONDecodeError 重试 1 次后降级
+2. ValidationError 直接降级，不重试
+3. 所有解析失败都记录 warning 级别日志
+4. 日志包含错误类型、错误消息、LLM 输出前 500 字符
+5. 所有测试通过
+</success_criteria>
+
+<output>
+执行完成后创建 `.planning/phases/14-robustness-cost/14-02-SUMMARY.md`
+</output>
@@ -0,0 +1,272 @@
+---
+phase: 14-robustness-cost
+plan: 03
+type: execute
+wave: 1
+depends_on: []
+files_modified:
+  - kb_workflow/srt/cache/semantic_cache.py
+  - kb_workflow/srt/cache/__init__.py
+autonomous: true
+requirements:
+  - COST-02
+must_haves:
+  truths:
+    - "相同内容的字幕使用缓存结果，避免重复 LLM 调用"
+    - "缓存键基于内容哈希（SHA256），精确匹配"
+    - "缓存存储在本地文件系统 (.planning/cache/semantic/)"
+  artifacts:
+    - path: "kb_workflow/srt/cache/semantic_cache.py"
+      provides: "SemanticCache 缓存管理器"
+      exports: ["SemanticCache", "compute_content_hash"]
+    - path: "kb_workflow/srt/cache/__init__.py"
+      provides: "缓存模块导出"
+      exports: ["SemanticCache"]
+  key_links:
+    - from: "SemanticCache.get()"
+      to: ".planning/cache/semantic/"
+      via: "文件系统读取"
+      pattern: "Path.read_text"
+    - from: "SemanticCache.set()"
+      to: ".planning/cache/semantic/"
+      via: "文件系统写入"
+      pattern: "Path.write_text"
+---
+
+<objective>
+实现基于内容哈希的智能缓存，避免重复 LLM 调用，降低成本和延迟。
+
+Purpose: 相同内容的字幕直接使用缓存结果，减少 LLM API 调用次数。
+Output: SemanticCache 缓存管理器，支持内容哈希、文件存储、TTL 永久有效。
+</objective>
+
+<execution_context>
+@/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md
+@/home/starmen1265/.claude/get-shit-done/templates/summary.md
+</execution_context>
+
+<context>
+@.planning/phases/14-robustness-cost/14-CONTEXT.md
+@.planning/phases/14-robustness-cost/14-RESEARCH.md
+@.planning/phases/14-robustness-cost/14-VALIDATION.md
+@.planning/ROADMAP.md
+@.planning/REQUIREMENTS.md
+@.planning/STATE.md
+
+@kb_workflow/srt/semantic_engine.py
+@kb_workflow/srt/corrector.py
+</context>
+
+<interfaces>
+<!-- Key contracts needed for cache implementation -->
+
+From 14-CONTEXT.md (User Decisions):
+```python
+# 缓存机制
+- 缓存键: 字幕内容哈希（SHA256）— 精确匹配
+- 缓存存储: 本地文件系统 — .planning/cache/semantic/
+- 缓存 TTL: 永久有效 — 手动清理
+- 缓存内容: 完整的 SegmentationResult（JSON 格式）
+```
+
+From kb_workflow/srt/semantic_engine.py:
+```python
+class SegmentationResult(BaseModel):
+    video_type: str
+    segments: List[VideoSegment]
+    actual_segment_count: int
+
+class VideoSegment(BaseModel):
+    title: Optional[str]
+    content: str
+    start_index: int
+    end_index: int
+```
+
+关键观察:
+- SegmentationResult 是 Pydantic 模型，可以序列化为 JSON
+- 需要缓存完整的 SegmentationResult 对象
+- 缓存键应该仅基于内容哈希，不包含元数据
+</interfaces>
+
+<tasks>
+
+<task type="auto" tdd="true">
+  <name>Task 1: 创建 SemanticCache 缓存管理器</name>
+  <files>kb_workflow/srt/cache/semantic_cache.py, kb_workflow/srt/cache/__init__.py</files>
+  <behavior>
+    - compute_content_hash(): 计算内容的 SHA256 哈希
+    - get(): 从缓存读取 SegmentationResult
+    - set(): 保存 SegmentationResult 到缓存
+    - 缓存目录不存在时自动创建
+    - 缓存文件命名: {hash}.json
+    - 缓存内容包含 version 字段（未来兼容性）
+    - 缓存命中/未命中记录 debug 日志
+  </behavior>
+  <action>
+    创建 kb_workflow/srt/cache/semantic_cache.py:
+
+    1. 导入依赖:
+       ```python
+       import hashlib
+       import json
+       import logging
+       from pathlib import Path
+       from typing import Optional
+
+       from kb_workflow.srt.semantic_engine import SegmentationResult
+       ```
+
+    2. 实现 compute_content_hash():
+       ```python
+       def compute_content_hash(content: str) -> str:
+           """计算内容的 SHA256 哈希作为缓存键"""
+           return hashlib.sha256(content.encode('utf-8')).hexdigest()
+       ```
+
+    3. 实现 SemanticCache 类:
+       ```python
+       class SemanticCache:
+           """语义分段结果缓存管理器
+
+           使用内容哈希（SHA256）作为缓存键，存储在本地文件系统。
+           """
+
+           CACHE_VERSION = "1.0"
+           DEFAULT_CACHE_DIR = Path(".planning/cache/semantic")
+
+           def __init__(self, cache_dir: Optional[Path] = None):
+               """初始化缓存管理器
+
+               Args:
+                   cache_dir: 缓存目录路径，默认为 .planning/cache/semantic/
+               """
+               self.cache_dir = cache_dir or self.DEFAULT_CACHE_DIR
+               self.logger = logging.getLogger(__name__)
+
+           def get(self, content: str) -> Optional[SegmentationResult]:
+               """从缓存读取分段结果
+
+               Args:
+                   content: 字幕内容
+
+               Returns:
+                   缓存的 SegmentationResult，如果不存在则返回 None
+               """
+               cache_key = compute_content_hash(content)
+               cache_file = self.cache_dir / f"{cache_key}.json"
+
+               if not cache_file.exists():
+                   self.logger.debug(f"Cache miss: {cache_key}")
+                   return None
+
+               try:
+                   data = json.loads(cache_file.read_text(encoding='utf-8'))
+
+                   # 验证版本
+                   if data.get('version') != self.CACHE_VERSION:
+                       self.logger.warning(
+                           f"Cache version mismatch: {data.get('version')} != {self.CACHE_VERSION}"
+                       )
+                       return None
+
+                   # 移除 version 字段，转换为 SegmentationResult
+                   data.pop('version', None)
+                   result = SegmentationResult(**data)
+
+                   self.logger.debug(f"Cache hit: {cache_key}")
+                   return result
+
+               except (json.JSONDecodeError, IOError, ValidationError) as e:
+                   self.logger.warning(f"Failed to load cache: {e}")
+                   return None
+
+           def set(self, content: str, result: SegmentationResult) -> None:
+               """保存分段结果到缓存
+
+               Args:
+                   content: 字幕内容
+                   result: 分段结果
+               """
+               cache_key = compute_content_hash(content)
+               cache_file = self.cache_dir / f"{cache_key}.json"
+
+               try:
+                   # 确保目录存在
+                   self.cache_dir.mkdir(parents=True, exist_ok=True)
+
+                   # 序列化为 JSON，添加 version 字段
+                   data = result.model_dump()
+                   data['version'] = self.CACHE_VERSION
+
+                   cache_file.write_text(
+                       json.dumps(data, ensure_ascii=False, indent=2),
+                       encoding='utf-8'
+                   )
+
+                   self.logger.debug(f"Cache saved: {cache_key}")
+
+               except IOError as e:
+                   self.logger.warning(f"Failed to save cache: {e}")
+       ```
+
+    4. 创建 kb_workflow/srt/cache/__init__.py:
+       ```python
+       """语义分段缓存模块"""
+
+       from kb_workflow.srt.cache.semantic_cache import (
+           SemanticCache,
+           compute_content_hash
+       )
+
+       __all__ = ['SemanticCache', 'compute_content_hash']
+       ```
+
+    遵守用户决策:
+    - 缓存键: SHA256 内容哈希
+    - 缓存存储: 本地文件系统 (.planning/cache/semantic/)
+    - 缓存 TTL: 永久有效
+    - 缓存内容: 完整的 SegmentationResult
+  </action>
+  <verify>
+    <automated>pytest tests/test_semantic_robustness.py::test_cache_hit -x</automated>
+  </verify>
+  <done>
+    - SemanticCache 类实现完整的缓存读写功能
+    - compute_content_hash() 使用 SHA256 计算内容哈希
+    - 缓存目录不存在时自动创建
+    - 缓存文件包含 version 字段，支持未来兼容性
+    - 缓存命中/未命中记录 debug 日志
+    - 缓存加载失败时返回 None，不中断流程
+  </done>
+</task>
+
+</tasks>
+
+<verification>
+## 整体验证标准
+
+1. **缓存键计算**: 相同内容产生相同的 SHA256 哈希
+2. **缓存命中**: 第二次调用时从缓存读取，不调用 LLM
+3. **缓存未命中**: 首次调用时保存结果到缓存
+4. **版本控制**: 缓存包含 version 字段，支持未来格式变更
+5. **错误处理**: 缓存读写失败时返回 None，不中断流程
+
+运行完整测试套件:
+```bash
+python3 -m pytest tests/test_semantic_robustness.py::test_cache_hit -v -x
+```
+</verification>
+
+<success_criteria>
+1. 相同内容第二次调用时从缓存读取
+2. 缓存文件存储在 .planning/cache/semantic/ 目录
+3. 缓存文件命名格式: {sha256_hash}.json
+4. 缓存内容包含完整的 SegmentationResult (JSON 格式)
+5. 缓存命中时记录 debug 级别日志
+6. 所有测试通过
+</success_criteria>
+
+<output>
+执行完成后创建 `.planning/phases/14-robustness-cost/14-03-SUMMARY.md`
+</output>
@@ -0,0 +1,352 @@
+---
+phase: 14-robustness-cost
+plan: 04
+type: execute
+wave: 1
+depends_on:
+  - 14-03
+files_modified:
+  - kb_workflow/srt/semantic_engine.py
+autonomous: true
+requirements:
+  - COST-01
+must_haves:
+  truths:
+    - "长视频字幕（>5000字符）自动分块处理"
+    - "分块按句子边界分割，保持语义完整性"
+    - "每个分块独立调用 LLM 分段，结果直接拼接"
+  artifacts:
+    - path: "kb_workflow/srt/semantic_engine.py"
+      provides: "UnifiedSegmentationStrategy 分块处理逻辑"
+      exports: ["_split_into_chunks", "_merge_segment_results"]
+  key_links:
+    - from: "_split_into_chunks()"
+      to: "ASRCorrector._correct_long_text()"
+      via: "复用分批模式"
+      pattern: "按句子边界分割"
+    - from: "segment()"
+      to: "_split_into_chunks()"
+      via: "长文本分块"
+      pattern: "len(content) > 5000"
+---
+
+<objective>
+实现长视频字幕的分块处理，避免 LLM 上下文窗口溢出。
+
+Purpose: 处理超过 5000 字符的长字幕，按句子边界分割后独立分段。
+Output: 带分块处理的 UnifiedSegmentationStrategy.segment() 方法。
+</objective>
+
+<execution_context>
+@/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md
+@/home/starmen1265/.claude/get-shit-done/templates/summary.md
+</execution_context>
+
+<context>
+@.planning/phases/14-robustness-cost/14-CONTEXT.md
+@.planning/phases/14-robustness-cost/14-RESEARCH.md
+@.planning/phases/14-robustness-cost/14-VALIDATION.md
+@.planning/ROADMAP.md
+@.planning/REQUIREMENTS.md
+@.planning/STATE.md
+
+@kb_workflow/srt/semantic_engine.py
+@kb_workflow/srt/corrector.py
+@.planning/phases/14-robustness-cost/14-03-PLAN.md
+</context>
+
+<interfaces>
+<!-- Key contracts from corrector.py for batch processing -->
+
+From kb_workflow/srt/corrector.py (分批处理模式):
+```python
+def _correct_long_text(self, text: str) -> str:
+    """分批处理长文本
+
+    按句子分割，每批不超过 batch_size 字符。
+    """
+    # 按句子分割（简单分割，基于句号、问号、感叹号）
+    sentences = []
+    current = ""
+    for char in text:
+        current += char
+        if char in '。！？\n':
+            sentences.append(current)
+            current = ""
+
+    # 添加最后一个不完整的句子
+    if current:
+        sentences.append(current)
+
+    # 分批处理
+    batches = []
+    current_batch = []
+    current_length = 0
+
+    for sentence in sentences:
+        if current_length + len(sentence) > self.batch_size:
+            if current_batch:
+                batches.append(''.join(current_batch))
+            current_batch = [sentence]
+            current_length = len(sentence)
+        else:
+            current_batch.append(sentence)
+            current_length += len(sentence)
+
+    if current_batch:
+        batches.append(''.join(current_batch))
+
+    # 处理每个批次
+    corrected_batches = []
+    for batch in batches:
+        corrected = self._correct_batch(batch)
+        corrected_batches.append(corrected)
+
+    return ''.join(corrected_batches)
+```
+
+关键观察:
+- 按标点符号（。！？\n）分割句子
+- 每批不超过 batch_size (5000)
+- 处理完每批后直接拼接结果
+
+From 14-CONTEXT.md (User Decisions):
+```python
+# 分块处理
+- 分块阈值: 5000 字符（约 10-15 分钟字幕）
+- 分块方式: 按句子边界分割（句号、问号、感叹号等）
+- 结果合并: 直接拼接各分块结果
+- 分块策略: 长字幕按句子边界分成多个 5000 字符以内的块
+```
+
+From 14-RESEARCH.md (Open Questions):
+> **分块后的分段索引如何计算？**
+> - What we know: 每个分块独立分段，结果直接拼接
+> - What's unclear: VideoSegment.start_index 和 end_index 是相对于原文还是分块？
+> - Recommendation: 相对于原文，需要在分块时记录每个分块的起始偏移量，分段后累加偏移量
+
+决策: start_index/end_index 相对于原文，需要记录分块偏移量。
+</interfaces>
+
+<tasks>
+
+<task type="auto" tdd="true">
+  <name>Task 1: 实现长文本分块处理</name>
+  <files>kb_workflow/srt/semantic_engine.py</files>
+  <behavior>
+    - 添加 _split_into_chunks() 方法，按句子边界分割长文本
+    - 分块阈值: 5000 字符
+    - 添加 _merge_segment_results() 方法，合并分块结果
+    - 修正 VideoSegment.start_index 和 end_index 为相对于原文的位置
+    - 短文本（<= 5000 字符）直接处理，不分块
+  </behavior>
+  <action>
+    在 UnifiedSegmentationStrategy 类中添加分块处理逻辑:
+
+    1. 添加 _split_into_chunks() 方法（复用 ASRCorrector 模式）:
+       ```python
+       def _split_into_chunks(self, text: str, max_chunk_size: int = 5000) -> List[tuple[str, int]]:
+           """按句子边界分割文本为多个块
+
+           Args:
+               text: 待分割的文本
+               max_chunk_size: 每块的最大字符数
+
+           Returns:
+               List of (chunk_text, offset) 元组，offset 是该块在原文中的起始位置
+           """
+           # 按句子分割
+           sentences = []
+           current = ""
+           for char in text:
+               current += char
+               if char in '。！？\n':
+                   sentences.append(current)
+                   current = ""
+
+           # 添加最后一个不完整的句子
+           if current:
+               sentences.append(current)
+
+           # 分块
+           chunks = []
+           current_chunk = []
+           current_length = 0
+           current_offset = 0
+
+           for sentence in sentences:
+               sentence_len = len(sentence)
+               if current_length + sentence_len > max_chunk_size:
+                   if current_chunk:
+                       chunk_text = ''.join(current_chunk)
+                       chunks.append((chunk_text, current_offset))
+                       current_offset += len(chunk_text)
+                   current_chunk = [sentence]
+                   current_length = sentence_len
+               else:
+                   current_chunk.append(sentence)
+                   current_length += sentence_len
+
+           # 添加最后一个块
+           if current_chunk:
+               chunk_text = ''.join(current_chunk)
+               chunks.append((chunk_text, current_offset))
+
+           return chunks
+       ```
+
+    2. 添加 _merge_segment_results() 方法:
+       ```python
+       def _merge_segment_results(
+           self,
+           chunk_results: List[tuple[List[VideoSegment], int]]
+       ) -> List[VideoSegment]:
+           """合并分块的分段结果
+
+           Args:
+               chunk_results: List of (segments, offset) 元组
+
+           Returns:
+               合并后的分段列表，修正 start_index 和 end_index 为相对于原文的位置
+           """
+           all_segments = []
+
+           for segments, offset in chunk_results:
+               for seg in segments:
+                   # 修正索引为相对于原文的位置
+                   corrected_seg = VideoSegment(
+                       title=seg.title,
+                       content=seg.content,
+                       start_index=seg.start_index + offset,
+                       end_index=seg.end_index + offset
+                   )
+                   all_segments.append(corrected_seg)
+
+           return all_segments
+       ```
+
+    3. 修改 segment() 方法，添加分块逻辑:
+       ```python
+       def segment(self, segments: List, metadata: dict) -> List[VideoSegment]:
+           # AI 不可用，直接使用规则分段
+           if self.ai_client is None:
+               logger.info("AI client unavailable, using fallback segmenter")
+               paragraphs = self.fallback_segmenter.segment(segments)
+               return [
+                   VideoSegment(title=None, content=para, start_index=0, end_index=0)
+                   for para in paragraphs
+               ]
+
+           # 合并字幕文本
+           content = " ".join(seg.text for seg in segments)
+
+           # 检查是否需要分块
+           if len(content) <= 5000:
+               # 短文本直接处理
+               return self._segment_single_chunk(content, metadata, segments)
+           else:
+               # 长文本分块处理
+               return self._segment_multiple_chunks(content, metadata, segments)
+       ```
+
+    4. 添加 _segment_single_chunk() 和 _segment_multiple_chunks() 辅助方法:
+       ```python
+       def _segment_single_chunk(
+           self,
+           content: str,
+           metadata: dict,
+           fallback_segments: List
+       ) -> List[VideoSegment]:
+           """处理单个文本块（不分块）"""
+       prompt = self._build_prompt(content=content, metadata=metadata)
+
+       # 调用 LLM（带重试和降级）
+       # ... (现有的 LLM 调用逻辑)
+
+       def _segment_multiple_chunks(
+           self,
+           content: str,
+           metadata: dict,
+           fallback_segments: List
+       ) -> List[VideoSegment]:
+           """处理多个文本块（分块）"""
+           logger.info(f"Content too long ({len(content)} chars), splitting into chunks")
+
+           # 分块
+           chunks = self._split_into_chunks(content, max_chunk_size=5000)
+           logger.info(f"Split into {len(chunks)} chunks")
+
+           # 处理每个分块
+           chunk_results = []
+           for i, (chunk_text, offset) in enumerate(chunks, 1):
+               logger.info(f"Processing chunk {i}/{len(chunks)} (offset={offset}, length={len(chunk_text)})")
+
+               # 构建该分块的 prompt
+               chunk_metadata = metadata.copy()
+               chunk_metadata['chunk_info'] = f"第 {i}/{len(chunks)} 块"
+               prompt = self._build_prompt(content=chunk_text, metadata=chunk_metadata)
+
+               try:
+                   # 调用 LLM
+                   llm_output = self.ai_client.generate(prompt, max_tokens=8000, disable_reasoning=True)
+                   segments = self._parse_result(llm_output, fallback_segments=fallback_segments)
+                   chunk_results.append((segments, offset))
+                   logger.info(f"Chunk {i} segmented successfully: {len(segments)} segments")
+               except Exception as e:
+                   logger.warning(f"Chunk {i} failed: {e}, using fallback for this chunk")
+                   # 该分块降级到规则分段
+                   chunk_segments = self._fallback_chunk(chunk_text, offset)
+                   chunk_results.append((chunk_segments, offset))
+
+           # 合并结果
+           return self._merge_segment_results(chunk_results)
+       ```
+
+    遵守用户决策:
+    - 分块阈值: 5000 字符
+    - 按句子边界分割（。！？\n）
+    - 直接拼接各分块结果
+    - start_index/end_index 相对于原文
+  </action>
+  <verify>
+    <automated>pytest tests/test_semantic_robustness.py::test_long_text_chunking -x</automated>
+  </verify>
+  <done>
+    - 长文本（>5000字符）自动分块处理
+    - 按句子边界分割，保持语义完整性
+    - 每个分块独立调用 LLM 分段
+    - 分块结果直接拼接，start_index/end_index 相对于原文
+    - 分块失败时该分块降级到规则分段
+    - 短文本（<=5000字符）不分块，直接处理
+  </done>
+</task>
+
+</tasks>
+
+<verification>
+## 整体验证标准
+
+1. **分块触发**: 超过 5000 字符的文本自动分块
+2. **句子边界**: 分块在标点符号（。！？\n）后分割
+3. **索引修正**: VideoSegment.start_index 和 end_index 相对于原文
+4. **分块降级**: 单个分块失败时降级到规则分段，不影响其他分块
+5. **日志记录**: 记录分块信息（块数、偏移量、长度）
+
+运行完整测试套件:
+```bash
+python3 -m pytest tests/test_semantic_robustness.py::test_long_text_chunking -v -x
+```
+</verification>
+
+<success_criteria>
+1. 长文本（>5000字符）自动分块处理
+2. 每个分块不超过 5000 字符
+3. 分块边界在句子末尾（标点符号后）
+4. VideoSegment.start_index/end_index 相对于原文位置
+5. 分块失败时该分块降级到规则分段
+6. 所有测试通过
+</success_criteria>
+
+<output>
+执行完成后创建 `.planning/phases/14-robustness-cost/14-04-SUMMARY.md`
+</output>
@@ -0,0 +1,381 @@
+---
+phase: 14-robustness-cost
+plan: 05
+type: execute
+wave: 2
+depends_on:
+  - 14-01
+  - 14-02
+  - 14-03
+  - 14-04
+files_modified:
+  - kb_workflow/srt/semantic_engine.py
+  - kb_workflow/srt/processor.py
+autonomous: true
+requirements:
+  - COST-03
+must_haves:
+  truths:
+    - "用户可通过配置开关启用/禁用 LLM 分段"
+    - "禁用时使用简单规则分段，不影响其他平台"
+    - "SRTProcessor.enable_semantic_segmentation 参数控制开关"
+  artifacts:
+    - path: "kb_workflow/srt/processor.py"
+      provides: "SRTProcessor 配置开关"
+      exports: ["SRTProcessor.__init__"]
+  key_links:
+    - from: "SRTProcessor.__init__()"
+      to: "SemanticSegmentationEngine"
+      via: "enable_semantic_segmentation 参数"
+      pattern: "if enable_semantic_segmentation"
+---
+
+<objective>
+集成所有鲁棒性和成本控制功能到 SRTProcessor，提供配置开关控制 LLM 分段。
+
+Purpose: 完成阶段 14 的集成，确保所有功能可通过配置开关控制。
+Output: 集成缓存、分块、重试、降级的 SRTProcessor，支持配置开关。
+</objective>
+
+<execution_context>
+@/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md
+@/home/starmen1265/.claude/get-shit-done/templates/summary.md
+</execution_context>
+
+<context>
+@.planning/phases/14-robustness-cost/14-CONTEXT.md
+@.planning/phases/14-robustness-cost/14-RESEARCH.md
+@.planning/phases/14-robustness-cost/14-VALIDATION.md
+@.planning/ROADMAP.md
+@.planning/REQUIREMENTS.md
+@.planning/STATE.md
+
+@kb_workflow/srt/processor.py
+@kb_workflow/srt/semantic_engine.py
+@.planning/phases/14-robustness-cost/14-01-PLAN.md
+@.planning/phases/14-robustness-cost/14-02-PLAN.md
+@.planning/phases/14-robustness-cost/14-03-PLAN.md
+@.planning/phases/14-robustness-cost/14-04-PLAN.md
+</context>
+
+<interfaces>
+<!-- Key contracts from processor.py for integration -->
+
+From kb_workflow/srt/processor.py (Phase 13 Plan 07):
+```python
+class SRTProcessor(BaseProcessor):
+    def __init__(
+        self,
+        config: Optional[dict] = None,
+        enable_semantic_segmentation: bool = False,
+        **kwargs
+    ):
+        """初始化 SRT 处理器
+
+        Args:
+            config: 配置字典
+            enable_semantic_segmentation: 是否启用语义分段（默认 False）
+        """
+        super().__init__(config=config, **kwargs)
+
+        # 初始化子模块
+        self.parser = SRTParser()
+        self.corrector = ASRCorrector()
+
+        # 语义分段开关
+        self.enable_semantic_segmentation = enable_semantic_segmentation
+        if self.enable_semantic_segmentation:
+            self.semantic_engine = SemanticSegmentationEngine()
+        else:
+            self.semantic_engine = None
+            self.segmenter = SemanticSegmenter()
+
+    def _segment_subtitles(self, segments: List, metadata: dict) -> List[str]:
+        """分段字幕
+
+        Args:
+            segments: 字幕分段列表
+            metadata: 视频元数据
+
+        Returns:
+            分段后的文本列表
+        """
+        if self.enable_semantic_segmentation and self.semantic_engine:
+            # 使用 LLM 语义分段
+            video_segments = self.semantic_engine.segment(segments, metadata)
+            # 转换为纯文本列表（保留标题）
+            paragraphs = []
+            for seg in video_segments:
+                if seg.title:
+                    paragraphs.append(f"## {seg.title}\n\n{seg.content}")
+                else:
+                    paragraphs.append(seg.content)
+            return paragraphs
+        else:
+            # 使用规则分段
+            return self.segmenter.segment(segments)
+```
+
+关键观察:
+- enable_semantic_segmentation 参数控制是否启用 LLM 分段
+- True: 使用 SemanticSegmentationEngine
+- False: 使用 SemanticSegmenter (规则分段)
+- 输出格式: 带标题的分段使用 "## {title}\n\n{content}" 格式
+
+From 14-CONTEXT.md (User Decisions):
+```python
+# Claude's Discretion
+- 具体的重试逻辑实现细节 (Plan 01)
+- 缓存文件命名规则和目录结构 (Plan 03)
+- 分块算法的具体实现 (Plan 04)
+- 日志消息的具体措辞和格式 (Plan 01, 02)
+```
+
+本计划需要:
+1. 集成 SemanticCache 到 UnifiedSegmentationStrategy
+2. 确保配置开关 (enable_semantic_segmentation) 正常工作
+3. 添加集成测试验证完整流程
+</interfaces>
+
+<tasks>
+
+<task type="auto" tdd="true">
+  <name>Task 1: 集成缓存到 UnifiedSegmentationStrategy</name>
+  <files>kb_workflow/srt/semantic_engine.py</files>
+  <behavior>
+    - UnifiedSegmentationStrategy 接受可选的 cache 参数
+    - 在 segment() 方法中，调用 LLM 前先检查缓存
+    - 缓存命中: 直接返回缓存结果
+    - 缓存未命中: 调用 LLM，然后保存结果到缓存
+    - 缓存键仅基于字幕内容，不包含元数据
+  </behavior>
+  <action>
+    修改 UnifiedSegmentationStrategy 类，集成缓存:
+
+    1. 修改 __init__() 方法，添加 cache 参数:
+       ```python
+       def __init__(
+           self,
+           ai_client: Optional[AIClientProtocol] = None,
+           fallback_segmenter: Optional['SemanticSegmenter'] = None,
+           cache: Optional['SemanticCache'] = None
+       ):
+           """初始化统一分段策略
+
+           Args:
+               ai_client: AI 客户端实例
+               fallback_segmenter: 降级时使用的规则分段器
+               cache: 缓存管理器，如果为 None 则不使用缓存
+           """
+           self.ai_client = ai_client
+           self.fallback_segmenter = fallback_segmenter
+           self.cache = cache
+
+           # Lazy AI Client 初始化（保持不变）
+           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()
+       ```
+
+    2. 修改 segment() 方法，添加缓存逻辑:
+       ```python
+       def segment(self, segments: List, metadata: dict) -> List[VideoSegment]:
+           # AI 不可用，直接使用规则分段
+           if self.ai_client is None:
+               logger.info("AI client unavailable, using fallback segmenter")
+               paragraphs = self.fallback_segmenter.segment(segments)
+               return [
+                   VideoSegment(title=None, content=para, start_index=0, end_index=0)
+                   for para in paragraphs
+               ]
+
+           # 合并字幕文本
+           content = " ".join(seg.text for seg in segments)
+
+           # 检查缓存
+           if self.cache is not None:
+               cached_result = self.cache.get(content)
+               if cached_result is not None:
+                   logger.info("Using cached segmentation result")
+                   return cached_result.segments
+
+           # 检查是否需要分块
+           if len(content) <= 5000:
+               # 短文本直接处理
+               video_segments = self._segment_single_chunk(content, metadata, segments)
+           else:
+               # 长文本分块处理
+               video_segments = self._segment_multiple_chunks(content, metadata, segments)
+
+           # 保存到缓存
+           if self.cache is not None:
+               # 构建完整的 SegmentationResult
+               result = SegmentationResult(
+                   video_type=self._infer_video_type(metadata),
+                   segments=video_segments,
+                   actual_segment_count=len(video_segments)
+               )
+               self.cache.set(content, result)
+               logger.debug("Saved segmentation result to cache")
+
+           return video_segments
+       ```
+
+    3. 修改 SemanticSegmentationEngine.__init__()，传递 cache 参数:
+       ```python
+       def __init__(
+           self,
+           strategy: Optional[SegmentationStrategy] = None,
+           fallback_segmenter: Optional['SemanticSegmenter'] = None,
+           cache: Optional['SemanticCache'] = None
+       ):
+           """初始化语义分段引擎
+
+           Args:
+               strategy: 分段策略实例
+               fallback_segmenter: 降级时使用的规则分段器（已弃用）
+               cache: 缓存管理器
+           """
+           if strategy is None:
+               self.strategy = UnifiedSegmentationStrategy(
+                   fallback_segmenter=fallback_segmenter,
+                   cache=cache
+               )
+           else:
+               self.strategy = strategy
+       ```
+
+    遵守用户决策:
+    - 缓存是可选的，通过构造函数传入
+    - 缓存键仅基于内容哈希
+    - 缓存命中时记录 info 日志
+  </action>
+  <verify>
+    <automated>pytest tests/test_semantic_robustness.py::test_cache_hit -x</automated>
+  </verify>
+  <done>
+    - UnifiedSegmentationStrategy 支持可选缓存
+    - 缓存命中时直接返回，不调用 LLM
+    - 缓存未命中时调用 LLM 并保存结果
+    - SemanticSegmentationEngine 支持传递 cache 参数
+  </done>
+</task>
+
+<task type="auto" tdd="true">
+  <name>Task 2: 在 SRTProcessor 中集成缓存和验证配置开关</name>
+  <files>kb_workflow/srt/processor.py</files>
+  <behavior>
+    - SRTProcessor 接受可选的 enable_cache 参数
+    - 启用语义分段时创建 SemanticCache 实例
+    - 禁用语义分段时不创建缓存
+    - 配置开关正确控制 LLM 分段和规则分段
+  </behavior>
+  <action>
+    修改 SRTProcessor 类，集成缓存:
+
+    1. 修改 __init__() 方法，添加缓存支持:
+       ```python
+       def __init__(
+           self,
+           config: Optional[dict] = None,
+           enable_semantic_segmentation: bool = False,
+           enable_cache: bool = True,
+           **kwargs
+       ):
+       """初始化 SRT 处理器
+
+       Args:
+           config: 配置字典
+           enable_semantic_segmentation: 是否启用语义分段（默认 False）
+           enable_cache: 是否启用缓存（默认 True，仅在启用语义分段时有效）
+       """
+       super().__init__(config=config, **kwargs)
+
+       # 初始化子模块
+       self.parser = SRTParser()
+       self.corrector = ASRCorrector()
+
+       # 语义分段开关
+       self.enable_semantic_segmentation = enable_semantic_segmentation
+       if self.enable_semantic_segmentation:
+           # 初始化缓存
+           cache = None
+           if enable_cache:
+               from kb_workflow.srt.cache import SemanticCache
+               cache = SemanticCache()
+
+           self.semantic_engine = SemanticSegmentationEngine(cache=cache)
+           self.segmenter = None
+       else:
+           self.semantic_engine = None
+           self.segmenter = SemanticSegmenter()
+
+           logger.info("Using rule-based segmenter (semantic segmentation disabled)")
+       ```
+
+    2. 验证 _segment_subtitles() 方法正确使用配置开关:
+       - 检查 enable_semantic_segmentation 为 True 时使用 SemanticSegmentationEngine
+       - 检查 enable_semantic_segmentation 为 False 时使用 SemanticSegmenter
+       - 输出格式正确（带标题的分段使用 "## {title}\n\n{content}"）
+
+    遵守用户决策:
+    - 配置开关控制是否启用 LLM 分段
+    - 默认禁用语义分段（向后兼容）
+    - 缓存默认启用（当语义分段启用时）
+  </action>
+  <verify>
+    <automated>pytest tests/test_srt_processor.py::test_semantic_segmentation_toggle -x</automated>
+  </verify>
+  <done>
+    - SRTProcessor 支持配置开关控制语义分段
+    - 启用时创建 SemanticCache 和 SemanticSegmentationEngine
+    - 禁用时使用 SemanticSegmenter，不创建缓存
+    - 集成测试验证完整流程
+  </done>
+</task>
+
+</tasks>
+
+<verification>
+## 整体验证标准
+
+1. **配置开关**: enable_semantic_segmentation=True 启用 LLM 分段，False 使用规则分段
+2. **缓存集成**: 启用语义分段时缓存工作，禁用时不创建缓存
+3. **向后兼容**: 默认禁用语义分段，不影响现有功能
+4. **完整流程**: 从 SRT 处理到 Markdown 生成的完整流程正常工作
+
+运行完整测试套件:
+```bash
+python3 -m pytest tests/test_srt_processor.py::test_semantic_segmentation_toggle -v -x
+```
+
+运行鲁棒性测试:
+```bash
+python3 -m pytest tests/test_semantic_robustness.py -v -x
+```
+</verification>
+
+<success_criteria>
+1. enable_semantic_segmentation=True 时使用 LLM 分段
+2. enable_semantic_segmentation=False 时使用规则分段
+3. 缓存在启用语义分段时正常工作
+4. 禁用语义分段时不创建缓存
+5. 所有功能集成测试通过
+6. 所有鲁棒性测试通过
+</success_criteria>
+
+<output>
+执行完成后创建 `.planning/phases/14-robustness-cost/14-05-SUMMARY.md`
+</output>