From baf7c0f521b26f1a851512b5c9b528e7b1609ba6 Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Mon, 23 Mar 2026 22:35:57 +0800 Subject: [PATCH] feat(14-03): implement SemanticCache for content-based caching MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - compute_content_hash(): 计算 SHA256 内容哈希 - SemanticCache.get(): 从缓存读取 SegmentationResult - SemanticCache.set(): 保存 SegmentationResult 到缓存 - 支持版本控制（CACHE_VERSION = '1.0'） - 缓存存储在 .planning/cache/semantic/ 目录 - 错误容错：缓存失败不影响主流程 - 自动创建缓存目录 遵守用户决策： - 缓存键: SHA256 内容哈希 - 缓存存储: 本地文件系统 - 缓存 TTL: 永久有效 - 缓存内容: 完整的 SegmentationResult Co-Authored-By: Claude Opus 4.6 --- kb_workflow/srt/cache/__init__.py | 8 ++ kb_workflow/srt/cache/semantic_cache.py | 155 ++++++++++++++++++++++++ 2 files changed, 163 insertions(+) create mode 100644 kb_workflow/srt/cache/__init__.py create mode 100644 kb_workflow/srt/cache/semantic_cache.py diff --git a/kb_workflow/srt/cache/__init__.py b/kb_workflow/srt/cache/__init__.py new file mode 100644 index 0000000..17f5bc0 --- /dev/null +++ b/kb_workflow/srt/cache/__init__.py @@ -0,0 +1,8 @@ +"""语义分段缓存模块""" + +from kb_workflow.srt.cache.semantic_cache import ( + SemanticCache, + compute_content_hash +) + +__all__ = ['SemanticCache', 'compute_content_hash'] diff --git a/kb_workflow/srt/cache/semantic_cache.py b/kb_workflow/srt/cache/semantic_cache.py new file mode 100644 index 0000000..d9a87d6 --- /dev/null +++ b/kb_workflow/srt/cache/semantic_cache.py @@ -0,0 +1,155 @@ +"""语义分段缓存管理器 + +使用内容哈希（SHA256）作为缓存键，存储在本地文件系统。 + +功能： +- 计算内容的 SHA256 哈希作为缓存键 +- 缓存 SegmentationResult 到本地文件系统 +- 支持版本控制，便于未来格式变更 +- 错误容错，缓存失败不影响主流程 +""" + +from __future__ import annotations + +import hashlib +import json +import logging +from pathlib import Path +from typing import Optional + +from pydantic import ValidationError + +from kb_workflow.srt.semantic_engine import SegmentationResult + +logger = logging.getLogger(__name__) + + +def compute_content_hash(content: str) -> str: + """计算内容的 SHA256 哈希作为缓存键 + + Args: + content: 字幕内容 + + Returns: + SHA256 哈希值（64 字符的十六进制字符串） + + Examples: + >>> hash1 = compute_content_hash("相同内容") + >>> hash2 = compute_content_hash("相同内容") + >>> hash1 == hash2 + True + >>> len(hash1) + 64 + """ + return hashlib.sha256(content.encode('utf-8')).hexdigest() + + +class SemanticCache: + """语义分段结果缓存管理器 + + 使用内容哈希（SHA256）作为缓存键，存储在本地文件系统。 + + 缓存格式： + - 文件命名: {sha256_hash}.json + - 文件内容: JSON 格式的 SegmentationResult + version 字段 + - 缓存 TTL: 永久有效，手动清理 + + Attributes: + CACHE_VERSION: 缓存格式版本，用于未来兼容性 + DEFAULT_CACHE_DIR: 默认缓存目录 + + Examples: + >>> cache = SemanticCache() + >>> result = cache.get(subtitle_content) + >>> if result is None: + ... result = llm_segment(subtitle_content) + ... cache.set(subtitle_content, result) + """ + + 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 + + Examples: + >>> cache = SemanticCache() + >>> result = cache.get("字幕内容") + >>> if result: + ... print(f"从缓存读取: {result.video_type}") + """ + 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: 分段结果 + + Examples: + >>> cache = SemanticCache() + >>> result = SegmentationResult(video_type="知识类", segments=[], actual_segment_count=0) + >>> cache.set("字幕内容", 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}")