第 3 章：设计你的第一个 Agent Loop

本章解锁：你能独立设计一个包含观察-规划-执行-验证四步骨架的 Agent Loop 配置，正确选择触发模式，并运用 Maker-Checker 模式设计停止条件。
预计时间：25 分钟

---

起点与环境

前置章节：第 2 章（五个构建模块）

你已经认识了 Loop 的五个构建模块和外部记忆。但认识零件不等于会造机器。这一章我们把零件组装起来，设计一个完整的 Agent Loop。

---

为什么现在做

你可能在想：我理解了五个模块，但真正设计一个 loop 的时候，从哪里开始？模块之间怎么连接？Agent 跑起来之后，什么时候该停？如果它跑偏了怎么办？

这些问题不是靠"理解概念"能回答的，需要一个具体的设计过程。而这个过程有一个明确的骨架。

---

动手实践

3.1 Loop 的四步骨架

任何 Agent Loop，不管多复杂，都可以拆解为四步：

┌──────────────────────────────────────────────┐
│                                              │
│    ┌──────────┐     ┌──────────┐            │
│    │ 1. 观察  │────→│ 2. 规划  │            │
│    │ Observe  │     │  Plan    │            │
│    └──────────┘     └────┬─────┘            │
│         ↑                │                   │
│         │                ↓                   │
│    ┌──────────┐     ┌──────────┐            │
│    │ 4. 验证  │←────│ 3. 执行  │            │
│    │ Verify   │     │ Execute  │            │
│    └──────────┘     └──────────┘            │
│         │                                    │
│         ↓                                    │
│    [停止条件满足?]                            │
│    ├── 是 → 退出 Loop                        │
│    └── 否 → 回到 1. 观察                     │
│                                              │
└──────────────────────────────────────────────┘

逐步拆解：

1. 观察（Observe）：Loop 开始时，agent 需要了解当前状态。

• 读取外部记忆，知道上次做到哪里了
• 通过 Plugin 获取新的输入（新 issue、新 PR、新告警）
• 检查环境状态（CI 结果、测试覆盖率、依赖版本）

2. 规划（Plan）：基于观察结果，agent 决定这一轮做什么。

• 有哪些任务需要处理？
• 优先级如何？
• 需要调用哪些 sub-agent？

3. 执行（Execute）：按照计划行动。

• 分派 sub-agent 完成具体工作
• 修改代码、创建 PR、发送通知
• 在 worktree 中并行处理多个任务

4. 验证（Verify）：检查执行结果是否符合预期。

• 测试是否通过？
• 代码质量是否达标？
• 结果是否和预期一致？
• 更新外部记忆，记录本轮结果

关键洞察：验证步骤是 Loop 区别于普通自动化脚本的核心。脚本只执行不验证；Loop 执行后必须验证，验证失败会触发新一轮循环。

概念理解了，但只有跑起来才能真正体会四步骨架的运转方式。下面是一个使用 OpenAI function calling 的最小实现，把 observe / plan / execute / verify 编进一个循环——你可以直接 pip install openai 后运行。

# agent_loop.py
"""Agent Loop 最小实现：观察-规划-执行-验证四步骨架"""

import json
import subprocess
from openai import OpenAI

client = OpenAI()

