s01: Agent Loop — learn-openclaw

动机：为什么需要循环

你可能以为 LLM Agent 是某种很复杂的架构。实际上，核心就是一个 while 循环。

没有这个循环，你就是那个循环——手动复制粘贴 LLM 的输出到终端，执行命令，把结果贴回去。Agent loop 把这个手动过程自动化了。

“Agents are just LLMs using tools in a loop.” — Anthropic, Building Effective Agents

OpenClaw 的源码直接印证了这一点。在 src/agents/pi-embedded-runner/run.ts 第 538 行，可以找到一个清晰的 while (true) 循环——它不断调用 runEmbeddedAttempt()，处理 tool call 结果，直到模型停止请求工具。没有复杂的状态机，没有 if/else 分支树。

交互演示

在阅读任何代码之前，先操作一下这个可视化。点击 Play 或 Next 逐步观察 agent loop 的执行过程：

while (has_tool_call) { execute → append → repeat }

messages[]

[ empty ]

1 / 7

Agent Loop 概览

每个 LLM Agent 的核心都是一个 while 循环——不断调用 LLM，直到它不再需要工具。

观察要点：

调用 LLM
检查是否有 tool call
有 → 执行工具，追加结果，回到步骤 1
没有 → 结束

退出条件只有一个：stop_reason != "tool_use"。

Ground Truth：真实代码库中的实现

我们不发明抽象——看真实产品怎么做。以下是同一个 pattern 在三个不同代码库中的实现：

Agent Loop——跨代码库对比

同一个 while(tool_call) pattern 在 Python 和 TypeScript 中的实现

pythonagents/default.py

1class DefaultAgent:
2  def run(self, task: str = "", **kwargs) -> dict:
3      self.messages = []
4      self.add_messages(
5          format_message(role="system", content=system),
6          format_message(role="user", content=instance),
7      )
8      while True:
9          try:
10              self.step()
11          except InterruptAgentFlow as e:
12              self.add_messages(*e.messages)
13          if self.messages[-1].get("role") == "exit":
14              break
15      return self.messages[-1].get("extra", {})
16 
17  def step(self) -> list[dict]:
18      return self.execute_actions(self.query())
19 
20  def query(self) -> dict:
21      message = self.model.query(self.messages)
22      self.add_messages(message)
23      return message

pythonagent/loop.py

1async def _run_agent_loop(self, initial_messages):
2  messages = initial_messages
3  iteration = 0
4 
5  while iteration < self.max_iterations:
6      iteration += 1
7      response = await self.provider.chat(
8          messages=messages,
9          tools=self.tools.get_definitions(),
10          model=self.model,
11      )
12 
13      if response.has_tool_calls:
14          messages = self.context.add_assistant_message(
15              messages, response.content, tool_calls,
16          )
17          for tool_call in response.tool_calls:
18              result = await self.tools.execute(
19                  tool_call.name, tool_call.arguments,
20              )
21              messages = self.context.add_tool_result(
22                  messages, tool_call.id, result,
23              )
24      else:
25          final_content = response.content
26          break
27 
28  return final_content, tools_used, messages

typescriptsrc/agents/pi-embedded-runner/run.ts

1// OpenClaw 主循环 (开源 TypeScript)
2// 文件: src/agents/pi-embedded-runner/run.ts:538
3 
4while (true) {
5  if (runLoopIterations >= MAX_RUN_LOOP_ITERATIONS) {
6      return { payloads: [{ text: "Exceeded retry limit", isError: true }] };
7  }
8  runLoopIterations += 1;
9 
10  const attempt = await runEmbeddedAttempt({
11      sessionId, sessionKey, prompt,
12      provider, model, tools,
13      workspaceDir: resolvedWorkspace,
14      // ... 更多参数
15  });
16 
17  // 根据 attempt 结果决定是否继续
18  // stopReason === "tool_calls" 时继续循环
19  // 否则返回结果
20}

关键观察：

