📚 课程目录第1章构建第一个ReAct模式Agent

1. 构建第一个ReAct模式Agent

任务目标

构建一个基于 ReAct（Reasoning + Acting）模式的 Agent，使其能够根据用户问题，通过“思考-行动-观察”循环调用外部工具，最终给出答案。本案例以“查询城市天气”为场景，Agent 需要自主决定调用天气查询工具，并基于工具返回结果生成最终回答。

核心实现

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

# ---------- 工具定义 ----------
def get_weather(city: str) -> str:
    """模拟天气查询工具（真实场景可替换为 API 调用）"""
    mock_data = {
        "北京": "晴，25°C，微风",
        "上海": "多云，28°C，东南风3级",
        "广州": "雷阵雨，30°C，南风2级",
    }
    return mock_data.get(city, f"暂无{city}的天气数据")

# ---------- ReAct Agent 核心 ----------
class ReActAgent:
    def __init__(self, tools: Dict[str, Callable]):
        self.tools = tools
        self.max_steps = 5  # 防止无限循环

    def _parse_action(self, text: str) -> Dict[str, str] | None:
        """从模型输出中解析 Action 与 Action Input"""
        action_match = re.search(r"Action:\s*(\w+)", text)
        input_match = re.search(r"Action Input:\s*(\{.*?\}|[^\n]+)", text, re.DOTALL)
        if not action_match:
            return None
        action = action_match.group(1)
        raw_input = input_match.group(1).strip() if input_match else ""
        # 支持 JSON 或纯文本输入
        try:
            action_input = json.loads(raw_input) if raw_input.startswith("{") else {"city": raw_input}
        except json.JSONDecodeError:
            action_input = {"city": raw_input}
        return {"action": action, "action_input": action_input}

    def run(self, question: str) -> str:
        """执行 ReAct 循环：Thought → Action → Observation → ... → Final Answer"""
        # 模拟 LLM 的推理与决策（真实场景接入 LLM API）
        observation = ""
        for step in range(self.max_steps):
            # 构造提示（此处用规则模拟 LLM 输出，真实场景替换为 LLM 调用）
            if step == 0:
                thought = f"用户想知道：{question}。我需要查询天气工具。"
                action_text = f"Action: get_weather\nAction Input: {{\"city\": \"{question.replace('天气', '').strip()}\"}}"
            else:
                # 基于观察结果生成最终答案
                return f"根据查询结果：{observation}"

            print(f"Step {step+1}: {thought}")
            print(action_text)

            parsed = self._parse_action(action_text)
            if not parsed or parsed["action"] not in self.tools:
                return "无法理解或执行该操作，请重新提问。"

            # 执行工具
            tool_func = self.tools[parsed["action"]]
            observation = tool_func(**parsed["action_input"])
            print(f"Observation: {observation}\n")

        return "已达到最大步骤数，未能完成回答。"

# ---------- 运行入口 ----------
if __name__ == "__main__":
    agent = ReActAgent(tools={"get_weather": get_weather})
    user_question = "北京天气怎么样？"
    answer = agent.run(user_question)
    print(f"Final Answer: {answer}")

运行与验证

运行方式：将代码保存为 react_agent.py，执行 python react_agent.py。
预期输出（基于代码逻辑可唯一推出）：

Step 1: 用户想知道：北京天气怎么样？。我需要查询天气工具。
Action: get_weather
Action Input: {"city": "北京"}
Observation: 晴，25°C，微风

Final Answer: 根据查询结果：晴，25°C，微风

通过条件：程序输出包含 Action: get_weather 与 Observation 的循环过程，且最终答案与工具返回的天气数据一致。修改 user_question 为“上海天气”或“广州天气”，输出应相应变化。

关键说明

本案例用规则模拟 LLM 的 Thought/Action 生成，以聚焦 ReAct 循环机制本身。接入真实 LLM 时，将 run() 中构造 thought 与 action_text 的部分替换为 LLM API 调用即可，解析与工具执行逻辑无需改动。
工具函数 get_weather 使用模拟数据，实际工程中替换为真实天气 API 或数据库查询。
ReAct 循环的核心是 Thought → Action → Observation → 重复 → Final Answer，本实现通过 max_steps 限制循环次数，避免模型陷入无限推理。