diff --git a/.planning/phases/14-robustness-cost/14-00-PLAN.md b/.planning/phases/14-robustness-cost/14-00-PLAN.md
new file mode 100644
index 0000000..138046e
--- /dev/null
+++ b/.planning/phases/14-robustness-cost/14-00-PLAN.md
@@ -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"
+---
+
+
+创建阶段 14 的测试基础设施，覆盖所有鲁棒性和成本控制功能。
+
+Purpose: 确保所有计划任务都有对应的自动化测试，遵循 TDD 原则。
+Output: 完整的测试套件，包括鲁棒性测试、成本控制测试、集成测试。
+
+
+
+@/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md
+@/home/starmen1265/.claude/get-shit-done/templates/summary.md
+
+
+
+@.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
+
+
+
+
+
+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 |
+
+
+
+
+
+ Task 1: 创建 test_semantic_robustness.py 测试文件
+ tests/test_semantic_robustness.py
+
+ 创建 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 状态（功能尚未实现）。
+
+
+ pytest tests/test_semantic_robustness.py -v --tb=no
+
+
+ - test_semantic_robustness.py 文件创建，包含所有测试
+ - 所有测试都是 RED 状态（预期失败）
+ - 测试覆盖所有需求（ROBUST-01 到 COST-03）
+ - Mock fixtures 正确模拟 LLM 客户端行为
+
+
+
+
+ Task 2: 更新 conftest.py 添加新 fixtures
+ tests/conftest.py
+
+ 更新 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
+ ```
+
+
+ pytest tests/conftest.py --collect-only
+
+
+ - conftest.py 添加所有需要的 fixtures
+ - Fixtures 包括: 模拟失败的 AI 客户端、长文本样本、缓存目录
+ - 所有 fixtures 可以正确导入和使用
+
+
+
+
+ Task 3: 在 test_srt_processor.py 中添加配置开关测试
+ tests/test_srt_processor.py
+
+ 在 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 状态（功能尚未实现）。
+
+
+ pytest tests/test_srt_processor.py::test_semantic_segmentation_toggle -v --tb=no
+
+
+ - test_srt_processor.py 添加配置开关测试
+ - 测试覆盖启用/禁用语义分段、启用/禁用缓存
+ - 所有测试都是 RED 状态（预期失败）
+
+
+
+
+
+
+## 整体验证标准
+
+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
+```
+
+
+
+1. test_semantic_robustness.py 文件创建，包含 6 个测试类
+2. conftest.py 添加 10+ 个新 fixtures
+3. test_srt_processor.py 添加 TestSemanticSegmentationToggle 测试类
+4. 所有测试都是 RED 状态（预期失败）
+5. 测试覆盖所有需求（ROBUST-01 到 COST-03）
+
+
+
diff --git a/.planning/phases/14-robustness-cost/14-01-PLAN.md b/.planning/phases/14-robustness-cost/14-01-PLAN.md
new file mode 100644
index 0000000..70385d8
--- /dev/null
+++ b/.planning/phases/14-robustness-cost/14-01-PLAN.md
@@ -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"
+---
+
+
+实现 LLM 调用的优雅降级策略，确保 AI 不可用时系统仍能正常运行。
+
+Purpose: 提高系统稳定性，避免 LLM 故障导致整个处理流程失败。
+Output: 带重试和降级逻辑的 UnifiedSegmentationStrategy.segment() 方法。
+
+
+
+@/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md
+@/home/starmen1265/.claude/get-shit-done/templates/summary.md
+
+
+
+@.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
+
+
+
+
+
+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
+
+
+
+
+
+ Task 1: 实现可恢复错误检测和重试逻辑
+ kb_workflow/srt/semantic_engine.py
+
+ - 添加 _is_recoverable_error() 方法判断错误类型
+ - 可恢复错误: ConnectionError, Timeout, HTTPError with status 429/500/502/503/504
+ - 不可恢复错误: 其他 HTTPError (4xx), JSONDecodeError, ValidationError
+ - 重试逻辑: 仅对可恢复错误重试 1 次
+ - 重试延迟: 立即重试（无延迟）
+ - 最后一次失败或不可恢复错误: 降级到规则分段
+
+
+ 在 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 次, 仅可恢复错误重试, 立即重试。
+
+
+ pytest tests/test_semantic_robustness.py::test_llm_failure_fallback -x
+
+
+ - _is_recoverable_error() 方法正确识别可恢复/不可恢复错误
+ - LLM 调用失败时重试 1 次（可恢复错误）或直接降级（不可恢复错误）
+ - 降级时记录具体的错误类型和消息 (warning 级别)
+ - AI 不可用时 (ai_client is None) 直接使用规则分段，无重试
+
+
+
+
+
+
+## 整体验证标准
+
+1. **功能验证**: LLM 调用失败时系统仍能返回分段结果（使用规则分段）
+2. **重试验证**: 可恢复错误（网络错误、超时、限流）触发重试
+3. **降级验证**: 不可恢复错误（API 错误、解析失败）直接降级，不重试
+4. **日志验证**: 降级时记录具体原因，便于问题诊断
+
+运行完整测试套件:
+```bash
+python3 -m pytest tests/test_semantic_robustness.py::test_llm_failure_fallback -v -x
+```
+
+
+
+1. LLM 调用失败时自动降级到规则分段，不中断处理流程
+2. 可恢复错误（网络、超时、限流）重试 1 次后降级
+3. 不可恢复错误（API 错误、解析失败）直接降级，不浪费配额
+4. AI 客户端未配置时直接使用规则分段，无 LLM 调用
+5. 降级时记录 warning 级别日志，包含错误类型和原因
+6. 所有测试通过
+
+
+
diff --git a/.planning/phases/14-robustness-cost/14-02-PLAN.md b/.planning/phases/14-robustness-cost/14-02-PLAN.md
new file mode 100644
index 0000000..9184b45
--- /dev/null
+++ b/.planning/phases/14-robustness-cost/14-02-PLAN.md
@@ -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)"
+---
+
+
+增强 LLM 响应解析的错误处理，区分解析失败类型并采取不同策略。
+
+Purpose: 提高 LLM 输出格式不稳定时的容错能力，记录详细日志便于诊断。
+Output: 改进的 _parse_result() 方法，带详细的错误日志和重试逻辑。
+
+
+
+@/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md
+@/home/starmen1265/.claude/get-shit-done/templates/summary.md
+
+
+
+@.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
+
+
+
+
+
+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 捕获并降级
+
+
+
+
+
+ Task 1: 改进解析失败的错误处理和日志
+ kb_workflow/srt/semantic_engine.py
+
+ - 区分 JSONDecodeError (无法提取 JSON) 和 ValidationError (Pydantic 验证失败)
+ - JSONDecodeError 是可恢复错误（可能 LLM 输出格式问题），可以重试
+ - ValidationError 是不可恢复错误（数据结构错误），直接降级
+ - 所有解析失败都记录详细日志:
+ - 错误类型 (JSONDecodeError/ValidationError/ValueError)
+ - 错误消息
+ - LLM 输出前 500 字符（用于诊断）
+
+
+ 修改 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 次。
+
+
+ pytest tests/test_semantic_robustness.py::test_parse_failure_retry -x
+
+
+ - JSONDecodeError 和 ValueError 触发重试 1 次
+ - ValidationError 直接降级，不重试
+ - 所有解析失败都记录详细日志（错误类型 + LLM 输出片段）
+ - 测试覆盖各种解析失败场景
+
+
+
+
+
+
+## 整体验证标准
+
+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
+```
+
+
+
+1. JSONDecodeError 重试 1 次后降级
+2. ValidationError 直接降级，不重试
+3. 所有解析失败都记录 warning 级别日志
+4. 日志包含错误类型、错误消息、LLM 输出前 500 字符
+5. 所有测试通过
+
+
+
diff --git a/.planning/phases/14-robustness-cost/14-03-PLAN.md b/.planning/phases/14-robustness-cost/14-03-PLAN.md
new file mode 100644
index 0000000..377fab2
--- /dev/null
+++ b/.planning/phases/14-robustness-cost/14-03-PLAN.md
@@ -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"
+---
+
+
+实现基于内容哈希的智能缓存，避免重复 LLM 调用，降低成本和延迟。
+
+Purpose: 相同内容的字幕直接使用缓存结果，减少 LLM API 调用次数。
+Output: SemanticCache 缓存管理器，支持内容哈希、文件存储、TTL 永久有效。
+
+
+
+@/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md
+@/home/starmen1265/.claude/get-shit-done/templates/summary.md
+
+
+
+@.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
+
+
+
+
+
+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 对象
+- 缓存键应该仅基于内容哈希，不包含元数据
+
+
+
+
+
+ Task 1: 创建 SemanticCache 缓存管理器
+ kb_workflow/srt/cache/semantic_cache.py, kb_workflow/srt/cache/__init__.py
+
+ - compute_content_hash(): 计算内容的 SHA256 哈希
+ - get(): 从缓存读取 SegmentationResult
+ - set(): 保存 SegmentationResult 到缓存
+ - 缓存目录不存在时自动创建
+ - 缓存文件命名: {hash}.json
+ - 缓存内容包含 version 字段（未来兼容性）
+ - 缓存命中/未命中记录 debug 日志
+
+
+ 创建 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
+
+
+ pytest tests/test_semantic_robustness.py::test_cache_hit -x
+
+
+ - SemanticCache 类实现完整的缓存读写功能
+ - compute_content_hash() 使用 SHA256 计算内容哈希
+ - 缓存目录不存在时自动创建
+ - 缓存文件包含 version 字段，支持未来兼容性
+ - 缓存命中/未命中记录 debug 日志
+ - 缓存加载失败时返回 None，不中断流程
+
+
+
+
+
+
+## 整体验证标准
+
+1. **缓存键计算**: 相同内容产生相同的 SHA256 哈希
+2. **缓存命中**: 第二次调用时从缓存读取，不调用 LLM
+3. **缓存未命中**: 首次调用时保存结果到缓存
+4. **版本控制**: 缓存包含 version 字段，支持未来格式变更
+5. **错误处理**: 缓存读写失败时返回 None，不中断流程
+
+运行完整测试套件:
+```bash
+python3 -m pytest tests/test_semantic_robustness.py::test_cache_hit -v -x
+```
+
+
+
+1. 相同内容第二次调用时从缓存读取
+2. 缓存文件存储在 .planning/cache/semantic/ 目录
+3. 缓存文件命名格式: {sha256_hash}.json
+4. 缓存内容包含完整的 SegmentationResult (JSON 格式)
+5. 缓存命中时记录 debug 级别日志
+6. 所有测试通过
+
+
+
diff --git a/.planning/phases/14-robustness-cost/14-04-PLAN.md b/.planning/phases/14-robustness-cost/14-04-PLAN.md
new file mode 100644
index 0000000..0fb9d2a
--- /dev/null
+++ b/.planning/phases/14-robustness-cost/14-04-PLAN.md
@@ -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"
+---
+
+
+实现长视频字幕的分块处理，避免 LLM 上下文窗口溢出。
+
+Purpose: 处理超过 5000 字符的长字幕，按句子边界分割后独立分段。
+Output: 带分块处理的 UnifiedSegmentationStrategy.segment() 方法。
+
+
+
+@/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md
+@/home/starmen1265/.claude/get-shit-done/templates/summary.md
+
+
+
+@.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
+
+
+
+
+
+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 相对于原文，需要记录分块偏移量。
+
+
+
+
+
+ Task 1: 实现长文本分块处理
+ kb_workflow/srt/semantic_engine.py
+
+ - 添加 _split_into_chunks() 方法，按句子边界分割长文本
+ - 分块阈值: 5000 字符
+ - 添加 _merge_segment_results() 方法，合并分块结果
+ - 修正 VideoSegment.start_index 和 end_index 为相对于原文的位置
+ - 短文本（<= 5000 字符）直接处理，不分块
+
+
+ 在 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 相对于原文
+
+
+ pytest tests/test_semantic_robustness.py::test_long_text_chunking -x
+
+
+ - 长文本（>5000字符）自动分块处理
+ - 按句子边界分割，保持语义完整性
+ - 每个分块独立调用 LLM 分段
+ - 分块结果直接拼接，start_index/end_index 相对于原文
+ - 分块失败时该分块降级到规则分段
+ - 短文本（<=5000字符）不分块，直接处理
+
+
+
+
+
+
+## 整体验证标准
+
+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
+```
+
+
+
+1. 长文本（>5000字符）自动分块处理
+2. 每个分块不超过 5000 字符
+3. 分块边界在句子末尾（标点符号后）
+4. VideoSegment.start_index/end_index 相对于原文位置
+5. 分块失败时该分块降级到规则分段
+6. 所有测试通过
+
+
+
diff --git a/.planning/phases/14-robustness-cost/14-05-PLAN.md b/.planning/phases/14-robustness-cost/14-05-PLAN.md
new file mode 100644
index 0000000..db8f103
--- /dev/null
+++ b/.planning/phases/14-robustness-cost/14-05-PLAN.md
@@ -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"
+---
+
+
+集成所有鲁棒性和成本控制功能到 SRTProcessor，提供配置开关控制 LLM 分段。
+
+Purpose: 完成阶段 14 的集成，确保所有功能可通过配置开关控制。
+Output: 集成缓存、分块、重试、降级的 SRTProcessor，支持配置开关。
+
+
+
+@/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md
+@/home/starmen1265/.claude/get-shit-done/templates/summary.md
+
+
+
+@.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
+
+
+
+
+
+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. 添加集成测试验证完整流程
+
+
+
+
+
+ Task 1: 集成缓存到 UnifiedSegmentationStrategy
+ kb_workflow/srt/semantic_engine.py
+
+ - UnifiedSegmentationStrategy 接受可选的 cache 参数
+ - 在 segment() 方法中，调用 LLM 前先检查缓存
+ - 缓存命中: 直接返回缓存结果
+ - 缓存未命中: 调用 LLM，然后保存结果到缓存
+ - 缓存键仅基于字幕内容，不包含元数据
+
+
+ 修改 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 日志
+
+
+ pytest tests/test_semantic_robustness.py::test_cache_hit -x
+
+
+ - UnifiedSegmentationStrategy 支持可选缓存
+ - 缓存命中时直接返回，不调用 LLM
+ - 缓存未命中时调用 LLM 并保存结果
+ - SemanticSegmentationEngine 支持传递 cache 参数
+
+
+
+
+ Task 2: 在 SRTProcessor 中集成缓存和验证配置开关
+ kb_workflow/srt/processor.py
+
+ - SRTProcessor 接受可选的 enable_cache 参数
+ - 启用语义分段时创建 SemanticCache 实例
+ - 禁用语义分段时不创建缓存
+ - 配置开关正确控制 LLM 分段和规则分段
+
+
+ 修改 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 分段
+ - 默认禁用语义分段（向后兼容）
+ - 缓存默认启用（当语义分段启用时）
+
+
+ pytest tests/test_srt_processor.py::test_semantic_segmentation_toggle -x
+
+
+ - SRTProcessor 支持配置开关控制语义分段
+ - 启用时创建 SemanticCache 和 SemanticSegmentationEngine
+ - 禁用时使用 SemanticSegmenter，不创建缓存
+ - 集成测试验证完整流程
+
+
+
+
+
+
+## 整体验证标准
+
+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
+```
+
+
+
+1. enable_semantic_segmentation=True 时使用 LLM 分段
+2. enable_semantic_segmentation=False 时使用规则分段
+3. 缓存在启用语义分段时正常工作
+4. 禁用语义分段时不创建缓存
+5. 所有功能集成测试通过
+6. 所有鲁棒性测试通过
+
+
+