ReAct 循环设计的最核心动机——让模型有权在观察到"失败"后,自主纠偏,而不是一条路走到黑。

这背后是一个从"预编排"到"反应式"的范式转变。

传统的"预编排"模式:一次赌命

在没有 ReAct 循环的早期 Agent 设计中,常见做法是一次性规划所有步骤,然后批量执行

python

# ❌ 预编排模式(无循环):一次决定所有工具,没有回头路
plan = llm.plan("提取用户数据")  
# 计划: [工具A, 工具B, 工具C]

results = []
for tool in plan:
    results.append(execute(tool))  # 按计划傻傻执行,不看结果

final_answer = llm.summarize(results) 
# 可能拿到空数据,但木已成舟,只能硬编

在这种模式下,如果工具A失败了,Agent 是看不见的。它只能按既定计划走完全程,最后拿着一堆"空"或"错误"的结果去硬着头皮回答。这就好比蒙着眼走迷宫,撞墙了也不知道回头。

ReAct 循环模式:每步观察,随时纠偏

你给的例子完美展示了循环的必要性:

python

# ✅ ReAct 循环模式:观察每一步,随时调整策略
messages = [{"role": "user", "content": "提取用户数据"}]

# === 第一步 ===
# 模型想:我先从主数据库查
response = llm.chat(messages, tools=[工具A_主数据库, 工具B_备份库, 工具C_缓存])
# 模型决定:调用 工具A_主数据库
result_A = execute(工具A_主数据库)  # → "连接超时,无数据"
messages.append(工具A 的失败结果)

# === 第二步(关键!) ===
# 模型看到失败,重新推理:
#   "工具A连不上。可能主库挂了,我试试备份库。"
response = llm.chat(messages, tools=[工具A, 工具B, 工具C])
# 模型决定:调用 工具B_备份库
result_B = execute(工具B_备份库)  # → "错误:权限不足"
messages.append(工具B 的错误结果)

# === 第三步(再次纠偏!) ===
# 模型看到错误,再次推理:
#   "备份库也没权限。可能是配置问题。我看看缓存里有没有旧数据。"
response = llm.chat(messages, tools=[工具A, 工具B, 工具C])
# 模型决定:调用 工具C_缓存
result_C = execute(工具C_缓存)  # → "找到用户数据:{...}"
messages.append(工具C 的成功结果)

# === 第四步 ===
# 模型看到数据了,整理输出
response = llm.chat(messages, tools=[...])
# 最终回复:"为您找到数据如下..."

如果没有循环,故事在第一步就结束了——Agent 拿到一个空结果,只能对你说"没找到"。

有了循环,Agent 展现了三种关键能力:

  1. 错误识别:它知道"空"和"权限不足"是两种不同性质的失败

  2. 策略调整:根据错误类型,自主选择不同的备选方案

  3. 目标坚持:不会因为一两次失败就放弃,直到真正无计可施

循环结构赋予模型的,本质上是"试错权"

你的例子揭示了一个更深层的设计哲学:

ReAct 循环的本质,是将控制流从代码中剥离,交给模型本身。

在传统编程中,if-else 分支是我们开发者提前写死的:

python

result_A = 工具A()
if not result_A:
    result_B = 工具B()  # 这个fallback是程序员硬编码的

在 ReAct 循环中,if-else 逻辑是模型实时推理出来的。这意味着:

  • 面对从未预见的错误,模型也可能想出合理的下一步

  • 工具集可以动态变化,模型能在运行时决定用哪个

  • 错误恢复策略不是写死的,而是模型基于对错误信息的理解即时生成的

你这个问题问到了 ReAct 循环的"灵魂"——不是为了让 Agent 多走几步,而是为了赋予它在行动中观察、在失败后调整、在不确定中寻找确定性的能力。循环只是这种能力的物理载体。

从最简实现到接近 OpenClaw 风格的工程化版本,逐步递进。

一、极简核心:ReAct 的本质

Agent 不是什么魔法,本质就是一个 while 循环:

python

# 最简 ReAct 实现(伪代码级别的核心逻辑)
messages = [{"role": "user", "content": user_input}]

while True:
    response = llm.chat(messages, tools=available_tools)
    
    if not response.has_tool_calls():
        # 没有工具调用 → 任务完成,返回最终回答
        return response.content
    
    # 有工具调用 → 执行工具 → 结果追加到消息历史 → 继续循环
    for tool_call in response.tool_calls:
        result = execute_tool(tool_call.name, tool_call.arguments)
        messages.append({
            "role": "tool",
            "tool_call_id": tool_call.id,
            "content": result
        })

二、TypeScript 实现:接近 OpenClaw 源码风格

OpenClaw 的核心 Agent Loop 用 TypeScript 实现,来自 runEmbeddedPiAgent 函数:

typescript

// 接近 OpenClaw 源码的 Agent 主循环
async function runAgentLoop(params: AgentLoopParams): Promise<string> {
  const messages: MessageParam[] = [
    { role: "user", content: params.userInput }
  ];
  
  while (true) {
    // 1. THINK:调用 LLM 进行推理
    const response = await client.messages.create({
      model: "claude-sonnet-4-20250514",
      max_tokens: 8096,
      tools: toolDefinitions,  // 所有可用工具的 JSON Schema
      messages,
    });
    
    // 2. REFLECT:判断模型是否要调工具
    if (response.stop_reason === "tool_use") {
      // 3. ACT + OBSERVE:执行工具并收集结果
      const toolResults = await Promise.all(
        response.content
          .filter((block) => block.type === "tool_use")
          .map(async (block) => ({
            type: "tool_result" as const,
            tool_use_id: block.id,
            content: await executeTool(block.name, block.input),
          }))
      );
      
      // 将助手消息和工具结果追加到历史
      messages.push({ role: "assistant", content: response.content });
      messages.push({ role: "user", content: toolResults });
      
      // 继续循环——让 LLM 根据工具结果再想一轮
      continue;
    }
    
    // 4. 无工具调用 → 任务完成
    return response.content.find((b) => b.type === "text")?.text ?? "";
  }
}

