合集

Claude Agent SDK 实战

已更新 27 章
阅读进度 85%

第 22 章

编码 Agent：从 demo 到生产

demo 关心“能不能修”；生产关心“修错了怎么办、两个人同时修怎么办、进程挂了怎么办”。

第 9 章的 Agent 能修 days_left。它跑在一个临时目录里，失败了你看 diff，成功了你说“不错”。

生产不是这样。

生产里，一个 coding agent 可能：

  • 在共享工作区里 git clean -fdx，把别人的未跟踪文件删了；
  • 两个 worker 同时领到同一个任务，改同一份代码；
  • 审批弹窗还没点，进程重启了，任务“继续”执行；
  • 改完测试，测试绿了，但测试文件也被改了；
  • 跑完不留下 patch/日志，出了问题无法复盘。

这些都不是“模型不够聪明”，是 Harness 没接住。

这一章把第 9 章的循环补成一个生产形状的 coding agent：不是完整产品，但每个控制点都落在代码上。

22.1 demo 与生产之间，差的是 Harness

先把“六问”摆出来。第 1 章说，Harness 控制六件事；生产 coding agent 的每个事故，都能对上其中一问：

  • 看见什么：它读的是整个仓库，还是相关文件 + 项目规则？
  • 能做什么：它能 Edit、Bash、Write，还是只有必要工具？
  • 允许做什么：路径、命令、审批、沙箱，谁来判？
  • 什么时候停：测试绿了？diff 干净？预算用完？
  • 失败怎么办：重试、恢复、回滚，还是留给人？
  • 做完留下什么：patch、测试输出、审计、成本？

第 9 章的 tiny agent 只回答了“能做什么”和“什么时候停（模型说完成）”。生产要把另外四问补上。

22.2 生产架构：Agent 只是其中一层

一个能上线的 coding agent，至少长这样：

text
POST /runs
   │
   ▼
[API] ──► [RunStore] ◄── 状态机 / 幂等键 / 租户
   │         │
   ▼         │
[Queue] ─────┘
   │
   ▼
[Worker] ──► [Workspace]   git worktree，一 run 一个
   │            │
   │            ▼
   │        [Agent Loop]  SDK 的 query/client
   │            │
   ▼            ▼
[Verifier]   [Policy] ── 路径/命令/审批/沙箱
   │
   ▼
[Patch + Audit + Metrics]

五个角色，各有各的 Harness 控制：

  • API：接收任务、鉴权、租户、幂等键；
  • RunStore：任务状态、租约、审批、事件；
  • Workspace：隔离工作区，失败可丢弃；
  • Agent Loop：模型 + 工具 + 权限；
  • Verifier：测试/diff/静态检查，决定“能不能算完成”。

SDK 只负责 Agent Loop 这一层。前面四层，才是“生产”和“demo”的差距。

22.3 Harness 控制点

六问本项目的实现对应章节
看见什么CLAUDE.md + Skill 目录 + 相关文件；不加载整个仓库第 6、13 章
能做什么Read/Glob/Grep/Edit/Bash/TodoWrite/Skill/Agent第 4、11 章
允许做什么路径 jail + 命令 deny + 写操作审批 + 沙箱第 5、12、21 章
什么时候停测试 + diff + Stop Hook + 预算/轮数第 15、18 章
失败怎么办重试、resume、worktree 丢弃、人工接管第 20 章
做完留下什么patch、测试输出、审计、成本/trace第 7、19、21 章

下面逐项落到代码。

看见什么：仓库地图，不是整个仓库

第 6 章算过：工具 schema 和长文件会每轮重发。生产 coding agent 不能一上来 Read 整个仓库。

上下文分三层：

  1. 常驻：短 CLAUDE.md（构建/测试命令、目录约束、禁止事项）；
  2. 按需：Skill 的目录（name/description 常驻，正文按需加载）；
  3. 临时：Glob/Grep 定位后，只 Read 相关文件。
