【大模型】ReAct Agent知识梳理及代码解析

最近实习开始从零学习Agent以及大模型训练，不定期总结梳理一下最近学到的东西。

什么是Agent？

简单理解就是，Agent是一种能够感知环境并自主决策的实体。Agent可以被看作是基于大模型的“核心调度器“，借助任务规划、记忆及外部工具等能力，大模型能够识别出应该执行的任务以及执行方式，从而实现自主决策。

所以可以很明显看到，Agent不同于我们之前学习的常规数据流（规定先执行A操作，再执行B操作），Agent不是单纯地“执行任务的工具”，而具备人类一样的主观能动性。Agent模仿人类的方式，实现从“计划”、“组织”、“执行”、“学习”等过程，直至完成一个任务。

从上面的图可以看到，Agent作为调度器，可以思考planning、调用外部工具tools、行动action、记忆memory。其中tools是通过action执行的，中间的思考过程和工具执行结果都会存储在memory中，让大模型可以根据上下文去不断规划和反思。

从上面的介绍可以感受到Agent的重要性，相比于单个大模型，Agent可以自主地去处理事情（我的感受是Agent可以自主构建一套适用于当前场景的数据流，这是动态且在线完成的，相比于预设的流程更加灵活）。

关于工具调用的理解：

Agent最重要的能力之一是可以去触发工具调用，那为什么需要工具？原因很简单，因为大模型本身的知识来源于预训练的数据，而数据总会过时，比如用户询问“明天天气怎么样？”，大模型要么无法回答，要么胡乱回答。那如果模型有调用工具的能力（function calling），就可以生成标准格式的工具调用（比如json，包含tool name和param/argument），然后工具返回执行结果后（比如json），模型进行解析，然后将解析结果拼回上下文中。于是我们认为Agent就具有了回答这种训练数据分布外的问题的能力。

Function calling的能力怎么来？一种是训练的时候具备，另一种是prompt引导（但我觉得如果模型不大时，指令遵循的能力应该很有限）

什么是ReAct？

ReAct是一种非常典型Agent的推理模式（当然对我来说其实不是很典型，去业界才第一次接触到这个名词），也就是reasoning -> action （-> observation），这里observation其实也是action后工具的执行结果。假设我们把这个过程称作RAO，那这只是一轮的RAO，在ReAct推理中，可能会调用多轮RAO过程。我们称这个过程是multi-turn，或者mulit-turn tool call。

ReAct的核心能力：

• 自动决策与工具调用‌：Agent根据query分析是否需要调用工具。比如用户输入“请告诉我明天的天气怎么样？”时，Agent会自动触发调用工具
• Multi-turn的多轮迭代推理：也就是上面提到的RAO过程，（我目前了解到的）通常Agent会通过不同方式去决定是否结束推理
  • 达到预设的轮数
  • 大模型明确生成了类似answer的标签
  • 某一轮没有调用任何工具

那么到这里，其实还是不清楚Agent为什么可以看到工具，怎么去调用工具，工具如何执行，以及返回结果怎么用。那就接着看后面的Agent实现部分。

用LangChain框架实现ReAct Agent

LangChain是非常常用的一个大模型框架，能快速上手Agent的构建，所以这里先用这个框架来实现一个Agent。

导入Python库：

# 导入必要的库

import json
import time
import os
from dotenv import load_dotenv
from langchain_core.tools import tool
from langchain.chat_models import init_chat_model
from langchain_core.prompts import ChatPromptTemplate
from langchain.agents import AgentExecutor, create_react_agent
from langchain import hub

最重点的部分，规范地定义工具！在ReAct Agent系统中，工具的定义必须非常规范，主要包含以下几个部分：

• 工具接口定义：也就是定义好函数的输入输出参数
• 工具描述：描述都要在函数开头的这段注释文字中，其中要包括工具的功能、输入参数、输出参数
• 工具内容：也就是函数体，正常按功能流程去实现就好了

值得注意的是，工具接口和描述都是可以被大模型看到的，这也就是为什么要定义地非常清晰才行。那如何被大模型看到呢？答案就是把可用的工具列表放在prompt中。

