#!/usr/bin/env python3 """01_agent_loop_simulator.py — 带停止条件的智能体循环模拟器（无需 API key） 用「脚本化策略」代替真实 Claude，完整演示官方 agent loop 五要素： 目标 Goal -> 推理 Reasoning -> 规划 Planning -> 行动 Action(tool_use) -> 观察 Observation 以及全部五类停止条件 (stopping conditions)： ① 任务完成(可验证成功标准) ② 最大迭代次数 ③ 预算上限 ④ 人工检查点(不可逆操作前) ⑤ 阻塞升级(连续失败转人工) 场景：一个内部「代码修复 agent」在模拟仓库里定位并修复一个 bug。 对照官方描述（访问于 2026-08-30）: https://www.anthropic.com/engineering/building-effective-agents ("The task often terminates upon completion, but it's also common to include stopping conditions (such as a maximum number of iterations) to maintain control.") 运行: python3 01_agent_loop_simulator.py """ from __future__ import annotations from dataclasses import dataclass, field # ---------- 模拟环境（真实系统中 = 文件系统 + 测试运行器 = ground truth 的来源） ---------- FAKE_REPO = { "src/auth.py": "def login(user, pwd):\n t = make_token(user)\n return t\n", "src/tokens.py": ( "def make_token(user):\n" " # BUG: expiry hardcoded to 1 second -> refresh instantly 401s\n" " return {'exp': 1}\n" ), "tests/test_tokens.py": ( "def test_expiry():\n" " t = make_token('u')\n" " assert t['exp'] > 60, 'token must live > 60s'\n" ), } APPROVAL = {"approved": True} # 模拟人工审批结果（生产中应来自审批 UI） @dataclass class Trace: """一轮循环 = 一次 Reasoning->Planning->Action->Observation 记录。""" turn: int reasoning: str plan: str tool: str observation: str @dataclass class AgentLoop: """agent loop 本体：模型策略 + 确定性外壳（工具执行 / 权限门 / 停止条件）。 关键设计（对照模块 08 README §2）： - 「下一步做什么」由 *模型*（此处为脚本化策略）决定 —— 这是 agent 的定义; - 但停止条件、权限门、工具执行永远是 *应用层确定性代码* —— 三样不交给模型。 """ max_iterations: int = 8 budget_tokens: int = 8_000 tokens_per_turn: int = 900 script: dict[int, dict[str, str]] = field(default_factory=dict) traces: list[Trace] = field(default_factory=list) # ---- 应用层：工具执行（真实系统 = 模块 04 的 tool_use -> tool_result 回路） ---- def execute(self, tool: str, repo: dict[str, str]) -> str: """确定性执行工具并返回观察。真实系统中对应 tool_result content。""" if tool == "run_tests": src = repo["src/tokens.py"] passed = "'exp': 3600" in src return f"tests: 1 passed, 0 failed" if passed else "tests: 0 passed, 1 failed (test_expiry)" if tool == "read_file": return repo["src/tokens.py"] if tool == "edit_file": repo["src/tokens.py"] = repo["src/tokens.py"].replace("'exp': 1", "'exp': 3600") return "edited src/tokens.py: exp 1 -> 3600" if tool == "ask_human": return "HUMAN: approved" if APPROVAL["approved"] else "HUMAN: rejected" if tool == "deploy": return "deployed to staging" raise ValueError(f"unknown tool {tool}") # ---- 模型层（脚本化策略）：真实系统中 = 一次 Claude Messages 调用 ---- def model_step(self, turn: int, last_obs: str) -> dict[str, str]: """返回 reasoning / plan / tool。真实系统中由 Claude 基于 (system prompt + 全部历史 + 工具定义) 自主决定下一步。""" return self.script[turn] # ---- 外壳：把循环跑起来，停止条件全部在此裁决 ---- def run(self, goal: str, repo: dict[str, str]) -> str: tokens = 0 last_obs = "(start)" failure_streak = 0 print(f"GOAL: {goal}") print(f"stop conditions: max_iterations={self.max_iterations}, " f"budget_tokens={self.budget_tokens}, human gate before deploy\n") for turn in range(1, self.max_iterations + 1): if tokens + self.tokens_per_turn > self.budget_tokens: # ③ 预算 return self._stop("BUDGET_EXHAUSTED", tokens) tokens += self.tokens_per_turn step = self.model_step(turn, last_obs) # 模型决定下一步 print(f"[turn {turn}] reasoning: {step['reasoning']}") print(f" plan: {step['plan']}") if step["tool"] == "deploy" and not APPROVAL["approved"]: # ④ 人工检查点 print(" HUMAN CHECKPOINT: deploy rejected -> stop") return self._stop("HUMAN_REJECTED", tokens) last_obs = self.execute(step["tool"], repo) # 行动+观察 print(f" observation: {last_obs}\n") self.traces.append(Trace(turn, step["reasoning"], step["plan"], step["tool"], last_obs)) if step["tool"] == "deploy": # ① 任务完成 return self._stop("TASK_COMPLETE", tokens) if "failed" in last_obs: failure_streak += 1 if failure_streak >= 3: # ⑤ 阻塞升级 return self._stop("ESCALATED_TO_HUMAN", tokens) else: failure_streak = 0 return self._stop("MAX_ITERATIONS", tokens) # ② 最大迭代 @staticmethod def _stop(reason: str, tokens: int) -> str: print(f"=== LOOP STOPPED: {reason} (tokens spent ~{tokens}) ===\n") return reason # ---------- 脚本化策略：成功路径 ---------- SUCCESS_SCRIPT = { 1: {"reasoning": "任务是修 bug；先拿环境 ground truth——跑测试确认失败", "plan": "调用 run_tests 建立基线", "tool": "run_tests"}, 2: {"reasoning": "测试失败于 test_expiry，观察指向 token 过期时间", "plan": "读 src/tokens.py 定位", "tool": "read_file"}, 3: {"reasoning": "发现 'exp': 1，与断言 exp>60 不符，这是根因", "plan": "编辑为 3600", "tool": "edit_file"}, 4: {"reasoning": "修改后必须用环境反馈验证，不能自说自话", "plan": "重跑测试验证", "tool": "run_tests"}, 5: {"reasoning": "测试通过；部署是不可逆操作，先过人工检查点", "plan": "ask_human 请求批准", "tool": "ask_human"}, 6: {"reasoning": "人类已批准", "plan": "部署到 staging", "tool": "deploy"}, } # ---------- 脚本化策略：跑飞路径（触发 max_iterations 保护） ---------- RUNAWAY_SCRIPT = { t: {"reasoning": "还没找到根因…再读一遍文件", "plan": "read_file", "tool": "read_file"} for t in range(1, 9) } def main() -> None: repo = dict(FAKE_REPO) print("========== 场景 1：正常路径（完成 + 人工检查点 + 部署） ==========") loop = AgentLoop(script=SUCCESS_SCRIPT) r1 = loop.run("Fix the failing token-expiry test and deploy to staging", repo) assert r1 == "TASK_COMPLETE", r1 assert "'exp': 3600" in repo["src/tokens.py"], "repo must be fixed" print("========== 场景 2：跑飞路径（max iterations 兜底） ==========") runaway = AgentLoop(script=RUNAWAY_SCRIPT, max_iterations=4, budget_tokens=100_000) r2 = runaway.run("Fix the failing test", dict(FAKE_REPO)) assert r2 == "MAX_ITERATIONS", r2 print("========== 场景 3：预算耗尽（budget guard） ==========") cheap = AgentLoop(script=SUCCESS_SCRIPT, budget_tokens=1_800) # 只够 2 轮 r3 = cheap.run("Fix the failing test", dict(FAKE_REPO)) assert r3 == "BUDGET_EXHAUSTED", r3 print("========== 场景 4：人工拒绝（human gate） ==========") APPROVAL["approved"] = False gated = AgentLoop(script=SUCCESS_SCRIPT) r4 = gated.run("Fix and deploy", dict(FAKE_REPO)) assert r4 == "HUMAN_REJECTED", r4 print("全部停止条件行为验证通过：") print(" ① TASK_COMPLETE / ② MAX_ITERATIONS / ③ BUDGET_EXHAUSTED / ④ HUMAN gate / (⑤ 见下)") APPROVAL["approved"] = True # ⑤ 阻塞升级：连续 3 次失败观察后转人工 fail_script = {t: {"reasoning": "retry", "plan": "run tests", "tool": "run_tests"} for t in range(1, 9)} stuck = AgentLoop(script=fail_script) r5 = stuck.run("Fix the failing test", dict(FAKE_REPO)) assert r5 == "ESCALATED_TO_HUMAN", r5 print(" ⑤ ESCALATED_TO_HUMAN — 连续失败转人工，同样通过") if __name__ == "__main__": main()