$npx -y skills add virgo777/buddyme --skill content-hash-cache-pattern使用 SHA-256 内容哈希缓存高昂的文件处理结果 —— 与路径无关、自动失效且服务层分离。
| 1 | # 内容哈希文件缓存模式 (Content-Hash File Cache Pattern) |
| 2 | |
| 3 | 使用 SHA-256 内容哈希(而非文件路径)作为缓存键,缓存高昂的文件处理结果(如 PDF 解析、文本提取、图像分析)。与基于路径的缓存不同,这种方法在文件移动/重命名后依然有效,并在内容变化时自动失效。 |
| 4 | |
| 5 | ## 何时激活 |
| 6 | |
| 7 | - 构建文件处理流水线(Pipeline),如 PDF、图像或文本提取 |
| 8 | - 处理成本高昂且需要重复处理相同文件 |
| 9 | - 需要提供 `--cache/--no-cache` 命令行选项 |
| 10 | - 希望在不修改现有纯函数(Pure Functions)的情况下为其添加缓存功能 |
| 11 | |
| 12 | ## 核心模式 |
| 13 | |
| 14 | ### 1. 基于内容哈希的缓存键 |
| 15 | |
| 16 | 使用文件内容(而非路径)作为缓存键: |
| 17 | |
| 18 | ```python |
| 19 | import hashlib |
| 20 | from pathlib import Path |
| 21 | |
| 22 | _HASH_CHUNK_SIZE = 65536 # 大文件采用 64KB 分块读取 |
| 23 | |
| 24 | def compute_file_hash(path: Path) -> str: |
| 25 | """计算文件内容的 SHA-256 哈希值(针对大文件进行分块)。""" |
| 26 | if not path.is_file(): |
| 27 | raise FileNotFoundError(f"File not found: {path}") |
| 28 | sha256 = hashlib.sha256() |
| 29 | with open(path, "rb") as f: |
| 30 | while True: |
| 31 | chunk = f.read(_HASH_CHUNK_SIZE) |
| 32 | if not chunk: |
| 33 | break |
| 34 | sha256.update(chunk) |
| 35 | return sha256.hexdigest() |
| 36 | ``` |
| 37 | |
| 38 | **为什么使用内容哈希?** 文件重命名/移动 = 缓存命中。内容更改 = 自动失效。无需索引文件。 |
| 39 | |
| 40 | ### 2. 用于缓存条目的冻结数据类 (Frozen Dataclass) |
| 41 | |
| 42 | ```python |
| 43 | from dataclasses import dataclass |
| 44 | |
| 45 | @dataclass(frozen=True, slots=True) |
| 46 | class CacheEntry: |
| 47 | file_hash: str |
| 48 | source_path: str |
| 49 | document: ExtractedDocument # 缓存的结果内容 |
| 50 | ``` |
| 51 | |
| 52 | ### 3. 基于文件的缓存存储 |
| 53 | |
| 54 | 每个缓存条目存储为 `{hash}.json` —— 通过哈希实现 O(1) 查询,无需索引文件。 |
| 55 | |
| 56 | ```python |
| 57 | import json |
| 58 | from typing import Any |
| 59 | |
| 60 | def write_cache(cache_dir: Path, entry: CacheEntry) -> None: |
| 61 | cache_dir.mkdir(parents=True, exist_ok=True) |
| 62 | cache_file = cache_dir / f"{entry.file_hash}.json" |
| 63 | data = serialize_entry(entry) |
| 64 | cache_file.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") |
| 65 | |
| 66 | def read_cache(cache_dir: Path, file_hash: str) -> CacheEntry | None: |
| 67 | cache_file = cache_dir / f"{file_hash}.json" |
| 68 | if not cache_file.is_file(): |
| 69 | return None |
| 70 | try: |
| 71 | raw = cache_file.read_text(encoding="utf-8") |
| 72 | data = json.loads(raw) |
| 73 | return deserialize_entry(data) |
| 74 | except (json.JSONDecodeError, ValueError, KeyError): |
| 75 | return None # 将损坏的缓存视为未命中 |
| 76 | ``` |
| 77 | |
| 78 | ### 4. 服务层封装 (满足单一职责原则 SRP) |
| 79 | |
| 80 | 保持处理函数为纯函数。将缓存逻辑作为独立的服务层添加。 |
| 81 | |
| 82 | ```python |
| 83 | def extract_with_cache( |
| 84 | file_path: Path, |
| 85 | *, |
| 86 | cache_enabled: bool = True, |
| 87 | cache_dir: Path = Path(".cache"), |
| 88 | ) -> ExtractedDocument: |
| 89 | """服务层逻辑:检查缓存 -> 提取内容 -> 写入缓存。""" |
| 90 | if not cache_enabled: |
| 91 | return extract_text(file_path) # 纯函数,不感知缓存逻辑 |
| 92 | |
| 93 | file_hash = compute_file_hash(file_path) |
| 94 | |
| 95 | # 检查缓存 |
| 96 | cached = read_cache(cache_dir, file_hash) |
| 97 | if cached is not None: |
| 98 | logger.info("Cache hit: %s (hash=%s)", file_path.name, file_hash[:12]) |
| 99 | return cached.document |
| 100 | |
| 101 | # 缓存未命中 -> 提取 -> 存储 |
| 102 | logger.info("Cache miss: %s (hash=%s)", file_path.name, file_hash[:12]) |
| 103 | doc = extract_text(file_path) |
| 104 | entry = CacheEntry(file_hash=file_hash, source_path=str(file_path), document=doc) |
| 105 | write_cache(cache_dir, entry) |
| 106 | return doc |
| 107 | ``` |
| 108 | |
| 109 | ## 关键设计决策 |
| 110 | |
| 111 | | 决策 | 理由 | |
| 112 | |----------|-----------| |
| 113 | | SHA-256 内容哈希 | 与路径无关,内容更改时自动失效 | |
| 114 | | `{hash}.json` 文件命名 | O(1) 查询速度,无需维护索引文件 | |
| 115 | | 服务层封装 (Wrapper) | 满足单一职责原则(SRP):提取逻辑保持纯净,缓存作为独立关注点 | |
| 116 | | 手动 JSON 序列化 | 对冻结数据类的序列化拥有完全控制权 | |
| 117 | | 损坏返回 `None` | 优雅降级,在下次运行时重新处理 | |
| 118 | | `cache_dir.mkdir(parents=True)` | 在首次写入时延迟创建目录 | |
| 119 | |
| 120 | ## 最佳实践 |
| 121 | |
| 122 | - **哈希内容而非路径** —— 路径会变,内容身份(Identity)不变。 |
| 123 | - **对大文件进行分块哈希** —— 避免将整个文件加载进内存。 |
| 124 | - **保持处理函数纯净** —— 它们不应知道任何关于缓存的信息。 |
| 125 | - **记录缓存命中/未命中日志** —— 使用截断后的哈希值以便调试。 |
| 126 | - **优雅处理缓存损坏** —— 将无效的缓存条目视为未命中,绝不要因此崩溃。 |
| 127 | |
| 128 | ## 应避免的反模式 (Anti-Patterns) |
| 129 | |
| 130 | ```python |
| 131 | # 错误做法:基于路径的缓存(文件移动/重命名后失效) |
| 132 | cache = {"/path/to/file.pdf": result} |
| 133 | |
| 134 | # 错误做法:在处理函数内部添加缓存逻辑(违反 SRP) |
| 135 | def extract_text(path, *, cache_enabled=False, cache_dir=None): |
| 136 | if cache_enabled: # 该函数现在有了双重职责 |
| 137 | ... |
| 138 | |
| 139 | # 错误做法:对嵌套的冻结数据类直接使用 dataclasses.asdict() |
| 140 | # (在处理复杂的嵌套类型时可能会出现问题) |
| 141 | data = dataclasses.asdict(entry) # 推荐使用手动序列化 |
| 142 | ``` |
| 143 | |
| 144 | ## 适用场景 |
| 145 | |
| 146 | - 文件处理流水线(PDF 解析、OCR、文本提取、图像分析) |
| 147 | - 受益于 `--cache/--no-cache` 选项的命令行工具 (CLI) |
| 148 | - 在不同运行周期中会出现相同文件的批处理任务 |
| 149 | - 在不修改现有纯函数的情况下为其添加缓存功能 |
| 150 | |
| 151 | ## 不适用场景 |
| 152 | |
| 153 | - 必须保持绝对实时的数据(实时数据流) |
| 154 | - 缓存条目体积极其庞大(此时应考虑流式处理) |
| 155 | - 结果取决于文件内容以外的参数(例如不同的提取配置) |