diff --git a/kb_workflow/srt/semantic_engine.py b/kb_workflow/srt/semantic_engine.py index 3af8e1f..d36d5ce 100644 --- a/kb_workflow/srt/semantic_engine.py +++ b/kb_workflow/srt/semantic_engine.py @@ -86,16 +86,19 @@ class UnifiedSegmentationStrategy: def __init__( self, ai_client: Optional[AIClientProtocol] = None, - fallback_segmenter: Optional['SemanticSegmenter'] = None + fallback_segmenter: Optional['SemanticSegmenter'] = None, + cache: Optional['SemanticCache'] = None ): """初始化统一分段策略 Args: ai_client: AI 客户端实例，如果为 None 则尝试创建 SimpleAIClient fallback_segmenter: 降级时使用的规则分段器 + cache: 缓存管理器，如果为 None 则不使用缓存 """ self.ai_client = ai_client self.fallback_segmenter = fallback_segmenter + self.cache = cache # Lazy AI Client 初始化（与 ASRCorrector 相同模式） if self.ai_client is None: @@ -184,83 +187,33 @@ class UnifiedSegmentationStrategy: # 合并字幕文本 content = " ".join(seg.text for seg in segments) - # 构建 Prompt - prompt = self._build_prompt( - content=content, - metadata=metadata - ) + # 检查缓存（Plan 14-05: 缓存集成） + 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 - # 调用 LLM，带重试逻辑（最多 2 次尝试） - max_attempts = 2 - last_error = None - last_llm_output = None + # 检查是否需要分块（Plan 14-04: 分块阈值 5000 字符） + if len(content) <= 5000: + # 短文本直接处理 + video_segments = self._segment_single_chunk(content, metadata, segments) + else: + # 长文本分块处理 + video_segments = self._segment_multiple_chunks(content, metadata, 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 + # 保存到缓存（Plan 14-05: 缓存结果） + 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") - # 解析和验证 - video_segments = self._parse_result( - llm_output, - fallback_segments=segments - ) - - 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 - ] + return video_segments def _build_prompt(self, content: str, metadata: dict) -> str: """构建语义分段 Prompt @@ -399,6 +352,240 @@ class UnifiedSegmentationStrategy: return result.segments + def _segment_single_chunk( + self, + content: str, + metadata: dict, + fallback_segments: List + ) -> List[VideoSegment]: + """处理单个文本块（不分块） + + Args: + content: 字幕内容 + metadata: 视频元数据 + fallback_segments: 降级时使用的原始分段 + + Returns: + VideoSegment 对象列表 + """ + # 构建 Prompt + prompt = self._build_prompt(content=content, metadata=metadata) + + # 调用 LLM，带重试逻辑（最多 2 次尝试） + max_attempts = 2 + last_error = None + last_llm_output = None + + 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 + + # 解析和验证 + video_segments = self._parse_result( + llm_output, + fallback_segments=fallback_segments + ) + + 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(fallback_segments) + return [ + VideoSegment(title=None, content=para, start_index=0, end_index=0) + for para in paragraphs + ] + + def _segment_multiple_chunks( + self, + content: str, + metadata: dict, + fallback_segments: List + ) -> List[VideoSegment]: + """处理多个文本块（分块） + + Args: + content: 字幕内容 + metadata: 视频元数据 + fallback_segments: 降级时使用的原始分段 + + Returns: + 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) + + 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 + + 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 + + def _fallback_chunk(self, chunk_text: str, offset: int) -> List[VideoSegment]: + """对单个分块使用规则分段 + + Args: + chunk_text: 分块文本 + offset: 该块在原文中的起始位置 + + Returns: + VideoSegment 对象列表 + """ + # 简单实现：将整个块作为一个分段 + return [ + VideoSegment( + title=None, + content=chunk_text, + start_index=offset, + end_index=offset + len(chunk_text) + ) + ] + # ============================================================================ # Pydantic Models @@ -597,18 +784,21 @@ class SemanticSegmentationEngine: def __init__( self, strategy: Optional[SegmentationStrategy] = None, - fallback_segmenter: Optional['SemanticSegmenter'] = None + fallback_segmenter: Optional['SemanticSegmenter'] = None, + cache: Optional['SemanticCache'] = None ): """初始化语义分段引擎 Args: strategy: 分段策略实例，如果为 None 则使用统一策略 fallback_segmenter: 降级时使用的规则分段器（已弃用，保留兼容性） + cache: 缓存管理器 """ # 默认使用统一策略 if strategy is None: self.strategy = UnifiedSegmentationStrategy( - fallback_segmenter=fallback_segmenter + fallback_segmenter=fallback_segmenter, + cache=cache ) else: self.strategy = strategy