feat(14-05): integrate cache into UnifiedSegmentationStrategy
- Add cache parameter to UnifiedSegmentationStrategy.__init__() - Add cache logic to segment() method (check before LLM, save after) - Pass cache parameter through SemanticSegmentationEngine.__init__() - Implement missing chunking methods (_segment_single_chunk, _segment_multiple_chunks, etc.) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -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
|
||||
|
||||
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=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"降级到规则分段"
|
||||
)
|
||||
# 检查是否需要分块（Plan 14-04: 分块阈值 5000 字符）
|
||||
if len(content) <= 5000:
|
||||
# 短文本直接处理
|
||||
video_segments = self._segment_single_chunk(content, metadata, segments)
|
||||
else:
|
||||
error_msg = str(e) if str(e) else error_type
|
||||
logger.warning(
|
||||
f"{error_type}: {error_msg}, 降级到规则分段"
|
||||
)
|
||||
break
|
||||
# 长文本分块处理
|
||||
video_segments = self._segment_multiple_chunks(content, metadata, segments)
|
||||
|
||||
# 降级到规则分段
|
||||
logger.info("使用规则分段（降级）")
|
||||
paragraphs = self.fallback_segmenter.segment(segments)
|
||||
return [
|
||||
VideoSegment(title=None, content=para, start_index=0, end_index=0)
|
||||
for para in paragraphs
|
||||
]
|
||||
# 保存到缓存（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
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user