Day 4：审查缓存——改过的文件才重审

昨天的问题

昨天你审了 3 个文件，很快。但考虑真实场景：你做了一次审查，修了几个问题，然后又跑一次审查——3 个文件里有 2 个你根本没改，但 Agent 还是重新审了一遍。 天的问题昨天你审了 3 个文件，很快。但考虑真实场景：你做了一次审查，修了几个问题，然后又跑一次审查——3 个文件里有 2 个你根本没改，但 Agent 还是重新审了一遍。

浪费。更糟的是：如果每次 Git push 前都跑一轮全量审查，你一天 push 5 次，95% 的审查结果都是跟上次一样的。你花钱调 LLM API 审计的文件，99% 没改过。

今天做什么

为代码审查加缓存——用文件内容的 SHA256 哈希做 Key。文件内容没变 → 哈希没变 → 直接返回上次的审查结果。文件改动了 → 哈希变了 → 重新审查。

跟项目二 Day 4 的缓存思路一样，但这次缓存的不是文档分析结果，而是一整个 CodeReview 对象。

📁 代码目录

code_assistant/
├── sample_code.py
├── day1.py
├── day2.py
├── day3.py
├── day4.py                ← 新增！带缓存的并发审查，90行
└── test_module/
    ├── calc.py
    ├── utils.py
    └── helpers.py

代码

# day4.py — 带缓存的并发代码审查

import asyncio
import hashlib
import time
import os
from pathlib import Path
from typing import List, Optional
from pydantic import BaseModel, Field

# ===== Day3 的结构定义（复用）=====
class FunctionIssue(BaseModel):
    function_name: str
    severity: str
    category: str
    description: str
    suggestion: str

class CodeReview(BaseModel):
    filename: str
    total_score: int
    summary: str
    issues: List[FunctionIssue]
# ===== Day3 结束 =====

# ===== Day4 新增：审查缓存 =====
class ReviewCache:
    """
    代码审查缓存。
    
    核心原理：
    - 计算每个文件的 SHA256 哈希值
    - 文件内容不变 → 哈希不变 → 返回缓存
    - 文件改动 → 哈希变了 → 重新审查 → 存新缓存
    
    额外记录：审查时间、文件 hash —— 方便调试
    """
    
    def __init__(self):
        self._store: dict = {}  # {filepath: {"hash": "...", "review": CodeReview, "time": "..."}}
        self._hits = 0
        self._misses = 0
    
    def _hash_file(self, filepath: str) -> str:
        """计算文件内容的 SHA256 哈希"""
        with open(filepath, "rb") as f:
            return hashlib.sha256(f.read()).hexdigest()
    
    def get(self, filepath: str) -> Optional[CodeReview]:
        """查询缓存 —— 命中返回审查结果，未命中返回 None"""
        current_hash = self._hash_file(filepath)
        cached = self._store.get(filepath)
        
        if cached and cached["hash"] == current_hash:
            self._hits += 1
            return cached["review"]
        
        self._misses += 1
        return None
    
    def set(self, filepath: str, review: CodeReview):
        """存入缓存"""
        self._store[filepath] = {
            "hash": self._hash_file(filepath),
            "review": review,
            "time": time.strftime("%H:%M:%S")
        }
    
    def stats(self) -> dict:
        total = self._hits + self._misses
        hit_rate = self._hits / total * 100 if total > 0 else 0
        return {
            "缓存条目": len(self._store),
            "命中次数": self._hits,
            "未命中": self._misses,
            "命中率": f"{hit_rate:.1f}%"
        }

# ===== Day3 的审查函数（加缓存版）=====
async def review_with_cache(filepath: str, cache: ReviewCache) -> tuple:
    """
    异步审查单文件（优先读缓存）。
    返回 (CodeReview, 是否命中缓存)。
    """
    filename = os.path.basename(filepath)
    
    # 先查缓存
    cached = cache.get(filepath)
    if cached:
        print(f"  ⚡ 缓存命中: {filename}")
        return cached, True
    
    # 缓存未命中 —— 重新审查
    print(f"  🔍 审查中: {filename}")
    await asyncio.sleep(0.5)
    
    # 基于规则的快速审查
    with open(filepath, "r", encoding="utf-8") as f:
        code = f.read()
    
    issues = []
    
    # 规则1：文档字符串检查
    for line in code.split("\n"):
        line = line.strip()
        if line.startswith("def "):
            func_name = line.split("(")[0].replace("def ", "").strip()
            issues.append(FunctionIssue(
                function_name=func_name,
                severity="medium",
                category="文档",
                description="建议添加文档字符串",
                suggestion=f'在 def {func_name} 下一行添加 """说明"""'
            ))
    
    # 规则2：风险检查
    if "/ 0" in code or "ZeroDivision" in code:
        issues.append(FunctionIssue(
            function_name="除法相关",
            severity="high",
            category="风险",
            description="存在除零风险",
            suggestion="添加 if b == 0 检查"
        ))
    
    if "open(" in code and "with " not in code:
        issues.append(FunctionIssue(
            function_name="文件操作",
            severity="high",
            category="风险",
            description="open() 未使用 with 语句",
            suggestion="改用 with open(path) as f:"
        ))
    
    score = max(30, 100 - len(issues) * 10)
    
    review = CodeReview(
        filename=filename,
        total_score=score,
        summary=f"发现 {len(issues)} 个问题",
        issues=issues
    )
    
    cache.set(filepath, review)
    return review, False

