ReAct智能体架构学习

阅读前须知-点我下拉

简单易懂的ReAct框架，纯小白跟着做，也能快速搭建出属于自己的小智能体；
不同于LangChain工作链，以Agent AI底层逻辑进行展开。

所需工具及版本：
Python 3.10
OpenAI 2.32.0
任意大模型API

引言

本篇文章为作者阶段性的智能体学习笔记。将带大家从0了解Agent AI(ReAct架构)。

ReAct(Reasoning and Acting)是一种大模型(LLM)设计框架/思维模式，核心思想是：让模型一边思考，一边行动，而不是只思考或者只执行。

关于ReAct框架逻辑拆解：

1. 推理(Reasoning)：分析问题，规划步骤，决定下一步。
2. 行动(Acting)：调用工具，代码执行。
3. 反馈：获取结果，更新认知，修正策略。

与普通LLM大语言模型的区别：

对比	普通LLM	ReAct
思考方式	一次性	多轮迭代
是否使用工具	不适用	使用
准确性	容易出现幻觉	更可靠
灵活性	低	高

框架结构概览：

• config.py 环境配置(API KEY，Base URL)，代码与秘钥分离，提升安全性
• get_weather.py 元工具，负责被智能体调用，执行底层逻辑
• tools/__init__.py工具定义与映射层，连接大模型和本地函数元工具的桥梁
• agent.py 核心部分，调度模型，决策并处理工作流
• main.py 负责交互界面和程序主循环

tools/__init__.py部分代码：

tools = [
    {
        "type": "function",
        # 函数
        "function": {
            "name": "tool_name", # 函数的名字，必须和 Python 里的函数名一致
            "description": "我是干啥的？？？", # 告诉模型这个工具是干嘛的
            # 函数的参数
            "parameters": {
                "type": "object",
                # 函数参数的属性
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "我输入啥模型能理解？？" # 告诉模型这个参数填什么
                    }
                },
                "required": ["xxxx"] # 告诉模型，这个参数是必填的
            }
        }
    }
]

一上来有点懵，没关系，我们只需要知道它是一个工具包，并且用规范的JSON格式编写。该__init__.py用于初始化，定义了tools列表，发送给OpenAI的JSON描述，后续会再定义tool_map字典，用于调用对应的工具。

Agent两次调用逻辑：

• 第一次调用：分析用户意图，触发tool_calls（工具调用包）。
• 中间层执行：代码截获模型请求，本地运行get_weather函数，拿到结果。
• 第二次调用：将函数运行结果投喂给模型，生成最终答案。

优缺点：

• 优点：结构清晰。
• 缺点：缺乏记忆功能(Memory)。

构建ReAct架构过程：

引言部分只是给大家了解一下什么是ReAct架构，和它的基本工作逻辑，接下来，感兴趣的朋友可以跟着实际操作，看看能不能搭建出自己的ReAct智能体。

1. 配置环境：创建配置文件存放API密钥与请求API代理地址

config.py完整代码如下：

BASE_URL = "你的 API 代理地址"
API_KEY = "你的 API 密钥"

如何获取API KEY？

注：本作者仅提供获取渠道，关于如何获取请自行上网寻找教程，涉及付费购买token为自愿购买。

①大模型服务平台，可提供免费API，比如千问之类的AI
https://bailian.console.aliyun.com/
②CloseAI亚洲区，可提供Kimi，OpenAI等API，需购买token
https://www.closeai-asia.com/

2. 搭建Agent骨架：实现基本的初始化

导入OpenAI库，定义Agent类，初始化客户端(密钥，地址)；
定义run函数，用于执行智能体主体部分；
写一个messages对话列表，用于存放模型能够认识的JSON格式语言，这样做可以实现多轮对话，区分“User”和“AI”；
尝试进行一次基础对话，response变量用于存放大模型输出的内容，通过client.chat.completions.create()方法，response为json形式输出；
然后调用模型名称，使用api能用的模型名称即可；
此处的messages=messages含义是把前面构造好的对话数据，作为参数传给模型接口，怎么理解呢？左边的messages为规定格式的参数名，右边的messages为自己定义的变量。即把自己定义的messages对话列表，传入OpenAI模型；
由return response.choices[0].message.content返回response中content(内容)部分；
最后写main()函数，运行程序，实例化Agent后，尝试进行与AI进行聊天。
agent.py完整代码如下：

