2.3 ReAct 循环完整实现：思考、行动、观察的最小代码

2.3 ReAct 循环完整实现：思考、行动、观察的最小代码

本节摘要：ReAct（Reasoning + Acting）是智能体最核心的运行模式：模型交替产出"思考"与"行动"，程序执行行动并回填"观察"，循环直到任务完成。本节给出一份不依赖任何框架的完整实现，把步数预算、错误恢复、死循环识别、终止判断这些工程兜底一次讲全——这是一百五十行以内的代码，却是整册的实验底座。

前两节解决了一轮调用：模型怎么知道工具、怎么报菜名、程序怎么执行回填。但真实任务几乎从不需要一轮——查完订单还要查物流，查完物流才能答复用户。把多轮串起来的机制就是 ReAct 循环，名字来自它每轮的三段输出：Thought（现在什么情况、下一步为什么这么走）、Action（调用哪个工具、什么参数）、Observation（工具返回的现场结果）。

图：ReAct 循环状态机——正常路径与全部兜底出口

完整实现：一百五十行以内

import json

class ReActAgent:
    def __init__(self, tools: dict, tool_specs: list, max_steps: int = 15,
                 max_fails: int = 3):
        self.tools = tools              # {工具名: 可执行函数}
        self.tool_specs = tool_specs    # JSON Schema 定义清单
        self.max_steps = max_steps      # 步数预算
        self.max_fails = max_fails      # 连续失败上限

    def run(self, goal: str) -> dict:
        messages = [{"role": "system", "content": build_contract(self.tool_specs)},
                    {"role": "user", "content": goal}]
        fails, seen = 0, set()          # 连续失败计数 / 已见(工具,参数)指纹
        for step in range(1, self.max_steps + 1):
            reply = llm.chat(messages, tools=self.tool_specs)

            if not reply.tool_calls:                       # 无动作 = 直接回答
                return {"status": "done", "answer": reply.content,
                        "steps": step, "trace": messages}

            for call in reply.tool_calls:
                name = call.function.name
                args = json.loads(call.function.arguments)
                fingerprint = (name, json.dumps(args, sort_keys=True))

                if fingerprint in seen:                    # 疑似原地打转
                    messages.append(observe(call.id,
                        {"error": "重复调用：同样的工具与参数已经执行过。"
                                  "请换思路或调用 finish。"}))
                    continue
                seen.add(fingerprint)
                try:
                    result = self.tools[name](**validate(name, args))
                    obs = slim(result)                     # 清洗截断（2.2 节）
                    fails = 0                              # 成功一次即清零
                except ToolError as e:
                    fails += 1
                    obs = {"error": str(e)}
                    if fails >= self.max_fails:            # 连续失败兜底
                        return {"status": "failed", "reason": f"连续 {fails} 次失败",
                                "steps": step, "trace": messages}
                messages.append(observe(call.id, obs))     # 观察回填

        return {"status": "timeout",                       # 步数预算兜底
                "reason": f"超过 {self.max_steps} 步未完成",
                "steps": self.max_steps, "trace": messages}
​

这份实现里每个看似多余的判断都对应一类线上事故：max_steps 对应预算烧穿，fails 对应工具故障时的无限重试，seen 指纹对应最常见的死法——模型在两个工具间来回横跳。循环的健壮性不来自模型聪明，来自这些丑陋的 if。

案例：一个三步任务的完整轨迹

背景：目标"把用户王芳最近一笔订单的物流单号发到她的注册手机上"，挂三个工具：query_orders（查订单）、get_logistics（查物流）、send_sms（发短信）。

轨迹：第一轮 Thought 判断要先查人名对应的订单，Action 调 query_orders(name="王芳")，Observation 返回两笔订单；第二轮 Thought 选取最近那笔并取单号，Action 调 get_logistics(order_id=...)，Observation 给出运单号；第三轮 Thought 确认手机号在订单返回里，Action 调 send_sms(...)，Observation 返回发送成功；第四轮模型不再输出工具调用，而是直接给出文字答复并结束。

解读：注意第二步的 Thought——"选最近那笔"是个业务判断，模型做这类判断的成功率远高于做算术；也注意如果 query_orders 的返回里没有手机号字段，第三轮模型会面临"想发短信却没有号码"的窘境，多数模型会硬编一个——这正是 2.4 节观察完整性议题的伏笔。

变式：任务失败时（比如查无此人），好的轨迹是模型调用 finish 说明"未找到用户王芳的订单，请确认姓名"，而不是连续调三次 query_orders 换三种拼写。前者靠契约里"承认做不到"的授权（2.1 节），后者靠 fails 兜底——两层防线都要在。

什么时候别用裸 ReAct

裸循环对所有任务一视同仁地"走一步看一步"，这在两类场景是劣势：步骤本可预知的结构化任务（先查再发，顺序固定），每轮都让模型重新决策纯属浪费，第 4 章的 Plan-and-Execute 会先定计划再执行；需要回头反思的任务（失败了要总结教训再战），纯 ReAct 没有反思环节，第 5 章的 Reflexion 模式会补上。ReAct 是地基不是终点，它的价值在于让你彻底理解循环结构，之后用任何框架都是在给这份理解付接口费。

⚠️ 常见坑：把 trace（完整轨迹）当调试垃圾扔掉。上线后每一单客诉，你唯一能依赖的排查材料就是当时的思考与观察序列；没有轨迹的智能体故障，和没有黑匣子的空难一样只能靠猜。

本节要点回顾

• ReAct 三段：Thought 定方向、Action 报调用、Observation 回现场——循环的本质是让决策始终基于最新事实。
• 三类兜底：步数预算、连续失败上限、重复调用指纹，分别防预算烧穿、故障重试、原地打转。
• 无动作即结束：模型不再输出工具调用时视为直接回答，这是最自然的终止信号。
• 轨迹即黑匣子：生产环境的每次运行都要留全量轨迹，这是排查与评测的共同原料。

循环能跑通了，下一节解决一个隐蔽但致命的问题：喂回循环的观察，本身可能就是脏的、残缺的、过大的。