Content hash cache pattern

何时使用

适用：

• 构建文件处理流水线（PDF 解析、OCR、文本抽取、图像分析），处理成本高且同一文件会被反复处理。
• CLI 工具需要 --cache/--no-cache 开关，按内容跨次运行复用结果。
• 批处理中同一文件在多个批次反复出现。
• 想给一个已有的纯处理函数加缓存，但不愿改动它的签名与职责。

不该用（负边界）：

• 数据必须始终最新（实时行情、消息流等）——缓存会返回过期结果。
• 单条缓存条目极大——{hash}.json 全量读写会吃内存，改用流式/分块存储。
• 结果不只取决于文件内容，还依赖内容之外的参数（如不同的抽取配置、模型版本、语言选项）——纯内容哈希会错误命中；需把这些参数一并并入缓存键。
• 文件内容本身高频变动，几乎每次都未命中——缓存收益为负。

步骤 / 指令

1. 用内容（非路径）算缓存键：对文件内容做 SHA-256，大文件分块读取，避免整文件入内存。
2. 缓存条目用 frozen dataclass 表示（file_hash + source_path + 处理结果）。
3. 按 {hash}.json 落盘：O(1) 按哈希查，无需索引文件；首次写入时惰性建目录。
4. 用独立服务层包裹纯处理函数：服务层做「查缓存 -> 未命中则处理 -> 写缓存」，纯函数对缓存零感知（SRP）。
5. 读缓存时优雅降级：JSON 损坏/反序列化失败一律当未命中返回 None，不崩溃，下次运行自动重算。
6. 用截断哈希（hash[:12]）记录命中/未命中日志，便于排查。

核心约束：

• 哈希内容而非路径——路径会变，内容身份不变。改名/移动 = 缓存命中；内容改动 = 自动失效；无需任何索引文件。
• 处理函数保持纯净，不得在其内部塞入缓存逻辑（否则一函数两职责，违反 SRP）。
• 嵌套 frozen dataclass 不要用 dataclasses.asdict()（复杂嵌套类型会出问题），改用手写序列化以完全掌控格式。

示例

1. 内容哈希缓存键（大文件分块）：

import hashlib
from pathlib import Path

_HASH_CHUNK_SIZE = 65536  # 64KB 分块，适配大文件

def compute_file_hash(path: Path) -> str:
    """文件内容的 SHA-256（分块读取）。"""
    if not path.is_file():
        raise FileNotFoundError(f"File not found: {path}")
    sha256 = hashlib.sha256()
    with open(path, "rb") as f:
        while chunk := f.read(_HASH_CHUNK_SIZE):
            sha256.update(chunk)
    return sha256.hexdigest()

2. 缓存条目（frozen dataclass）+ 文件存储：

from dataclasses import dataclass
import json

@dataclass(frozen=True, slots=True)
class CacheEntry:
    file_hash: str
    source_path: str
    document: "ExtractedDocument"   # 被缓存的处理结果

def write_cache(cache_dir: Path, entry: CacheEntry) -> None:
    cache_dir.mkdir(parents=True, exist_ok=True)  # 惰性建目录
    cache_file = cache_dir / f"{entry.file_hash}.json"
    data = serialize_entry(entry)                 # 手写序列化
    cache_file.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")

def read_cache(cache_dir: Path, file_hash: str) -> CacheEntry | None:
    cache_file = cache_dir / f"{file_hash}.json"
    if not cache_file.is_file():
        return None
    try:
        data = json.loads(cache_file.read_text(encoding="utf-8"))
        return deserialize_entry(data)
    except (json.JSONDecodeError, ValueError, KeyError):
        return None  # 损坏当未命中，下次重算

3. 服务层包裹纯函数（保持 SRP）：

def extract_with_cache(
    file_path: Path,
    *,
    cache_enabled: bool = True,
    cache_dir: Path = Path(".cache"),
) -> "ExtractedDocument":
    """服务层：查缓存 -> 抽取 -> 写缓存。"""
    if not cache_enabled:
        return extract_text(file_path)        # 纯函数，对缓存无感

    file_hash = compute_file_hash(file_path)
    cached = read_cache(cache_dir, file_hash)
    if cached is not None:
        logger.info("Cache hit: %s (hash=%s)", file_path.name, file_hash[:12])
        return cached.document

    logger.info("Cache miss: %s (hash=%s)", file_path.name, file_hash[:12])
    doc = extract_text(file_path)             # 纯函数
    write_cache(cache_dir, CacheEntry(file_hash, str(file_path), doc))
    return doc

反模式（避免）：

# 坏：路径作键 —— 文件移动/改名即失效
cache = {"/path/to/file.pdf": result}

# 坏：把缓存逻辑塞进处理函数 —— 一函数两职责，违反 SRP
def extract_text(path, *, cache_enabled=False, cache_dir=None):
    if cache_enabled: ...

# 坏：嵌套 frozen dataclass 用 asdict() —— 复杂嵌套类型会出错，改用手写序列化
data = dataclasses.asdict(entry)

注意事项

• 哈希内容不哈希路径；大文件务必分块，别整文件读入内存。
• 处理函数保持纯净——缓存是独立关注点，放服务层。
• 缓存损坏/反序列化失败优雅降级为未命中（返回 None），绝不崩溃。
• 命中/未命中均记日志（用截断哈希）便于调试。
• 关键设计取舍：SHA-256 内容哈希=路径无关+内容变即失效；{hash}.json 命名=O(1) 查、免索引；手写 JSON 序列化=完全掌控 frozen dataclass 的格式；mkdir(parents=True)=首写惰性建目录。
• 若结果依赖内容外参数（抽取配置/模型版本），把这些参数一并并入缓存键，否则会错误命中。

互见

• related：file-upload-storage —— 文件上传与存储管理，缓存层常落在其下游处理环节。
• related：python-performance-optimization —— 缓存是消除重复高成本计算的性能手段之一。
• related：clean-code-principles —— 服务层包裹纯函数即单一职责（SRP）的具体落地。
• combines_with：bullmq-job-queue —— 在批处理/后台队列中复用内容哈希缓存，跳过已处理文件。

---

采编自 affaan-m/everything-claude-code（MIT 许可）。