diff --git a/kb_workflow/srt/semantic_engine.py b/kb_workflow/srt/semantic_engine.py index 1a65e79..3af8e1f 100644 --- a/kb_workflow/srt/semantic_engine.py +++ b/kb_workflow/srt/semantic_engine.py @@ -17,6 +17,7 @@ import logging import re from typing import List, Optional, Protocol +import requests.exceptions from pydantic import BaseModel, Field, field_validator, ValidationError logger = logging.getLogger(__name__) @@ -113,6 +114,46 @@ class UnifiedSegmentationStrategy: from kb_workflow.srt.segmenter import SemanticSegmenter self.fallback_segmenter = SemanticSegmenter() + @staticmethod + def _is_recoverable_error(error: Exception) -> bool: + """判断错误是否可恢复 + + 可恢复错误：网络错误、超时、限流、服务器错误（500/502/503/504）、 + JSON 解析错误（JSONDecodeError, 非 ValidationError 的 ValueError） + 不可恢复错误：其他 HTTP 错误（4xx）、Pydantic 验证错误（ValidationError） + + Args: + error: 异常对象 + + Returns: + True 如果错误可恢复，False 否则 + """ + # Pydantic 验证错误（不可恢复） + # 注意：ValidationError 继承自 ValueError，所以需要先检查 + if isinstance(error, ValidationError): + return False + + # 网络层错误（可恢复） + if isinstance(error, requests.exceptions.ConnectionError): + return True + if isinstance(error, requests.exceptions.Timeout): + return True + + # JSON 解析错误（可恢复 - 可能是 LLM 输出格式问题） + if isinstance(error, (json.JSONDecodeError, ValueError)): + return True + + # HTTP 错误（根据状态码判断） + if isinstance(error, requests.exceptions.HTTPError): + status_code = error.response.status_code if hasattr(error, 'response') and error.response else None + if status_code in (429, 500, 502, 503, 504): + return True + # 其他 4xx 错误不可恢复 + return False + + # 其他错误不可恢复 + return False + def segment( self, segments: List, @@ -149,31 +190,77 @@ class UnifiedSegmentationStrategy: metadata=metadata ) - # 调用 LLM - try: - llm_output = self.ai_client.generate( - prompt, - max_tokens=8000, - disable_reasoning=True # 分段不需要思考模式 - ) + # 调用 LLM，带重试逻辑（最多 2 次尝试） + max_attempts = 2 + last_error = None + last_llm_output = None - # 解析和验证 - video_segments = self._parse_result( - llm_output, - fallback_segments=segments - ) + for attempt in range(max_attempts): + try: + llm_output = self.ai_client.generate( + prompt, + max_tokens=8000, + disable_reasoning=True # 分段不需要思考模式 + ) + last_llm_output = llm_output - logger.info(f"LLM semantic segmentation: {len(video_segments)} segments") - return video_segments + # 解析和验证 + video_segments = self._parse_result( + llm_output, + fallback_segments=segments + ) - except Exception as e: - logger.error(f"LLM segmentation failed: {e}, using fallback") - # 将规则分段结果转换为 VideoSegment 格式（无标题） - paragraphs = self.fallback_segmenter.segment(segments) - return [ - VideoSegment(title=None, content=para, start_index=0, end_index=0) - for para in paragraphs - ] + logger.info(f"LLM semantic segmentation: {len(video_segments)} segments") + return video_segments + + except Exception as e: + last_error = e + is_recoverable = self._is_recoverable_error(e) + is_last_attempt = (attempt == max_attempts - 1) + + # 可恢复错误且不是最后一次尝试：重试 + if is_recoverable and not is_last_attempt: + logger.info( + f"LLM call failed (attempt {attempt + 1}/{max_attempts}): " + f"{type(e).__name__}: {e}, retrying..." + ) + continue + + # 不可恢复错误或最后一次尝试：降级 + error_type = type(e).__name__ + + # 为 HTTPError 提取更详细的信息 + if isinstance(e, requests.exceptions.HTTPError) and hasattr(e, 'response') and e.response: + status_code = e.response.status_code + response_text = getattr(e.response, 'text', '') or '' + error_msg = f"HTTP {status_code}" + if response_text and isinstance(response_text, str): + error_msg += f" ({response_text[:100]})" + logger.warning( + f"{error_type}: {error_msg}, 降级到规则分段" + ) + # 对于 JSON 解析错误（排除 ValidationError），记录 LLM 输出片段（前 500 字符） + elif isinstance(e, (json.JSONDecodeError, ValueError)) and not isinstance(e, ValidationError) and last_llm_output: + error_msg = str(e) if str(e) else error_type + logger.warning( + f"{error_type}: {error_msg}\n" + f"LLM 输出片段: {last_llm_output[:500]}...\n" + f"降级到规则分段" + ) + else: + error_msg = str(e) if str(e) else error_type + logger.warning( + f"{error_type}: {error_msg}, 降级到规则分段" + ) + break + + # 降级到规则分段 + logger.info("使用规则分段（降级）") + paragraphs = self.fallback_segmenter.segment(segments) + return [ + VideoSegment(title=None, content=para, start_index=0, end_index=0) + for para in paragraphs + ] def _build_prompt(self, content: str, metadata: dict) -> str: """构建语义分段 Prompt @@ -295,33 +382,22 @@ class UnifiedSegmentationStrategy: VideoSegment 对象列表 Raises: - Exception: 解析或验证失败时 + json.JSONDecodeError: JSON 解析失败 + ValueError: 无法提取 JSON + ValidationError: Pydantic 验证失败 """ - try: - # 提取 JSON - data = extract_json_from_llm_response(llm_output) + # 提取 JSON + data = extract_json_from_llm_response(llm_output) - # Pydantic 验证 - result = SegmentationResult(**data) + # Pydantic 验证 + result = SegmentationResult(**data) - logger.info( - f"LLM 分段成功: 类型={result.video_type}, " - f"分段数={result.actual_segment_count}" - ) + logger.info( + f"LLM 分段成功: 类型={result.video_type}, " + f"分段数={result.actual_segment_count}" + ) - return result.segments - - except (json.JSONDecodeError, ValidationError) as e: - logger.error(f"LLM 分段结果验证失败: {e}") - logger.error(f"原始 LLM 输出: {llm_output[:500]}...") - - # 降级到规则分段 - logger.info("降级到规则分段") - raise - - except Exception as e: - logger.error(f"未知错误: {e}") - raise + return result.segments # ============================================================================ diff --git a/tests/test_semantic_robustness.py b/tests/test_semantic_robustness.py index 3b12989..b417e17 100644 --- a/tests/test_semantic_robustness.py +++ b/tests/test_semantic_robustness.py @@ -471,8 +471,11 @@ class TestParseFailureRetry: # 验证调用了两次 LLM（初始 + 1 次重试） assert mock_client.generate.call_count == 2 - # 验证日志包含重试信息 - assert any("重试" in record.message for record in caplog.records) + # 验证日志包含重试信息（可能是"重试"或"retrying"） + assert any( + "重试" in record.message or "retrying" in record.message.lower() + for record in caplog.records + ) def test_validation_error_no_retry(self, caplog): """测试 ValidationError（Pydantic 验证失败）直接降级，不重试""" @@ -510,8 +513,10 @@ class TestParseFailureRetry: assert mock_client.generate.call_count == 1 # 验证日志包含验证失败信息 - assert any("Pydantic 验证失败" in record.message or "验证失败" in record.message - for record in caplog.records) + assert any( + "ValidationError" in record.message or "验证失败" in record.message + for record in caplog.records + ) def test_parse_error_logs_llm_output(self, caplog): """测试解析失败时记录 LLM 输出片段（前 500 字符）""" @@ -528,17 +533,17 @@ class TestParseFailureRetry: segments = [Segment(1, 0.0, 1.0, "测试")] - with caplog.at_level(logging.ERROR): + with caplog.at_level(logging.WARNING): result = strategy.segment(segments, {"title": "测试"}) - # 验证日志包含 LLM 输出片段 - error_logs = [record for record in caplog.records if record.levelno == logging.ERROR] - assert len(error_logs) > 0 + # 验证日志包含 LLM 输出片段（使用 WARNING 级别） + warning_logs = [record for record in caplog.records if record.levelno == logging.WARNING] + assert len(warning_logs) > 0 # 验证日志包含 LLM 输出片段信息 assert any( - "LLM 输出片段" in record.message or "原始 LLM 输出" in record.message - for record in error_logs + "LLM 输出片段" in record.message + for record in warning_logs )