# ---- 工具定义（传给模型的 function schema）----
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "read_file",
            "description": "读取指定路径的文件内容",
            "parameters": {
                "type": "object",
                "properties": {"path": {"type": "string", "description": "文件路径"}},
                "required": ["path"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "run_command",
            "description": "执行 shell 命令并返回输出",
            "parameters": {
                "type": "object",
                "properties": {"command": {"type": "string", "description": "要执行的命令"}},
                "required": ["command"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "search_code",
            "description": "在项目中搜索包含关键字的代码行",
            "parameters": {
                "type": "object",
                "properties": {
                    "keyword": {"type": "string", "description": "搜索关键词"},
                    "directory": {"type": "string", "description": "搜索目录，默认当前目录"},
                },
                "required": ["keyword"],
            },
        },
    },
]

def execute_tool(name: str, args: dict) -> str:
    """工具的真正实现：根据函数名分派到对应操作"""
    if name == "read_file":
        with open(args["path"], "r") as f:
            return f.read()
    elif name == "run_command":
        result = subprocess.run(args["command"], shell=True, capture_output=True, text=True, timeout=30)
        return result.stdout + result.stderr
    elif name == "search_code":
        result = subprocess.run(
            f"grep -rn '{args['keyword']}' {args.get('directory', '.')}",
            shell=True, capture_output=True, text=True, timeout=30,
        )
        return result.stdout or "未找到匹配项"
    return f"未知工具: {name}"

SYSTEM_PROMPT = "你是一个自主编程助手。你可以调用工具完成任务。当你认为任务已完成时，回复 [DONE]。"

def agent_loop(task: str, max_iterations: int = 10):
    """
    主循环：反复执行 观察→规划→执行→验证 直到完成或达到上限。
    - Observe + Plan：模型读取对话历史，决定调用哪些工具
    - Execute：实际运行工具，拿到结果
    - Verify：把结果回传给模型，让它判断是否需要继续
    """
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": task},
    ]

    for i in range(max_iterations):
        # 1-2. Observe + Plan：模型观察历史并决定下一步调用哪个工具
        response = client.chat.completions.create(model="gpt-4o", messages=messages, tools=TOOLS)
        msg = response.choices[0].message
        messages.append(msg)

        # 如果模型没调用工具 → 它在做最终判断
        if not msg.tool_calls:
            if "[DONE]" in (msg.content or ""):
                print(f"第 {i+1} 轮：任务完成。")
                return
            messages.append({"role": "user", "content": "请继续，或回复 [DONE] 表示完成。"})
            continue

        # 3. Execute：逐个执行模型选择的工具
        for tc in msg.tool_calls:
            args = json.loads(tc.function.arguments)
            print(f"第 {i+1} 轮：调用 {tc.function.name}({args})")
            result = execute_tool(tc.function.name, args)
            # 4. Verify：把执行结果塞回对话，让模型判断是否继续
            messages.append({"role": "tool", "tool_call_id": tc.id, "content": result})

    print(f"达到最大迭代次数 {max_iterations}，强制停止。")

if __name__ == "__main__":
    agent_loop("读取 main.py，检查是否有未处理的异常，如有则列出位置")

3.2 两种触发模式：/goal vs /loop

Addy Osmani 在 Loop Engineering 中区分了两种触发模式，它们对应不同的使用场景：

/goal 模式：反复跑直到条件满足

mode: goal
trigger:
  type: manual  # 手动启动一次
goal:
  description: "所有测试通过且覆盖率 > 80%"
  verification:
    command: "npm test -- --coverage"
    success_criteria: "exit code 0 且覆盖率 >= 80%"
max_iterations: 10
on_max_iterations: "notify_human"

/goal 的特点：

• 有一个明确的终止条件
• 启动后自动迭代，直到目标达成或达到最大迭代次数
• 适合收敛型任务：修 bug、达到覆盖率目标、通过所有 lint 规则

/loop 模式：按周期重复运行

mode: loop
trigger:
  type: scheduled
  cron: "0 9 * * 1-5"  # 工作日每天上午 9 点
task:
  description: "扫描新 issue，分类标记，简单问题自动回复"
memory:
  file: ".agent/issue-triage-memory.md"
  read_on_start: true
  write_on_end: true

/loop 的特点：

• 没有"完成"的概念，是持续运行的
• 每次 run 处理当前批次的工作，然后等待下次触发
• 适合周期型任务：issue 分类、依赖检查、日报生成

怎么选？ 问自己一个问题：这个任务有没有"做完了"的状态？

判断维度	/goal	/loop
有明确完成条件	是（如"测试全过"）	否（如"持续监控"）
需要迭代纠错	是（跑不过就改了再跑）	不主要（每次 run 相对独立）
运行频率	一次性（启动后持续迭代）	周期性（定时触发）
典型场景	修 bug、重构、达标	巡检、分类、报告

3.3 Maker-Checker 模式

这里有一个微妙但重要的设计决策：谁来判断 Loop 是否该停下来？

如果让执行任务的 agent 自己判断"我做完了"，就像让学生自己给自己打分。它可能因为过度自信而过早停止，也可能因为陷入某个方向而永远不停。

