用纯Python构建ReAct智能体：思考、行动与观察

规划执行型智能体有一个巧妙的技巧：模型会事先列出带编号的步骤清单，然后由普通的 Python 代码依次执行这些步骤。只要现实情况与计划相符，这种方法就有效。但当环境状况发生变化——比如目标位置移动、搜索返回 404 错误，或者模型对第二步的判断出错时——该智能体仍会继续执行第三到第十步，因为其设计中并未包含回溯机制。

ReAct 通过用循环取代完整的计划来解决这个问题。下文中展示的是用纯 Python 编写的循环，未使用任何框架；随后会说明当真实用户介入时该循环可能出现的两种故障情况。

从固定行程到反馈循环

想象一下穿越一片陌生森林的两种方法。第一位旅行者写下“向北1公里，然后向东500米”，并严格照做，即便在200米处出现悬崖也继续前进。而另一位旅行者则每隔几步就停下来，大声分析：向北的路被悬崖挡住了，那就试着向西绕行，没错，那边有安全的路径。

第一位旅行者属于“计划与执行型”智能体。它使用的计算资源较少且条理清晰，但很脆弱：如果第二步失败，第三步到第十步就都毫无意义了。第二位则是“推理与行动型”智能体（先推理再行动）。它一次只执行一步，并在下一步之前重新审视周围环境。这虽然需要更多的计算资源，但在情况出现变故时能表现得更好。

第二位旅行者所记录的实时评论其实是一种内心独白，正是这种机制让智能体显得具有适应性而非按固定脚本行事。不过，要想真正发挥作用，这种独白必须能够触发实际的动作。

一段文字中的工具调用

语言模型只能输出文本。为了让它具备“动手能力”，你需要一个简单的字典，将名称映射到 Python 函数，比如 {"get_weather": get_weather_api}。当模型的输出中出现字符串 get_weather 时，你的代码会检测到它并调用对应的函数。在 ReAct 智能体中，这些工具本身保持不变，变化的是模型决定如何使用它们。

整个设计中有点奇怪的地方在于：为了让模型看起来像是在独立思考，就必须对其施加严格的限制。

内心独白背后的契约

与其要求模型直接给出答案，不如将其约束在严格的输出格式中，直到它按照该格式生成内容后才允许其使用任何工具。这个循环包含三个步骤：

1. 思考。模型会审视问题、迄今为止收集的所有信息以及之前的失败案例，然后写下下一步该做什么以及原因。
2. 行动。基于上述思考，它从工具列表中选择一种工具，并写出对应的精确输入内容。
3. 观察。这是至关重要的环节。模型必须停止工作，由你的Python代码接手，读取行动指令，执行实际功能（如实时天气查询、数据库检索等），并将原始结果返回给模型。

每次循环都会为模型提供新的推理依据。假设、执行与验证的循环会不断重复，直到最终得出用户能够理解的结论。简而言之，模型负责推理与决策，而普通代码则充当裁判的角色，拦截每一次工具调用并反馈实际发生的情况。

理论上这听起来很完美，但实际问题在于如何让文本预测器在运行过程中暂停、等待脚本，然后再继续之前的推理流程。

手动编写循环

LangChain、CrewAI和AutoGen之类的框架将所有这些操作都隐藏在一次.run()调用背后。这样固然方便，但一旦智能体陷入死循环或编造出并不存在的工具，问题就出现了。调试这种黑箱非常困难，因此了解其内部机制十分重要。

首先是从系统提示开始，它规定了交互规则并明确了契约内容：

system_prompt = """
You are a dynamic AI Agent. You work in a continuous
loop of
- Thought,
- Action,
- Observation.
- PAUSE

Your available tools are:
- get_weather(location: str): Returns the current weather.
- calculate(expression: str): Evaluates a math expression.

Use the following format exactly:
Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [get_weather, calculate]
Action Input: the input to the action
PAUSE
(You will then receive an Observation from the environment)
Thought: I now know the answer
Final Answer: the final answer to the original input question
"""

PAUSE指令负责执行核心操作，同时提示词也明确了具体的字段名称（Question、Thought、Action、Action Input、Final Answer），这样正则表达式就能在自由格式的输出中准确找到这些字段。该指令会告知模型停止生成内容，从而将控制权交还给Python，使工具能够运行，并将其输出作为Observation写入上下文中。

模型并不能完全遵循格式要求，下文将会展示这一点。不过提示词仅相当于规则手册而已，还需要有某种机制来执行这些规则：捕获PAUSE信号、触发相应函数并让流程持续运转。这个机制就是while循环，它是智能体的核心所在。每次循环都会向模型发送提示、解析其回复、运行所需的工具，并将整个交互过程记录到历史中，直到模型给出最终答案或轮次预算用尽为止。

import re

