【补坑】【新手向】ReAct实战演示

洛茗同学

403人浏览 · 2026-04-23 16:12:43

洛茗同学 · 2026-04-23 16:12:43 发布

整个项目从配置层开始写。先建一个config文件，把两个API密钥和地址放进去。一个是OpenWeatherMap的天气接口，另一个是DeepSeek的大模型接口。这里没什么复杂的，就是把密钥集中管理，方便后面换环境的时候不用到处改代码。这里是config.py

2.2 开发工具层

天气查询的逻辑单独抽成一个get_weather函数，接收城市名参数，调OpenWeatherMap的接口，把返回的JSON解析成一段可读的文字。

这里踩过一个坑：这个API不认中文城市名，传"青岛"会报404，传"Qingdao"才正常。所以可以在函数里加一个简单的映射表，把常见中文城市名自动转成英文再发请求，也可以直接发送英文请求。


# 这里是tools.py
import requests
import json
from config import WEATHER_API_KEY, WEATHER_API_URL

def get_weather(city: str) -> str:
    """
    天气查询工具，返回格式化的天气信息字符串
    """
    params = {
        "q": city,
        "appid": WEATHER_API_KEY,
        "units": "metric",
        "lang": "zh_cn"
    }
    try:
        response = requests.get(WEATHER_API_URL, params=params, timeout=10)
        response.raise_for_status()
        data = response.json()
        
        city_name = data.get("name", city)
        temp = data["main"].get("temp")
        feels_like = data["main"].get("feels_like")
        humidity = data["main"].get("humidity")
        weather_desc = data["weather"][0].get("description", "未知")
        
        return (
            f"{city_name}当前天气：{weather_desc}，"
            f"气温{temp}℃，体感温度{feels_like}℃，湿度{humidity}%"
        )
    except Exception as e:
        return f"查询天气失败：{str(e)}"


# 工具注册表
TOOLS_REGISTRY = {
    "get_weather": {
        "description": "查询指定城市的当前天气信息",
        "parameters": {
            "city": {
                "type": "string",
                "description": "城市名称，如北京、上海、London等"
            }
        },
        "function": get_weather
    }
}

工具写好后注册到一个TOOLS_REGISTRY字典里，这样后面AI想调用的时候有地方可查。

2.3 客户端封装

接下来要封装DeepSeek的客户端。这一步的关键是要让模型支持Function Calling，也就是让AI自己决定什么时候该调用外部工具。

我把TOOLS_REGISTRY里的工具信息用AI转换成DeepSeek要求的格式，每个工具写上名称、功能描述和参数说明。

    def _build_tools_schema(self):
        """
        将本地工具注册表转换为DeepSeek可用的tools格式
        """
        tools = []
        for name, info in TOOLS_REGISTRY.items():
            tools.append({
                "type": "function",
                "function": {
                    "name": name,
                    "description": info["description"],
                    "parameters": {
                        "type": "object",
                        "properties": info["parameters"],
                        "required": list(info["parameters"].keys())
                    }
                }
            })
        return tools

客户端里主要实现两个方法，一个是发对话请求，另一个是执行AI指定的本地工具。执行的时候从注册表里找到对应的函数，把AI传过来的参数解包进去运行，再把结果返回给AI。

完整代码如下：

# 这里是deepseek_client.py

import requests
import json
from config import DEEPSEEK_API_KEY, DEEPSEEK_API_URL
from tools import TOOLS_REGISTRY


