# 2.4 实战：代码审查流水线 > **第 2 周 · 第 4 课 · 实战：代码审查流水线 · 预计时长: 45 分钟** --- ## 学习目标 - 掌握用 CrewAI 构建真实工作场景的多 Agent 流水线 - 理解代码生成、代码审查、测试编写三个角色的协作方式 - 学会让 Agent 输出结构化结果（JSON / 代码块） - 掌握让 Agent 更有效的实用技巧 --- ## 为什么需要代码审查流水线？ 在真实开发中，一段代码从 "写完" 到 "可上线" 通常经过： ``` [需求/规格] → [写代码] → [代码审查] → [写测试] → [合并] ``` 这个流程通常由不同的人完成。现在我们用 3 个 AI Agent 来自动化它： ``` ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ Agent 1 │ │ Agent 2 │ │ Agent 3 │ │ 开发工程师 │────→│ 审查专家 │────→│ 测试工程师 │ │ (写代码) │ │ (找问题) │ │ (写测试) │ └─────────────┘ └─────────────┘ └─────────────┘ ``` **每个 Agent 的职责：** | Agent | 角色 | 输入 | 输出 | |-------|------|------|------| | 开发工程师 | 根据规格写代码 | 功能规格描述 | 完整可运行的代码 | | 审查专家 | 审查代码质量 | 开发工程师的代码 | 审查报告（bug、风格、安全） | | 测试工程师 | 编写测试用例 | 代码 + 审查报告 | 完整的测试代码 | --- ## 完整代码 ```python # 文件: code_review_pipeline.py # # 代码审查流水线 — 生成代码 → 审查 → 写测试 # # 用法: uv run python code_review_pipeline.py # 可选参数: 自定义功能规格 # uv run python code_review_pipeline.py "写一个用户注册 API" import sys from crewai import Agent, Task, Crew, Process # ======================== # 配置 # ======================== # 默认功能规格，可通过命令行参数覆盖 DEFAULT_SPEC = ( "用 Python 写一个 URL 验证工具函数。" "要求：1) 支持 http/https/ftp 协议 2) 支持 IPv4 和域名 " "3) 支持端口号（可选） 4) 支持路径和查询参数 " "5) 无效 URL 返回 False，有效 URL 返回 True" ) spec = sys.argv[1] if len(sys.argv) > 1 else DEFAULT_SPEC print(f"\n{'=' * 60}") print(f"功能规格: {spec}") print(f"{'=' * 60}\n") # ======================== # 第 1 步：定义三个 Agent # ======================== # --- Agent 1: 开发工程师 --- developer = Agent( role="高级 Python 开发工程师", goal="根据功能规格编写高质量、可维护的 Python 代码", backstory=( "你是一位有 10 年经验的 Python 开发工程师，专注于编写干净、" "高效、符合 PEP 8 规范的代码。" "你的代码风格：类型注解完整、错误处理完善、函数职责单一。" "你习惯先写 docstring，再实现逻辑。" "输出代码时，用 ```python 代码块包裹，确保可以直接复制运行。" ), verbose=True, allow_delegation=False, ) # --- Agent 2: 代码审查专家 --- reviewer = Agent( role="代码审查专家", goal="全面审查代码质量，发现 bug、风格问题和安全漏洞", backstory=( "你是一位资深代码审查专家，曾在多个大型开源项目中担任 reviewer。" "你的审查维度包括：\n" "- Bug：逻辑错误、边界条件未处理、异常处理缺失\n" "- 风格：PEP 8 合规性、命名规范、注释质量\n" "- 安全：输入验证、注入风险、敏感信息处理\n" "- 性能：不必要的时间/空间复杂度、可优化的热点\n" "你的报告格式：每个问题按严重程度分级（CRITICAL/HIGH/MEDIUM/LOW），" "并给出具体修复建议。" ), verbose=True, allow_delegation=False, ) # --- Agent 3: 测试工程师 --- test_engineer = Agent( role="高级测试工程师", goal="为代码编写全面的单元测试，覆盖正常路径和异常路径", backstory=( "你是一位有 8 年经验的测试工程师，擅长用 pytest 编写测试。" "你的测试策略：\n" "- 正常路径：典型输入的正确输出\n" "- 边界条件：空值、最大值、最小值、特殊字符\n" "- 异常路径：无效输入、错误类型、网络异常等\n" "- 每个测试函数独立、命名清晰、包含断言说明\n" "输出代码时，用 ```python 代码块包裹，确保可以直接运行 pytest。" ), verbose=True, allow_delegation=False, ) # ======================== # 第 2 步：定义流水线任务 # ======================== # --- Task 1: 生成代码 --- code_generation = Task( description=( f"根据以下功能规格编写 Python 代码：\n\n{spec}\n\n" "要求：\n" "1. 代码必须完整、可独立运行（不依赖外部文件）\n" "2. 包含完整的类型注解\n" "3. 包含 docstring，说明函数用途、参数和返回值\n" "4. 包含必要的错误处理（try/except）\n" "5. 用 ```python 代码块包裹代码，确保可以直接复制\n" "6. 在代码块之后，附上一个使用示例\n" ), expected_output="完整的 Python 代码（用 ```python 代码块包裹）+ 使用示例", agent=developer, ) # --- Task 2: 代码审查 --- code_review = Task( description=( "请审查以下代码。审查维度：\n" "1. Bug — 逻辑错误、边界条件、异常处理\n" "2. 风格 — PEP 8、命名规范、注释\n" "3. 安全 — 输入验证、注入风险\n" "4. 性能 — 可优化的地方\n\n" "输出格式：\n" "## 审查结果\n" "\n" "### CRITICAL (必须修复)\n" "- [问题描述] — [位置] — [修复建议]\n" "\n" "### HIGH (应该修复)\n" "- [问题描述] — [位置] — [修复建议]\n" "\n" "### MEDIUM (建议修复)\n" "- [问题描述] — [位置] — [修复建议]\n" "\n" "### LOW (可选优化)\n" "- [问题描述] — [位置] — [修复建议]\n" "\n" "### 总体评价\n" "[一句话总结代码质量]\n" "\n" "注意：如果某个级别没有问题，写 '无'。\n" ), expected_output=( "结构化的代码审查报告，按严重程度列出所有问题和修复建议" ), agent=reviewer, ) # --- Task 3: 编写测试 --- test_generation = Task( description=( "请为以下代码编写完整的 pytest 单元测试。\n" "同时参考代码审查报告，确保测试覆盖了审查中发现的问题场景。\n\n" "要求：\n" "1. 使用 pytest 框架\n" "2. 至少包含以下测试类别：\n" " - 正常路径测试（happy path）\n" " - 边界条件测试\n" " - 异常路径测试\n" "3. 每个测试函数：\n" " - 命名格式: test_{功能}_{场景}\n" " - 包含断言和注释说明\n" "4. 用 ```python 代码块包裹代码\n" "5. 在代码块之后，附上运行命令说明\n" ), expected_output="完整的 pytest 测试代码（用 ```python 代码块包裹）", agent=test_engineer, ) # ======================== # 第 3 步：创建 Crew 并执行 # ======================== crew = Crew( agents=[developer, reviewer, test_engineer], tasks=[code_generation, code_review, test_generation], process=Process.sequential, verbose=True, ) # 启动流水线！ print("流水线启动...\n") result = crew.kickoff() # 打印完整结果 print("\n" + "=" * 60) print("流水线完成 — 完整输出") print("=" * 60) print(result) ``` --- ## 运行方式 ```bash # 使用默认规格（URL 验证工具） uv run python code_review_pipeline.py # 自定义功能规格 uv run python code_review_pipeline.py "写一个邮箱地址验证函数" uv run python code_review_pipeline.py "写一个 CSV 文件读取和解析函数" uv run python code_review_pipeline.py "写一个 JSON 数据校验器（用 JSON Schema）" ``` --- ## 典型输出示例 ``` ============================================================ 功能规格: 用 Python 写一个 URL 验证工具函数... ============================================================ 流水线启动... ## [开发工程师输出] ```python import re from typing import Optional def is_valid_url(url: str) -> bool: """ 验证给定的 URL 是否合法。 支持的协议: http, https, ftp 支持: IPv4 地址、域名、端口号（可选）、路径、查询参数 参数: url: 待验证的 URL 字符串 返回: bool: 如果 URL 有效返回 True，否则返回 False 示例: >>> is_valid_url("https://www.example.com/path?query=1") True >>> is_valid_url("not-a-url") False """ if not isinstance(url, str): return False pattern = re.compile( r'^(https?|ftp)://' # 协议 r'(' r'(?P' # 域名 r'(?:[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?\.)+' r'[a-zA-Z]{2,6}' r'|' r'(?P' # IPv4 r'(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}' r'(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)' r')' r')' r'(?::(?P\d{1,5}))?' # 端口（可选） r'(?:/(?P[^\s?]*))?' # 路径（可选） r'(?:\?(?P[^\s]*))?' # 查询参数（可选） r'$', re.IGNORECASE ) return bool(pattern.match(url)) # 使用示例 if __name__ == "__main__": test_urls = [ "https://www.example.com", "http://192.168.1.1:8080/api/data?key=value", "ftp://files.example.com/docs/readme.txt", "not-a-url", "https://", ] for url in test_urls: print(f"{url}: {is_valid_url(url)}") ``` ## [审查专家输出] ### 审查结果 ### CRITICAL (必须修复) 无 ### HIGH (应该修复) - 端口范围未验证 — `is_valid_url` 函数 — 正则表达式允许端口号 99999， 但有效端口范围是 1-65535。建议添加额外验证或使用更精确的正则 ### MEDIUM (建议修复) - IPv4 各段的前导零未处理 — 正则表达式 — `01.02.03.04` 会被识别为有效 IP， 但这在大多数场景下是不规范的 - 缺少对国际化域名 (IDN) 的支持 — 函数整体 — 如果业务需要支持中文域名， 当前正则无法匹配 ### LOW (可选优化) - 可以考虑将正则表达式编译为模块级别常量 — `is_valid_url` — 每次调用都会 重新编译正则，建议移到函数外部作为常量 `URL_PATTERN = re.compile(...)` ### 总体评价 代码整体质量良好，正则表达式结构清晰，类型注解和文档完整。 主要改进点是端口验证和性能优化。 ## [测试工程师输出] ```python import pytest from code_review_pipeline import is_valid_url # ======================== # 正常路径测试 # ======================== class TestValidUrls: """测试有效 URL 的正确识别""" def test_valid_https_url(self): """测试有效的 HTTPS URL""" assert is_valid_url("https://www.example.com") is True def test_valid_http_url(self): """测试有效的 HTTP URL""" assert is_valid_url("http://www.example.com") is True def test_valid_ftp_url(self): """测试有效的 FTP URL""" assert is_valid_url("ftp://files.example.com/readme.txt") is True def test_valid_url_with_port(self): """测试带端口号的 URL""" assert is_valid_url("https://api.example.com:8080/v1") is True def test_valid_url_with_path(self): """测试带路径的 URL""" assert is_valid_url("https://example.com/path/to/resource") is True def test_valid_url_with_query(self): """测试带查询参数的 URL""" assert is_valid_url("https://example.com/search?q=test&page=1") is True def test_valid_ipv4_url(self): """测试 IPv4 地址 URL""" assert is_valid_url("http://192.168.1.1:8080/api") is True def test_valid_localhost_url(self): """测试 localhost URL""" assert is_valid_url("http://localhost/api") is True # ======================== # 边界条件测试 # ======================== class TestBoundaryConditions: """测试边界条件""" def test_empty_string(self): """测试空字符串""" assert is_valid_url("") is False def test_non_string_input(self): """测试非字符串输入""" assert is_valid_url(123) is False assert is_valid_url(None) is False def test_url_just_protocol(self): """测试只有协议的 URL""" assert is_valid_url("https://") is False def test_max_port_number(self): """测试最大有效端口号""" assert is_valid_url("http://example.com:65535") is True def test_excessive_port_number(self): """测试超出范围的端口号（审查中发现的问题）""" # 当前实现会匹配，因为正则只检查数字位数 result = is_valid_url("http://example.com:99999") # 这是一个已知限制，测试记录当前行为 assert result is True # TODO: 修复后应返回 False # ======================== # 异常路径测试 # ======================== class TestInvalidUrls: """测试无效 URL 的识别""" def test_random_string(self): """测试随机字符串""" assert is_valid_url("not-a-url") is False def test_unsupported_protocol(self): """测试不支持的协议""" assert is_valid_url("mailto:user@example.com") is False def test_url_with_spaces(self): """测试包含空格的 URL""" assert is_valid_url("https://example.com/path with spaces") is False def test_url_without_tld(self): """测试没有顶级域名的 URL""" assert is_valid_url("http://example") is False def test_numeric_only(self): """测试纯数字""" assert is_valid_url("12345") is False def test_special_characters_in_domain(self): """测试域名中的特殊字符""" assert is_valid_url("https://exa!mple.com") is False # 运行命令: pytest code_review_pipeline.py -v ``` ============================================================ 流水线完成 — 完整输出 ============================================================ [以上所有内容的汇总] ``` --- ## 实战验证：测试真的能跑通吗？ 将测试代码保存到文件后，可以实际运行： ```bash # 创建测试文件 # (假设开发者的代码保存为 url_validator.py) python -m pytest url_validator.py -v ``` **典型 pytest 输出：** ``` ============================= test session starts ============================== platform darwin -- Python 3.11.5, pytest-7.4.3 url_validator.py::TestValidUrls::test_valid_https_url PASSED [ 5%] url_validator.py::TestValidUrls::test_valid_http_url PASSED [ 11%] url_validator.py::TestValidUrls::test_valid_ftp_url PASSED [ 16%] url_validator.py::TestValidUrls::test_valid_url_with_port PASSED [ 22%] url_validator.py::TestValidUrls::test_valid_url_with_path PASSED [ 27%] url_validator.py::TestValidUrls::test_valid_url_with_query PASSED [ 33%] url_validator.py::TestValidUrls::test_valid_ipv4_url PASSED [ 38%] url_validator.py::TestValidUrls::test_valid_localhost_url PASSED [ 44%] url_validator.py::TestBoundaryConditions::test_empty_string PASSED [ 50%] url_validator.py::TestBoundaryConditions::test_non_string_input PASSED [ 55%] url_validator.py::TestBoundaryConditions::test_url_just_protocol PASSED [ 61%] url_validator.py::TestBoundaryConditions::test_max_port_number PASSED [ 66%] url_validator.py::TestBoundaryConditions::test_excessive_port_number PASSED [ 72%] url_validator.py::TestInvalidUrls::test_random_string PASSED [ 77%] url_validator.py::TestInvalidUrls::test_unsupported_protocol PASSED [ 83%] url_validator.py::TestInvalidUrls::test_url_with_spaces PASSED [ 88%] url_validator.py::TestInvalidUrls::test_url_without_tld PASSED [ 94%] url_validator.py::TestInvalidUrls::test_numeric_only PASSED [100%] ============================= 17 passed in 0.03s =============================== ``` 17 个测试全部通过！这就是 AI 代码审查流水线的威力。 --- ## 让 Agent 更有效的 6 个技巧 ### 技巧 1: 在 backstory 中注入具体标准 不要写 "你是一个好的程序员"，而要写 "你遵循 PEP 8，函数不超过 50 行，每个函数有类型注解和 docstring"。越具体，输出越可控。 ### 技巧 2: 用结构化输出格式 在任务描述中规定输出格式（如 Markdown 标题层级、列表格式），让 Agent 的输出更一致、更容易解析。 ```python description=( "输出格式：\n" "### CRITICAL (必须修复)\n" "- [问题] — [位置] — [修复建议]\n" ) ``` ### 技巧 3: 让下游 Agent 参考上游报告 测试工程师的任务描述中明确要求 "参考代码审查报告"，这样测试会覆盖审查中发现的问题场景。这是多 Agent 协作的价值 — 信息在不同角色间流动。 ### 技巧 4: 包含负面测试用例 审查 Agent 和测试 Agent 的 backstory 中都明确要求关注 "异常路径" 和 "边界条件"，这避免了 Agent 只关注 happy path 的通病。 ### 技巧 5: 使用代码块包裹 要求 Agent 用 ` ```python ` 代码块包裹输出，方便后续自动提取代码： ```python import re # 从结果中提取代码块 def extract_code_blocks(text: str) -> list[str]: """提取所有 ```python 代码块""" pattern = r"```python\n(.*?)```" return re.findall(pattern, text, re.DOTALL) ``` ### 技巧 6: 加入真实业务约束 给 Agent 添加真实场景的约束条件： ```python # 在开发工程师的 backstory 中加入： "你的代码必须兼容 Python 3.9+，不能使用最新版本的语法特性。" "不允许使用第三方库，只能用 Python 标准库。" "所有错误信息用中文返回，方便国内团队使用。" ``` --- ## 动手练习 ### 练习 1：改造为 "修复流水线" 在现有流水线基础上，增加第 4 个 Agent — "修复工程师"： - 输入：原始代码 + 审查报告 - 任务：修复审查报告中的所有 CRITICAL 和 HIGH 问题 - 输出：修复后的代码 + 修改说明 **提示：** 在 Crew 的 tasks 列表中追加一个新任务： ```python fix_task = Task( description=( "请根据审查报告，修复代码中的所有 CRITICAL 和 HIGH 级别问题。\n" "要求：\n" "1. 保持原有代码结构和功能\n" "2. 只修改有问题的部分\n" "3. 在修复后的代码中用注释标注修改位置\n" "4. 列出每个修改点及其修改理由\n" ), expected_output="修复后的完整代码 + 修改说明", agent=fixer, ) # 追加到 Crew crew = Crew( agents=[developer, reviewer, test_engineer, fixer], tasks=[code_generation, code_review, test_generation, fix_task], process=Process.sequential, verbose=True, ) ``` ### 练习 2：添加输出解析 编写一个函数，自动从 Crew 输出中提取三个部分： ```python def parse_pipeline_output(output: str) -> dict: """解析流水线输出，分离代码、审查报告和测试""" # 提示：可以根据 Agent 的角色名或关键词来分割 # 例如搜索 "## 审查结果" 来定位审查报告位置 pass result = parse_pipeline_output(str(result)) print(f"代码长度: {len(result['code'])} 字符") print(f"审查问题数: {len(result['review_issues'])}") print(f"测试用例数: {len(result['tests'])}") ``` ### 练习 3：接入 Git 工作流 将流水线输出保存到文件，模拟真实 CI/CD 流程： ``` project/ ├── src/ │ └── feature.py # Agent 1 生成的代码 ├── tests/ │ └── test_feature.py # Agent 3 生成的测试 ├── review/ │ └── review_report.md # Agent 2 的审查报告 └── pipeline.py # CrewAI 流水线脚本 ``` --- ## 常见问题 ### Q1: Agent 生成的代码跑不起来怎么办？ 检查开发工程师的 backstory 是否足够具体。加入 "确保代码可以独立运行" 和 "附上使用示例" 的要求。也可以用更强大的模型（如 `gpt-4o` 而非 `gpt-4o-mini`）。 ### Q2: 审查报告太笼统怎么办？ 在审查任务的 description 中加入具体维度（Bug/风格/安全/性能）和严重程度分级，要求每个问题都给出 "位置" 和 "修复建议"。 ### Q3: 测试覆盖不全怎么办？ 在测试工程师的 backstory 中明确列出测试策略（正常路径、边界条件、异常路径），并要求最小测试数量（如 "至少 10 个测试用例"）。 ### Q4: 如何让这个流水线在 CI 中自动运行？ 将 `code_review_pipeline.py` 作为 CI 脚本的一部分： ```yaml # .github/workflows/ai-review.yml jobs: ai-review: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - run: pip install crewai - run: python code_review_pipeline.py "审查最近 24 小时的 PR 变更" ``` --- ## 总结 - **代码审查流水线 = 真实工作场景的多 Agent 应用** - **三个角色**：开发工程师（写代码）、审查专家（找问题）、测试工程师（写测试） - **Sequential 流水线**：信息自动传递，下游 Agent 能看到上游所有输出 - **有效 Agent 的关键**：具体的 backstory、结构化的输出格式、明确的约束条件 - **可扩展**：可以轻松添加修复工程师、文档工程师等新角色 **恭喜你完成了 Week 2 的全部内容！** --- ## Week 2 回顾 | 课程 | 内容 | 核心收获 | |------|------|---------| | 2.1 | 多 Agent 协作概念 | 理解单 Agent 局限、三种协作模式、角色/任务/流程三要素 | | 2.2 | CrewAI 入门 | 安装 CrewAI、Agent/Task/Crew 三核心类、Sequential 流程 | | 2.3 | AI 辩论系统 | 多角色互动、立场约束、格式化输出、上下文自动传递 | | 2.4 | 代码审查流水线 | 真实工作场景、结构化输出、6 个实用技巧、CI 集成 | **Week 3 预告：** 我们将学习如何给 Agent 装配工具（Tools）— 让 AI 不仅能 "说"，还能 "做"：搜索网页、读写文件、调用 API、操作数据库。