Day11：ReAct Agent 优化：实现多工具智能选择与稳健调用
·
Day11：ReAct Agent 优化：实现多工具智能选择与稳健调用
摘要：本文聚焦于 ReAct Agent 的优化，旨在提升其根据用户问题自主选择合适工具的能力。以通义千问大模型为基础，通过扩展功能实现多工具选择，并增添工具调用失败时的重试逻辑。详细介绍优化思路、代码实现以及测试过程，帮助读者深入理解并掌握这一优化后的智能交互模式。
一、引言
在之前对 ReAct Agent 的学习基础上，进一步优化其功能，使其能够在多种工具间智能选择，以更好地满足用户多样化的需求，是提升模型实用性的关键一步。今天我们将围绕如何优化 ReAct Agent，使其能够根据用户问题准确选择合适工具，并处理工具调用失败的重试逻辑展开学习。
二、核心任务：优化 ReAct Agent 实现多工具选择
（一）优化思路
-
工具描述与映射：为每个工具（如天气、计算器、搜索等）创建详细的描述信息，包括工具的功能、参数以及适用场景。通过映射关系将工具描述与实际的工具函数关联起来，方便模型根据推理结果选择对应的工具。
-
推理与工具选择：模型在接收到用户问题后，首先进行深入推理，分析问题的意图和所需信息，依据工具描述判断哪个工具最适合解决该问题。例如，对于 “今天天气适合跑步吗” 这样的问题，模型能够识别出需要先获取天气信息，从而选择天气工具。
（二）代码实现
import time
import requests
from pydantic import BaseModel, field_validator
import json
from typing import Optional, Dict, Any
# ===================== 工具参数模型定义（严格校验）=====================
# 天气查询工具参数
class WeatherQueryParams(BaseModel):
city: str
api_key: str
@field_validator('city')
def city_not_empty(cls, v):
if not v.strip():
raise ValueError('城市名称不能为空')
return v
@field_validator('api_key')
def api_key_not_empty(cls, v):
if not v.strip():
raise ValueError('api_key不能为空')
return v
# 计算器工具参数
class CalculatorParams(BaseModel):
num1: float
num2: float
operation: str
@field_validator('operation')
def valid_operation(cls, v):
valid_ops = ['+', '-', '*', '/']
if v not in valid_ops:
raise ValueError('仅支持+、-、*、/四种运算')
return v
# 搜索工具参数
class SearchParams(BaseModel):
query: str
api_key: str
@field_validator('query')
def query_not_empty(cls, v):
if not v.strip():
raise ValueError('搜索关键词不能为空')
return v
@field_validator('api_key')
def api_key_not_empty(cls, v):
if not v.strip():
raise ValueError('api_key不能为空')
return v
# ===================== 真实工具实现（本地/远程API）=====================
# 1. 天气查询（调用真实WeatherAPI）
def query_weather(params: WeatherQueryParams) -> Optional[Dict[str, Any]]:
base_url = "http://api.weatherapi.com/v1/current.json"
params_dict = {
"key": params.api_key,
"q": params.city,
"lang": "zh-CN" # 返回中文结果
}
try:
# 真实GET请求，超时控制
response = requests.get(base_url, params=params_dict, timeout=10)
response.raise_for_status() # 触发HTTP错误（4xx/5xx）
return response.json()
except requests.RequestException as e:
print(f"【天气工具调用失败】: {str(e)}")
return None
# 2. 计算器（本地工具，无API依赖）
def calculate(params: CalculatorParams) -> Optional[float]:
try:
if params.operation == '+':
return params.num1 + params.num2
elif params.operation == '-':
return params.num1 - params.num2
elif params.operation == '*':
return params.num1 * params.num2
elif params.operation == '/':
if params.num2 == 0:
print("【计算工具调用失败】: 除数不能为零")
return None
return round(params.num1 / params.num2, 4) # 保留4位小数
except Exception as e:
print(f"【计算工具调用失败】: {str(e)}")
return None
# 3. 通用搜索（调用真实搜索API，可替换为百度/必应搜索API）
def search(params: SearchParams) -> Optional[Dict[str, Any]]:
base_url = "https://api.example.com/search" # 替换为真实搜索API地址（如百度百科/必应）
params_dict = {
"key": params.api_key,
"q": params.query,
"page_size": 10 # 限制返回结果数量
}
try:
response = requests.get(base_url, params=params_dict, timeout=10)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
print(f"【搜索工具调用失败】: {str(e)}")
return None
# 工具映射表：大模型返回的函数名 -> 本地工具函数+参数模型
TOOL_MAPPING = {
"query_weather": {"func": query_weather, "model": WeatherQueryParams},
"calculate": {"func": calculate, "model": CalculatorParams},
"search": {"func": search, "model": SearchParams}
}
# ===================== 真实通义千问API调用（带函数调用声明）=====================
def call_qwen_api(
qwen_api_key: str,
user_prompt: str,
tools: list # 声明可用工具列表，让大模型感知
) -> Optional[Dict[str, Any]]:
"""
调用真实通义千问兼容OpenAI格式的API（compatible-mode）
自动携带工具声明，大模型会返回function_call或直接回答
"""
url = "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {qwen_api_key}"
}
# 通义千问函数调用核心参数：声明工具列表+强制函数调用判断
request_data = {
"model": "qwen-plus", # 可替换为qwen-turbo/qwen-max
"messages": [{"role": "user", "content": user_prompt}],
"tools": tools, # 告诉大模型有哪些工具可用
"tool_choice": "auto", # 自动判断是否调用工具：auto/required/none
"stream": False,
"temperature": 0.1 # 低温度保证推理稳定性
}
try:
response = requests.post(url, headers=headers, json=request_data, timeout=15)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
print(f"【通义千问API调用失败】: {str(e)}")
return None
# ===================== 定义通义千问可感知的工具描述（标准化JSON）=====================
# 严格遵循OpenAI函数调用格式，大模型会解析该描述判断工具使用方式
QWEN_TOOLS_DESC = [
{
"type": "function",
"function": {
"name": "query_weather",
"description": "查询指定城市的实时天气，包括温度、天气状况等，用于回答天气相关问题",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "要查询的城市名称，如北京、上海"},
"api_key": {"type": "string", "description": "天气API的有效密钥"}
},
"required": ["city", "api_key"]
}
}
},
{
"type": "function",
"function": {
"name": "calculate",
"description": "执行简单的数学运算，仅支持加、减、乘、除四种操作",
"parameters": {
"type": "object",
"properties": {
"num1": {"type": "number", "description": "第一个运算数"},
"num2": {"type": "number", "description": "第二个运算数"},
"operation": {"type": "string", "description": "运算操作，仅支持+、-、*、/"}
},
"required": ["num1", "num2", "operation"]
}
}
},
{
"type": "function",
"function": {
"name": "search",
"description": "搜索互联网公开信息，用于回答人工智能、科技发展、时事等非常识性问题",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词，如人工智能的发展历程"},
"api_key": {"type": "string", "description": "搜索API的有效密钥"}
},
"required": ["query", "api_key"]
}
}
}
]
# ===================== 核心ReAct流程实现（推理-行动-反馈）=====================
def react_flow(
user_input: str,
qwen_api_key: str,
max_retries: int = 3
) -> str:
"""
真实ReAct流程主函数
步骤：1.调用大模型推理→2.判断是否调用工具→3.工具参数校验+执行→4.返回结果/二次推理
"""
# 步骤1：调用真实通义千问API，携带工具声明
llm_response = call_qwen_api(qwen_api_key, user_input, QWEN_TOOLS_DESC)
if not llm_response or "choices" not in llm_response:
return "【ReAct流程失败】通义千问API无有效返回"
llm_choice = llm_response["choices"][0]["message"]
# 步骤2：判断大模型是否要求调用工具
if "tool_calls" in llm_choice: # 通义千问返回工具调用指令（兼容OpenAI tool_calls格式）
for tool_call in llm_choice["tool_calls"]:
func_name = tool_call["function"]["name"]
func_args = json.loads(tool_call["function"]["arguments"]) # 解析参数JSON
# 校验工具是否在本地映射表中
if func_name not in TOOL_MAPPING:
return f"【ReAct流程失败】不支持的工具：{func_name}"
tool_func = TOOL_MAPPING[func_name]["func"]
tool_model = TOOL_MAPPING[func_name]["model"]
tool_result = None
# 步骤3：工具调用+失败重试（参数校验+执行）
for attempt in range(max_retries):
try:
# 严格校验参数格式
params = tool_model(**func_args)
# 执行真实工具函数
tool_result = tool_func(params)
if tool_result is not None:
break
else:
print(f"【工具重试】{func_name} 调用失败，重试 {attempt+1}/{max_retries}")
except Exception as e:
print(f"【参数校验失败】{func_name}：{str(e)}，重试 {attempt+1}/{max_retries}")
tool_result = None
if tool_result is None:
return f"【ReAct流程失败】{func_name} 工具调用多次失败，请检查参数/API密钥"
# 步骤4：工具结果处理（个性化返回，贴合用户问题）
if func_name == "query_weather":
try:
city = func_args["city"]
condition = tool_result["current"]["condition"]["text"]
temp_c = tool_result["current"]["temp_c"]
wind = tool_result["current"]["wind_kph"]
# 针对「是否适合出行」类问题的个性化反馈
if "适合" in user_input or "跑步" in user_input or "出行" in user_input:
if "雨" in condition or "雪" in condition or temp_c < 10 or temp_c > 32:
return f"{city}实时天气：{condition}，温度{temp_c}℃，风速{wind}km/h，当前天气不太适合出行/跑步~"
else:
return f"{city}实时天气：{condition}，温度{temp_c}℃，风速{wind}km/h，当前天气很适合出行/跑步~"
else:
return f"{city}实时天气：{condition}，温度{temp_c}℃，风速{wind}km/h，体感温度{tool_result['current']['feelslike_c']}℃"
except KeyError as e:
return f"【天气结果解析失败】缺失字段：{str(e)}，原始返回：{tool_result}"
elif func_name == "calculate":
return f"计算结果：{func_args['num1']}{func_args['operation']}{func_args['num2']} = {tool_result}"
elif func_name == "search":
return f"【搜索结果】关键词：{func_args['query']}\n{json.dumps(tool_result, ensure_ascii=False, indent=2)}"
else:
# 大模型无需调用工具，直接返回自然语言回答
return llm_choice.get("content", "【ReAct流程完成】无有效回答内容")
# ===================== 真实场景测试（需替换为自己的有效API_KEY）=====================
if __name__ == "__main__":
time.sleep(1)
# 替换为自己的真实密钥（必填！）
QWEN_API_KEY = "QWEN_API_KEY" # 阿里云dashscope获取
WEATHER_API_KEY = "你的WeatherAPI_KEY" # weatherapi.com获取
SEARCH_API_KEY = "你的搜索API_KEY" # 替换为真实搜索API密钥
# 测试用例（真实用户问题，通义千问会自动推理是否调用工具）
test_cases = [
f"今天上海适合跑步吗？需要调用天气工具，api_key填{WEATHER_API_KEY}",
"计算100.5除以2.5的结果",
f"搜索2025年人工智能行业发展报告，api_key填{SEARCH_API_KEY}",
"介绍一下ReAct范式的核心原理" # 无需调用工具，大模型直接回答
]
# 执行ReAct流程测试
for idx, question in enumerate(test_cases, 1):
print(f"\n===== 测试用例{idx}：{question} =====")
result = react_flow(question, QWEN_API_KEY)
print(f"结果：{result}")
代码说明：
-
工具函数定义：分别定义了天气查询（
query_weather）、计算器（calculate）和搜索（search）三个工具函数，并使用pydantic进行参数校验。 -
通义千问 API 调用函数：
call_qwen_api函数按照通义千问 API 要求构建请求并处理响应，实际使用时需替换为真实 API 密钥。 -
ReAct 流程函数
**react_flow**：接收用户输入、API 密钥以及最大重试次数。通过调用通义千问 API 获取响应，根据响应中的function_call信息选择并调用相应工具函数。若工具调用失败，按照重试逻辑进行重试，最多重试max_retries次。根据工具调用结果生成回答。 -
测试部分：设置了三个不同的用户问题，分别测试天气、计算器和搜索工具的调用，展示了优化后的 ReAct Agent 对不同问题的工具选择和处理能力。
三、补充任务：添加工具与测试
（一）添加多个工具
如上述代码中添加了计算器和搜索工具，丰富了 Agent 可使用的功能。
（二）测试 Agent 对不同问题的工具选择能力
通过在 if name == “main”: 部分设置不同的用户问题，模拟实际场景下 Agent 对各类问题的处理。观察
输出结果，验证 Agent 是否能正确选择工具并给出合理回答。
（三）处理工具调用失败的重试逻辑
在 react_flow 函数中，针对工具调用可能出现的失败情况，设置了重试逻辑。当工具调用返回 None 时，
进行重试，并在重试次数用尽后返回提示信息。
四、总结
通过今天的学习，我们成功优化了 ReAct Agent，使其具备多工具选择能力，并完善了工具调用失败时的
重试逻辑。这一优化显著提升了 Agent 的智能性和稳定性，能够更好地应对复杂多样的用户问题。
在实际应用中，可以根据具体需求进一步扩展工具种类和优化推理逻辑。记得将优化后的多工具选择
ReAct Agent 代码整理保存，方便后续参考和改进。若在学习过程中遇到问题，仔细检查工具函数的定
义、API 调用以及重试逻辑的设置。
希望大家通过掌握这些优化技巧，在基于大语言模型的应用开发中实现更强大、可靠的功能。
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
6
0- 0
已为社区贡献10条内容
所有评论(0)