Maker-Checker 模式把"做事"和"判断做没做好"分开：

Maker（执行者）：写代码、修 bug、生成报告
    ↓ 产出结果
Checker（验证者）：一个独立的模型/agent，判断结果是否达标
    ↓ 判断结果
    ├── 通过 → Loop 结束
    └── 不通过 → 反馈问题，Maker 继续

为什么要用不同的模型实例？

这不只是"多一层检查"。当同一个 agent 既做又查时，它会有确认偏误——倾向于认为自己的输出是对的。用一个独立的 checker（甚至用不同的模型），可以获得真正独立的判断。

Checker 的配置例子：

checker:
  model: "claude-sonnet"  # 不需要和 maker 用同一个模型
  prompt: |
    你是一个代码审查专家。以下是一个 agent 修改的代码差异。
    判断这个修改是否：
    1. 正确解决了问题
    2. 没有引入新的 bug
    3. 符合项目编码规范
    4. 有足够的测试覆盖
    
    回答 PASS 或 FAIL，并给出具体原因。
  on_fail:
    action: "feedback_to_maker"
    include: "checker 的具体反馈"

上面的 YAML 描述了 Checker 的配置思路，但实际项目中更常见的做法是直接用代码实现 Maker-Checker 的调用逻辑。下面这段代码展示了如何用两个独立的模型调用分别做"生成"和"验证"，并在 FAIL 时把反馈塞回 Maker 重试。

# maker_checker.py
"""Maker-Checker 模式：用两个独立模型调用分别负责生成和验证"""

from openai import OpenAI

client = OpenAI()

def maker(task: str, feedback: str = "") -> str:
    """Maker：根据任务（和上轮反馈）生成代码"""
    prompt = f"请完成以下任务：\n{task}"
    if feedback:
        prompt += f"\n\n上一轮 Checker 的反馈：\n{feedback}\n请根据反馈修正。"
    resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
    )
    return resp.choices[0].message.content

def checker(task: str, output: str) -> tuple[bool, str]:
    """Checker：独立判断 Maker 的输出是否合格，返回 (是否通过, 反馈)"""
    prompt = (
        f"任务要求：\n{task}\n\n待检查的输出：\n{output}\n\n"
        "请判断输出是否满足要求。第一行回复 PASS 或 FAIL，"
        "第二行起说明原因；如果不合格，给出具体修改建议。"
    )
    # 实践中建议 checker 换用不同模型，以减少确认偏误
    resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
    )
    text = resp.choices[0].message.content
    passed = text.strip().upper().startswith("PASS")
    feedback = text.split("\n", 1)[1].strip() if "\n" in text else ""
    return passed, feedback

def maker_checker_loop(task: str, max_retries: int = 3):
    """主循环：FAIL 时把 Checker 反馈塞回 Maker 重试"""
    feedback = ""
    for attempt in range(1, max_retries + 1):
        output = maker(task, feedback)                  # Maker 生成
        passed, feedback = checker(task, output)        # Checker 验证
        if passed:
            print(f"第 {attempt} 轮通过，最终输出：\n{output}")
            return output
        print(f"第 {attempt} 轮未通过，反馈：{feedback}")
    print(f"达到最大重试次数 {max_retries}，停止。")

if __name__ == "__main__":
    maker_checker_loop("写一个 Python 函数，输入列表返回去重后保持顺序的列表")

3.4 停止条件的设计

Loop 最危险的地方不是跑不动，而是停不下来。一个设计良好的 loop 需要明确的停止条件。

三类停止条件：

类型	描述	例子
成功停止	目标达成	所有测试通过、覆盖率达标
失败停止	达到最大迭代次数或最大时间	跑了 10 轮还没通过，停下来通知人
异常停止	检测到 loop 在原地打转	连续 3 轮 checker 反馈相同的问题

原地打转检测是最容易被忽视的。如果 agent 连续几轮都在做同样的修改、得到同样的错误，它很可能卡在了一个它无法自己解决的问题上。这时候 loop 应该停下来，把问题交给人类。