def react_agent(self, user_question, max_turns=5):
    # Initialize our conversational memory with the system prompt
    messages = [
        {"role": "system", "content": self.system_prompt},
        {"role": "user", "content": f"Question: {user_question}"}
    ]

    turn_count = 0

    while turn_count < max_turns:
        print(f"\n--- Turn {turn_count + 1} ---")

        # 1. Get the LLM's response (The Thought + Action)
        response = my_react_agent.call_llm(messages)
        print(response)

        # Add the LLM's generation to the memory
        messages.append({"role": "assistant", "content": response})

        # 2. Check if the agent has reached a conclusion
        if "Final Answer:" in response:
            print("\n✅ Task Complete.")
            #print(json.dumps(messages, indent=2))
            return response.split("Final Answer:")[-1].strip()

        # 3. Parse the Action and Action Input using Regex
        action_match = re.search(r"Action: (.*)", response)
        input_match = re.search(r"Action Input: (.*)", response)

        if action_match and input_match:
            action = action_match.group(1).strip()
            action_input = input_match.group(1).strip()

            # 4. The Observation Phase (Python takes control)
            print(f"⚙️ System Executing: {action}({action_input})")
            try:
                action_input = json.loads(action_input)
                observation_result = my_react_agent.execute_tool(action, action_input)
            except (TypeError, ValueError, json.JSONDecodeError) as error_messgae:
                observation_result = error_messgae
            # Format the observation and feed it back to the agent
            observation_text = f"Observation: {observation_result}"
            messages.append({"role": "user", "content": observation_text})
            print(observation_text)

        else:
            # If the LLM breaks the API contract, gently correct it
            messages.append({"role": "user", "content": "Error: Invalid format. Please provide an Action and Action Input, or a Final Answer."})

        turn_count += 1
    return "❌ Agent timed out before reaching a final answer."

拥有该循环能带来什么

由于该循环是您自己的代码，因此您可以决定出现故障时该如何处理。使用类似天气分析工具的运行记录可以说明这一点的重要性。

例如，模型正确地判断出需要一个名为 get_planet_mass 的工具，但它用单引号来书写输入值 {'planet': 'Earth'}。这不是有效的 JSON 格式，因此 Python 端的解析器会抛出标准错误：它期望属性名使用双引号。

普通的脚本会在这一处直接终止运行。而这个循环不会抛出异常，它会将原始错误信息直接作为下一个 Observation 放入对话中。

在第二步中，模型读取该消息并在内部意识到输入格式有误，必须遵循JSON规则。它使用{"planet": "Earth"}重新发起请求，路由器执行相应工具，任务即完成。

注意，这种情况并没有专门的处理代码——没有针对格式错误的JSON的重试机制，只有将错误以观察结果形式返回的习惯，其余工作则由模型的理解能力完成。这便是实际中的内心独白表现形式。如果是静态规划器，会在第一步就崩溃，需要人工重新启动；而ReAct循环让模型能够在运行过程中自行诊断并修复语法错误。它并非仅仅遵循计划，而是会对实际发生的情况作出反应。

ReAct在生产环境中的局限性

第一次运行上述代码时，它几乎会给人一种神奇的感觉。在将其投入实际使用之前，你应当了解它会出问题的两个环节。

幻觉循环。有时模型会固执地认为一个已损坏的工具即将恢复正常功能。它会调用get_weather，遇到错误后再次调用该函数，还是得到同样的错误结果，如此反复下去，导致你的API额度不断被消耗。对此的防范措施就是在循环中设置max_turns=5的限制。这样的强制断路机制是不可或缺的，任何ReAct智能体都不应被允许无限制地运行。

上下文膨胀问题。每个轮次都会将自身的想法、行动和观察结果添加到对话中。如果某个工具返回包含10,000行的JSON数据，所有内容都会进入上下文窗口。到了下一轮时，模型就会被海量数据淹没，推理质量下降，甚至可能忘记最初的问题。切勿直接将原始API响应或完整的网页抓取内容作为观察结果输入。设定一个大致的上限，比如前几千个字符，并注明其余内容已被截断，总比让模型承受过多数据更好。应在中间加入一个小型中间件函数，先对结果进行总结或截断处理，这样对话内容才能保持简短清晰。

还有一个没有任何防护措施能解决的局限。当一个智能体需要同时处理太多大型工具以及大量累积的上下文时，它的推理过程会变得混乱，从而偏离任务目标。ReAct智能体是出色的战术家，但却不是合格的执行者。

结合两种风格

这两种模式解决的是不同的问题。规划与执行模式提供结构，而ReAct模式则带来弹性。许多实际系统最终会同时使用这两种方式：规划器先制定高层级的步骤，然后将每个复杂的子任务交给独立的ReAct智能体，由后者即时处理细节问题。像这样在不依赖任何框架的情况下从零开始协调多个智能体，是这一循环之后的自然下一步。