关键的终止条件来自 stop_reason 字段:当模型返回 "end_turn" 而不是 "tool_use" 时,说明它认为任务已经完成,循环退出。

三、Python 实现:可运行的多工具示例

下面是一个可直接运行的 Python 版本,模拟了两个工具:

python

import json
from openai import OpenAI

client = OpenAI()

# ========== 工具定义 ==========
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的当前天气信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名称"}
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "read_file",
            "description": "读取本地文件的文本内容",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "文件绝对路径"}
                },
                "required": ["path"]
            }
        }
    }
]


# ========== 工具实现 ==========
def execute_tool(name: str, args: dict) -> str:
    """实际执行工具调用(生产环境中这里可能执行 shell 命令、数据库查询等)"""
    if name == "get_weather":
        return f"{args['city']}今天晴,气温 18-26°C,适合外出"
    elif name == "read_file":
        try:
            with open(args['path'], 'r', encoding='utf-8') as f:
                return f.read()
        except FileNotFoundError:
            return f"错误:文件 {args['path']} 不存在"
    return "未知工具"


# ========== ReAct 循环核心 ==========
def run_agent_loop(user_message: str, max_iterations: int = 15):
    """
    Agent 主循环——ReAct 范式的直接实现。
    
    循环逻辑:
    1. Think:LLM 推理下一步该做什么
    2. Reflect:判断是否需要调用工具
    3. Act + Observe:执行工具并收集结果
    4. 结果追加到消息历史,回到步骤 1
    5. 当 LLM 不再请求工具调用时,返回最终回答
    """
    messages = [{"role": "user", "content": user_message}]
    
    for step in range(1, max_iterations + 1):
        print(f"\n{'='*50}")
        print(f" Step {step}: Think(思考)")
        print(f"{'='*50}")
        
        # ① Think:调用 LLM
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools,
            tool_choice="auto"
        )
        
        assistant_msg = response.choices[0].message
        messages.append(assistant_msg)
        
        # ② Reflect:是否有工具调用?
        if not assistant_msg.tool_calls:
            print(f"\n{'='*50}")
            print(f" 任务完成!")
            print(f"{'='*50}")
            return assistant_msg.content
        
        print(f"LLM 决定调用 {len(assistant_msg.tool_calls)} 个工具")
        
        # ③ Act + Observe:执行每个工具调用
        for tool_call in assistant_msg.tool_calls:
            func_name = tool_call.function.name
            func_args = json.loads(tool_call.function.arguments)
            
            print(f"\n  Act: 调用 {func_name}({func_args})")
            result = execute_tool(func_name, func_args)
            print(f"  Observe: {result[:100]}...")
            
            # 将工具结果追加到消息历史
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result
            })
        # 循环继续:让 LLM 根据新的工具结果再次推理
    
    return "达到最大迭代次数,任务未完成"


# ========== 运行示例 ==========
if __name__ == "__main__":
    result = run_agent_loop(
        "北京今天天气怎么样?同时帮我读取 /tmp/notes.txt 文件"
    )
    print(f"\n最终回答:{result}")

这段代码的运行过程清晰展示了 ReAct 的每个轮次:LLM 先决定调用工具 → 执行工具得到结果 → 结果追加到上下文 → LLM 再次推理决定下一步 → 最终输出完整答案。

四、OpenClaw 源码中的真实结构

OpenClaw 的实际源码比上面复杂得多,但核心循环体始终保持简洁。额外能力不是塞进循环内部,而是叠加在循环外部:

text

runEmbeddedPiAgent()          ← 外层:重试、容错、故障转移
  └── while (true) {          ← 主重试循环
        ├── 检查重试次数限制
        ├── 调用 runEmbeddedAttempt()  ← 单次推理尝试
        ├── 处理 context overflow → 自动压缩
        ├── 处理 auth failure → profile 轮换
        ├── 处理 timeout → 重试或报错
        └── 成功则返回结果
      }

而 runEmbeddedAttempt() 内部的 ReAct 循环本质就是前面那 20 行代码的模式——一次 LLM 调用,有工具调工具,没工具就返回。

扩展能力(子 Agent、上下文压缩、Skills 加载、多模型轮换)全部通过三种方式接入:扩展工具集和 handler、调整系统提示结构、把状态外化到文件或数据库,循环体本身基本不动。

五、关键设计要点总结

要素 设计决策 目的
while True 循环 让 LLM 自主决定何时结束 不预设步骤数,模型自己判断任务完成度
stop_reason 判断 依赖 API 的停止原因字段 区分"我要调工具"vs"我回答完了"
工具结果追加 结果作为 tool_result 角色追加到 messages 让 LLM 在下一轮看到执行反馈
最大迭代限制 硬性上限(默认 15-40 次) 防止无限循环烧 token
工具并行调用 Promise.all 同时执行 不相关的工具调用可以并行,提高效率

如果你需要更具体的实现细节——比如工具定义怎么写 LLM 才不容易用错、上下文压缩的触发逻辑、或者子 Agent 的 spawn 机制——可以继续展开。

Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