# 定义工具函数并提供详细的工具描述

# 1、商品工具  
@tool
def query_goods(good_detail: str, goods_type: str=None, price_range: str = None) -> str:
    """根据商品名称以及价格范围商品数据库查找匹配的商品。
        
        Args:
            good_detail: 商品信息总参数，用来接受传参，真实的参数是后面的字段
            goods_type: 商品类别，如"水果"、"手机"、"洗衣机"等
            price_range: 可选，价格范围，如"1000-2000"
        
        return:
            查询到的商品列表
    """
    goodss = [
        {"goods_name": "苹果","goods_type": "水果","price": 1.00,"description": "苹果","id":"11111111"},
        {"goods_name": "香蕉","goods_type": "水果","price": 0.50,"description": "香蕉","id":"22222222"},
        {"goods_name": "小米15", "goods_type": "手机","price": 1500,"description": "小米手机","id":"33333333"},
        {"goods_name": "iPhone 13","goods_type": "手机","price": 4099,"description": "苹果手机","id":"44444444"},
        {"goods_name": "华为mate 70","goods_type": "手机","price": 7800,"description": "华为手机","id":"55555555"}
    ]
   
    goodsDic = json.loads(good_detail)
    goods_type = goodsDic["goods_type"]

    
    new_goods = [goods for goods in goodss if goods["goods_type"] == goods_type]
    if len(new_goods) == 0:
        return f"\n\n抱歉，未找到相关的产品：{goods_type}\n\n"
    
    if "price_range" in goodsDic:
        try:
            min_price, max_price = map(int, price_range.split("-"))
            new_goods = [p for p in new_goods if min_price <= p["price"] <= max_price]
        except:
            pass
        
 
    formatted_results = "\n".join([
        f"{p['goods_name']} - ¥{p['price']}" for p in new_goods
    ])
    
    return f"\n\n获取到以下的{goods_type}：\n{formatted_results}\n\n"


# 2、进行下单操作
@tool
def create_order(good_detail: str, goods_name: str = None, price: int= None, quantity:int=None) -> str:
    """创建订单，并返回订单编号。
    
        Args:
            good_detail: 商品信息,用来接受传参，正真参数是后面的字段
            goods_name: 商品名称
            price: 商品价格
            quantity: 购买数量
        return:
            订单编号
    """
 
    goodsDetail = json.loads(good_detail)
    goods_name = goodsDetail["goods_name"]
    price = goodsDetail["price"]
    if "quantity" not in goodsDetail: 
        quantity = 1
    else:
        quantity = goodsDetail["quantity"]
 
    # 当前时间的毫秒数 作为订单编号
    order_num = int(time.time() * 1000)
    
    # 模拟下单操作，返回订单编号
    return f"\n\n订单创建成功，商品{goods_name}订单编号：{order_num},购买数量为:{quantity},单价:{price},订单总价:{price*quantity}\n\n"


# 3、进行支付
@tool
def order_pay(order_detail: str, order_num: int=None,order_price: int=None, payment_method: str = None) -> str:
    """进行支付，并返回支付结果。
    
        Args:
            order_detail: 订单信息，用来接受传参，正真参数是后面的字段
            order_num: 订单编号
            order_price: 订单价格
            payment_method: 支付方式，如"支付宝"、"信微"、"银行卡"
        return:
            支付结果
    """
    payDetail = json.loads(order_detail)
    order_num = payDetail["order_num"]
    order_price = payDetail["order_price"]
    payment_method = payDetail["payment_method"]
 
    # 模拟支付操作，返回支付结果
    return f"\n\n订单{order_num}支付成功,支付金额为{order_price}，支付方式：{payment_method}\n\n"

这一步构建工具列表、Agent实体以及引导Agent去调工具的prompt。

# 构建react prompt和agent

# 工具列表
tools = [
    query_goods,
    create_order,
    order_pay
]

