第 13 章：部署形态

草稿章节。跑通格式用的第一版，正式索引前会再打磨。

本章目标

为 PR reviewer 制定一份部署方案。此前它只在你的笔记本上跑，由你触发，用你的个人凭证。本章决定它真正在哪里运行，并写清随之而来的约束：同时跑多少个评审，如何不撞模型限流，密钥放在哪里，以及 agent 被允许触碰什么。读完本章，你会有：四种常见形态的对比表、一个选定的形态（带 worker 的队列）、以及一份 webhook 触发 worker 的具体伪代码。

部署形态不是无关紧要的外观选择。它决定了你的延迟特征、成本上限、负载下的失败行为，以及密钥一旦泄露的爆炸半径。认真选还是随便选，结果就是流量高峰时这个 agent 能扛住、还是直接崩掉的区别。

前置准备

• 第三部分完成的 agent，已带护栏与沙箱。
• 模型与代码托管平台的 API 密钥（例如一个 GitHub token）。
• 对流量的大致判断：每小时多少次评审，以及有多突发（一分钟内涌进十个 PR，和一小时内均匀分布十个，是完全不同的形态）。

动手做

1. 列出形态及其取舍

常见有四种形态。每种在延迟、成本与运维负担之间做出不同的取舍。有一点四种都适用：生产级 agent 不应在进程内存里持有重要状态，因为任何实例随时可能被杀掉再替换。状态要放在队列、数据库和代码托管平台里。

形态	触发方式	延迟	并发控制	适合
同步 API 服务	HTTP 请求，调用方等待	最低，但调用方阻塞	连接/线程池	交互式、快调用
任务队列 + worker	立即入队，worker 拉取	秒到分钟	worker 数量	webhook 驱动的后台工作
定时 / 批跑	定时触发	跑起来才算	批大小	夜间扫描、摘要
Serverless 函数	事件触发，函数拉起	冷启动开销	平台并发上限	突发、短、低频的工作

一次 PR 评审要花几十秒到几分钟（它要读文件并做好几次模型调用），所以让 webhook 调用方同步等待并不合适：代码托管平台会把这个 webhook 判超时。这就排除了同步服务。

2. 形态匹配触发方式

PR reviewer 由代码托管平台的 webhook 触发。正确的模式是：webhook 处理器几乎什么都不干，只校验载荷、入队一个任务、立刻返回 200；worker 拉取任务，在带外完成缓慢的评审工作。cron 适合夜间批处理（每周一次评审质量摘要）。Serverless 适合突发、短、低频的评审，你不想常驻一个 worker，愿意接受冷启动成本。对稳定的 webhook 流量，带 worker 的队列形态在可控性和成本可预测性上胜出，所以我们就建这个。

3. 限制并发

设定一个固定的 worker 数量与单 worker 在途上限，这样你永远不会无界扇出模型请求。这是防限流和失控成本的第一道措施：N 个 worker 每个最多同时做一次模型调用，把峰值负载封在一个你自己定的数上，不管涌进多少 webhook。第 10 课的生产指引把这件事说成防止 agent 变成黑盒，它列的可观测指标里就有请求错误，而无界扇出恰恰会大量制造这种错误。

review_pr 就是把 agent 回路包成一个函数：在 worker 里调用 query()，按事件触发，跑完一次评审就返回。

# review_pr：把 agent 回路包进一个可被 worker 调用的函数
from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, TextBlock

async def review_pr(job):
    options = ClaudeAgentOptions(
        system_prompt=REVIEW_CONTRACT,
        allowed_tools=["Read"],   # 只读：无状态、无副作用
        max_turns=6,
        cwd=job["repo"],
    )
    out = ""
    async for message in query(prompt=f"审查 PR #{job['pr']}（{job['sha']}）。", options=options):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    out += block.text
    return out

# Worker 池：N 个 worker，每个一次拉一个任务 -> 峰值并发 = N
import asyncio

async def worker(name, queue):
    while True:
        job = await queue.get()           # 阻塞直到有活
        try:
            await review_pr(job)           # 每个 worker 同时只跑一个评审
        except Exception as e:
            await handle_failure(job, e)   # 重试/退避或进死信队列
        finally:
            queue.task_done()

async def main(concurrency=4):
    queue = await connect_queue()
    await asyncio.gather(*[worker(f"w{i}", queue) for i in range(concurrency)])

// review_pr：把 agent 回路包进一个可被 worker 调用的函数
import { query } from "@anthropic-ai/claude-agent-sdk";

async function reviewPr(job) {
  const options = {
    systemPrompt: REVIEW_CONTRACT,
    allowedTools: ["Read"],   // 只读：无状态、无副作用
    maxTurns: 6,
    cwd: job.repo,
  };
  let out = "";
  for await (const message of query({ prompt: `审查 PR #${job.pr}（${job.sha}）。`, options })) {
    if (message.type === "assistant") {
      for (const block of message.message.content) {
        if (block.type === "text") out += block.text;
      }
    }
  }
  return out;
}

// Worker 池：N 个 worker，每个一次拉一个任务 -> 峰值并发 = N
async function worker(name, queue) {
  while (true) {
    const job = await queue.get();        // 阻塞直到有活
    try {
      await reviewPr(job);                // 每个 worker 同时只跑一个评审
    } catch (e) {
      await handleFailure(job, e);        // 重试/退避或进死信队列
    } finally {
      queue.taskDone();
    }
  }
}

async function main(concurrency = 4) {
  const queue = await connectQueue();
  await Promise.all(
    Array.from({ length: concurrency }, (_, i) => worker(`w${i}`, queue)),
  );
}

4. 搭建 webhook 到 worker 的链路