# 导包
from openai import OpenAI
from config import API_KEY, BASE_URL
class Agent:

    def __init__(self):
        # 初始化客户端，存放好API_KEY和BASE_URL
        self.client = OpenAI(
            api_key=API_KEY,
            base_url=BASE_URL
        )

    def run(self, user_input):
        # 第一步：把用户的输入包装成模型能理解的消息格式
        messages = [
            {"role": "user", "content": user_input}
        ]
        # 尝试进行一次基础对话
        response = self.client.chat.completions.create(
            model="gpt-4o", # 或者你使用的其他模型
            messages=messages
        )

        return response.choices[0].message.content

# 运行程序执行主体代码
if __name__ == '__main__':
    agent = Agent()  # 实例化Agent
    text = agent.run("你好！")  #此处"你好"为user_input实参
    print(text) # 最终输出text，检验智能体能否实现聊天功能

运行后，如果终端返回出类似“你好！😊有什么我可以帮忙的吗？”那么，恭喜你，你的AI模型已经跑通了，能够实现最基本的人机对话了，相当于是？你输出了”Hello World!”
接下来，我们要为我们的小智能体增添新的功能，联系引言的tools工具部分，为其添加工具包。

3. 撰写技能说明书：让智能体使用自己编写的工具

当我们要给智能体添加一个功能的时候，就需要给它新增一个函数；假设我们现在要给它增添描述城市天气温度的函数getweather。
那么先用自然语言描述getweather函数：“这是一个专门获取天气信息的工具。当你需要知道某个具体城市当天的天气状况或温度时，请调用此函数。你必须提供一个名为city的参数，他应该是你想查询的城市名称。”
当然，本文章不会为了演示架构去真正写一个可以实时查询当前城市天气温度的函数，仅供参考，我们会写一个静态的温度，主要是要理解如何调用工具。
我们把刚刚的自然语言描述转化成模型能听懂的json格式。
我们创建一个名为tools的软件包，在__init__.py文件中定义tools工具包，并创建get_weather.py文件用于编写获取天气的函数工具。
以下为get_weather.py代码示例：

def get_weather(city):
    return f"{city}今天是晴天，25度"

以下为`__init.py`代码示例：

from .get_weather import get_weather

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather", # 函数的名字，必须和 Python 里的函数名一致
            "description": "获取某个城市的天气", # 告诉模型这个工具是干嘛的
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，例如：北京" # 告诉模型这个参数填什么
                    }
                },
                "required": ["city"] # 告诉模型，这个参数是必填的
            }
        }
    }
]

4. 通过实现映射表，让程序根据给定的函数名和参数内容自动执行get_weather函数。

在tools/__init__.py工具包中建立工具映射表，具体代码如下：

# 核心：建立工具映射表
tool_map = {
    "get_weather": get_weather
}

此时回到agent.py升级一下run方法。
如果输入地名相关的提示词，比如“常州”，“徐州”，“北京”等，message.tool_calls会检查工具映射表，并调用get_weather函数，此时条件判断为True，可以检查是否有打印输出。

def run(self,user_input):
        messages= [
            {"role":"user","content":user_input}
        ]

        # 第一次调用：询问模型是否需要使用工具
        response = self.client.chat.completions.create(
            model="gpt-4.1-mini",
            messages=messages,
            tools=tools  # 把上面的工具说明书传进去
        )

        message = response.choices[0].message  # 拿到模型的初步回复

        # 检查模型是否想调用工具
        if message.tool_calls:
            print("...")

5. 解释有关message.tool_calls的含义，并实现多个或多次工具调用。

message.tool_calls是一个列表，有时候模型会非常聪明，可能一次性调用多个工具（比如同时查询常州和北京的天气），或一次性调用多次同一个工具，所以它返回的是一个列表。
这里取出列表里的第一个任务：tool_call = message.tool_calls[0]
如果我这么说很模糊的话，可以写一段main()函数代码去验证一下message.tool_calls里的内容

def run(self,user_input):
...
    return response.choices[0].message.tool_calls
if __name__ == '__main__':
    agent = Agent()
    text = agent.run("北京")
print(text)

注意这里返回的是message.tool_calls，为了方便理解，这里返回的是工具调用包这个列表，而不是message.content的内容。
输出后可以看到关于tool_calls的列表：

