Google AI智能体实战指南:如何用ReAct框架打造你的第一个智能助手(附代码)
从理论到实践:用ReAct框架构建你的第一个AI智能体
你是否曾对着那些关于AI智能体的前沿论文和技术博客感到兴奋,却又在动手时不知从何开始?我们常常被各种新概念包围——智能体、工具调用、自主规划——听起来很酷,但代码该怎么写?环境怎么配?一个能真正“干活”的助手,到底是如何从零开始搭建起来的?
今天,我们就抛开那些宏大的叙事,直接切入实战。我将带你一步步,用Google AI白皮书中重点提及的ReAct框架,构建一个能处理真实世界任务的智能助手。我们不止步于“Hello World”,而是要打造一个能理解你的意图、调用外部工具、并经过多轮“思考”后给出精准答案的智能体。无论你是想为个人项目添加自动化能力,还是探索下一代应用的可能性,这篇文章都将提供一条清晰的路径。
我们将使用Python作为主要语言,并借助一些当前最活跃的开源库。整个过程会从最基础的环境搭建讲起,涵盖工具的定义、智能体循环的构建、以及如何让AI学会“三思而后行”。最重要的是,我会分享在构建过程中容易踩到的“坑”和调试技巧,这些都是文档里不会写的实战经验。
1. 环境搭建与核心库选型
在开始编写第一行智能体代码之前,一个稳定且高效的环境是成功的基石。与许多教程直接让你pip install一堆包不同,我更建议先理解每个核心库的职责,再根据你的具体需求进行选择。盲目安装所有依赖,往往会导致版本冲突和难以排查的环境问题。
我个人的开发环境通常基于 Python 3.10+,这是一个在稳定性和新特性支持上取得很好平衡的版本。虚拟环境是必须的,我强烈推荐使用 venv 或 conda 来隔离项目依赖。
注意:大型语言模型(LLM)的调用是智能体的核心。你可以选择云服务API(如OpenAI的GPT-4、Google的Gemini),也可以部署本地模型(如Llama 3、Qwen)。对于学习和原型开发,云API更方便;对于数据安全和成本控制有要求的项目,本地模型是更好的选择。本文示例将主要使用OpenAI API,但其架构完全兼容其他模型。
接下来,我们看看需要哪些核心库。一个典型的ReAct智能体项目依赖可以归纳为以下几层:
| 层级 | 功能 | 推荐库 | 说明 |
|---|---|---|---|
| LLM交互层 | 提供与语言模型通信的接口 | openai, langchain, litellm |
openai 是官方库,langchain封装更全面,litellm支持多模型统一接口。 |
| 智能体框架层 | 实现ReAct等智能体逻辑 | langchain, agents |
langchain的AgentExecutor是当前最成熟的实现之一。 |
| 工具层 | 定义和执行外部工具 | langchain.tools, 自定义工具类 |
包括搜索、计算、API调用等具体功能。 |
| 编排与记忆层 | 管理对话历史与状态 | langchain.memory |
用于让智能体拥有上下文感知能力。 |
对于初学者,我建议从 langchain 开始,因为它提供了一个相对完整的抽象,让你能快速看到效果。但为了深入理解原理,我们也会剖析其底层运作机制。安装命令很简单:
pip install langchain openai
如果你打算使用本地模型,可能还需要安装 transformers, accelerate 等库。但今天我们聚焦于架构,先使用API模型来保证流程的顺畅。
环境配置的最后一步,是设置你的API密钥。永远不要将密钥硬编码在代码中。我习惯使用环境变量来管理:
# 在终端中设置(临时)
export OPENAI_API_KEY='your-api-key-here'
# 或者在代码中通过python-dotenv加载
# 示例:使用dotenv加载配置
from dotenv import load_dotenv
import os
load_dotenv() # 从 .env 文件加载环境变量
api_key = os.getenv("OPENAI_API_KEY")
至此,你的开发环境已经就绪。我们拥有了与大脑(LLM)对话的能力,接下来要为这个大脑配备可以操纵世界的“手脚”——也就是工具。
2. 定义智能体的“手脚”:工具(Tools)的创建与集成
工具是智能体超越纯文本生成,与外部世界交互的桥梁。在白皮书的比喻中,工具是厨师手中的刀和锅。没有工具,再聪明的厨师也只能空想菜谱。在ReAct框架中,模型通过“思考”(Reason)来决定使用哪个工具,然后“行动”(Act)去执行它。
一个工具本质上是一个函数,它有着明确的名称、描述、输入参数和返回格式。清晰的描述至关重要,因为LLM就是根据这些描述来判断在什么情况下调用这个工具的。我们来创建一个最简单的工具——一个能进行单位换算的工具。
首先,我们不依赖高级框架,用最原始的方式理解工具的定义:
def unit_converter(amount: float, from_unit: str, to_unit: str) -> str:
"""
将数值从一个单位转换为另一个单位。
支持的长度单位:米(m)、千米(km)、英尺(ft)。
支持的重量单位:千克(kg)、克(g)、磅(lb)。
参数:
amount: 要转换的数值。
from_unit: 原始单位。
to_unit: 目标单位。
返回:
转换后的数值和单位字符串。
"""
conversion_rates = {
('m', 'km'): 0.001,
('km', 'm'): 1000,
('m', 'ft'): 3.28084,
('ft', 'm'): 0.3048,
('kg', 'g'): 1000,
('g', 'kg'): 0.001,
('kg', 'lb'): 2.20462,
('lb', 'kg'): 0.453592,
}
key = (from_unit.lower(), to_unit.lower())
if key in conversion_rates:
result = amount * conversion_rates[key]
return f"{result:.2f} {to_unit}"
else:
# 尝试通过中间单位(如米)进行间接转换
# 这里简化处理,直接返回错误
return f"抱歉,暂不支持从 {from_unit} 到 {to_unit} 的转换。"
这只是一个普通的Python函数。要让它成为智能体可用的“工具”,我们需要用框架能理解的方式包装它。在LangChain中,我们可以使用 Tool 类或者继承 BaseTool 类。
from langchain.tools import Tool
unit_conversion_tool = Tool(
name="UnitConverter",
func=unit_converter,
description="""当你需要进行单位换算时使用此工具。例如,将米转换为千米,或将千克转换为磅。
输入应该是一个包含三个参数的字符串,用逗号分隔:数值,原单位,目标单位。
例如:'5, km, m' 或 '10, kg, lb'。
"""
)
工具描述的艺术:你会发现,description 字段写得非常具体,甚至给出了输入格式的例子。这是因为LLM并不“理解”代码,它只是根据你的描述进行模式匹配。一个模糊的描述会导致模型无法正确调用工具。好的描述应包含:
- 工具的核心功能:用一句话说清它能干什么。
- 适用场景:在什么情况下应该被调用。
- 输入格式:明确要求输入的格式,是JSON字符串还是逗号分隔。
- 输出说明:让模型知道会得到什么。
一个智能体通常需要多个工具才能处理复杂任务。我们再添加一个获取当前天气的工具(模拟版)和一个执行简单计算的计算器工具。
import requests
from datetime import datetime
def get_weather(city: str) -> str:
"""模拟获取城市天气。在实际应用中,这里会调用如OpenWeatherMap的API。"""
# 模拟数据
weather_data = {
"beijing": {"temp": 22, "condition": "晴朗", "humidity": 40},
"shanghai": {"temp": 25, "condition": "多云", "humidity": 65},
"new york": {"temp": 18, "condition": "小雨", "humidity": 80},
}
city_lower = city.lower()
if city_lower in weather_data:
data = weather_data[city_lower]
return f"{city}的天气:温度{data['temp']}°C,{data['condition']},湿度{data['humidity']}%。"
else:
return f"未找到{city}的天气信息,目前支持北京、上海、纽约。"
def simple_calculator(expression: str) -> str:
"""执行安全的数学表达式计算。支持加减乘除和括号。"""
try:
# 警告:在生产环境中,直接使用eval是危险的,应使用ast.literal_eval或专用库如`numexpr`
# 此处为演示简化处理,确保表达式仅包含数字和运算符
allowed_chars = set("0123456789+-*/(). ")
if not all(c in allowed_chars for c in expression):
return "错误:表达式包含不安全字符。"
result = eval(expression)
return f"{expression} = {result}"
except Exception as e:
return f"计算错误:{e}"
# 创建工具列表
weather_tool = Tool(name="GetWeather", func=get_weather, description="获取指定城市的当前天气信息。输入是城市名称,例如:'北京'。")
calc_tool = Tool(name="Calculator", func=simple_calculator, description="计算一个数学表达式的结果。输入是一个字符串表达式,例如:'(3+5)*2'。")
tools = [unit_conversion_tool, weather_tool, calc_tool]
现在,我们的智能体拥有了三样“武器”:测量世界(单位换算)、感知环境(天气查询)、逻辑演算(数学计算)。接下来,我们需要一个“大脑”来指挥它们协同工作。
3. 构建智能体核心:ReAct循环的实现
ReAct框架的精髓在于其名称所揭示的循环:Reason(推理) 和 Act(行动)。模型不是直接给出最终答案,而是生成一个“思考链”,其中穿插着对工具的调用。这个过程类似于人类解决问题:先分析(我需要知道什么?),再行动(去查资料或计算),然后根据结果继续分析,直到得出结论。
LangChain提供了高级的 initialize_agent 函数来快速创建智能体,但为了深入理解,我们先拆解这个循环。一个简化的ReAct步骤序列如下:
- 接收用户输入:例如,“北京今天天气怎么样?如果温度是25摄氏度,相当于多少华氏度?”
- 模型推理:模型分析问题,决定第一步需要调用
GetWeather工具获取北京的温度。 - 执行行动:智能体执行工具调用,得到结果:“北京天气:温度25°C,晴朗,湿度40%。”
- 再次推理:模型将工具结果纳入上下文,分析出下一步需要将25摄氏度转换为华氏度。它知道需要调用
UnitConverter,但发现工具库中没有直接的摄氏转华氏工具。 - 可能的分支:模型可能会尝试用已知公式推理(需要计算能力),或者意识到无法完成而向用户求助。在我们的例子中,它应该调用
Calculator工具来计算转换公式(25 * 9/5) + 32。 - 最终回答:综合所有信息,给出最终答案。
下面,我们使用LangChain来构建这个智能体。首先,初始化LLM和工具:
from langchain.agents import initialize_agent, AgentType
from langchain.chat_models import ChatOpenAI
from langchain.memory import ConversationBufferMemory
# 1. 初始化LLM
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0, openai_api_key=api_key)
# temperature=0 使输出更确定,更适合执行任务
# 2. 初始化记忆(让智能体有上下文)
memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)
# 3. 创建智能体
agent = initialize_agent(
tools,
llm,
agent=AgentType.CHAT_CONVERSATIONAL_REACT_DESCRIPTION, # 专为对话设计的ReAct智能体
verbose=True, # 设置为True可以看到详细的思考过程,调试时非常有用
memory=memory,
handle_parsing_errors=True # 优雅处理模型输出格式错误
)
AgentType.CHAT_CONVERSATIONAL_REACT_DESCRIPTION 是一个关键选择。它基于ReAct框架,并针对多轮对话进行了优化,能够很好地利用历史记忆。verbose=True 是学习和调试阶段的神器,它会打印出模型的完整思考过程。
现在,让我们运行第一个完整的交互:
response = agent.run("北京今天天气怎么样?如果温度是25摄氏度,相当于多少华氏度?")
print(response)
当 verbose=True 时,你会在控制台看到类似下面的输出(已简化):
> Entering new AgentExecutor chain...
Thought: 用户问了两个问题。第一个是北京今天的天气,第二个是温度单位换算。我需要先回答第一个问题,这需要调用GetWeather工具。
Action:
{ "action": "GetWeather", "action_input": "北京" }
Observation: 北京的天气:温度25°C,晴朗,湿度40%。
Thought: 我已经得到了北京的天气,温度是25°C。现在需要回答第二个问题:将25摄氏度转换为华氏度。我的工具里没有直接的摄氏转华氏工具,但我有Calculator工具。我知道转换公式是 F = (C * 9/5) + 32。
Action:
{ "action": "Calculator", "action_input": "(25 * 9/5) + 32" }
Observation: (25 * 9/5) + 32 = 77.0
Thought: 现在我有了所有信息。我可以组织回答了。
Final Answer: 北京今天的天气是晴朗,温度25°C,湿度40%。25摄氏度相当于77华氏度。
这个过程完美展示了ReAct的“思考-行动”循环。模型不仅回答了问题,还展示了其决策路径。这种透明性对于调试和建立信任至关重要。你可以看到,当缺少直接工具时,模型能够利用已有的计算工具和领域知识(转换公式)来解决问题,这体现了其一定的组合推理能力。
4. 进阶实战:打造一个多功能个人助理
掌握了基础构建后,我们可以挑战更复杂的场景:打造一个能处理复合任务、拥有长期记忆并能从错误中学习的个人助理。我们将引入几个新概念:自定义工具类、结构化输出解析以及错误处理与重试机制。
4.1 创建更强大的自定义工具
之前的工具很简单。现在,我们创建一个能与外部Web API交互的工具,例如,一个获取新闻头条的工具。我们将继承 BaseTool 类,这能给我们更多的控制权。
from langchain.tools import BaseTool
from pydantic import Field
from typing import Type, Optional
import aiohttp
import asyncio
class NewsFetcherTool(BaseTool):
name = "FetchNews"
description = "获取指定类别的最新新闻头条。输入是新闻类别,例如:'technology', 'business', 'sports'。"
categories: list = Field(default_factory=lambda: ["technology", "business", "sports", "entertainment"])
def _run(self, category: str) -> str:
"""同步执行的方法。这里我们模拟API调用。"""
# 注意:这是一个模拟函数。真实情况下应调用如NewsAPI的接口。
# 为安全起见,我们避免进行真实的网络调用演示。
if category.lower() not in self.categories:
return f"错误:不支持的类型 '{category}'。请使用以下类型之一:{', '.join(self.categories)}"
mock_news = {
"technology": ["AI芯片取得新突破", "量子计算云平台上线测试"],
"business": ["全球市场迎来波动", "新能源产业投资增长"],
"sports": ["国际赛事即将开幕", "知名球队完成转会"],
"entertainment": ["新电影票房创纪录", "音乐节阵容公布"]
}
news_list = mock_news.get(category.lower(), ["暂无该类别新闻"])
return f"{category} 类别最新头条:\n" + "\n".join([f"- {news}" for news in news_list])
async def _arun(self, category: str) -> str:
"""异步执行的方法(可选)。"""
return self._run(category)
# 定义输入模式(Pydantic模型),帮助LangChain进行参数验证
args_schema: Optional[Type] = None # 可以在这里定义更详细的输入模型
# 将自定义工具加入列表
news_tool = NewsFetcherTool()
tools.append(news_tool)
使用 BaseTool 的好处是我们可以更精细地控制工具的行为,比如添加参数验证、异步支持等。现在我们的智能体可以获取新闻了。
4.2 处理复杂查询与结构化任务分解
用户的问题可能很复杂:“帮我查一下科技新闻,然后告诉我今天北京的天气,如果天气好,就计算一下从我家到公司的距离(假设10公里)需要走多少步(假设一步0.7米)。”
这需要智能体进行任务分解和规划。虽然高级的ReAct智能体有一定规划能力,但对于非常复杂的任务,我们可以在前端进行预处理,或者使用更高级的规划器(如Plan-and-Execute模式)。这里,我们依赖模型自身的推理能力,并通过清晰的提示工程来引导。
我们可以通过修改智能体的系统提示(System Prompt)来增强其规划能力。在LangChain中,我们可以自定义 AgentExecutor 的提示模板。
from langchain.agents import AgentExecutor
from langchain.agents.conversational_chat.prompt import PREFIX, SUFFIX
from langchain.prompts import MessagesPlaceholder
# 自定义前缀,指导智能体更好地规划和思考
CUSTOM_PREFIX = PREFIX + """
你是一个强大的个人助理。请遵循以下准则:
1. 当用户提出复杂、多步骤的问题时,先在脑中规划步骤。
2. 一次只执行一个清晰、简单的动作。
3. 充分利用之前的对话历史和工具返回的结果。
4. 如果某个步骤需要计算,优先使用Calculator工具。
5. 如果工具返回错误或信息不足,请基于已有信息进行合理推断或询问用户澄清。
"""
# 构建提示模板
from langchain.prompts import ChatPromptTemplate, HumanMessagePromptTemplate, SystemMessagePromptTemplate
from langchain.schema import SystemMessage
prompt = ChatPromptTemplate.from_messages([
SystemMessage(content=CUSTOM_PREFIX),
MessagesPlaceholder(variable_name="chat_history"),
HumanMessagePromptTemplate.from_template("{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad") # 这里是模型放置思考和行动记录的地方
])
# 重新初始化智能体
agent_executor = AgentExecutor.from_agent_and_tools(
agent=agent.agent, # 复用之前的agent逻辑
tools=tools,
memory=memory,
prompt=prompt,
verbose=True,
max_iterations=5, # 防止无限循环
handle_parsing_errors=True
)
现在,让我们用这个增强版的智能体来处理那个复杂查询:
complex_query = """
帮我查一下科技新闻,然后告诉我今天北京的天气。
如果天气好(晴朗或多云),就计算一下从我家到公司的距离(假设10公里)需要走多少步(假设一步0.7米)。
"""
response = agent_executor.run(complex_query)
print(response)
观察 verbose 输出,你会看到智能体是如何一步步分解任务的:先获取科技新闻,再查询北京天气,判断天气条件,最后触发一系列计算(公里转米,再除以步长)。这个过程可能涉及多次工具调用和中间推理。
4.3 错误处理与智能体鲁棒性
在实际应用中,工具调用可能失败(API超时、返回意外格式),模型也可能输出无法解析的指令。提高智能体的鲁棒性至关重要。
1. 工具调用错误处理:我们可以在工具函数内部做好异常捕获,返回清晰的错误信息供模型理解。
def robust_calculator(expression: str) -> str:
"""更健壮的计算器,包含错误处理。"""
import ast
import operator
# 定义安全的运算符
allowed_operators = {
ast.Add: operator.add,
ast.Sub: operator.sub,
ast.Mult: operator.mul,
ast.Div: operator.truediv,
ast.Pow: operator.pow,
ast.USub: operator.neg,
}
def eval_expr(node):
if isinstance(node, ast.Num):
return node.n
elif isinstance(node, ast.BinOp):
left_val = eval_expr(node.left)
right_val = eval_expr(node.right)
op_func = allowed_operators.get(type(node.op))
if op_func is None:
raise ValueError(f"不支持的运算符: {type(node.op)}")
return op_func(left_val, right_val)
elif isinstance(node, ast.UnaryOp):
operand_val = eval_expr(node.operand)
op_func = allowed_operators.get(type(node.op))
if op_func is None:
raise ValueError(f"不支持的运算符: {type(node.op)}")
return op_func(operand_val)
else:
raise ValueError(f"不支持的AST节点: {type(node)}")
try:
# 使用ast解析,比eval安全
tree = ast.parse(expression, mode='eval')
result = eval_expr(tree.body)
return f"{expression} = {result}"
except (SyntaxError, ValueError, ZeroDivisionError) as e:
return f"计算失败:{e}。请检查表达式格式。"
2. 解析错误处理:handle_parsing_errors=True 参数能帮助智能体在模型输出不符合工具调用格式时进行重试或降级处理。你还可以自定义一个错误处理函数。
3. 设置迭代上限:max_iterations 参数防止智能体陷入无限循环。当步骤过多时,它会自动停止并总结当前信息。
经过这些增强,你的智能体已经从一个简单的问答机器,进化成了一个具备初步规划能力、错误恢复能力的实用助手原型。你可以继续为其添加更多工具,如日历管理、邮件发送、数据查询等,构建出真正属于你的个性化AI伙伴。
构建AI智能体的过程,就像教一个天赋异禀但缺乏经验的新手。你需要清晰地定义任务(工具),耐心地引导思考过程(提示工程),并为其建立从错误中学习的机制。ReAct框架提供了一种让大语言模型将“思考”过程外化的优雅方法,使得智能体的决策变得可追溯、可调试。
我在最初尝试时,常常纠结于工具描述的准确性。一个模糊的描述会导致模型频繁误调用或不敢调用。后来我发现,用自然语言清晰地列举几个正例和反例,效果比抽象的定义好得多。例如,在计算器工具的描述里加上“适用于:(10+5)/2;不适用于:请总结这篇文章”,能显著提升模型的理解。
另一个常见的坑是对话历史的管理。默认的 ConversationBufferMemory 可能会让上下文变得过长,影响性能和模型关注点。对于长对话,可以考虑使用 ConversationSummaryMemory 或 ConversationBufferWindowMemory(只保留最近N轮对话)。这需要根据你的应用场景进行权衡。
最后,别忘了测试各种边缘情况。问一些模糊的、矛盾的、甚至错误的问题,看看你的智能体如何应对。它的反应往往能揭示出提示词或工具设计中的薄弱环节。智能体的开发是一个迭代过程,每一次与它的交互,都是对其能力的一次打磨。
更多推荐

所有评论(0)