AI Agent 的灵魂 —— ReAct 工作流全解析

一、 引言：为什么我们需要 ReAct？

• 痛点引入：传统的大模型（LLM）就像在做“闭卷考试”，遇到不懂的只能靠幻觉瞎编；或者像传统的 RAG，只能被动接受人类喂给它的资料。
• 概念提出：ReAct (Reasoning + Acting) 范式打破了这个僵局。它让大模型长出了“手脚”，不仅能思考（Reason），还能主动决定去调用工具（Act），真正从一个“聊天机器人”进化成了“智能体（Agent）”。
• 核心比喻：ReAct 就像是给大模型安排了一场“开卷考试”。大模型是考生，Python 代码是监考老师。遇到不会的题，考生（大模型）会在草稿纸上写“我要查某某书”，老师（Python）看到后把书找来翻到那一页递给考生，考生看完后再写出最终答案。

二、 ReAct 的“三步走”魔法：T-A-O 循环

ReAct 的底层逻辑非常优雅，它是一个不断循环的状态机，由三个核心基石组成：

1. Thought（思考）
   • 作用：大模型的“内心独白”。它分析当前的用户目标，评估自己已有的知识，并决定下一步该干什么。
   • 例子：Thought: 用户问紫微星在命宫代表什么，这属于专业命理知识，我不确定，需要调用搜索引擎工具。
2. Action（行动指令）
   • 作用：大模型向外部系统（Python 代码）下达的指令，通常包含“工具名称”和“JSON格式的参数”。
   • 例子：Action: search_ziwei \n Action Input: {"query": "紫微星在命宫"}
3. Observation（观察反馈）
   • 作用：外部系统（如你的 RAG 检索器、计算器、天气 API）执行完任务后，将结果以文本形式返回给大模型。
   • 例子：Observation: 检索到以下资料：紫微星入命宫，主人气质高雅，有领导欲...

三、 完整工作流程图解（以一次查询为例）

• [第 0 步] 设定规则 (System Prompt)：我们通过极其严格的 Prompt 告诉模型：“你拥有工具 A 和 B。你必须按照 Thought -> Action -> Observation 的格式输出。”
• [第 1 步] 用户提问：用户：“廉贞化忌会怎样？”
• [第 2 步] 模型初次生成 (Thought + Action)：
  模型输出思考过程，并下达 JSON 格式的查询指令。此时，神奇的事情发生了——模型停止了生成。
• [第 3 步] 代码拦截与执行 (Observation)：
  Python 代码监听到了 Action，拦截了模型的输出。Python 在后台运行 RAG 数据库查询，拿到资料。
• [第 4 步] 结果回传 (Second Thought)：
  Python 把拿到的资料拼接成 Observation: [资料内容] 重新发给大模型。
• [第 5 步] 输出结论 (Final Answer)：
  大模型结合拿到的 Observation 进行二次思考：Thought: 我已经拿到充足资料，可以回答了。 然后输出最终的 Final Answer。

四、 藏在代码深处的工程秘密（硬核加分项）

1. 怎么让大模型“停下来”？（Stop Words 的妙用）
   • 很多人以为是 Python 在实时打断大模型，其实不然。核心技巧是在调用 LLM 的 API 时传入 stop=["Observation:"] 参数。这样大模型一旦企图自己编造观察结果，底层推理引擎就会自动切断生成，乖乖把控制权交还给 Python。
2. 如何防止系统崩溃？（JSON 幻觉与自愈机制）
   • 大模型输出 Action Input 时，经常会少写一个括号或多写一个逗号，导致 json.loads() 报错。
   • 工业级的做法不是直接报错退出，而是捕获异常（try...except），把错误日志包装成 Observation: JSON格式错误... 扔回给大模型，让大模型自己意识到语法错误并重新生成。
3. 防止死循环（熔断机制）
   • 如果工具一直查不到数据，大模型可能会陷入“思考->查询->查不到->再思考->再查询”的死循环。必须在外层加一个 for step in range(MAX_ITER) 的循环器，达到最大次数强制结束，保护系统资源。

以下是关键代码

import re
import json

