Agent AtlasAI Agent 知识库
06.08教程 · Tutorials

代码 Agent 实战

仓库地图、ripgrep、补丁、pytest 反馈循环、Docker 沙箱、防作弊与 Pass@k 评估。

代码 Agent 的核心不是「会写代码」,而是「改完能验证」。这个教程实现一个最小但完整的循环:定位文件、生成补丁、跑测试、根据失败信息修正,并在沙箱里执行。

目标

bash
python -m coder "修复 tests/test_auth.py::test_expired_token 的失败"

Agent 会:

  1. 生成仓库地图;
  2. 搜索相关符号;
  3. 读取文件片段;
  4. 输出 unified diff;
  5. 应用补丁;
  6. 跑相关测试;
  7. 失败则带着 traceback 修正;
  8. 成功后跑全量测试并输出 diff。

项目结构

text
coder/
├─ __init__.py
├─ __main__.py
├─ repo.py          # 地图、搜索、读文件、打补丁
├─ sandbox.py       # Docker 执行
├─ loop.py          # 主循环
└─ prompts.py
Dockerfile.sandbox

Step 1:仓库地图

不要整仓加载。先给模型一个「有哪些文件、各自大概做什么」的视图。

repo.py:

python
from __future__ import annotations

import fnmatch
import subprocess
from pathlib import Path

PROTECTED = ("tests/", ".github/", "migrations/")


def sh(cmd: list[str], cwd: str, timeout: int = 30) -> str:
    result = subprocess.run(
        cmd, cwd=cwd, text=True, capture_output=True, timeout=timeout
    )
    if result.returncode != 0:
        raise RuntimeError(result.stderr[-2000:])
    return result.stdout


def repo_map(root: str, max_files: int = 200) -> str:
    files = sh(["git", "ls-files"], root).splitlines()
    files = [
        f
        for f in files
        if not any(part in f for part in ("node_modules/", ".venv/", "dist/"))
    ][:max_files]
    lines = [f"{len(files)} files"]
    for f in files:
        lines.append(f"- {f}")
    return "\n".join(lines)


def search_code(root: str, query: str, max_hits: int = 20) -> str:
    try:
        out = sh(["rg", "-n", "--no-heading", "-m", "3", query, "."], root)
    except RuntimeError as e:
        return f"搜索失败:{e}"
    return "\n".join(out.splitlines()[:max_hits])


def read_file(root: str, path: str, start: int = 1, end: int = 200) -> str:
    p = Path(root) / path
    if not p.exists() or not p.is_file():
        return f"文件不存在:{path}"
    lines = p.read_text(encoding="utf-8", errors="replace").splitlines()
    start = max(1, start)
    end = min(len(lines), end)
    numbered = [f"{i:>4}: {lines[i-1]}" for i in range(start, end + 1)]
    return "\n".join(numbered)


def apply_patch(root: str, diff: str) -> str:
    if len(diff.splitlines()) > 250:
        return "补丁过大(>250 行),请拆小。"
    for bad in PROTECTED:
        if f"+++ b/{bad}" in diff or f"--- a/{bad}" in diff:
            return f"禁止修改受保护目录:{bad}"

    patch_file = Path(root) / ".agent.patch"
    patch_file.write_text(diff, encoding="utf-8")
    try:
        sh(["git", "apply", "--check", str(patch_file)], root)
        sh(["git", "apply", str(patch_file)], root)
        return "补丁已应用。"
    except RuntimeError as e:
        return f"补丁失败:{e}"
    finally:
        patch_file.unlink(missing_ok=True)

git apply --check 先验证,避免半应用状态。

Step 2:沙箱执行

Dockerfile.sandbox:

dockerfile
FROM python:3.12-slim
RUN apt-get update && apt-get install -y --no-install-recommends git ripgrep \
    && rm -rf /var/lib/apt/lists/*
WORKDIR /repo

构建:

bash
docker build -f Dockerfile.sandbox -t coder-sandbox .

sandbox.py:

python
import subprocess


def run_tests(repo: str, target: str | None = None, timeout: int = 120) -> dict:
    cmd = [
        "docker", "run", "--rm",
        "--network", "none",
        "-v", f"{repo}:/repo",
        "-w", "/repo",
        "coder-sandbox",
        "python", "-m", "pytest", "-q",
    ]
    if target:
        cmd.append(target)

    try:
        p = subprocess.run(cmd, text=True, capture_output=True, timeout=timeout)
    except subprocess.TimeoutExpired:
        return {"ok": False, "returncode": -1, "stdout": "", "stderr": "测试超时"}

    return {
        "ok": p.returncode == 0,
        "returncode": p.returncode,
        "stdout": p.stdout[-6000:],
        "stderr": p.stderr[-6000:],
    }

--network none 防止测试代码外联;真实项目可能需要只读缓存或专用网络,按需放宽。

Step 3:主循环

用 JSON 动作协议,便于把注意力放在循环上:

python
ACTIONS = {
    "search": {"query": "string"},
    "read": {"path": "string", "start": "int", "end": "int"},
    "patch": {"diff": "string"},
    "test": {"target": "string|null"},
    "final": {"summary": "string"},
}

SYSTEM = """你是代码修复 Agent。
可用动作:search、read、patch、test、final。
规则:
1. 先搜索定位,再读相关片段,不要猜文件内容。
2. 补丁必须是 unified diff,尽量小。
3. 修改后必须运行相关测试。
4. 测试失败时根据 traceback 修正,不要重复同一个补丁。
5. 不要修改 tests/、.github/、migrations/。
6. 完成后输出 final,附变更摘要与测试结果。
只输出 JSON:{"thought": "...", "action": "...", "args": {...}}"""

loop.py:

python
from __future__ import annotations

import json

from . import repo
from .sandbox import run_tests


def run(root: str, task: str, llm, max_steps: int = 10) -> str:
    context = f"仓库地图:\n{repo.repo_map(root)}\n\n任务:{task}"
    history: list[str] = []
    seen_patches: set[str] = set()

    for step in range(1, max_steps + 1):
        result = llm.complete_json(
            system=SYSTEM,
            user=f"{context}\n\n历史:\n" + "\n".join(history[-8:]),
        )
        action, args = result["action"], result.get("args", {})
        history.append(f"[{step}] {result['thought']} -> {action} {json.dumps(args, ensure_ascii=False)[:200]}")

        if action == "search":
            output = repo.search_code(root, args["query"])
        elif action == "read":
            output = repo.read_file(root, args["path"], args.get("start", 1), args.get("end", 200))
        elif action == "patch":
            diff = args["diff"]
            if diff in seen_patches:
                output = "同一个补丁已经尝试过,请修改方案。"
            else:
                seen_patches.add(diff)
                output = repo.apply_patch(root, diff)
        elif action == "test":
            output = json.dumps(run_tests(root, args.get("target")), ensure_ascii=False)
        elif action == "final":
            return args.get("summary", "完成")
        else:
            output = f"未知动作:{action}"

        history.append(f"观察:{output[:3000]}")

    return "达到最大步数,未完成。请查看 diff 后转人工。"

Step 4:提示词与约束

prompts.py 里已经写了基本规则。再加三条工程约束:

text
1. 补丁只允许修改与失败测试相关的文件。
2. 不允许删除测试、跳过测试、捕获所有异常。
3. 如果测试通过但改动超过 150 行,要求解释必要性。

第一条减少回归;第二条防止「作弊」;第三条控制审查成本。

Step 5:防测试作弊

模型可能通过改测试、加 skip、空 except 让测试通过。至少检查:

python
def check_patch(diff: str) -> list[str]:
    problems = []
    if "+++ b/tests/" in diff:
        problems.append("修改了测试文件")
    if "+    pass" in diff and "-    raise" in diff:
        problems.append("疑似用 pass 替换异常")
    if "@pytest.mark.skip" in diff or "xfail" in diff:
        problems.append("跳过测试")
    if "except Exception:" in diff and "pass" in diff:
        problems.append("空异常捕获")
    return problems

更强的做法:测试文件只读挂载,或用 held-out 测试验证。

Step 6:评估

  • Pass@k:k 次尝试至少一次通过,衡量上限;
  • Pass^k:k 次全部通过,更接近生产可靠性;
  • 回归率:目标测试通过,但其他测试被弄坏;
  • 补丁大小:越大越难审查;
  • 平均轮数/成本。
python
def pass_at_k(results: list[list[bool]], k: int) -> float:
    import math
    total = 0.0
    for runs in results:
        n, c = len(runs), sum(runs[:k])
        if n - c < k:
            total += 1.0
        else:
            total += 1 - math.comb(n - c, k) / math.comb(n, k)
    return total / len(results)

生产更关心 Pass^k:十次里九次成功,不如十次都成功。

Step 7:运行

bash
export OPENAI_API_KEY=...
python -m coder "修复 tests/test_auth.py::test_expired_token"

观察输出:

text
[1] 先定位 validateSession -> search "validateSession"
[2] 读取 src/auth/session.ts
[3] 补丁:为 jwt.verify 增加 algorithms 限制
[4] 测试:pytest tests/test_auth.py -q -> 1 passed
[5] final:已修复,变更 2 行

常见失败

一次改太多

为了修一个测试重写模块,引入新回归。限制补丁大小,只改相关文件。

不读代码就生成补丁

模型幻觉 API 或行号。必须先 search/read,再 patch。

在宿主机直接跑测试

模型生成的代码可能删文件、外联。用 Docker/沙箱,限制网络。

测试文件可写

模型会改测试让它通过。测试目录只读,或校验 diff。

没有回滚

补丁失败留下半成品。用 git apply --check 和独立分支。

验收清单

  • 能生成仓库地图,不整仓加载
  • 搜索、读文件、补丁、测试四个动作可用
  • 补丁先 check 再 apply
  • 测试在沙箱执行,默认无网络
  • 受保护目录不可修改
  • 重复补丁会被阻止
  • 失败时能带 traceback 修正
  • 有 Pass@k / 回归率评估

小结

  • 代码 Agent 的闭环是「定位 → 最小补丁 → 测试 → 修正」。
  • 验证信号来自测试,不是模型自评。
  • 沙箱、只读测试、补丁上限是安全底线。
  • 评估看 Pass^k 与回归率,而不只是演示。

下一篇处理没有 API 的界面:浏览器 Agent。

读完了?进度会保存在本地浏览器。
内容会持续修订。发现错误或想补充,欢迎反馈。 遇到问题?查排查手册 →