学习链接：https://www.datawhale.cn/learn/content/220/5019
github hello-agent项目第4章

一、什么是ReAct

ReAct（Reasoning + Acting），先观察环境，在结合用户问题思考推理，接着按照思考的结果采取行动，最后观察行动的结果，结果如果满足用户需求，则输出，如果不满足继续上面步骤循环。
特点：
1.将思考和行动融合在每个步骤中
2.通过观察-思考-行动的循环实现决策
3.适合需要实时响应的动态任务

下面是代码实现简单的ReAct：

第1步：定义一个LLM类

定义一个专属的LLM客户端类。这个类将封装所有与模型服务交互的细节，让我们的主逻辑可以更专注于智能体的构建。
注意！！！需要在同目录下创建一个.env文件来存重要不能外露的信息
然后在脚本中使用os.getenv(“参数名称”)来获取

IsaacAgentsLLM.py

# IsaacAgentsLLM.py
import os
from openai import OpenAI
from dotenv import load_dotenv
from typing import List, Dict

# 加载.env文件中的环境变量
load_dotenv()

class IsaacAgentsLLM:
    """
    创建一个LLM客户端。
    它用于调用任何兼容openAI接口的服务，并默认使用流式响应。
    """

    def __init__(self, model: str = None, apiKey: str =None, baseUrl: str = None, timeout: int = None):
        """
        初始化客户端。有限使用传入参数，如果未提，则从环境变量加载。
        """
        self.model = model or os.getenv("LLM_MODEL_ID")
        apiKey = apiKey or os.getenv("LLM_API_KEY")
        baseUrl = baseUrl or os.getenv("LLM_BASE_URL")
        timeout = timeout or int(os.getenv("LLM_TIMEOUT", 60))

        if not all([self.model, apiKey, baseUrl]):
            print(11)
            raise ValueError("模型的ID、API密钥和服务地址必须被提供或者在.env文件中定义")
        self.client = OpenAI(api_key=apiKey, base_url=baseUrl, timeout=timeout)

    def think(self, messages: List[Dict[str, str]], temperature: float = 0) -> str:
        """
        调用大语言模型进行思考，并返回其响应。
        """
        print(f"正在调用{self.model}模型...")
        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                temperature=temperature,
                stream=True,
            )
        
            # 处理流式响应
            print("大语言模型响应成功：")
            collected_content = []

            for chunk in response:
                content = chunk.choices[0].delta.content or ""
                print(content, end="", flush=True)
                collected_content.append(content)
            print()
            return "".join(collected_content)
        except Exception as e:
            print(f"调用LLM API时发生错误:{e}")
            return None

if __name__ == "__main__":
    try:
        # 注意调用think模式需要模型有这个能力
        llmClient = IsaacAgentsLLM(model='deepseek-r1')

        exampleMessages = [
            {"role": "system", "content": "You are a helpful assistant that writes Python code"},
            {"role": "user", "content": "写一个快速排序算法"}
        ]

        print("---调用LLM---")
        
        responseText = llmClient.think(exampleMessages)
        if responseText:
            print("\n\n---完整模型响应---")
            print(responseText)
        
    except ValueError as e:
        print(e) 

第2步：定义一个search函数

我们的第一个工具是 search 函数，它的作用是接收一个查询字符串，然后返回搜索结果。
文件名：search_tool.py

# search_tool.py
from ddgs import DDGS

def search(query: str) -> str:
    """
    一个网页搜索工具
    如果不存在,它使用DuckDuckGo来搜索并返回排名前三的结果摘要。
    """
    print(f"正在执行真实网页搜索：{query}")
    try:
        # 使用with上下文管理器确保资源被正确处理
        with DDGS() as ddgs:
            # max_results控制返回结果数量
            results = [r for r in ddgs.text(query, max_results=3)]
        
        if not results:
            return f"对不起，没有找到关于‘{query}’的信息"
        # 将结果格式化为LLM友好的字符串
        result_string = []
        for i, result in enumerate(results):
            result_string.append(f"[{i+1}]{result['title']}\n{result['body']}")

        return "\n\n".join(result_string)
    
    except Exception as e:
        return f"搜索时发生错误：{e}"

第3步：构建通用的工具执行器

当智能体需要使用多种工具时（例如，除了搜索，还可能需要计算、查询数据库等），我们需要一个统一的管理器来注册和调度这些工具。为此，我们创建一个 ToolExecutor 类。

from typing import Dict, Any
from serch_tool import search

