在人工智能领域快速发展的背景下,大型语言模型(LLM)已成为处理从回答问题到生成代码等各种任务的强大工具。然而,处理需要同时进行推理和与外部工具交互的复杂多步骤任务一直是挑战。2022 年 10 月的论文《ReAct:Synergizing Reasoning and Acting in Language Models》(2023 年 3 月修订)提出了 ReAct(推理 + 行动)框架,旨在通过动态结合推理和行动,使语言模型更智能、通用和可解释。本文将深入探讨 ReAct 框架的核心理念、工作原理、优势与局限性,并结合实际代码示例说明如何在智能代理中实现 ReAct 风格的多意图任务处理。
什么是 ReAct 框架?
ReAct 全称为 Reasoning + Acting(推理 + 行动),是一个用于提示大型语言模型的框架,旨在通过以下方式提升模型性能:
推理(Reasoning):模型通过逻辑分析逐步分解任务,生成中间步骤或假设。例如,回答“北京今天适合户外活动吗?”时,模型可能推理:“需要先查询北京的天气,然后根据天气条件判断是否适合户外活动。”
行动(Acting):模型根据推理结果执行具体任务,例如调用外部工具(如天气 API)或采取其他动作(如搜索信息)。在上述例子中,行动可能是调用 get_weather 工具获取北京天气数据。
通过在推理和行动之间动态交替,ReAct 使模型能够:
动态调整:根据任务需求灵活切换推理和行动,而不是仅依赖单一的回答生成。
增强可解释性:记录推理步骤,使模型的决策过程更透明。
提高通用性:适用于多种任务场景,如问答、决策制定和交互式任务,尤其是在需要外部工具或环境的场景。
ReAct 的工作原理
ReAct 的核心是通过精心设计的提示(Prompting)引导模型在推理和行动之间循环。以下是其典型工作流程:
观察(Observation):
模型接收用户输入或任务描述,例如“查询北京和上海的天气并计算 2+3*4”。
输入可能包含单一意图或多个意图,模型需要解析任务需求。
推理(Reasoning):
模型生成文字描述,记录其思考过程。例如:
推理:用户请求包含三个意图: 1. 查询北京的天气 → 调用 get_weather("北京")。 2. 查询上海的天气 → 调用 get_weather("上海")。 3. 计算 2+3*4 → 调用 calculate("2+3*4")。
行动(Acting):
模型根据推理结果调用外部工具(在 OpenAI API 中通过 tool_calls 实现)。例如:
调用 get_weather("北京") → 返回“北京的天气是晴天,温度 25°C”。
调用 get_weather("上海") → 返回“上海的天气是多云,温度 27°C”。
调用 calculate("2+3*4") → 返回“计算结果:14”。
更新上下文:
工具结果作为新信息反馈到模型的上下文(例如 role: tool 消息)。
模型根据结果继续推理,决定是否需要进一步行动或生成最终回答。
循环迭代:
如果任务复杂(例如需要多次工具调用或多步推理),模型重复推理-行动循环,直到任务完成。
最终生成综合回答,例如:“北京今天天气晴朗,气温 25 摄氏度;上海天气多云,气温 27 摄氏度;另外,2+3*4 的计算结果是 14。”
ReAct 与 OpenAI 的 tool_calls 机制
在 OpenAI API 中,tool_calls 机制是实现 ReAct 风格推理和行动的理想工具。tool_calls 允许模型根据用户输入动态调用外部函数(如 get_weather 或 calculate),并将结果反馈到对话历史中。以下是 ReAct 如何与 tool_calls 结合:
推理:第一次 API 调用分析用户输入,决定是否需要调用工具(生成 tool_calls)。系统提示(system 消息)指导模型识别意图,例如:
{"role": "system", "content": "你是一个 ReAct 智能助手,遵循推理-行动-推理流程..."}
行动:如果模型生成 tool_calls,代码执行工具并将结果存为 role: tool 消息。
综合推理:第二次 API 调用让模型根据工具结果(role: tool 消息)生成自然语言回答。
示例代码(基于 ReAct 的智能代理)
以下是一个基于 ReAct 框架的 Python 代码示例,扩展了你的 AIAgent 类,支持多意图任务处理:
import os
import json
import logging
import requests
from openai import OpenAI
# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class ReActAgent:
def __init__(self):
# 初始化 OpenAI 客户端
self.client = OpenAI(api_key=os.getenv('OPENAI_API_KEY', 'your-api-key'))
self.conversation_history = []
# 定义工具
self.tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "查询指定城市的天气信息,包括天气状况和温度。",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名称(如北京、上海)"}
},
"required": ["city"]
}
}
},
{
"type": "function",
"function": {
"name": "calculate",
"description": "计算数学表达式的结果,支持加减乘除。",
"parameters": {
"type": "object",
"properties": {
"expression": {"type": "string", "description": "数学表达式(如 2+3*4)"}
},
"required": ["expression"]
}
}
}
]
def get_weather(self, city):
"""查询城市天气信息"""
try:
api_key = os.getenv('OPENWEATHER_API_KEY', 'your-openweather-api-key')
url = f"http://api.openweathermap.org/data/2.5/weather?q={city}&appid={api_key}&units=metric&lang=zh_cn"
response = requests.get(url, timeout=5)
response.raise_for_status()
data = response.json()
return f"{city}的天气是 {data['weather'][0]['description']},温度 {data['main']['temp']}°C。"
except requests.RequestException as e:
logger.error(f"获取 {city} 天气失败:{str(e)}")
return f"无法获取 {city} 的天气信息,请稍后重试。"
def calculate(self, expression):
"""计算数学表达式"""
try:
import ast
result = ast.literal_eval(expression)
return f"计算结果:{result}"
except (ValueError, SyntaxError) as e:
logger.error(f"计算表达式 {expression} 失败:{str(e)}")
return "无法计算,请检查输入表达式。"
def execute_tool(self, tool_call, previous_results=None):
"""执行工具调用"""
function_name = tool_call.function.name
try:
arguments = json.loads(tool_call.function.arguments)
if function_name == "get_weather":
return self.get_weather(arguments["city"])
elif function_name == "calculate":
return self.calculate(arguments["expression"])
return f"未知的工具函数:{function_name}"
except (json.JSONDecodeError, KeyError) as e:
logger.error(f"工具调用失败:{str(e)}")
return f"工具调用失败:{str(e)}"
def process_task(self, prompt):
"""处理用户任务(ReAct 风格)"""
# 添加用户输入到历史
self.conversation_history.append({"role": "user", "content": prompt})
logger.info(f"用户输入: {prompt}")
# 第一次 API 调用(推理 + 行动)
try:
response = self.client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "system",
"content": """你是一个 ReAct 智能助手,遵循推理-行动-推理流程:
1. 推理:分析用户输入,分解任务,识别所有意图,列出需要执行的步骤。
2. 行动:根据推理调用必要的工具(如 get_weather 或 calculate),记录每个工具的结果。
3. 综合:根据工具结果生成自然、连贯的回答。
对于多意图请求,分别处理每个意图,并整合结果。如果输入不明确,请求澄清。
示例:
输入:查询北京天气并计算 2+3
推理:1. 调用 get_weather("北京") 查询天气。2. 调用 calculate("2+3") 计算结果。
行动:执行工具,获取结果。
综合:生成回答,如“北京天气晴朗,气温 25°C;2+3=5”。"""
},
*self.conversation_history[-5:]
],
tools=self.tools,
tool_choice="auto",
max_tokens=1024,
temperature=0.7,
)
except Exception as e:
logger.error(f"第一次 API 调用失败:{str(e)}")
return f"发生错误:{str(e)}"
# 获取响应
response_message = response.choices[0].message
logger.info(f"第一次模型响应: {response_message}")
# 检查是否触发工具调用
if response_message.tool_calls:
self.conversation_history.append(response_message)
results = []
for tool_call in response_message.tool_calls:
result = self.execute_tool(tool_call, results)
results.append(result)
self.conversation_history.append({
"role": "tool",
"content": result,
"tool_call_id": tool_call.id
})
# 第二次 API 调用(综合推理)
try:
history_length = max(5, len(response_message.tool_calls) * 2 + 3)
second_response = self.client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "根据工具调用结果,生成自然、连贯的回答,将多个结果整合成一句话或段落。"},
*self.conversation_history[-history_length:]
],
max_tokens=1024,
temperature=0.7,
)
final_response = second_response.choices[0].message.content
logger.info(f"第二次模型响应: {final_response}")
self.conversation_history.append({"role": "assistant", "content": final_response})
return final_response
except Exception as e:
logger.error(f"第二次 API 调用失败:{str(e)}")
return f"发生错误:{str(e)}"
else:
response_content = response_message.content
if response_content is None:
return "无内容返回,请澄清您的请求。"
return response_content
def run(self):
"""运行 Agent"""
while True:
try:
prompt = input('\n用户提问: ')
if prompt.lower() == '退出':
break
answer = self.process_task(prompt)
print(f"模型回复: {answer}")
self.conversation_history.append({"role": "assistant", "content": answer})
except Exception as e:
logger.error(f"运行时错误:{str(e)}")
print(f"错误: {str(e)}")
self.conversation_history.append({"role": "assistant", "content": f"发生错误:{str(e)}"})
if __name__ == "__main__":
agent = ReActAgent()
agent.run()
示例运行结果
输入:查询北京和上海天气并计算 2+3*4
第一次 API 调用:
模型推理:
推理:用户请求包含三个意图: 1. 查询北京的天气 → 调用 get_weather("北京")。 2. 查询上海的天气 → 调用 get_weather("上海")。 3. 计算 2+3*4 → 调用 calculate("2+3*4")。
生成 tool_calls:
[ {"id": "call_789", "type": "function", "function": {"name": "get_weather", "arguments": "{\"city\": \"北京\"}"}}, {"id": "call_012", "type": "function", "function": {"name": "get_weather", "arguments": "{\"city\": \"上海\"}"}}, {"id": "call_456", "type": "function", "function": {"name": "calculate", "arguments": "{\"expression\": \"2+3*4\"}"}} ]
工具执行:
get_weather("北京") → “北京的天气是晴天,温度 25°C”。
get_weather("上海") → “上海的天气是多云,温度 27°C”。
calculate("2+3*4") → “计算结果:14”。
第二次 API 调用:
模型生成综合回答:“北京今天天气晴朗,气温 25 摄氏度;上海天气多云,气温 27 摄氏度;另外,2+3*4 的计算结果是 14。”
输出:
模型回复: 北京今天天气晴朗,气温 25 摄氏度;上海天气多云,气温 27 摄氏度;另外,2+3*4 的计算结果是 14。
ReAct 的优势与局限性
优势
动态推理与行动:ReAct 允许模型在推理和行动之间循环,适合复杂任务和多意图场景。
可解释性:推理步骤记录了模型的思考过程,便于调试和理解。
通用性:适用于问答、工具调用、决策制定等多种任务。
多意图支持:通过显式推理,模型可以分解多意图请求,调用多个工具。
局限性
提示依赖:ReAct 的效果高度依赖精心设计的提示。需要清晰的 system 提示和示例来指导模型。
模型能力限制:如果输入模糊或任务复杂,模型可能无法正确分解意图。
计算成本:多次 API 调用(推理-行动循环)会增加延迟和成本。
工具依赖:复杂任务可能需要工具之间有依赖关系,需额外逻辑支持。
ReAct 与多意图任务
ReAct 框架特别适合处理单次请求中的多意图任务。例如,用户输入“查询北京和上海天气并计算 2+3*4”,ReAct 通过推理分解任务,调用多个工具(get_weather 和 calculate),并综合结果生成自然回答。相比传统的 tool_calls 机制,ReAct 的优势在于:
显式推理:记录思考步骤,提高可解释性。
动态调整:支持多步推理和行动,适应复杂任务。
上下文整合:通过第二次 API 调用综合多个工具结果,生成连贯回答。
结论
ReAct 框架通过协同推理和行动,显著提升了大型语言模型在复杂任务中的表现。结合 OpenAI 的 tool_calls 机制,开发者可以通过提示设计和代码实现 ReAct 风格的智能代理,支持多意图任务处理。上述代码示例展示了如何在 Python 中实现 ReAct,处理天气查询和数学计算等多意图请求。未来,可以通过优化提示、支持工具依赖和持久化存储进一步增强 ReAct 代理的功能。