把它们串起来：一个返回很快的轻量 webhook 端点、一个持久化队列、以及上面的 worker 池。端点在信任载荷之前必须校验 webhook 签名（不鉴权的端点谁都能往里塞数据），然后入队并返回。在请求处理器里直接做评审是个经典错误：处理器阻塞一分钟，代码托管平台等不及就重发，你就把同一个 PR 评审了两次。

# Webhook 端点：校验、入队、返回。绝不内联评审。
@app.post("/webhook/pr")
async def on_pr_event(request):
    body = await request.body()
    if not valid_signature(body, request.headers["X-Hub-Signature-256"], WEBHOOK_SECRET):
        return Response(status=401)        # 拒绝伪造事件
    event = parse(body)
    if event.action in ("opened", "synchronize"):
        await queue.put({
            "pr": event.pr_number,
            "repo": event.repo,
            "sha": event.head_sha,
            "idempotency_key": f"{event.repo}:{event.pr_number}:{event.head_sha}",
        })
    return Response(status=200)            # 立刻返回，活儿稍后再干

idempotency_key 很关键：webhook 可能被投递不止一次，worker 也可能在任务中途崩溃后重试。用「仓库 + PR + 提交 SHA」做键，能让 worker 跳过它已经为那个确切提交发过的评审，这样重复投递不会产出重复评论。

5. 为限流做准备

worker 有时会撞上模型服务商的每分钟限额，突发期间尤其容易。在模型调用外面加上带指数退避和抖动的重试，让有界队列去吸收突发，而不是猛砸 API。队列起的是缓冲作用：涌进来的 webhook 安全地堆在里面，worker 以一个能持续的速率把它们处理掉。当前限额请查模型服务商的官方文档，别把某个限额当成事实写死，去查。

async def call_model_with_backoff(req, max_tries=5):
    delay = 1.0
    for attempt in range(max_tries):
        try:
            return await model.call(req)
        except RateLimited as e:
            sleep_for = e.retry_after or (delay + random.random())  # 尊重服务端提示
            await asyncio.sleep(sleep_for)
            delay *= 2                                               # 指数退避
    raise PermanentFailure("rate limit not clearing")               # 显式失败

6. 放置密钥并画出权限边界

密钥放在密钥管理服务或平台环境变量里，别写进代码，也别写进日志。每个环境（dev、staging、prod）用各自的密钥，这样 staging 的泄露碰不到 production。然后把托管平台 token 的权限收窄到 agent 恰好需要的范围：读取 diff、发表评审评论，就这两样。一个还能推代码或合并的 token，会把一次提示注入 bug 变成一次安全事故。agent 能造成多大损害，上限就是它的权限，所以权限给得越窄越好。

# 按环境、最小权限的配置（示意）。
env: production
secrets:                       # 由密钥管理服务注入，不写在本文件里
  MODEL_API_KEY: ${vault:prod/model_key}
  GITHUB_TOKEN:  ${vault:prod/gh_review_token}
github_token_scopes:           # 只给 reviewer 需要的
  - pull_requests: read        # 读 diff
  - pull_requests: write       # 发评审评论
  # 没有 contents:write，没有 workflow，没有 admin

习得「独当一面」小满第一次从你的笔记本搬到了真服务器上，开始接陌生流量：由 webhook 触发、在 worker 池里干活，并发被定死、密钥按环境隔离、token 收窄到只读和只评论，流量高峰时也扛得住。

如何验证

• 同时打十个 webhook，确认在途评审停在你的 worker 数，而不是十个。观察队列深度上升再排空。
• 把同一个 webhook 投两次，确认幂等键阻止了重复评论。
• 在预发环境吊销一个密钥，确认 agent 显式失败（报错并停止），而不是静默地什么都不发。
• grep 日志，确认任何密钥都不出现其中，连错误栈里也没有。

习得「扛住压力」你现在能一次打十个 webhook，确认在途评审稳在 worker 数而不是失控扇出，重复投递不会发出第二条评论，吊销密钥时它显式报错而不是默默不动，日志里一个密钥也漏不出去。

原理

队列把快而不可信的 webhook 和慢而昂贵的评审解耦。就这一步解耦，把后面几件事都顺带解决了：并发控制变成「开多少个 worker」，限流吸收变成「队列能堆多深」，重试也变得安全，因为任务是幂等且持久的。无状态让 worker 随时可以丢掉，于是你能在负载高时扩容、在空闲时杀掉，都不丢活儿。

小结

你对比了四种部署形态，选了带 worker 的队列来匹配 webhook 触发，然后限制并发、让任务幂等、用退避处理限流、按环境隔离密钥、把托管平台 token 收窄到只读和只评论。形态不只是代码跑在哪里：它决定了你的成本、延迟与爆炸半径。下一章讲版本与回滚。

常见坑

• 在 webhook 处理器里内联评审，导致代码托管平台超时重发，把活儿翻倍。
• 无界扇出：没有并发上限，一个繁忙时段就会耗尽限额或预算。
• 没有幂等键，重发的 webhook 会在同一个提交上再发一遍评审。
• 跨环境共用密钥：预发环境一旦泄露，生产环境随之沦陷。
• token 权限过宽：一个能推代码而非只评论的 token，会把一个 bug 变成一次事故。

小满第一次离开你的本地，跑在真服务器上，面对从没见过的用户。头几批请求它反复自我核对，迟迟不敢交付，紧张得很。你在屏幕这头，像送孩子第一天上学。门外天光，亮了。

0123456789101112131415

刚点亮 门外天光 · 地图已点亮 14 / 16

它会改版、会长大，那旧的小满去哪了？下一站：年轮室。

来源

1. Microsoft: AI Agents in Production · official
2. Microsoft: AI Agents in Production（第 10 课） · official