Skip to content
🔗 分享本题
查看我的学习进度 →

工具调用从结构化参数、应用校验到执行回传,并按错误类型重试降级

🧠 记忆锚点:先校验再执行;按错误分类重试,失败要能降级退出。

💡 答案要点

工具调用完整流程: 识别 → 参数提取 → 执行 → 结果处理

阶段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_result

4. 监控与告警

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。"

📚 参考:OpenAI Function Calling 指南(并行调用与错误处理)