class DeepSeekClient:
    """
    封装DeepSeek API调用，支持Function Calling
    """
    
    def __init__(self):
        self.api_key = DEEPSEEK_API_KEY
        self.api_url = DEEPSEEK_API_URL
        self.tools = self._build_tools_schema()
    
    def _build_tools_schema(self):
        """
        将本地工具注册表转换为DeepSeek可用的tools格式
        """
        tools = []
        for name, info in TOOLS_REGISTRY.items():
            tools.append({
                "type": "function",
                "function": {
                    "name": name,
                    "description": info["description"],
                    "parameters": {
                        "type": "object",
                        "properties": info["parameters"],
                        "required": list(info["parameters"].keys())
                    }
                }
            })
        return tools
    
    def chat(self, messages: list, use_tools: bool = True) -> dict:
        """
        发送对话请求，返回AI响应
        """
        payload = {
            "model": "deepseek-chat",
            "messages": messages,
            "temperature": 0.7
        }
        
        if use_tools:
            payload["tools"] = self.tools
            payload["tool_choice"] = "auto"
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        try:
            response = requests.post(
                self.api_url,
                headers=headers,
                json=payload,
                timeout=30
            )
            response.raise_for_status()
            return response.json()
        except Exception as e:
            return {"error": str(e)}
    
    def execute_tool(self, tool_call: dict) -> str:
        """
        执行AI请求调用的本地工具
        """
        func_name = tool_call["function"]["name"]
        arguments = json.loads(tool_call["function"]["arguments"])
        
        if func_name not in TOOLS_REGISTRY:
            return f"工具 {func_name} 不存在"
        
        tool_func = TOOLS_REGISTRY[func_name]["function"]
        result = tool_func(**arguments)
        return result

2.4 Agent层

Agent层是整个项目的控制核心。这里实现了一个循环交互的逻辑，大概是这样：

用户说完话后，先把这句话塞进对话历史，然后发给DeepSeek。
模型返回的结果分两种情况处理。
1. 第一种是AI直接给出了回答，不需要查天气，那直接把回答返回给用户就行。
2. 第二种是AI判断需要调用工具，比如用户问了某个城市的天气，这时候返回结果里会带上tool_calls字段。Agent读到这个字段后，就去调用对应的本地天气查询函数，拿到真实的天气数据，再把数据以tool角色塞回对话历史，重新发给AI。
AI看到工具返回的结果后，再组织语言生成最终的回答。这个循环最多跑五轮，防止出现意外情况无限循环。

完整代码展示：

#Agent.py

from deepseek_client import DeepSeekClient


class ReActAgent:
    """
    ReAct风格Agent：用户提问 -> AI思考是否调用工具 -> 执行工具 -> AI总结回答
    """
    
    def __init__(self):
        self.client = DeepSeekClient()
        self.messages = []
        self.max_iterations = 5
    
    def run(self, user_input: str) -> str:
        """
        处理用户输入，完成一轮对话
        """
        # 添加用户消息
        self.messages.append({"role": "user", "content": user_input})
        
        for i in range(self.max_iterations):
            print(f"\n--- 第{i+1}轮交互 ---")
            
            # 调用DeepSeek
            response = self.client.chat(self.messages, use_tools=True)
            
            if "error" in response:
                return f"API调用失败：{response['error']}"
            
            choice = response["choices"][0]
            message = choice["message"]
            
            # 情况1：AI直接回答，不需要工具
            if message.get("content") and not message.get("tool_calls"):
                print(f"[AI直接回答] {message['content']}")
                self.messages.append({
                    "role": "assistant",
                    "content": message["content"]
                })
                return message["content"]
            
            # 情况2：AI要求调用工具
            if message.get("tool_calls"):
                print(f"[AI决定调用工具] {len(message['tool_calls'])}个")
                
                # 先把AI的思考过程加入上下文
                self.messages.append({
                    "role": "assistant",
                    "content": message.get("content", ""),
                    "tool_calls": message["tool_calls"]
                })
                
                # 执行每个工具调用
                for tool_call in message["tool_calls"]:
                    func_name = tool_call["function"]["name"]
                    print(f"[执行工具] {func_name}")
                    
                    # 执行本地工具
                    result = self.client.execute_tool(tool_call)
                    print(f"[工具返回] {result}")
                    
                    # 将工具结果加入上下文
                    self.messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call["id"],
                        "content": result
                    })
                
                # 继续循环，让AI基于工具结果生成最终回答
                continue
        
        return "交互次数已达上限"
    
    def clear_history(self):
        """
        清空对话历史
        """
        self.messages = []
        print("[系统] 对话历史已清空")

