"""语义分段引擎 使用 LLM 实现智能视频字幕语义分段。 功能： - 视频类型识别（知识类/故事类/访谈类/新闻类/vlog 等） - 语义边界检测 - 内容完整性保护 - 主题标题生成 - 可扩展的分段策略架构 """ from __future__ import annotations import json 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__) # ============================================================================ # AI Client Protocol # ============================================================================ class AIClientProtocol(Protocol): """AI 客户端协议（与 ASRCorrector 保持一致）""" def generate(self, prompt: str, max_tokens: int = 16000, disable_reasoning: bool = False) -> str: """生成文本""" ... # ============================================================================ # Segmentation Strategy Pattern (Plan 13-06) # ============================================================================ class SegmentationStrategy(Protocol): """分段策略接口 定义了分段策略的通用接口，支持未来扩展不同的分段策略。 Examples: >>> strategy = UnifiedSegmentationStrategy() >>> segments = strategy.segment(subtitle_segments, metadata) Note: 遵守用户决策：Phase 13 实现统一策略，不区分视频类型。 """ def segment( self, segments: List, metadata: dict ) -> List[Union[str, 'VideoSegment']]: """对字幕分段进行语义分段 Args: segments: 字幕分段列表（来自 parser.py 的 Segment 对象） metadata: 视频元数据（title, duration 等） Returns: 分段列表（可以是纯文本列表或 VideoSegment 对象列表） """ ... class UnifiedSegmentationStrategy: """统一分段策略 适用于所有视频类型（知识类、故事类、访谈类等）。 视频类型仅用于 Prompt 上下文，不改变分段策略。 遵守用户决策：无类型区分，使用统一 Prompt。 Attributes: ai_client: AI 客户端（实现 AIClientProtocol） """ def __init__( self, ai_client: Optional[AIClientProtocol] = 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: 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 @staticmethod def _is_recoverable_error(error: Exception) -> bool: """判断错误是否可恢复 可恢复错误：网络错误、超时、限流、服务器错误（500/502/503/504）、 JSON 解析错误（JSONDecodeError, 非 ValidationError 的 ValueError） 不可恢复错误：其他 HTTP 错误（4xx） 注意：Pydantic ValidationError（如标题超长）现在被视为可恢复错误， 因为这是LLM生成内容的问题，应该允许重试。 Args: error: 异常对象 Returns: True 如果错误可恢复，False 否则 """ # 注意：ValidationError 现在被视为可恢复错误 # 因为这可能是LLM偶然生成的内容问题，重试可能解决 # 网络层错误（可恢复） 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, metadata: dict ) -> List['VideoSegment']: """对字幕分段进行语义分段 使用统一的 LLM Prompt 进行语义分段，不区分视频类型。 视频类型仅用于 Prompt 上下文。 Args: segments: 字幕分段列表（来自 parser.py 的 Segment 对象） metadata: 视频元数据（title, duration 等） Returns: VideoSegment 对象列表（包含标题和内容） """ # AI 不可用，直接报错 if self.ai_client is None: raise RuntimeError( "AI客户端未配置，无法进行LLM语义分段。请检查AI_API_KEY环境变量。" ) # 合并字幕文本 content = " ".join(seg.text for seg in segments) # 检查缓存（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 # 检查是否需要分块（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) # 保存到缓存（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") return video_segments def _build_prompt(self, content: str, metadata: dict) -> str: """构建语义分段 Prompt Args: content: 字幕内容 metadata: 视频元数据 Returns: 构建好的 Prompt 字符串 """ # 推断视频类型 video_type = self._infer_video_type(metadata) title = metadata.get("title", "") or "" # 构建 Prompt total_chars = len(content) prompt = f"""你是一个专业的视频内容分析师。请将以下字幕内容按语义分段。 【视频信息】 类型：{video_type} 标题：{title} 总字数：{total_chars}字 【字幕内容】 {content} 【分段要求】 1. 建议将内容分为 5-8 个段落（可根据语义完整性灵活调整） 2. 每个段落约 1000-2000 字（为保持语义完整性可以突破此范围） 3. 在话题转换处创建新分段，不要混入不同话题的内容 4. **重要**：保持句子完整性，分段边界应放在自然停顿处（句号。、问号？、感叹号！、省略号……等标点符号后） - 基于语义停顿判断，不仅依赖标点符号 - 保持内容的连贯性和可读性 5. 保持论点完整性，不要在前提-结论结构中间截断 6. 为主题/章节类型的分段生成简洁标题（最多30字），论点/观点类型分段不加标题 7. **广告识别**：识别UP主的广告植入内容（包括但不限于"转转""京东""得物""淘宝""拼多多"等电商平台推广，以及其他明显的商业推广内容） - 将广告内容单独分为一个段落 - 广告段落的 title 必须设为"广告"或"推广" - 不要将广告内容混入正常内容段落中 【输出格式】 必须返回以下 JSON 格式（不要添加任何额外文本）： ```json {{ "video_type": "{video_type}", "segments": [ {{ "title": "分段标题（如果此分段不需要标题则设为 null）", "content": "分段内容", "start_index": 0, "end_index": 123 }} ], "actual_segment_count": 6 }} ``` 【注意事项】 - 优先保证语义完整性，不要为了凑数而强行截断 - 标题应简洁准确，如「机器学习概述」「梯度下降原理」 - 如果某个分段无法生成合适的标题（如过渡性内容），将 title 设为 null - **广告识别提示**：常见的广告植入特征包括： * 突然话题转换到商品推荐 * 提及"回收""卖""优惠""下单"等电商相关词汇 * 提及具体品牌名称（如转转、京东、得物等） * 说话风格明显变化（如口语化的推销语气） * 内容与视频主题无关的商品推荐""" return prompt def _infer_video_type(self, metadata: dict) -> str: """从元数据推断视频类型 Args: metadata: 视频元数据，包含 title, category, tags 等字段 Returns: 推断出的视频类型，默认为"知识类" Examples: >>> strategy = UnifiedSegmentationStrategy() >>> strategy._infer_video_type({"title": "Python教程"}) '知识类' >>> strategy._infer_video_type({"category": "影视"}) '故事类' """ if not metadata: return DEFAULT_VIDEO_TYPE # 提取所有文本字段 title = metadata.get("title", "") or "" category = metadata.get("category", "") or "" tags = metadata.get("tags", []) or [] description = metadata.get("description", "") or "" # 合并所有文本用于匹配 all_text = " ".join([title, category, description] + list(tags)) # 如果没有有效文本，返回默认类型 if not all_text or all_text.isspace(): return DEFAULT_VIDEO_TYPE # 按优先级匹配类型关键词 for video_type, keywords in VIDEO_TYPE_KEYWORDS.items(): for keyword in keywords: if keyword in all_text: logger.debug( f"Detected video type '{video_type}' " f"from keyword '{keyword}'" ) return video_type # 未匹配到任何类型，返回默认 logger.debug("No specific video type detected, using default") return DEFAULT_VIDEO_TYPE def _parse_result( self, llm_output: str, fallback_segments: List ) -> List[VideoSegment]: """解析 LLM 输出并验证 Args: llm_output: LLM 原始输出 fallback_segments: 已弃用参数，保留仅为向后兼容 Returns: VideoSegment 对象列表 Raises: json.JSONDecodeError: JSON 解析失败 ValueError: 无法提取 JSON ValidationError: Pydantic 验证失败 """ # 提取 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 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 # 分段不需要思考模式 ) # 解析和验证 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: 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]})" raise RuntimeError( f"{error_type}: {error_msg}, LLM语义分段失败" ) from e else: error_msg = str(e) if str(e) else error_type raise RuntimeError( f"{error_type}: {error_msg}, LLM语义分段失败" ) from e 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: # 分块处理失败，直接抛出异常 raise RuntimeError( f"分块 {i}/{len(chunks)} LLM语义分段失败: {type(e).__name__}: {e}" ) from e # 合并结果 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 对象列表 Raises: RuntimeError: 此方法已弃用 """ raise RuntimeError("_fallback_chunk 方法已弃用，LLM语义分段失败将直接抛出异常") # 转换为 VideoSegment，并设置正确的偏移量 segments = [] current_pos = offset for para in paragraphs: # 计算该分段在原文中的位置 seg = VideoSegment( title=None, content=para, start_index=current_pos, end_index=current_pos + len(para) ) segments.append(seg) current_pos += len(para) return segments # ============================================================================ # Pydantic Models # ============================================================================ def extract_json_from_llm_response(response: str) -> dict: """从 LLM 响应中提取 JSON 处理以下情况： - 纯 JSON - 包裹在 ```json ``` 中的 JSON - 前后包含额外说明文本的 JSON Args: response: LLM 响应文本 Returns: 解析后的 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 VideoSegment(BaseModel): """视频分段 Attributes: title: 分段标题（可选） content: 分段内容（必填） start_index: 起始字幕索引 end_index: 结束字幕索引 """ title: Optional[str] = Field( None, description="分段标题，简洁描述型（最多30字）" ) content: str = Field( ..., description="分段内容，合并后的字幕文本" ) start_index: int = Field( ..., ge=0, description="在原文中的起始字符位置" ) end_index: int = Field( ..., ge=0, description="在原文中的结束字符位置" ) @field_validator('title') @classmethod def validate_title_length(cls, v: Optional[str]) -> Optional[str]: """验证标题长度（最多30字）""" if v is not None and len(v) > 30: raise ValueError(f"标题长度不能超过30字，当前为{len(v)}字") return v model_config = { "json_schema_extra": { "examples": [ { "title": "机器学习概述", "content": "机器学习是人工智能的一个分支...", "start_index": 1, "end_index": 10 } ] } } class SegmentationResult(BaseModel): """分段结果 Attributes: video_type: 视频类型（知识类/故事类/访谈类/新闻类/vlog 等） segments: 分段列表 actual_segment_count: 实际分段数量 """ video_type: str = Field( ..., description="识别出的视频类型" ) segments: List[VideoSegment] = Field( default_factory=list, description="分段列表" ) actual_segment_count: int = Field( ..., ge=0, description="实际分段数量" ) model_config = { "json_schema_extra": { "examples": [ { "video_type": "知识类", "segments": [ { "title": "机器学习概述", "content": "...", "start_index": 1, "end_index": 10 } ], "actual_segment_count": 5 } ] } } # ============================================================================ # Video Type Detection # ============================================================================ # 视频类型关键词映射 VIDEO_TYPE_KEYWORDS = { "知识类": [ "教程", "课程", "教学", "入门", "基础", "进阶", "实战", "编程", "开发", "技术", "学习", "教育", "培训", "讲座" ], "故事类": [ "影视", "电影", "电视剧", "动漫", "动画", "小说", "故事", "剧情", "短片", "纪录片" ], "访谈类": [ "访谈", "对话", "采访", "专访", "聊天", "脱口秀", "播客" ], "新闻类": [ "新闻", "报道", "资讯", "时事", "热点", "快讯" ], "vlog": [ "vlog", "日常", "生活", "旅行", "美食", "探店", "分享", "体验" ] } # 默认视频类型 DEFAULT_VIDEO_TYPE = "知识类" # ============================================================================ # Semantic Segmentation Engine # ============================================================================ class SemanticSegmentationEngine: """语义分段引擎（使用策略架构） 使用策略模式实现可扩展的分段架构。 Attributes: strategy: 分段策略（实现 SegmentationStrategy 接口） Examples: >>> # 使用默认统一策略 >>> engine = SemanticSegmentationEngine() >>> paragraphs = engine.segment(segments, metadata) >>> # 使用自定义策略（未来扩展） >>> custom_strategy = KnowledgeTypeStrategy() >>> engine = SemanticSegmentationEngine(strategy=custom_strategy) >>> paragraphs = engine.segment(segments, metadata) Note: Plan 13-06: 实现可扩展策略架构，Phase 13 使用统一策略。 """ def __init__( self, strategy: Optional[SegmentationStrategy] = 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, cache=cache ) else: self.strategy = strategy logger.debug( f"SemanticSegmentationEngine initialized with strategy: " f"{self.strategy.__class__.__name__}" ) def _infer_video_type(self, metadata: dict) -> str: """从元数据推断视频类型（代理到策略） Args: metadata: 视频元数据 Returns: 推断出的视频类型 """ # 如果策略有 _infer_video_type 方法，使用它 if hasattr(self.strategy, '_infer_video_type'): return self.strategy._infer_video_type(metadata) # 否则返回默认类型 return DEFAULT_VIDEO_TYPE def _build_prompt(self, content: str, metadata: dict) -> str: """构建 Prompt（代理到策略） Args: content: 字幕内容 metadata: 视频元数据 Returns: 构建好的 Prompt """ # 如果策略有 _build_prompt 方法，使用它 if hasattr(self.strategy, '_build_prompt'): return self.strategy._build_prompt(content, metadata) # 否则返回基本 prompt return f"Content: {content}\nMetadata: {metadata}" def segment( self, segments: List, metadata: dict ) -> List['VideoSegment']: """对字幕分段进行语义分段 使用当前配置的策略进行分段。 Args: segments: 字幕分段列表（来自 parser.py 的 Segment 对象） metadata: 视频元数据（title, duration 等） Returns: VideoSegment 对象列表（包含标题和内容） """ return self.strategy.segment(segments, metadata)