从零构建一个基于大模型的 ReAct 智能体：原理、实现与调优

Thought -> Action -> Observation -> (Repeat) -> Final Answer

pip install openai python-dotenv

import json
import math
from datetime import datetime

def get_current_time():
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

def search_web(query: str):
    # 模拟搜索，实际可接入真实 API
    mock_db = {
        "石家庄天气": "2026-07-23 石家庄：多云，25°C ~ 32°C，东南风 3 级。",
        "Python 3.11 新特性": "包括异常组、性能优化等。"
    }
    return mock_db.get(query, f"未找到关于 '{query}' 的结果。")

def calculate(expression: str):
    try:
        # 安全起见，仅允许简单算术
        allowed = set("0123456789+-*/(). ")
        if not all(c in allowed for c in expression):
            return "错误：表达式包含非法字符。"
        result = eval(expression, {"__builtins__": {}}, {})
        return f"计算结果：{result}"
    except Exception as e:
        return f"计算错误：{e}"

TOOLS = {
    "get_current_time": {
        "name": "get_current_time",
        "description": "获取当前的精确时间（年-月-日 时:分:秒）。无需参数。",
        "func": get_current_time,
        "parameters": {}
    },
    "search_web": {
        "name": "search_web",
        "description": "模拟网络搜索，输入查询字符串，返回相关信息。",
        "func": search_web,
        "parameters": {"query": "str"}
    },
    "calculate": {
        "name": "calculate",
        "description": "执行数学计算，输入算式，如 '3 * (5 + 2)'。",
        "func": calculate,
        "parameters": {"expression": "str"}
    }
}

SYSTEM_PROMPT = """你是一个能够使用工具的智能助手。请严格按照以下格式回复：

Thought: 你当前的推理过程（中文）。
Action: 你要调用的工具名称，必须从以下列表中选择：{tool_names}
Action Input: 调用该工具所需的输入参数（JSON 格式）。

如果你已经有足够信息回答用户问题，则输出：
Final Answer: 最终回答内容（中文）。

注意：每次只调用一个工具，等待观察结果后再继续。
"""

def build_system_prompt():
    tool_names = ", ".join(TOOLS.keys())
    return SYSTEM_PROMPT.format(tool_names=tool_names)

import re

def parse_response(text: str):
    """解析 LLM 输出，返回 (thought, action, action_input, final_answer)"""
    thought_match = re.search(r"Thought:\s*(.*?)(?=\nAction:|\nFinal Answer:|$)", text, re.DOTALL)
    action_match = re.search(r"Action:\s*(\w+)", text)
    action_input_match = re.search(r"Action Input:\s*(\{.*\})", text, re.DOTALL)
    final_match = re.search(r"Final Answer:\s*(.*)", text, re.DOTALL)

    thought = thought_match.group(1).strip() if thought_match else None
    action = action_match.group(1).strip() if action_match else None
    action_input = action_input_match.group(1).strip() if action_input_match else None
    final_answer = final_match.group(1).strip() if final_match else None

    return thought, action, action_input, final_answer

import openai

class ReActAgent:
    def __init__(self, model="gpt-4o-mini", max_iterations=5):
        self.model = model
        self.max_iterations = max_iterations
        self.messages = [
            {"role": "system", "content": build_system_prompt()}
        ]

    def run(self, user_query: str):
        self.messages.append({"role": "user", "content": user_query})
        iteration = 0

        while iteration < self.max_iterations:
            iteration += 1
            print(f"\n--- 迭代 {iteration} ---")

            # 调用 LLM
            response = openai.chat.completions.create(
                model=self.model,
                messages=self.messages,
                temperature=0.0,  # 保持确定性
            )
            reply = response.choices[0].message.content
            self.messages.append({"role": "assistant", "content": reply})
            print(f"LLM 回复：\n{reply}")

            thought, action, action_input, final_answer = parse_response(reply)

            if final_answer:
                print(f"\n✅ 最终回答：{final_answer}")
                return final_answer

            if not action:
                print("⚠️ 未解析到 Action，强制终止。")
                return "抱歉，我无法继续推理。"

            # 执行工具
            tool = TOOLS.get(action)
            if not tool:
                observation = f"错误：未知工具 '{action}'。"
            else:
                try:
                    # 解析参数
                    if action_input:
                        params = json.loads(action_input)
                    else:
                        params = {}
                    func = tool["func"]
                    observation = func(**params)
                except Exception as e:
                    observation = f"工具执行异常：{e}"

            print(f"🔧 执行 {action} -> 观察：{observation}")
            # 将观察结果作为 user 消息追加
            self.messages.append({
                "role": "user",
                "content": f"Observation: {observation}\n请继续思考并给出下一步 Action 或 Final Answer。"
            })

        return "达到最大迭代次数，未得出最终答案。"

# 设置 OpenAI API Key（或兼容端点）
import os
openai.api_key = os.getenv("OPENAI_API_KEY")
openai.base_url = os.getenv("OPENAI_BASE_URL")  # 可选

agent = ReActAgent()
question = "现在石家庄天气如何？另外，计算 123 * 456 等于多少？"
answer = agent.run(question)
print(answer)

--- 迭代 1 ---
LLM 回复：
Thought: 用户想知道石家庄的天气，并计算一个乘法。我需要先搜索天气信息。
Action: search_web
Action Input: {"query": "石家庄天气"}
...
🔧 执行 search_web -> 观察：2026-07-23 石家庄：多云，25°C ~ 32°C，东南风 3 级。
...
--- 迭代 2 ---
LLM 回复：
Thought: 已获得天气信息。现在需要计算 123*456。
Action: calculate
Action Input: {"expression": "123 * 456"}
...
🔧 执行 calculate -> 观察：计算结果：56088
...
--- 迭代 3 ---
LLM 回复：
Final Answer: 当前石家庄天气为多云，气温 25°C ~ 32°C，东南风 3 级。123 乘以 456 等于 56088。