
🧠 记忆锚点:先校验再执行;按错误分类重试,失败要能降级退出。
💡 答案要点
工具调用完整流程: 识别 → 参数提取 → 执行 → 结果处理
阶段1: 工具定义
展开 Python 代码示例(39 行)
python
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取指定城市的天气",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名,如北京、上海"
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "温度单位"
}
},
"required": ["city"]
}
}
},
{
"type": "function",
"function": {
"name": "search_database",
"description": "搜索数据库",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string"},
"table": {"type": "string"}
},
"required": ["query", "table"]
}
}
}
]阶段2: LLM决策
python
response = openai.ChatCompletion.create(
model="qwen3.5-plus",
messages=[
{"role": "user", "content": "北京今天天气怎么样?"}
],
tools=tools,
tool_choice="auto" # 自动决定是否调用工具
)
# LLM返回:
{
"role": "assistant",
"tool_calls": [{
"id": "call_123",
"function": {
"name": "get_weather",
"arguments": '{"city": "北京", "unit": "celsius"}'
}
}]
}阶段3: 参数验证与执行
python
def execute_tool_call(tool_call):
function_name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)
# 1. 权限检查
if not has_permission(function_name):
return {"error": "权限不足"}
# 2. 参数验证
if function_name == "get_weather":
if "city" not in arguments:
return {"error": "缺少必需参数: city"}
if len(arguments["city"]) > 20:
return {"error": "城市名过长"}
# 3. 执行(带超时和重试)
try:
result = call_with_timeout(
function_map[function_name],
arguments,
timeout=5
)
return {"success": True, "data": result}
except TimeoutError:
return {"error": "工具调用超时"}
except Exception as e:
return {"error": str(e)}阶段4: 结果反馈
python
# 将工具结果返回给LLM
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(tool_result)
})
# LLM基于工具结果生成最终答案
final_response = openai.ChatCompletion.create(
model="qwen3.5-plus",
messages=messages
)失败处理策略:
1. 参数错误
python
if tool_result.get("error"):
# 让LLM修正参数
retry_prompt = f"""
工具调用失败: {tool_result['error']}
请修正参数后重试。
"""
# 重新调用2. 超时重试
python
def call_with_retry(func, args, max_retries=3):
for attempt in range(max_retries):
try:
return func(**args)
except TimeoutError:
if attempt == max_retries - 1:
return {"error": "多次重试失败"}
time.sleep(2 ** attempt) # 指数退避3. 降级策略
python
def execute_with_fallback(tool_call):
primary_result = try_tool(tool_call)
if primary_result.get("error"):
# 降级到备用工具
fallback_result = try_fallback_tool(tool_call)
if fallback_result.get("error"):
# 最终降级:返回缓存或默认值
return get_cached_or_default()
return fallback_result
return primary_result4. 监控与告警
python
import logging
def execute_tool(tool_call):
start_time = time.time()
try:
result = _execute(tool_call)
# 记录成功
logging.info({
"tool": tool_call.function.name,
"latency": time.time() - start_time,
"status": "success"
})
return result
except Exception as e:
# 记录失败
logging.error({
"tool": tool_call.function.name,
"error": str(e),
"status": "failure"
})
# 告警(错误率>5%)
if get_error_rate() > 0.05:
send_alert("工具调用错误率过高")
raise完整示例:
展开 Python 代码示例(42 行)
python
class AgentWithTools:
def __init__(self, tools):
self.tools = tools
self.messages = []
def run(self, user_input):
self.messages.append({
"role": "user",
"content": user_input
})
max_iterations = 5
for i in range(max_iterations):
# LLM决策
response = openai.ChatCompletion.create(
model="qwen3.5-plus",
messages=self.messages,
tools=self.tools
)
assistant_message = response.choices[0].message
self.messages.append(assistant_message)
# 检查是否需要调用工具
if not assistant_message.tool_calls:
return assistant_message.content
# 执行所有工具调用
for tool_call in assistant_message.tool_calls:
result = self.execute_tool(tool_call)
self.messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result)
})
return "达到最大迭代次数"
def execute_tool(self, tool_call):
# 带重试和降级的工具执行
pass面试话术:
"工具调用的关键是鲁棒性。我们做了4层防护:1)参数白名单防注入 2)超时+指数退避重试 3)主备工具降级 4)监控告警。生产环境工具调用成功率99.2%,P99延迟<2s。"