stop_conditions:
  success:
    - condition: "all_tests_pass"
      verification: "npm test"
  failure:
    - condition: "max_iterations"
      value: 10
      action: "notify_human"
    - condition: "max_duration"
      value: "30m"
      action: "notify_human"
  anomaly:
    - condition: "repeated_failure"
      threshold: 3  # 连续 3 轮相同错误
      action: "stop_and_report"

原地打转检测在 YAML 里只需一行 threshold: 3，但理解它的运行时逻辑同样重要。下面是一段最小的检测实现，可以嵌入到任何 agent loop 中。

# loop_guard.py
"""原地打转检测：连续相同错误时主动退出，避免无限循环"""

def detect_spinning(recent_errors: list[str], threshold: int = 3) -> bool:
    """检测最近 threshold 轮错误是否完全相同"""
    if len(recent_errors) < threshold:
        return False
    last_n = recent_errors[-threshold:]   # 取最近 threshold 条错误
    return len(set(last_n)) == 1           # 全部相同 → True

# 在 agent_loop 中集成使用
recent_errors = []
for attempt in range(max_iterations):
    error = run_iteration()           # 假设返回本轮 Checker 的 FAIL 原因
    if error is None:
        break                          # PASS → 正常退出
    recent_errors.append(error)
    if detect_spinning(recent_errors, threshold=3):
        print("检测到连续 3 轮相同错误，判定为原地打转，停止 Loop。")
        break

3.5 综合实践：设计一个代码审查 Loop

现在把所有内容综合起来，设计一个具体的 Agent Loop：自动代码审查系统。

场景：每当仓库收到新 PR，自动运行代码审查，提出修改建议。如果 PR 作者修改后重新提交，再次审查直到通过。

你的实践任务：先自己尝试设计，然后对比下面的参考方案。

参考方案：

# code-review-loop.yaml

name: "自动代码审查"
mode: loop
trigger:
  type: event
  event: "pull_request.opened OR pull_request.synchronize"

# 构建模块配置
skills:
  - path: ".agent/skills/code-review.md"
    description: "项目编码规范、审查标准"

plugins:
  github:
    server: "mcp-server-github"
    permissions: ["read:pull_requests", "write:reviews"]

memory:
  file: ".agent/code-review-memory.md"
  structure: |
    ## 已审查的 PR
    ## 已知的项目约定
    ## 常见问题模式

sub_agents:
  explorer:
    role: "理解 PR 变更的上下文"
    task: "阅读被修改文件的完整上下文、相关测试、调用链"
  reviewer:
    role: "执行代码审查"
    task: "基于项目约定和上下文，逐文件审查变更"
  verifier:
    role: "验证审查质量"
    task: "检查 review 意见是否有误判、是否遗漏关键问题"

# 四步骨架
loop_steps:
  observe:
    - "读取 memory：了解该 PR 之前的审查历史"
    - "通过 GitHub Plugin 获取 PR 详情和 diff"
    - "检查 CI 状态"
  
  plan:
    - "判断这是新 PR 还是修改后重新提交"
    - "如果是重新提交，聚焦上次提出但未修复的问题"
    - "确定审查优先级：安全问题 > 逻辑错误 > 风格问题"
  
  execute:
    - "explorer 阅读相关上下文"
    - "reviewer 基于上下文进行审查"
    - "verifier 检查审查结果质量"
    - "在 PR 上提交审查意见"
  
  verify:
    - "确认审查意见已提交"
    - "更新 memory：记录本次审查结果"
    - "如果发现严重安全问题，额外通知团队"

# 停止条件（对单个 PR 而言）
stop_conditions:
  success:
    - "PR 通过审查（没有严重问题）"
  failure:
    - condition: "max_review_rounds"
      value: 5
      action: "标记为需要人工审查"
  anomaly:
    - condition: "PR 作者连续 3 轮没有回应审查意见"
      action: "发送提醒通知"

设计复盘——注意这几个设计决策：

1. 三个 sub-agent 的分工：explorer 负责理解上下文，reviewer 负责审查，verifier 检查审查质量。这就是 Maker-Checker 模式——reviewer 是 maker，verifier 是 checker。

2. Memory 的结构：不仅记录"审查了哪些 PR"，还记录"已知的项目约定"和"常见问题模式"。后两者让 loop 随着运行时间增长变得越来越准确。