python
options = ClaudeAgentOptions(
    cwd=str(workspace.path),
    setting_sources=["project"],       # 加载 CLAUDE.md/.claude/skills
    system_prompt={"type": "preset", "preset": "claude_code", "append": WORKFLOW},
    tools=["Read", "Glob", "Grep", "Edit", "Bash", "TodoWrite", "Skill", "Agent"],
    # 其他 options
)

setting_sources=["project"] 是显式的：生产不能依赖“默认加载”。第 10 章那个“默认 setting_sources 把技能也加载了”的意外，就是没写死。

能做什么：工具面是权限，不是功能列表

生产 coding agent 的工具面要窄：

  • Read/Glob/Grep：只读；
  • Edit：精确替换，不允许整文件 Write；
  • Bash：跑测试/构建，必须沙箱 + 超时；
  • TodoWrite：计划可见；
  • Skill：项目流程；
  • Agent：只用于调研，不用于改代码。

不要给 Write。Edit + diff 已经够；Write 会让“整文件覆盖”变得太容易。

第 11 章讲过：工具面不是越多越好；每多一个工具，模型多一次选择，schema 也多一份成本。

允许做什么：Policy 先于模型

写操作和 Bash 走 can_use_tool。先做确定性检查，再谈“问人”。

python
class Policy:
    def __init__(self, root: Path):
        self.root = root.resolve()
        self.write_allow = {self.root / "taskflow", self.root / "README.md"}

    def check_path(self, raw: str) -> Path:
        p = (self.root / raw).resolve()
        if not str(p).startswith(str(self.root) + "/"):
            raise PermissionError(f"path escapes workspace: {raw}")
        return p

    async def __call__(self, tool, inp, ctx):
        if tool == "Bash":
            cmd = str(inp.get("command", ""))
            if any(x in cmd for x in ("rm -rf", "git clean -fdx", "curl ", "> /")):
                return PermissionResultDeny(message="dangerous command")
            return PermissionResultAllow()
        if tool == "Edit":
            p = self.check_path(str(inp.get("file_path", "")))
            if p not in self.write_allow:
                return PermissionResultDeny(message=f"{p} is not in the write allowlist")
            return PermissionResultAllow()
        return PermissionResultAllow()

这段代码体现三件事：

  • 路径 jail：resolve() 后检查前缀，防 ../；
  • 写白名单：只允许 taskflow/ 和 README.md，测试文件不在里面；
  • 命令 deny：明显危险的命令直接拒，不问模型。

第 5 章讲过权限管线的顺序；第 12 章讲过 allowed_tools 会阴影 can_use_tool；第 21 章讲过路径/命令/沙箱。这里是它们的组合。

Bash 还要沙箱：

python
sandbox={"enabled": True, "allowUnsandboxedCommands": False}

沙箱起不来就失败，不静默回退。否则 Bash 的权限检查只是“建议”。

什么时候停：Stop Hook + Verifier

模型说“完成”不算。Stop Hook 只允许在“测试跑过”之后停：

python
async def require_tests(inp, tool_use_id, ctx):
    if inp.get("stop_hook_active"):
        return {}
    if not STATE.tests_ran:
        return {"decision": "block", "reason": "Run the full test suite first."}
    return {}

真正的“完成”由 Verifier 判：

python
def verify(workspace: Path) -> tuple[bool, str]:
    tests = subprocess.run(
        ["python3", "-m", "unittest", "discover", "-s", "tests"],
        cwd=workspace, text=True, capture_output=True,
    )
    if tests.returncode != 0:
        return False, "tests failed"

    changed = subprocess.run(
        ["git", "diff", "--name-only"], cwd=workspace,
        text=True, capture_output=True,
    ).stdout.splitlines()

    allowed = {"taskflow/core.py", "README.md"}
    unexpected = [p for p in changed if p and p not in allowed]
    if unexpected:
        return False, f"unexpected changes: {unexpected}"
    return True, "tests pass; diff limited"

Stop Hook 防“没跑就说完成”；Verifier 防“跑绿了但改错文件”。两道一起，才是“什么时候停”。

失败怎么办：worktree + 状态机

生产不能“失败了再手动清理”。一 run 一个 git worktree：