# react_prompt = hub.pull("hwchase17/react")
react_prompt = ChatPromptTemplate.from_template("""
    Answer the following questions as best you can. You have access to the following tools:
    {tools}
    Use the following format:
    Question: the input question you must answer
    Thought: you should always think about what to do
    Action: the action to take, should be one of [{tool_names}]
    Action Input: the input to the action
    Observation: the result of the action
    ... (this Thought/Action/Action Input/Observation can repeat N times)
    Thought: I now know the final answer
    Final Answer: the final answer to the original input question
    
    Important: action Input must be a valid JSON object with the correct parameter names
                                                
    Begin!
    Question: {input}
    Thought:{agent_scratchpad}
"""
)
 
# 初始化大模型
client = init_chat_model(
    model="gpt-4.1-mini",
    api_key='sk-6GnJrlYCoqIzu1xFWg8jEMmEwIq5C4txN2WlKQ2rBdcbGSin',
    base_url="https://api.chatanywhere.tech/v1",
    model_provider="openai",
)

print("="*10, "LLM Loaded Successfully", "="*10)
 
# 创建ReAct agent
agent = create_react_agent(
    llm=client,
    tools=tools,
    prompt=react_prompt
)
 
# 创建agent执行器
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,  # 显示详细执行过程
    max_iterations=5,  # 最大工具调用次数
    # early_stopping_method="generate",  # 提前停止策略
    handle_parsing_errors=True  # 处理解析错误
)

最后则是给用户query到Agent，然后让Agent去解决问题（会需要调用定义的一个或多个工具）。下面是只需要调用一个工具的例子：

query = "我要买一个手机，我的预算是5000元，帮我推荐一款"

result = agent_executor.invoke({"input": query})
print(result)

下面是需要调用两个工具的例子：

query = "我要买一个手机，我的预算是5000元，帮我推荐一款，并进行下单支付，支付方式为微信"

result = agent_executor.invoke({"input": query})
print(result)

手写ReAct Agent

看了上面LangChain的实现后，基本能搞明白流程了，重点就是定义好工具函数，构造工具列表，构造好带工具列表的prompt，剩下的塞给LangChain就搞定了。
但是仔细一想，还有一些地方无法完全理解，比如multi-turn是如何迭代的？工具怎么就能整个塞进prompt了？如何决定迭代到什么时候停下来？
下面来看看核心的实现：

现在来依次回答这几个问题：

1. 如何迭代？最简单的方式就是用for循环进行
2. 工具如何塞进prompt？可以看到，其实prompt中只需要塞对工具的接口定义和描述（甚至可以给few-shot example），大模型真正去调用工具是通过getattr(self.toolkit, tool_name)实现的
3. 何时停止？要么是预设的循环次数达到了，要么是该轮不调用工具了，认为推理结束，要么是生成了Final Answer关键词
4. 其实我自己一直还有个 疑惑【关于生成格式和指令遵循】： 为什么大模型就能自己生成json格式的action呢，必须这样才能从中解析出tool name和param。可以看到prompt中已经其实给了action format（json格式），我目前的理解是，要么用是已经具有function calling能力的模型，比如GPT这些模型本身就具备这种调用工具的能力（已经训过了），要么就是模型本身足够强大，可以很好地遵循prompt，反之拿一个base model，肯定不会有这种效果。另外也可以看到，为什么能直接从response中去提Final Answer后面的内容作为答案？万一模型的没生成Final Answer怎么办？其实在prompt中已经告诉了模型要把答案放在Final Answer后面，GPT这样的强大模型可以很好的遵循，而同样地，如果拿一个base model，也不会有这种效果。

总结一句就是，要实现Agent调用模型，必须生成规范的action（json格式的tool name+param），要满足这一点，要么大模型本身具有function calling能力，要么大模型本身很强大，可以去follow prompt里的要求和example。不管怎样，应该给few-shot example都是有帮助的。

还在初学积累阶段，如有错误，欢迎随时指正！谢谢大家

一些很有帮助的参考资料

• ReAct | LangChain中文网
• langchain系列之基于create_react_agent 创建一个Agent
• GPT和DeepSeek免费API