...
[ChatCompletionMessageFunctionToolCall
(id='call_X3YmgmmqAKN72IB2LXRR64fS', 
function=Function(
  arguments='{"city":"北京"}', 
  name='get_weather'), 
  type='function')]

解读一下tool_calls列表：

1. id：唯一标识符，模型用它匹配后面返回的结果。
2. type：永远是”funciton”。
3. funciton：函数关键部分，包含了name（模型想调用的函数名 如”get_weather”）和arguments（模型生成的参数字符串 如{“city”:”北京”}）。

举例：
假设你现在问“北京的天气怎么样？”

变量	实际内容	含义
message.tool_calls	[tool_call_object]	一个包含任务的列表
tool_call.function.name	“get_weather”	决定去用哪个工具
tool_call.function.arguments	‘{“city”: “北京”}’	决定传什么参数

6. 实现字符串转字典：使用python内置标准库json.loads()

现在我们知道了message.tool_calls列表中的内容，接下来就是提取其中部分内容让agent去获取。
我们需要使用python内置标准库json.loads()，这里的loads是load string的缩写。
它能把一个符合JSON格式的字符串变成一个可以直接操作的字典对象。

import json
    ...
    if message.tool_calls:
        # 提取第一个工具任务
        tool_call = message.tool_calls[0]
        # 函数名 get_weather
        func_name = tool_call.function.name  

        # 解析参数字符串为字典
        args = json.loads(tool_call.function.arguments)

        # 执行真正的 Python 函数
        result = tool_map[func_name](**args) # 相当于get_weather(city):

7. 构建反馈：给模型加入刚才的决策，再加入工具的结果

构建工具响应消息，包含（role，tool_call_id，content）

• role：tool 表示工具返回的结果。
• tool_call_id：必须与之前模型发出的tool_call.id一致，收到对应id的请求。
• content：函数执行的真实结果result
  将构建工具响应信息加入对话历史，此历史并非Memory，而且为了让模型更好理解上下文。
• 模型之前的请求：包含tool_calls的那条message。
• 工具返回结果：构建的tool反馈信息。
  再次调用模型，会把最终的message输出。
  详细代码如下，直接添加在run函数内：

  # 构建反馈
  messages.append(message)
  messages.append({
      "role": "tool",
      "tool_call_id": tool_call.id,
      "content": result
  })
  # 第二次调用：让模型汇总事实给出最终答案
  final_response = self.client.chat.completions.create(
      model="gpt-4o",
      messages=messages
  )
  return final_response.choices[0].message.content

8. 模型同时执行多个任务

此时我们在main()中对run函数传入地名的参数，就已经可以实现输出xx今天的天气是25摄氏度的信息了。
但是当我们在text = agent.run("北京 常州")，输入2个或以上的地名，模型会给出什么答案？

会报错！

你会得到以下报错内容：

openai.BadRequestError: Error code: 400 - 
{'error': 
{'message': "An assistant message with 'tool_calls' must be followed by tool messages 
responding to each 'tool_call_id'. The following tool_call_ids did not have response 
messages: call_lZtuQiZ2yqVhOTWgwytcYYOv", 
'type': 'invalid_request_error', 
'param': 'messages', 'code': None}
}

原因是：
模型已经要调用工具了，但是没有把工具执行的结果返回给模型，当第一次工具(get_weather)调用了ID(call_xxx)，成功得到了对应的toll响应信息，然后第二次工具调用不到对应的ID，没有对应的tool响应信息。
那么如何解决呢？从（6）中的语句tool_call = message.tool_calls[0]可以知道，当只有一个地名的时候，tool_call为tool_calls列表中的第一个索引，所以当地名在2个以上的时候，需要用循环语句实现tool的响应信息，并加入message当中实现二次调用。
具体代码如下：

# 1. 首先把模型包含 tool_calls 的消息存入历史
messages.append(message)
# 2. 循环处理每一个工具请求
for tool_call in message.tool_calls:
    # 提取函数名和参数
    func_name = tool_call.function.name
    args = json.loads(tool_call.function.arguments)
    
    # 执行对应的函数
    result = tool_map[func_name](**args)
    
    # 这一步很关键：把每一个结果都存入 messages
    messages.append({
        "role": "tool",
        "tool_call_id": tool_call.id,
        "content": str(result) # 确保内容是字符串格式
})