3. 停止条件的三层设计：成功（审查通过）、失败（超过最大轮次）、异常（作者不回应）。覆盖了 loop 可能遇到的三类情况。

4. 重新提交时的聚焦：观察步骤中检查是否为"修改后重新提交"，如果是则聚焦上次的问题。这需要 Memory 的支持——没有 Memory，agent 不知道上次提过什么问题。

---

理解检查

1. 选择题：以下哪个任务更适合 /goal 模式？

   • A. 每天扫描代码中的 TODO 注释并生成报告
   • B. 把项目的 TypeScript 版本从 4.x 升级到 5.x，直到所有编译错误都修复

2. 设计题：一个 agent loop 的停止条件只设了"成功停止"（所有测试通过），没有设失败停止和异常停止。描述可能发生的最坏情况。

3. 辨析题：Maker-Checker 模式中，checker 可以和 maker 是同一个 agent 吗？为什么实践中建议分开？

参考答案：

1. B。TypeScript 升级有明确的完成条件（编译通过），适合 /goal。TODO 扫描是持续性的周期任务，没有"做完"的概念，适合 /loop。

2. 最坏情况：agent 进入无限循环。它修改代码 → 测试不通过 → 再修改 → 又不通过，无限下去。可能的后果包括：消耗大量 API 调用费用、长时间占用 CI 资源、甚至由于反复修改导致代码越改越烂。这就是为什么必须设 max_iterations 和 repeated_failure 检测。

3. 技术上可以是同一个 agent（发两次调用，一次做一次查）。但实践中建议分开，原因有二：(1) 确认偏误——同一个模型实例倾向于认为自己的输出是对的，即使换了 prompt 也很难完全消除这个偏差；(2) 关注点分离——checker 的 prompt 可以专注于"挑毛病"，不需要知道实现过程中的上下文和权衡，反而可能更客观。

---

发生了什么

这一章你从零设计了一个完整的 Agent Loop。

核心概念回顾：

• 四步骨架：观察 → 规划 → 执行 → 验证，验证是区别于普通脚本的关键
• /goal vs /loop：有完成条件用 /goal，周期性任务用 /loop
• Maker-Checker 模式：做事的和判断的分开，避免确认偏误
• 三类停止条件：成功停止、失败停止、异常停止（原地打转检测）
• Loop 的质量取决于验证步骤的设计和停止条件的完备性

---

常见误区

误区	纠正
"Loop 就是 while(true) 加个 LLM 调用"	Loop 的核心价值在验证步骤和停止条件，不是循环本身。一个没有验证的 loop 不比 cron job 高级。
"Maker 和 Checker 用同一个 prompt 就行"	Checker 的 prompt 应该完全不同于 Maker。Maker 的 prompt 关注"怎么做"，Checker 的 prompt 关注"做得对不对"。混在一起会削弱检查效果。
"停止条件不重要，大不了手动停"	如果 loop 在凌晨 3 点跑飞了呢？没有自动停止条件的 loop 就像没有断路器的电路。
"/goal 和 /loop 可以混用"	它们解决不同的问题。如果一个任务既有完成条件又需要周期运行，正确做法是用 /loop 定期触发，每次 run 内部用 /goal 式的迭代逻辑。
"四步骤必须严格按顺序"	骨架是概念模型，不是硬性约束。实际中观察和规划可能交织，执行中可能需要回到观察获取更多信息。重要的是每一步的职责明确。

---

本章检查点

完成以下清单后，你可以进入下一章：

• [ ] 我能画出四步骨架的流程图
• [ ] 我能区分 /goal 和 /loop 的适用场景
• [ ] 我理解 Maker-Checker 模式的设计动机
• [ ] 我能为一个具体任务设计三类停止条件
• [ ] 我已经完成（或至少仔细阅读）了代码审查 Loop 的设计

下一章预告：你现在能设计一个 Agent Loop 了。但当任务复杂到需要多个独立 Agent 协作时，单个 Loop 就不够了。下一章我们看看"什么信号"告诉你该从 Loop 走向 Graph，以及 Graph Engineering 到底在解决什么新问题。

---

← 上一章：理解 Loop Engineering 的五个构建模块 | 下一章：当一个 Loop 不够用 →