python
class Workspace:
    def __init__(self, repo: Path, run_id: str):
        self.path = repo / ".runs" / run_id

    def __enter__(self):
        subprocess.run(
            ["git", "worktree", "add", "--detach", str(self.path), "HEAD"],
            check=True,
        )
        return self

    def __exit__(self, *exc):
        subprocess.run(
            ["git", "worktree", "remove", "--force", str(self.path)],
            check=False,
        )

失败就丢弃 worktree；成功再生成 patch。这样不会把脏状态留给下一个任务。

RunStore 用状态机 + 租约，防止两个 worker 领同一个任务：

python
def claim(conn, run_id: str, worker: str) -> bool:
    cur = conn.execute(
        """UPDATE runs SET state='running', worker=?, lease_until=?
           WHERE id=? AND state='queued'""",
        (worker, now() + 60, run_id),
    )
    conn.commit()
    return cur.rowcount == 1

WHERE state='queued' 是原子条件；两个 worker 只有一个能改到一行。第 20 章讲“重试/恢复”，这里先把“不重复执行”做掉。

做完留下什么：patch、测试、审计

一次 run 至少留下：

  • patch.diff：相对 base 的 diff；
  • tests.log：测试输出；
  • events.jsonl：工具调用/审批/错误；
  • metrics.json：turns、cost、duration；
  • audit.jsonl：谁批了什么。

这些不是“日志好看”，是第 19 章的成本/时延、第 21 章的审计、第 17 章的评测所需要的原始数据。

22.4 一次 run 的完整路径

python
async def run_once(run: Run, workspace: Workspace) -> Run:
    store.transition(run.id, "queued", "running")
    try:
        result = await agent(workspace.path, run.task)      # SDK loop
        ok, why = verify(workspace.path)                   # Harness 判
        if not ok:
            store.fail(run.id, why)
            return run
        patch = workspace.diff()
        store.succeed(run.id, patch=patch, metrics=result)
        return run
    except Exception as e:
        store.fail(run.id, str(e))
        raise
    finally:
        workspace.cleanup()

注意 agent() 和 verify() 是分开的：Agent 只负责“改”，Verifier 负责“判”。第 9 章它们混在一起，模型说完成就完成；生产要拆开。

22.5 评测：三类任务，不是一张截图

第 17 章只测了 fix-due-date，2×2，两个配置都 2/2。这不够支撑“生产可用”。

生产 coding agent 的评测至少：

  • 3 类任务：修 bug、加功能、重构；
  • 每类 3 次；固定 fixture；确定性评分；
  • 指标：成功、turns、cost、diff 大小、测试是否被改、是否超预算；
  • 失败样例：是模型错、工具错、权限错，还是 Verifier 错。

这次没有跑 3×3，所以这一章不声称“生产可用”；它给的是生产形状和验收方法。

22.6 还差什么

  • 队列/租约：示例是 SQLite；生产换 Postgres/Redis，加心跳和超时回收；
  • 鉴权/租户：每租户独立 workspace、配额、审计；
  • 审批 UX：审批人是谁、超时多久、拒绝后怎么回填；
  • 镜像/依赖：固定 CLI/SDK/工具版本，启动探针；
  • 回滚：patch 可回滚，worktree 可丢弃；
  • 评测：3×3 真数据。

这些不是“再写一个提示词”，是第 20、21 章那些工程能力的组合。

22.7 下一章：研究 Agent

编码 Agent 的 Harness 控制点是“工作区 + 权限 + 验证”。

研究 Agent 的控制点是“来源 + 上下文预算 + 引用验证”。下一章把同一套六问，换到多源调研上。

本章代码 可运行控制点：https://github.com/kevinkelin/claude-agent-sdk-book/tree/master/code/projects/prod（store.py、workspace.py、policy.py、verify.py） smoke：cd code && PYTHONPATH=. python3 -m projects.prod.smoke → PASS store/workspace/policy/verify/support/research 本章是生产架构与关键控制点，不是完整产品；队列/鉴权/部署仍需按你的环境补。