mini-swe-agent 用 while True + step() + execute_actions(query())——最经典的分解
nanobot 加了 max_iterations 上限——防止无限循环，这是生产环境的标配
OpenClaw 用 while (true) + runEmbeddedAttempt() + 迭代上限——生产级的健壮性处理

三个代码库，两种语言，同一个核心 pattern。

构建：最小可行 Agent Loop

不到 30 行代码，实现完整的 agent loop：

def agent_loop(query: str):
    """核心: while(tool_call) -> execute -> append -> repeat"""
    messages = [{"role": "user", "content": query}]

    while True:
        response = client.messages.create(
            model=MODEL, system=SYSTEM, messages=messages,
            tools=TOOLS, max_tokens=4096,
        )
        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason != "tool_use":
            break

        results = []
        for block in response.content:
            if block.type == "tool_use":
                output = run_bash(block.input["command"])
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": output,
                })

        messages.append({"role": "user", "content": results})

分解这段代码：

messages 是整个对话历史——LLM 每次都能看到全部内容
stop_reason 是唯一的控制流——模型自己决定什么时候停
Tool result 作为 user message 追加——这是 API 的约定
循环是同步的——一次一个 tool call（生产系统可能并行执行）

测试：运行一个具体场景

用这个 agent 完成一个任务：“创建 hello.py 并验证它能运行”。

预期执行流程：

用户输入任务
Agent 调用 bash 写入文件
Agent 调用 bash 执行验证
Agent 返回结果文本

每一步，messages[] 都在增长。到结束时，它包含了完整的对话历史——这就是 agent 的”记忆”。

变更内容

组件	之前 (s00)	之后 (s01)
循环	无	`while(tool_call)` 主循环
工具	无	硬编码 `bash` 工具
消息	无	`messages[]` 累积对话
退出	无	`stop_reason` 控制流

本课代码: agents/s01_agent_loop.py — 69 行

试一试

cd public/code
python agents/s01_agent_loop.py "Create hello.py that prints Hello World"

可以尝试的提示:

“Create hello.py that prints Hello World”
“List all files”
“What is 2+2?”

距离生产

我们用 20 行写完了 agent loop。OpenClaw 的 run.ts 有 1165 行。多出的部分在做什么？

错误恢复。我们的循环任何异常直接崩溃。但 agent 执行一个复杂任务可能要 5 分钟和 20 次 tool call——中间任何一步失败都不应该丢弃全部进度。OpenClaw 的做法是 while (true) 外层再套 retry 循环（MAX_RUN_LOOP_ITERATIONS = 160），区分瞬时错误（网络超时，重试）、可恢复错误（context overflow，压缩后重试）和永久错误（鉴权失败，停止）。这不是过度工程——你可以在 run.ts 里搜索 isLikelyContextOverflowError、isTimeoutErrorMessage 看到完整的错误分类。

流式输出。我们等整个响应返回才展示。但 agent 生成长文本时用户会以为卡住了。流式不只是 UX 优化——它还允许用户在 agent 偏离方向时提前中断。OpenClaw 用 streamSimple() 包装 API 调用，中间状态通过 SSE 推送给客户端。

Provider failover。我们只调一个 API。但生产环境要处理 API 限流、区域故障、配额耗尽。OpenClaw 维护了一个 auth profile 池（resolveAuthProfileOrder），当某个 profile 失败时标记冷却期（markAuthProfileFailure），自动切换到下一个。这在 run.ts 第 530 行的循环中清晰可见。

第一性原理思考：为什么 agent loop 这么简单的东西，生产代码要膨胀 50 倍？因为 agent 的核心循环是 所有复杂性的汇聚点——每一轮循环都要处理网络、鉴权、context 大小、用户中断、并发安全等横切关注点。这些不是可选的装饰，而是”循环能持续跑下去”的前提。教学版忽略这些，是因为它只需要跑一次就行；生产版不能忽略，因为它要跑一万次。