class ToolExecutor:
    """
    一个工具执行器，负责管理和执行工具。
    """

    def __init__(self):
        self.tools: Dict[str, Dict[str, Any]] = {}

    def registerTool(self, name: str, description: str, func: callable):
        """
        向工具箱注册一个新工具。
        """
        if name in self.tools:
            print(f"警告：工具'{name}'已存在，将被覆盖。")
        
        self.tools[name] = {"description": description, "func": func}
        print(f"工具'{name}'已注册")
    
    def getTool(self, name: str) ->callable:
        """
        根据名称获取一个工具的执行函数。
        """
        return self.tools.get(name, {}).get("func")

    def getAvailavleTools(self) -> str:
        """
        获取所有可用工具的格式化描述字段。
        """

        return "\n".join([
            f"{name}: {info['description']}" for name, info in self.tools.items()
        ])

# --- 工具初始化与使用示例 ---
if __name__ == '__main__':
    # 1.初始化工具执行器
    ToolExecutor = ToolExecutor()

    # 2.注册我们的实际搜索工具
    search_description = '一个网页搜索引擎。当你需要回答关于时事、事实以及在你的知识库中找不到的信息时，应使用此工具。'
    ToolExecutor.registerTool('Search', search_description, search)

    # 3.打印可用工具
    print("\n---可用的工具---")
    print(ToolExecutor.getAvailavleTools())

    # 4.智能体的Action调用
    print("\n--- 执行 Action: Search['苹果最新的手机型号是什么'] ---")
    tool_name = "Search"
    tool_input = "苹果最新的手机型号是什么"
    tool_function = ToolExecutor.getTool(tool_name)

    if tool_function:
        observation = tool_function(tool_input)
        print("--- 观察 (Observation) ---")
        print(observation)
    else:
        print(f"错误：未找到名为 '{tool_name}' 的工具。")

第4步：ReAct 智能体的编码实现

现在，我们将所有独立的组件，LLM客户端和工具执行器组装起来，构建一个完整的 ReAct 智能体。我们将通过一个 ReActAgent 类来封装其核心逻辑。为了便于理解，我们将这个类的实现过程拆分为以下几个关键部分进行讲解。
文件名：ReAct.py

4.1系统提示词设计

提示词是整个 ReAct 机制的基石，它为大语言模型提供了行动的操作指令。我们需要精心设计一个模板，它将动态地插入可用工具、用户问题以及中间步骤的交互历史。

# ReAct 提示词模板
REACT_PROMPT_TEMPLATE = """
请注意，你是一个有能力调用外部工具的智能助手。

可用工具如下：
{tools}

请严格按照以下格式进行回应：

Thought: 你的思考过程，用于分析问题、拆解任务和规划下一步行动。
Action: 你决定采取的行动，必须是以下格式之一：
- `{tool_name}[{tool_input}]`：调用一个可用工具。
- `Finish[最终答案]`：当你认为已经获得最终答案时。

现在，请开始解决以下问题：
Question: {question}
History: {history}
"""

4.2核心循环的实现

from search_tool import search
from IsaacAgentsLLM import IsaacAgentsLLM
from toolexecutor import ToolExecutor

class ReActAgent:
    # ReAct 提示词模板
    REACT_PROMPT_TEMPLATE = """
    请注意，你是一个有能力调用外部工具的智能助手。

    可用工具如下：
    {tools}

    请严格按照以下格式进行回应：

    Thought: 你的思考过程，用于分析问题、拆解任务和规划下一步行动。
    Action: 你决定采取的行动，必须是以下格式之一：
    - `{tool_name}[{tool_input}]`：调用一个可用工具。
    - `Finish[最终答案]`：当你认为已经获得最终答案时。

    现在，请开始解决以下问题：
    Question: {question}
    History: {history}
    """
    def __init__(self, llm_client: IsaacAgentsLLM, tool_executor: ToolExecutor, max_steps: int = 5):
        self.llm_client = llm_client
        self.tool_executor = tool_executor
        self.max_steps = max_steps
        self.history = []

    def run(self, question: str):
        """
        运行ReAct智能体来回答一个问题。
        """

        self.history = []  # 每次运行重置历史记录
        current_step = 0

        while current_step < self.max_steps:
            current_step += 1
            print(f"---第{current_step}步---")

            # 1.格式化提示词
            tools_desc = self.tool_executor.getAvailavleTools()
            history_str = "\n".join(self.history)

            prompt = REACT_PROMPT_TEMPLATE.format(
                tools=tools_desc,
                question=question,
                history=history_str
            )

            # 2.调用LLM进行思考
            messages = [{"role": "user", "content": prompt}]
            response_text = self.llm_client.think(messages=messages)

            if not response_text:
                print("错误：LLM未能返回有效响应")
                break
    
    def _parse_out

…未完待续，纯手跟着敲，慢慢理解，有点慢。。。