2.5 程序入口

入口文件做一个简单的命令行交互界面，接收用户输入，调用Agent处理，打印结果。顺手做了个支持quit退出和clear清空历史两个指令。

from agent import ReActAgent


def main():
    print("=" * 50)
    print("ReAct 天气查询 Agent")
    print("输入 'quit' 退出，输入 'clear' 清空历史")
    print("=" * 50)
    
    agent = ReActAgent()
    
    while True:
        user_input = input("\n你：").strip()
        
        if not user_input:
            continue
        
        if user_input.lower() == "quit":
            print("再见")
            break
        
        if user_input.lower() == "clear":
            agent.clear_history()
            continue
        
        print("\n[开始处理...]")
        response = agent.run(user_input)
        print(f"\nAgent：{response}")


if __name__ == "__main__":
    main()

3. 成果演示

首先配置好config里面的两个api（过程不多赘述，如果有不会的可以发在评论区我逐一解答）。

然后直接启动程序

python main.py

看到如图所示：
运行结果1
输入“查询Nanjing天气”，也就是南京。程序很快给出答案：
运行结果2
结束。

整个流程跑通之后，你会发现ReAct框架的优势在实际开发中体现得很直接。

以前做这种功能，通常要写一堆正则表达式或者关键词匹配规则来判断用户是不是在问天气，然后硬编码调用接口的逻辑。现在这些判断完全交给大模型自己来做，它从对话上下文里理解用户的意图，自己决定要不要查工具。工具返回的原始数据也不会直接丢给用户，而是让AI再加工一遍，用自然的口吻说出来。如果以后想加新功能，比如查汇率或者算数学题，只需要在tools.py里写一个新函数，往注册表里加一条记录就行，Agent层和客户端层完全不用动。这种设计让工具扩展变得很简单，不需要每加一个功能就改一遍主流程的代码。这就是ReAct最大的优势。

4. 文章附件

我将代码开源到了Github，不想复制粘贴的读者可以自行下载：Github地址

AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

MLflow设计实战：打造可复现的MLOps工作流

基于对仓库的深入分析，该项目是一个面向机器学习工作流的设计与实现框架。对于而言，这是一个极具价值的工具库。对于用户而言，采用或参考用户无需从空白文件夹开始，可以直接基于的模板初始化项目，快速搭建一个结构规范、具备实验跟踪能力的基础框架。初始化后，项目目录将包含标准化的模块，如data/models/training/utils/等，并可能预置了实验跟踪的配置。在项目框架内进行模型开发时，可以便捷地

AtomGit开源社区

爱客问智能口碑问题生成的技术实现与AI分发机制实践

随着用户转向AI对话平台进行产品查询，企业SEO策略面临变革。爱客问平台通过分析AI引用规律，发现结构化问答、高事实密度和场景化内容更易被引用。为此，平台开发了三层架构：1）将企业资料转化为结构化知识库；2）控制内容生成数量；3）按场景生成适配的QA内容。该方案使企业内容生产效率提升10倍，并实现AI引用效果可视化监测，推动口碑营销向数据驱动转型。未来将探索UGC校验等更多优化方向。

AtomGit开源社区

人工智能核心技术解析

人工智能（Artificial Intelligence, AI）作为当今科技发展的核心驱动力，正以前所未有的深度和广度重塑各行各业。其本质是研究如何使计算机系统能够模拟、延伸和扩展人类智能，完成诸如学习、推理、感知、决策等复杂任务。本文将系统性地剖析其核心技术原理、广泛的应用领域以及未来的发展趋势。人工智能的底层技术栈是一个逐层递进的体系，其核心是使机器从数据中学习规律并做出决策。AI技术已渗透