此处代码的逻辑稍有改变，需要先将tool_calls的消息存入历史，再进行tool_call中参数的遍历，而不是直接进行第一次调用然后给把整个tool_calls列表存入历史。

9. 执行二次调用：将数据汇总成一句话发送给用户

此时如果main程序内输入2个地名，得到的输出大概为：
北京今天是晴天，气温为25度。
常州今天也是晴天，气温同样为25度。
至此，一个完整的ReAct架构的智能体已经搭建完成。

final_response = self.client.chat.completions.create(
    model="gpt-4o",
    messages=messages
)
return final_response.choices[0].message.content

10. 搭建ReAct框架过程中的混淆点

1. 结构转换：
   把JSON格式的字符串转换成字典，使用json.loads()方法，格式要求非常严格
   输入String：'{“city”: “BeiJing”, “unit”: “celsius”}'
   处理json.loads()扫描字符串
   输出Dict：{‘city’: ‘Beijing’, ‘unit’: ‘celsius’}
   如果内部的双引号换成单引号，在尝试运行json.loads(‘{“city”: ‘Beijing’,”unit”:’celsius}’)，将不会顺利通过，会直接抛出JSONDecodeError的报错。
   对于JSON解析器而言：”city”为合格的键，’Beijing’不符合JSON规范。

2. 结构不规范的预防措施：
   可以通过Python语法结构抛出异常来预防在转换时JSON结构不规范的问题。
   如果JSON格式正确，程序则正常运行；反之，直接捕获异常的信息。
   具体代码如下：

   import json
   try:
       # 尝试解析模型返回的字符串
       args = json.loads(tool_call.function.arguments)
       result = tool_map[func_name](**args)
   except json.JSONDecodeError:
       # 如果 JSON 格式不对，直接捕获异常
   result = "错误：工具调用的参数格式不正确，请重新生成。"

   绝大部分模型会处理的比较好，不会出现JSON错误的情况，但是一旦出现，可以直接通过抛出异常来精准捕获问题所在，提升代码可读性和减少调试的时间成本。

3. 解包调用：
   对于这行代码: result = tool_map[func_name](**args)
   此处的**为Python的“关键字参数解包”，它能自动把字典中的Key匹配到函数的参数名上。

   # 如果 args 是 {'city': 'Beijing'}
   # 那么 get_weather(**args) 
   # 等同于 get_weather(city='Beijing')

完整代码示例：

基础ReAct架构的树形项目目录结构：

ReactAgentAI/
├── tools/
│   ├── __init__.py
│   └── get_weather.py
├── config.py
├── agent.py
└── main.py

config.py代码示例：

BASE_URL = "https://api.xxxxx/v1"
API_KEY = "sk-T3xxxxxxxxw3o"

agent.py代码示例：

from openai import OpenAI
from config import BASE_URL,API_KEY
from tools import tools,tool_map
import json

class Agent:

    def __init__(self):
        self.client = OpenAI(
            api_key=API_KEY,
            base_url=BASE_URL
        )

    def run(self,user_input):
        messages= [
            {"role":"user","content":user_input}
        ]

        response = self.client.chat.completions.create(
            model="gpt-4.1-mini",
            messages=messages,
            tools=tools
        )

        message = response.choices[0].message  

        if message.tool_calls:

            messages.append(message)

            for tool_call in message.tool_calls:

                func_name = tool_call.function.name
                args = json.loads(tool_call.function.arguments)

                result = tool_map[func_name](**args)

                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": str(result)
                })

            final_response = self.client.chat.completions.create(
                model="gpt-4o",
                messages=messages
            )
            return final_response.choices[0].message.content

        return message.content

tools/__init__.py代码示例：

from .get_weather import get_weather

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取某个城市的天气",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，例如：北京"
                    }
                },
                "required": ["city"]
            }
        }
    }
]
tool_map = {
    "get_weather": get_weather
}

tools/get_weather.py代码示例：

def get_weather(city):
    return f"{city}今天是晴天，25度"

main.py代码示例：

from agent import Agent

agent = Agent()

while True:
    user_input = input("你：")
    if user_input.strip().lower() in ["exit", "quit", "q"]:
        print("AI：程序结束")
        break
    if not user_input.strip():
        print("AI：你还没输入内容呢")
        continue
    result = agent.run(user_input)
    print("AI：", result)