def run_react_agent_with_healing(question, llm_client, rag_system, max_iterations=5):
    """
    带有异常捕获与自愈机制的工业级 ReAct Agent 循环
    """
    print(f"👨‍💼 用户提问: {question}\n")
    
    # 1. 初始 System Prompt 与用户问题
    messages = [
        {"role": "system", "content": REACT_SYSTEM_PROMPT}, # 假设已定义工具列表
        {"role": "user", "content": question}
    ]
    
    # 2. 状态机主循环 (防止大模型无限发散的熔断机制)
    for step in range(max_iterations):
        print(f"--- [ Agent 思考轮次: {step + 1}/{max_iterations} ] ---")
        
        # ================= 第一阶段：思考与动作下发 =================
        # 💥 核心技巧 1：传入 stop 参数，一旦模型企图伪造观察结果，立刻硬中断！
        response = llm_client.chat.completions.create(
            model="qwen2.5-7b",
            messages=messages,
            temperature=0.1, # 降低温度，保证 JSON 格式输出的稳定性
            stop=["Observation:", "观察结果:"] 
        ).choices[0].message.content
        
        print(f"🤖 模型输出:\n{response}")
        messages.append({"role": "assistant", "content": response})
        
        # 解析模型意图 (利用正则抠出 Action 和 JSON)
        action_match = re.search(r"Action:\s*(.*?)\n", response)
        action = action_match.group(1).strip() if action_match else "None"
        
        # 如果大模型决定回答，直接退出循环
        if action == "None" or "Final Answer" in response:
            print("\n✅ 任务圆满结束！")
            return response
            
        # ================= 第二阶段：拦截、执行与自愈 =================
        # 提取 Action Input 中的 JSON 字符串
        input_match = re.search(r"Action Input:\s*(\{.*?\})", response, re.DOTALL)
        action_input_str = input_match.group(1).strip() if input_match else "{}"
        
        if "search_ziwei" in action:
            try:
                # 尝试将字符串解析为 JSON 对象
                args = json.loads(action_input_str)
                query = args.get("query", question)
                print(f"🛠️ [工具触发] 正在检索知识库: {query}")
                
                # 调用 RAG 获取真实数据
                _, docs = rag_system.query(query)
                observation = "\n".join([d.page_content for d in docs])
                print(f"👁️ [工具返回] 成功获取 {len(docs)} 条相关资料。\n")
                
                # 将真实的观察结果喂回给大模型
                messages.append({
                    "role": "user", 
                    "content": f"Observation: {observation}\n请根据以上资料继续思考，如果资料足够请给出 Final Answer。"
                })
                
            except json.JSONDecodeError as e:
                # 💥 核心技巧 2：JSON 幻觉自愈机制 (Self-Healing)
                print(f"⚠️ [系统警告] 模型输出了损坏的 JSON！正在触发自愈机制...")
                
                # 将详细的 Python 报错信息包装成提示词，强迫模型自我反思并重写
                error_feedback = (
                    f"Observation: Action Input 解析失败！"
                    f"Python 报错日志：{str(e)}。"
                    f"请检查你的 JSON 语法（是否遗漏了逗号或双引号？），修正后重新输出 Thought 和 Action。"
                )
                messages.append({"role": "user", "content": error_feedback})
                continue # 直接进入下一轮循环，让大模型重新生成
                
            except Exception as e:
                print(f"❌ [系统异常] 未知错误: {e}")
                messages.append({"role": "user", "content": "Observation: 工具调用发生内部错误，请直接根据你的常识回答。"})
        else:
            # 💥 核心技巧 3：防范“幻觉工具”
            print(f"⚠️ [系统警告] 模型调用了不存在的工具 [{action}]")
            messages.append({
                "role": "user", 
                "content": f"Observation: 工具 '{action}' 不存在！你只能使用 'search_ziwei'，或者输出 Action: None。"
            })
            continue

    # 如果走完了 for 循环还没得到答案，触发熔断
    return "抱歉，由于问题过于复杂，Agent 思考超时，已强行熔断。"

五、 结语：从手写 ReAct 到工业化编排

纯手写 ReAct 循环能让我们深刻理解 Agent 的底层原理。但在真实的复杂业务中（比如引入人工审批、多 Agent 协作），手写的 while 循环极其难以维护。这就引出了目前工业界的最优解——将线性循环升级为图结构的状态机（如 LangGraph）。