Day 3：并发审查——一个项目的代码一起审
昨天的问题
Day 2 的审查链一次只审一个文件。但真实项目通常有几个甚至几十个 Python 文件。审完一个再一个，跟
项目二的串行分析是一个问题——时间越来越长。 天的问题Day 2 的审查链一次只审一个文件。但真实项目通常有几个甚至几十个 Python 文件。审完一个再一个，跟项目二的串行分析是一个问题——时间越来越长。
更关键的是，代码审查有个天然属性：文件之间互不依赖。 审 models.py 不需要等 views.py 审完——两者内容独立，完全可以同时进行。这就是并发的最佳场景。
今天做什么
用 asyncio.gather() 同时审查多个 Python 文件。先扫描目录下所有 .py 文件，同时启动审查协程，等所有结果回来后生成一份汇总报告。
这个模式跟项目二 Day 3 几乎一样，但输入从"文档内容"变成了"文件路径 + 代码内容"。
📁 代码目录
code_assistant/
├── sample_code.py ← Day1 的示例代码
├── day1.py
├── day2.py
├── day3.py ← 新增！并发审查，60行
└── test_module/
├── calc.py ← 手动创建或自动生成
└── utils.py代码
python
# day3.py — 并发代码审查
import asyncio
import time
import os
from pathlib import Path
from typing import List
from pydantic import BaseModel, Field
# ===== Day2 的结构定义（复用）=====
class FunctionIssue(BaseModel):
function_name: str
severity: str
category: str
description: str
suggestion: str
class CodeReview(BaseModel):
filename: str = Field(description="文件名") # 重要：多文件时需要区分来源
total_score: int
summary: str
issues: List[FunctionIssue]
# ===== Day2 结束 =====
# ===== Day3 新增：准备测试文件 =====
TEST_DIR = "test_module"
os.makedirs(TEST_DIR, exist_ok=True)
files_to_create = {
"calc.py": '''"""计算模块"""
def add(a, b):
return a + b
def divide(a, b):
"""除法运算"""
return a / b # 缺少除零检查
def calculate_average(numbers):
result = sum(numbers) / len(numbers)
return result
''',
"utils.py": '''"""工具函数"""
def format_date(year, month, day):
return f"{year}-{month}-{day}"
def read_file(path):
f = open(path, "r") # 未用 with，可能忘记关闭
content = f.read()
f.close()
return content
def process_items(items):
result = []
for i in items:
result.append(i)
return result
''',
"helpers.py": '''"""辅助函数"""
def validate_email(email):
"""验证邮箱格式（不完整实现）"""
if "@" in email:
return True
return False
GREETING = "Hello"
def say_hello(name):
return GREETING + " " + name
'''
}
for fname, content in files_to_create.items():
fpath = os.path.join(TEST_DIR, fname)
if not os.path.exists(fpath):
with open(fpath, "w", encoding="utf-8") as f:
f.write(content)
print(f"📝 已创建: {fpath}")
# ===== Day3 新增：并发审查引擎 =====
async def review_one_file(filepath: str) -> CodeReview:
"""
异步审查单个文件。
在真实场景中，这里调用 LLM API 进行审查。
今天用简单规则模拟审查结果。
"""
filename = os.path.basename(filepath)
print(f" 🔍 审查中: {filename}")
# 读取代码
with open(filepath, "r", encoding="utf-8") as f:
code = f.read()
# 模拟审查耗时
await asyncio.sleep(0.5)
# 基于简单规则的审查
issues = []
# 规则1：检查是否缺少文档字符串
if 'def ' in code:
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="divide / calculate",
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)
print(f" ✅ 审查完成: {filename}（{len(issues)} 个问题，{score}分）")
return CodeReview(
filename=filename,
total_score=score,
summary=f"发现 {len(issues)} 个问题",
issues=issues[:5] # 最多5个
)
async def review_project(project_dir: str) -> List[CodeReview]:
"""
并发审查整个项目的所有 Python 文件。
"""
# 1. 扫描所有 .py 文件
py_files = list(Path(project_dir).glob("*.py"))
if not py_files:
print("❌ 没有找到 Python 文件")
return []
print(f"📂 找到 {len(py_files)} 个 Python 文件\n")
# 2. 并发审查 ⭐
tasks = [review_one_file(str(f)) for f in py_files]
results = await asyncio.gather(*tasks)
return results
# ===== 运行 =====
if __name__ == "__main__":
start = time.time()
results = asyncio.run(review_project(TEST_DIR))
elapsed = time.time() - start
# 汇总报告
print(f"\n{'=' * 50}")
print(f"📊 项目审查汇总")
print(f"{'=' * 50}")
print(f" 审查文件数: {len(results)}")
print(f" 总耗时: {elapsed:.1f} 秒")
if results:
print(f" 如果串行: {len(results) * 0.5:.0f} 秒")
print(f" 提速: {len(results) * 0.5 / elapsed:.1f} 倍")
# 每文件评分
total_issues = 0
for review in results:
total_issues += len(review.issues)
print(f"\n 📄 {review.filename}: {review.total_score}分")
for issue in review.issues:
icon = {"high": "🔴", "medium": "🟡", "low": "🟢"}.get(issue.severity, "⚪")
print(f" {icon} {issue.function_name}: {issue.description[:40]}...")
print(f"\n 总计发现: {total_issues} 个问题")运行
bash
python day3.py你应该看到：
📝 已创建: test_module/calc.py
📝 已创建: test_module/utils.py
📝 已创建: test_module/helpers.py
📂 找到 3 个 Python 文件
🔍 审查中: calc.py
🔍 审查中: utils.py
🔍 审查中: helpers.py
✅ 审查完成: calc.py（3 个问题，70分）
✅ 审查完成: utils.py（3 个问题，70分）
✅ 审查完成: helpers.py（2 个问题，80分）
==================================================
📊 项目审查汇总
==================================================
审查文件数: 3
总耗时: 0.5 秒
如果串行: 2 秒
提速: 3.0 倍
📄 calc.py: 70分
🟡 add: 建议添加文档字符串...
🟡 divide: 建议添加文档字符串...
🔴 divide: 存在除零风险...
📄 utils.py: 70分
🟡 format_date: 建议添加文档字符串...
🟡 read_file: 建议添加文档字符串...
🔴 文件操作: 使用 open() 但未使用 with...
📄 helpers.py: 80分
🟡 validate_email: 建议添加文档字符串...
🟡 say_hello: 建议添加文档字符串...
总计发现: 8 个问题你学到了什么
asyncio.gather() 的模式在三个项目中一模一样。 项目一用它同时处理多用户对话，项目二用它同时分析多份文档，项目三用它同时审查多个代码文件。底层原理相同——IO 等待期间切换协程——但应用场景完全不同。你现在已经能"条件反射"——遇到多个互不依赖的任务，第一时间想到 asyncio.gather()。
简单规则审查是真实 AI 审查的骨架。 今天用的规则（缺文档字符串、除零风险、文件未关闭）虽然简单，但它们是你未来 Prompt 里要告诉 LLM 的"审查标准"。把今天的规则翻译成自然语言放进 Prompt，就是你的第一条 LLM 审查指令。
多文件审查的评测价值： 三个文件一起审，你一眼就能看出哪个文件质量高（helper.py 80 分）、哪个需要多改改（calc.py 70 分）。这就是 CI/CD 里"代码质量门禁"的雏形——低于 60 分不让合入。
明天的预告
今天的审查规则是硬编码在 review_one_file() 里的。明天用缓存机制——只有修改过的文件才重新审查，没动过的直接拿上一次的结果。
Day 3 完成。现在你能审查整个项目——所有 Python 文件同时审，3 份文件不到 1 秒出结果。
