技术教程 · 阅读约 9 分钟

用 30 行 Python 让 Codex 帮你审完整个 Repo——多 Agent 联动实战

用 30 行 Python 让 Codex 帮你审完整个 Repo——多 Agent 联动实战

很多人跑 Codex 的姿势是:粘一段代码进去,等它给意见,再粘下一段。

这个方式的问题不是效果差,而是太慢、太贵、而且你的上下文窗口会撑死。

真正跑得起来的做法是:把 repo 拆成独立文件,用多个 Agent 并发审查,最后汇总报告。这篇文章就写这个——完整代码,可以直接跑。


为什么单 Agent 跑整个 repo 会翻车

V2EX 上有人跑了半个月、消耗十几亿 token,然后卡在多 Agent 联动上求助。根本原因在于:

  • 把整个 repo 塞进单次请求 → context 爆炸,模型开始"遗忘"前面的文件
  • 串行逐文件提交 → 慢,而且费用叠加没有优化空间
  • 没有任务分发层 → Agent 之间结果无法合并,报告碎片化

解法是任务拆分 + 并发 + 聚合,三层结构,30 行核心逻辑。


架构设计


repo/

├── src/auth.py

├── src/db.py

├── src/api.py

└── ...



分发层(Coordinator)

    ├── Worker Agent 1 → 审 auth.py

    ├── Worker Agent 2 → 审 db.py

    └── Worker Agent 3 → 审 api.py



聚合层(Summarizer)

    └── 汇总所有 Worker 输出,生成最终报告

每个 Worker 只看一个文件,上下文干净,结果可重复。Summarizer 只拿摘要,不需要原始代码。


完整代码


import os

import asyncio

from pathlib import Path

from openai import AsyncOpenAI



# 改一行 base_url,接入无量Api,价格比官方便宜 60%+

client = AsyncOpenAI(

    api_key=os.environ.get("WULIANG_API_KEY"),

    base_url="https://api2everything.xyz/v1"

)



REVIEW_PROMPT = """你是一个资深代码审查员。

请审查以下代码,重点关注:

1. 安全漏洞(SQL 注入、未授权访问、硬编码密钥等)

2. 明显的逻辑错误

3. 性能问题



文件名:{filename}

代码:

{code}



输出格式:

- 问题列表(每条一行,标明严重程度:HIGH/MED/LOW)

- 如果没有问题,输出 "✅ 无明显问题"

"""



async def review_file(filepath: Path) -> dict:

    """单个 Worker Agent:审查一个文件"""

    code = filepath.read_text(encoding="utf-8", errors="ignore")

    

    # 跳过太大的文件(>100KB 建议再拆)

    if len(code) > 100_000:

        return {"file": str(filepath), "result": "⚠️ 文件过大,跳过"}

    

    response = await client.chat.completions.create(

        model="gpt-4o",           # 换成 claude-sonnet-4-5 也行

        messages=[

            {"role": "user", "content": REVIEW_PROMPT.format(

                filename=filepath.name,

                code=code

            )}

        ],

        temperature=0.1           # 代码审查要低温,减少幻觉

    )

    

    return {

        "file": str(filepath),

        "result": response.choices[0].message.content

    }



async def summarize(reviews: list[dict]) -> str:

    """Summarizer Agent:汇总所有 Worker 的输出"""

    review_text = "\n\n---\n\n".join(

        f"**{r['file']}**\n{r['result']}" for r in reviews

    )

    

    response = await client.chat.completions.create(

        model="gpt-4o",

        messages=[

            {"role": "system", "content": "你是技术负责人,负责汇总代码审查报告。"},

            {"role": "user", "content": f"以下是各文件的审查结果,请生成一份执行摘要,"

             f"列出最需要立即修复的 TOP 5 问题:\n\n{review_text}"}

        ]

    )

    

    return response.choices[0].message.content



async def review_repo(repo_path: str, extensions: list[str] = None):

    """主入口:扫描 repo,并发审查,输出报告"""

    if extensions is None:

        extensions = [".py", ".js", ".ts", ".go", ".java"]

    

    repo = Path(repo_path)

    files = [

        f for f in repo.rglob("*")

        if f.suffix in extensions

        and ".git" not in f.parts

        and "node_modules" not in f.parts

        and "venv" not in f.parts

    ]

    

    print(f"📂 找到 {len(files)} 个文件,开始并发审查...")

    

    # 并发上限设 10,避免触发速率限制

    semaphore = asyncio.Semaphore(10)

    

    async def bounded_review(f):

        async with semaphore:

            result = await review_file(f)

            print(f"  ✓ {f.name}")

            return result

    

    reviews = await asyncio.gather(*[bounded_review(f) for f in files])

    

    print("\n🤖 正在生成汇总报告...")

    summary = await summarize(reviews)

    

    # 输出到文件

    report_path = repo / "REVIEW_REPORT.md"

    with open(report_path, "w", encoding="utf-8") as out:

        out.write("# Code Review Report\n\n")

        out.write("## Executive Summary\n\n")

        out.write(summary)

        out.write("\n\n---\n\n## Per-File Details\n\n")

        for r in reviews:

            out.write(f"### {r['file']}\n\n{r['result']}\n\n")

    

    print(f"\n✅ 报告已保存至 {report_path}")

    return summary



# 跑起来

if __name__ == "__main__":

    asyncio.run(review_repo("./your-repo"))


几个值得注意的细节

并发数别开太高。 Semaphore(10) 是比较保守的值。如果你的 API 额度充足,可以调到 20,速度会有明显提升。但超过 30 基本上是在和速率限制对抗,得不偿失。

模型选择有讲究。 审查安全漏洞用 gpt-4o 或 claude-sonnet-4-5,准确率更高。如果 repo 很大,Worker 层可以用 gpt-4o-mini 做初筛,只把标记了 HIGH 的结果送给 Summarizer 用强模型复核,成本能再降一半。

temperature=0.1 很关键。 代码审查不需要创意,需要的是一致性和准确性。低温能让模型老老实实找问题,而不是"发散"出一堆似是而非的建议。

文件过大的处理。 超过 100KB 的文件建议按函数/类拆分再送审,直接跳过会漏掉风险。可以用 ast 模块(Python)或 tree-sitter 做语法级拆分。


费用估算

一个中等规模的 repo(50个文件,每文件平均 200 行):

  • 输入约 150 万 token,输出约 20 万 token
  • 用 gpt-4o:官方约 ¥35,无量Api 约 ¥12
  • 换 gpt-4o-mini 做 Worker:总费用可以压到 ¥3 以内

用无量Api 跑这套流程

代码里已经写了 base_url="https://api2everything.xyz/v1",换上你的 key 就能跑。无量Api 支持 OpenAI / Claude / Gemini / DeepSeek 等 300+ 模型,国内直连不需要代理,注册送 ¥1 余额可以直接试。

价格比官方便宜 65%,余额永久有效,不用担心没用完过期。

👉 https://api2everything.xyz/?utm_source=v2ex&utm_medium=article&utm_campaign=daily&aff=wqUb


跑起来遇到问题的,评论区说一下,我看到都会回。觉得有用的话点个赞,后续还会写 Codex 在 CI/CD 里自动触发审查的方案。