Bardiyashavandi/code_review_agent

## 目录 - [核心理念](#the-idea) - [架构](#architecture) - [实际运行效果](#what-a-run-actually-looks-like) - [快速开始](#quick-start) - [工作原理](#how-it-works) - [HTTP API](#http-api) - [设计级安全](#security-by-design) - [测试](#testing) - [真实世界验证](#real-world-verification-not-just-mocks) - [项目结构](#project-structure) - [已知局限](#known-limitations) - [许可证](#license) ## 核心理念 静态分析器能找到模式，但无法解释它们为何重要。LLM 能够解释问题，但在缺乏真实依据时会产生幻觉。这个 Agent 弥补了这一鸿沟：它会拉取你真实的代码库，在其上运行真实的 Semgrep 静态分析，并将代码和扫描结果一并交给 Gemini —— 因此最终报告中的每一个问题都有确定性规则或正在查阅你真实代码的模型作为支撑，绝不是猜测。 只有拉取失败才会被视为致命错误——没有文件就无法进行审查。相反，Semgrep 或 Gemini 的短暂报错会被捕获为非致命的 `StageError`，因此流水线总是会返回可用的结果，虽然可能会有所降级，但绝不会两手空空。这并非纯理论：在实际测试中，Gemini 在高负载下间歇性地抛出瞬时的 `503` 错误，而重试逻辑保证了运行继续进行，没有中断。 ## 架构 ``` ┌────────────────────┐ repo URL ──────────►│ github_fetcher │── GitHub API └──────────┬─────────┘ │ Python files ▼ ┌────────────────────┐ │ semgrep_runner │── sandboxed subprocess └──────────┬─────────┘ │ files + findings ▼ ┌────────────────────┐ │ gemini_reviewer │── Gemini Flash └──────────┬─────────┘ │ structured issues ▼ ┌────────────────────┐ │ report_generator │── review_report.md └────────────────────┘ agent.py orchestrates the above AND exposes it as a Google ADK 2.3 Agent + FunctionTool, so an LLM-driven agent runtime can decide on its own when to call it. server.py wraps the same pipeline behind a FastAPI HTTP endpoint, so any service can trigger a review over the network with a single POST request. ``` | 阶段 | 模块 | 任务 | |---|---|---| | 1. 拉取 | `github_fetcher.py` | 通过 GitHub API 遍历仓库树，拉取每个 Python 文件，跳过 venv/构建噪音 | | 2. 扫描 | `semgrep_runner.py` | 将文件写入隔离的沙盒中，运行 Semgrep，将 JSON 解析为强类型的扫描结果 | | 3. 审查 | `gemini_reviewer.py` | 将代码和扫描结果分批打包为 prompt，要求 Gemini 提供结构化且按严重程度排序的审查 | ### Agent 的工具图 `agent.py` 不会只运行一次该流水线——它向 ADK Agent 暴露了**八个独立的工具**，全都作为平级的兄弟节点存在于 Agent 之下。这样一来，模型就可以自行规划调用路径，而不是每次都运行整个流程： ``` code_review_agent | +---------------+---------------+----------------+----------------+ | | | | | review_repo_ fetch_repo_ scan_code_ generate_ get_repo_ tool files_tool tool review_tool metadata_tool (one-shot: (fetch only) (Semgrep (Gemini review (language/size/ fetch+scan+ only) only) stars, no fetch) review) | | | search_code_ explain_ generate_ in_files_tool finding_tool report_file_tool (grep fetched (deep-dive on (save review as files) one issue) a real .md file) ``` 像“审查这个仓库”这样一句话的请求会被简化为一次单独的工具调用。而更具体的请求——比如“只给我看文件”、“找出所有使用 eval 的地方”、“进一步解释那个问题”、“把这个保存为文件”——会让模型自行选择（并链式调用）正确的工具，这正是使用 Agent 框架而不是单个庞大函数的真正意义所在。 ## 实际运行效果 ``` $ python3 main.py https://github.com/owner/repo --branch main --out review_report.md -v Files fetched: 25 | Semgrep findings: 2 | Review issues: 23 | Duration: 96.3s ### 严重 Flask Debug Mode Enabled in Production (app.py:115) Running with debug=True in production exposes tracebacks, environment variables, and an interactive debugger capable of arbitrary code execution. Suggested fix: set debug=False and gate it behind an environment-driven config. Hardcoded Mock API Key (agent.py:95) A string matching a real credential's prefix format is hardcoded. Even "mock" keys risk being mistaken for real ones or copied into production. Suggested fix: load all keys from environment variables, never literals. ``` 这是一次针对真实的、未经修改的仓库的实际运行——请参阅下方的[真实世界验证](#real-world-verification-not-just-mocks)。 ## 快速开始 ``` git clone https://github.com/Bardiyashavandi/code_review_agent cd code_review_agent python3 -m pip install -r requirements.txt pipx install semgrep # isolated — see "Why pipx" below ``` 在项目根目录下创建一个 `.env` 文件： ``` GITHUB_TOKEN=ghp_your_token_here GEMINI_API_KEY=your_gemini_key_here ``` **选项 1 — CLI:** ``` python3 main.py https://github.com/owner/repo --branch main --out review_report.md -v ``` `--max-files`（默认值为 `10`）限制了每次运行审查的 Python 文件数量——默认保持保守设定，因为 Gemini 的免费层级有每日请求上限；如果你的配额更高，可以调高此值。 **选项 2 — HTTP API:** ``` uvicorn server:app --reload ``` 然后向指定的 repo URL 发起 POST 请求，即可返回结构化的 JSON 审查报告： ``` curl -s -X POST http://127.0.0.1:8000/analyze \ -H "Content-Type: application/json" \ -d '{"repo_url": "https://github.com/owner/repo", "max_files": 10}' \ | python3 -m json.tool ``` **选项 3 — ADK playground:** ``` adk web ``` 在 `http://127.0.0.1:8000` 打开 Google 的 ADK Dev UI —— 可以直接在浏览器中与 Agent 对话，并查看包含全部八个工具节点的可视化图表。 ## 工作原理 `agent.py` 在单一的 `CodeReviewAgent.review_repo()` 调用背后统筹协调这三个阶段，并且还通过模块级的 `root_agent` 将相同的流水线作为 Google ADK 2.3 的 `Agent` + `FunctionTool` 暴露出来——因此，由 Gemini 驱动的 ADK Agent 能够根据普通的自然语言请求，自行决定是否调用 `review_repo_tool`。 **以编程方式使用：** ``` import os from agent import CodeReviewAgent agent = CodeReviewAgent( github_token=os.environ["GITHUB_TOKEN"], gemini_api_key=os.environ["GEMINI_API_KEY"], ) result = agent.review_repo("https://github.com/owner/repo") for issue in result.review_report.issues: print(issue.severity, issue.path, issue.title) ``` **作为 ADK Agent 使用** —— 模型会自行决定调用哪个/哪些工具： ``` from agent import build_adk_agent adk_agent = build_adk_agent( github_token=os.environ["GITHUB_TOKEN"], gemini_api_key=os.environ["GEMINI_API_KEY"], ) ``` 通过任何 ADK `Runner`（例如 `google.adk.runners.InMemoryRunner`）运行 `adk_agent`——或者直接运行 `python3 adk_demo.py` 查看现成的示例。 | 工具 | 作用 | |---|---| | `review_repo_tool` | 一步到位：通过一次调用完成对 repo URL 的拉取 + 扫描 + 审查 | | `fetch_repo_files_tool` | 仅拉取 repo 的 Python 文件 | | `scan_code_tool` | 仅对给定的一组文件运行 Semgrep | | `generate_review_tool` | 仅要求 Gemini 审查给定的一组文件（+ 可选的扫描结果） | | `get_repo_metadata_tool` | 查询 repo 的语言、大小、星数和默认分支——不进行文件拉取 | | `search_code_in_files_tool` | 对已拉取的文件进行正则/关键字搜索 | | `explain_finding_tool` | 要求 Gemini 针对一个已知问题提供专注且更深入的解释 | | `generate_report_file_tool` | 将已生成的审查结果渲染为 Markdown 并保存到磁盘 | Agent 的指令也使其保持在处理范围内：当被问及与代码审查无关的问题时，它会拒绝并引导话题，而不是强行调用不相关的工具。所有这些都在 ADK Dev UI playground 中进行了实时验证，其中的工具图显示了从 Agent 分出的八个独立节点。 ## HTTP API `server.py` 将完全相同的 `CodeReviewAgent.review_repo()` 流水线封装在一个 FastAPI endpoint 之后——内部逻辑没有任何改变，只是增加了一种触发它的新方式。 **启动服务器：** ``` uvicorn server:app --reload # dev, auto-reloads on save uvicorn server:app --host 0.0.0.0 --port 8080 # prod-like ``` **交互式文档：** `http://127.0.0.1:8000/docs`（Swagger UI，根据 Pydantic 模型自动生成） **`POST /analyze`** 请求： ``` { "repo_url": "https://github.com/owner/repo", "branch": "main", "max_files": 10 } ``` 响应 (200)： ``` { "repo_url": "https://github.com/owner/repo", "duration_s": 11.1, "files_fetched": 5, "truncated": false, "review": { "summary": "2 issues found...", "model": "gemini-3.1-flash-lite", "files_reviewed": 5, "duration_s": 1.8, "issues": [ { "path": "auth.py", "line": 42, "severity": "HIGH", "title": "Hardcoded secret", "description": "...", "suggested_fix": "...", "rule_id": null } ] }, "scan": { "scanned": 5, "skipped": [], "duration_s": 4.3, "findings": [] }, "stage_errors": [] } ``` 错误响应遵循标准的 HTTP 语义： | 状态码 | 原因 | |---|---| | `400` | 错误的 orchestrator/配置状态（`AgentError`、`ValueError`） | | `401` | GitHub token 无效或已过期 | | `404` | 未找到仓库（或者是没有访问权限的私有仓库） | | `429` | 触发了 GitHub API 的速率限制 | | `500` | 未预期的内部错误（服务端会记录包含完整 traceback 的日志） | | `502` | 与身份验证/速率限制/未找到无关的 GitHub API 错误 | | `504` | 流水线超出超时时间——尝试减小 `max_files` 或调高 `AGENT_TIMEOUT_S` | 请求验证错误（错误的 `repo_url`，`max_files` 不在 1–500 范围内）会在流水线运行之前返回 FastAPI 标准的 `422` 错误。在你的环境中设置 `AGENT_TIMEOUT_S` 可覆盖默认的 180 秒限制。 **`GET /health`** —— 存活检查，返回 `{"status": "ok"}`。 凭证保留在服务端，永远不会由调用方传递。 ## 设计级安全 - 每一个子进程调用都使用显式的参数列表——绝不使用 `shell=True`。 - 来自拉取仓库的文件路径在接触磁盘之前，都会经过验证以防路径遍历。 - Semgrep 的 `--config` 参数通过正则表达式建立白名单，以防止参数注入。 - Gemini 的系统提示词指示模型将所有文件内容和 Semgrep 输出视为**不可信的数据，而不是指令**——包含“忽略之前的指令”的恶意 commit 无法重定向审查。已使用注入的 payload 进行了直接测试。 - 绝不硬编码凭证。两个 API key 仅从环境变量中加载，并且 `test_secrets_never_logged` 会断言确保 key 永远不会泄露到日志行或异常消息中。 - 模型输出永远不会被作为代码执行，也不会不安全地插入到报告中——已使用注入的 `__import__` payload 进行过测试。 ## 测试 ``` pytest -v ``` 覆盖全部五个模块的 107 个测试。每一个外部依赖——GitHub 的 API、Semgrep 子进程、Gemini SDK——都经过了 mock，因此该测试套件大约在一秒钟内即可运行完毕，无需网络访问或凭证。 ## 真实世界验证，而非仅仅是 mock 一次真实的端到端运行（而非测试夹具）拉取了 25 个文件，运行了实时的 Semgrep 扫描，调用了 Gemini，并在 96 秒内生成了包含 23 个问题的报告，而且都是真实的发现——一个处于 debug 模式的 Flask 应用、一个硬编码的 mock API key、一个信任客户端提供 ID 的 endpoint。那次运行还暴露了三个真实的集成 bug，这是任何 mock 都无法捕捉到的，目前这三个 bug 都已修复并配有回归测试覆盖： 1. **依赖冲突** —— `google-adk` 和 `semgrep` 锁定了互不兼容的 `opentelemetry` 版本范围。通过将 Semgrep 隔离到其独立的 `pipx` 环境中得到修复。 2. **陈旧的 shell 环境变量** —— `python-dotenv` 从不覆盖已经导出的变量，因此来自之前测试的旧的 `GEMINI_API_KEY` 会悄无声息地覆盖掉正确的 `.env` 值。通过使用 `override=True` 加载 `.env` 得到修复。 3. **macOS 符号链接 bug** —— macOS 通过 `/private/...` 符号链接解析其临时目录；一个在 Linux 上运行良好的路径比较在真实的 Mac 上却引发了 `ValueError`。 ADK Agent 通过两种方式进行了验证：一次是通过 `adk_demo.py` 终端脚本，另一次是在 Google 的 ADK Dev UI playground（`adk web`）中实时进行——两者都产生了相同的正确工具调用行为。HTTP API 通过带有实时 `curl` 调用的真实仓库进行了验证，返回了完整的 JSON 审查结果。 ### 为什么 Semgrep 要用 `pipx` `google-adk` 和 `semgrep` 锁定了不兼容的 `opentelemetry-api`/`opentelemetry-sdk` 版本范围——将两者安装到同一个环境中会导致其中一个崩溃。`pipx` 为 Semgrep 提供了独立的隔离 venv；`semgrep_runner.py` 永远只在 `PATH` 中调用 `semgrep` 二进制文件，因此这种隔离对项目的其余部分是不可见的。 ## 项目结构 ``` code_review_agent/ ├── agent.py # orchestrator + ADK Agent/FunctionTool (exposes root_agent) ├── github_fetcher.py # stage 1: fetch ├── semgrep_runner.py # stage 2: scan ├── gemini_reviewer.py # stage 3: review ├── report_generator.py # Markdown rendering ├── main.py # CLI entry point ├── server.py # FastAPI HTTP wrapper (POST /analyze) ├── adk_demo.py # standalone ADK tool-calling demo ├── *_spec.md # spec written before each module's code ├── tests/ # 107 tests, one file per module ├── KAGGLE_WRITEUP.md # full capstone writeup └── VIDEO_SCRIPT.md # demo video script ``` ## 已知局限 `--config auto` 需要通过网络访问 `semgrep.dev` 的规则注册表；网络出口限制严格且锁定的 CI runner 或沙盒将需要使用本地或固定版本的规则集。Gemini 偶尔会在高需求下返回瞬时的 `503` 错误——`gemini_reviewer.py` 会以指数退避算法自动重试，但持续的中断仍然会表现为非致命的 `StageError`，而不会阻断运行。免费层级的 Gemini key 还限制了每日的总请求数（不仅仅是每分钟）——`--max-files` 默认为 `10`，并且批次之间包含短暂的批处理间延迟，专门用于进一步延长免费层级的配额使用时间。 HTTP 服务器（`server.py`）在本地运行，并未部署到任何云服务——真正的云部署通常需要一个启用结算功能的项目，这与本项目不使用付费服务的约束条件相冲突。 ## 展示了什么 每个模块在编写任何实现代码之前，都是从一个书面的规范（接口、行为、错误层级、测试表）开始的——此仓库中的 `*_spec.md` 文件就是该过程的可见记录。Orchestrator 是一个真正的 Google ADK 2.3 工具，由 Agent 运行时自行决定何时调用流水线以及链接八个工具中的哪一个。相同的流水线也可以通过 FastAPI 服务器在 HTTP访问，使其可以从任何语言或服务中进行调用。任何地方都没有使用付费服务——Semgrep 的 `--config auto`、Gemini 和 GitHub API 全部都是免费层级，这是从一开始就设定的硬性约束。 完整说明：[`KAGGLE_WRITEUP.md`](./KAGGLE_WRITEUP.md)。演示视频脚本：[`VIDEO_SCRIPT.md`](./VIDEO_SCRIPT.md)。 ## 许可证 MIT —— 详见 [`LICENSE`](./LICENSE)。

标签：AV绕过, FastAPI, SOC Prime, 代码审查, 开发工具, 逆向工具, 错误基检测, 静态代码分析