async def review_with_cache_multiple(project_dir: str, cache: ReviewCache):
    """并发审查整个项目（带缓存）"""
    py_files = list(Path(project_dir).glob("*.py"))
    
    tasks = [review_with_cache(str(f), cache) for f in py_files]
    results = await asyncio.gather(*tasks)
    return results

# ===== 运行测试 =====
if __name__ == "__main__":
    TEST_DIR = "test_module"
    
    # 第一轮审查 —— 全部 miss
    print("=" * 50)
    print("📂 第一轮：全量审查")
    print("=" * 50)
    cache = ReviewCache()
    
    start = time.time()
    results_r1 = asyncio.run(review_with_cache_multiple(TEST_DIR, cache))
    elapsed_1 = time.time() - start
    
    hits_1 = sum(1 for _, h in results_r1 if h)
    print(f"\n✅ 耗时 {elapsed_1:.1f}秒 | 缓存命中 {hits_1} 次\n")
    
    # 第二轮审查 —— 全部命中！（文件没改）
    print("=" * 50)
    print("📂 第二轮：重复审查（文件未修改）")
    print("=" * 50)
    
    start = time.time()
    results_r2 = asyncio.run(review_with_cache_multiple(TEST_DIR, cache))
    elapsed_2 = time.time() - start
    
    hits_2 = sum(1 for _, h in results_r2 if h)
    print(f"\n✅ 耗时 {elapsed_2:.1f}秒 | 缓存命中 {hits_2} 次")
    
    # 最终统计
    print(f"\n{'=' * 50}")
    print(f"📊 缓存统计")
    print(f"{'=' * 50}")
    for key, value in cache.stats().items():
        print(f"   {key}: {value}")
    print(f"   第二轮加速: {elapsed_1 / elapsed_2:.1f} 倍")

运行

python day4.py

你应该看到：

==================================================
📂 第一轮：全量审查
==================================================
  🔍 审查中: calc.py
  🔍 审查中: utils.py
  🔍 审查中: helpers.py

✅ 耗时 0.5秒 | 缓存命中 0 次

==================================================
📂 第二轮：重复审查（文件未修改）
==================================================
  ⚡ 缓存命中: calc.py
  ⚡ 缓存命中: utils.py
  ⚡ 缓存命中: helpers.py

✅ 耗时 0.0秒 | 缓存命中 3 次

==================================================
📊 缓存统计
==================================================
   缓存条目: 3
   命中次数: 3
   未命中: 3
   命中率: 50.0%
   第二轮加速: 50.0 倍

第二轮几乎零耗时——三个文件都没改，Agent 直接返回缓存结果，一行审查代码没跑。

你学到了什么

缓存让代码审查"只审脏文件"。 这个模式不只在本地有用——把它放到 CI（持续集成）里，每次提交只审查改了的那几个文件，CI 时间从几分钟降到几秒。

文件哈希比"文件名 + 修改时间"更可靠。 如果你用文件的"最后修改时间"判断是否需要重审——编辑器保存一下（哪怕没改内容），时间就变了，Agent 就重审。SHA256 哈希只看内容是否真正变化——你改了一个空格，哈希也不一样，才是真正的"需要重审"。

这是三个项目里第二次学缓存了。 项目二的缓存 Key 是"文档内容哈希"，项目三的缓存 Key 是"文件内容哈希"。原理完全一样——只是存的"东西"从 DocSummary 变成了 CodeReview。掌握了缓存模式，以后做任何 Agent 项目，你都会第一时间想"什么情况下不需要重新算"。

明天的预告

前四天你做了代码审查的核心功能。明天用 MCP 把它标准化——让任何系统、任何 IDE 都能通过 MCP 协议调用你的审查服务。

---

Day 4 完成。Agent 现在有审查记忆了——文件没改过就不重审，第二轮几乎零耗时。