📚 课程目录第1章构建一个基于ReAct模式的最小Agent

1. 构建一个基于ReAct模式的最小Agent

任务目标

实现一个基于ReAct（Reasoning + Acting）模式的最小Agent。该Agent能够针对用户问题，循环执行“思考（Thought）→ 行动（Action）→ 观察（Observation）”流程，通过调用外部工具获取信息，最终给出答案。

核心实现

本案例使用Python实现，核心依赖仅为标准库。Agent通过解析大语言模型（LLM）的文本输出来决定下一步动作。为便于独立运行验证，LLM部分使用一个可替换的模拟接口，实际接入时只需替换该接口为真实模型API调用。

import re
import json
from typing import Callable, Dict, List, Optional

# ---------- 1. 工具定义 ----------
def get_weather(city: str) -> str:
    """模拟天气查询工具"""
    data = {"北京": "晴，25°C", "上海": "多云，28°C", "广州": "雷阵雨，30°C"}
    return data.get(city, f"暂无{city}的天气数据")

def get_time() -> str:
    """模拟时间查询工具"""
    return "2025-06-14 10:30:00 (UTC+8)"

TOOLS: Dict[str, Callable] = {
    "get_weather": get_weather,
    "get_time": get_time,
}

# ---------- 2. 模拟LLM（ReAct提示词模板） ----------
# 实际使用时替换为真实LLM API调用，输入为完整对话历史，输出为模型文本
def mock_llm(prompt: str) -> str:
    """模拟LLM的ReAct推理输出，仅用于演示循环逻辑"""
    if "天气" in prompt and "北京" in prompt:
        return "Thought: 用户想知道北京天气，我需要调用get_weather工具。\nAction: get_weather\nAction Input: {\"city\": \"北京\"}"
    if "时间" in prompt:
        return "Thought: 用户想知道当前时间，直接调用get_time工具。\nAction: get_time\nAction Input: {}"
    # 默认终止分支
    return "Thought: 我已经获得足够信息。\nFinal Answer: 根据查询结果，北京今天晴朗，气温25°C。"

# ---------- 3. ReAct Agent核心循环 ----------
class ReActAgent:
    def __init__(self, llm: Callable[[str], str], tools: Dict[str, Callable], max_steps: int = 5):
        self.llm = llm
        self.tools = tools
        self.max_steps = max_steps

    def run(self, user_query: str) -> str:
        """执行ReAct循环，返回最终答案"""
        history = f"用户问题: {user_query}\n"
        for step in range(self.max_steps):
            # 生成当前步骤的提示词
            prompt = history + "\n请继续ReAct推理，输出格式为Thought/Action/Action Input或Final Answer。"
            response = self.llm(prompt)

            # 解析Final Answer
            final_match = re.search(r"Final Answer:\s*(.+)", response, re.DOTALL)
            if final_match:
                return final_match.group(1).strip()

            # 解析Action
            action_match = re.search(r"Action:\s*(\w+)", response)
            input_match = re.search(r"Action Input:\s*(\{.*?\})", response, re.DOTALL)

            if not action_match:
                return f"错误：模型输出无法解析（第{step+1}步）: {response}"

            action_name = action_match.group(1)
            action_input = {}
            if input_match:
                try:
                    action_input = json.loads(input_match.group(1))
                except json.JSONDecodeError:
                    return f"错误：Action Input JSON解析失败: {input_match.group(1)}"

            # 执行工具
            if action_name not in self.tools:
                return f"错误：未知工具 '{action_name}'"
            observation = self.tools[action_name](**action_input)

            # 追加到历史
            history += f"Thought: {response}\nObservation: {observation}\n"
        return "错误：超过最大推理步数"

# ---------- 4. 最小调用入口 ----------
if __name__ == "__main__":
    agent = ReActAgent(llm=mock_llm, tools=TOOLS)
    result = agent.run("北京天气怎么样？")
    print(result)

运行与验证

运行方式：将上述代码保存为 react_agent.py，在Python 3.8+环境中执行：

python react_agent.py

预期输出：由于代码中mock_llm对“北京天气”问题直接返回了包含最终答案的响应，因此预期输出为：

根据查询结果，北京今天晴朗，气温25°C。

验证路径：

循环机制验证：将mock_llm中“北京天气”分支的返回值改为仅包含Action和Action Input（去掉Final Answer），再次运行。程序应进入第二轮循环，此时history中已包含Observation，模型才会返回Final Answer。观察控制台输出，确认循环正常执行。
工具调用验证：将用户问题改为“现在几点？”，运行后应输出模拟时间2025-06-14 10:30:00 (UTC+8)。
异常处理验证：将mock_llm返回值改为Action: unknown_tool，运行后应输出错误：未知工具 'unknown_tool'。

接入真实LLM：将ReActAgent的llm参数替换为调用、Claude等API的函数，输入为prompt字符串，返回模型生成的文本即可。提示词模板需按模型要求调整，但ReAct循环逻辑无需改动。