Agent开发实战：如何高效调用工具（Function Calling）的完整指南

1次阅读

没有评论

共计 2427 个字符，预计需要花费 7 分钟才能阅读完成。

Agent的本质是一个智能调度中枢，它通过感知环境状态（输入数据）和预设目标（任务描述），动态决定何时调用哪些工具（Functions）。就像餐厅里的服务员，根据顾客需求协调后厨、收银等不同部门。

Function Calling则是 Agent 与外部工具交互的标准协议，包含三个关键要素：

工具注册：每个工具需要明确定义输入输出格式（类似 API 文档）
意图解析：Agent 将自然语言指令转换为结构化调用请求
执行反馈：工具返回结构化结果供 Agent 决策

典型交互流程如下：

Agent 接收用户请求（如 ” 预订下周五北京飞上海的航班 ”）
解析出需要调用航班查询工具（flight_search）
生成符合工具规范的参数（{date: “2023-11-17”, from: “PEK”, to: “SHA”}）
执行调用并处理返回结果
根据结果决定后续动作（如展示给用户或继续调用支付工具）

在实际开发中，我们团队遇到过这些典型问题：

接口不一致：不同工具返回的成功响应格式差异大，有的用code:200，有的用status:"OK"
错误雪崩：一个工具失败导致整个调用链中断，没有优雅降级方案
超时失控：第三方工具响应缓慢时，长时间阻塞主线程
上下文丢失：多步骤调用中，前序工具的返回结果无法自动传递给后续工具
监控盲区：难以统计各工具的调用成功率、耗时等关键指标

我们采用的架构如下图所示（图示说明）：

[用户请求] 
    → [Agent 路由层] 
    → [工具适配层]（协议转换 + 错误预处理）→ [具体工具执行]
    → [结果标准化层]
    → [响应组装]

关键设计原则：

统一接入规范 ：所有工具必须实现execute(input: dict) -> dict 接口
熔断机制：单个工具连续失败 N 次后自动隔离
超时分级：关键工具设置短超时（如 500ms），非关键工具可放宽（如 3s）
上下文管理 ：通过session_id 贯穿整个调用链

以下是带错误处理的核心代码：

class WeatherTool:
    @staticmethod
    def execute(params: dict) -> dict:
        """ 查询天气工具
        Args:
            params: {
                "city": str,   # 城市名称
                "date": str    # 日期 YYYY-MM-DD
            }
        Returns:
            {
                "temperature": float,
                "weather": str
            }
        """
        try:
            # 模拟实际 API 调用
            response = requests.get(f"https://api.weather.com/{params['city']}",
                timeout=2.0
            )
            response.raise_for_status()
            return {"temperature": response.json()["temp"],
                "weather": response.json()["condition"]
            }
        except Exception as e:
            # 标准化错误格式
            return {
                "__error__": True,
                "code": "WEATHER_API_FAILED",
                "message": str(e)
            }

class AgentCore:
    def __init__(self):
        self.tools = {"weather": WeatherTool.execute}

    def run_tool(self, tool_name: str, params: dict, retry=2):
        for attempt in range(retry + 1):
            try:
                result = self.tools[tool_name](params)
                if "__error__" in result:
                    raise ToolException(result["message"])
                return result
            except Exception as e:
                if attempt == retry:
                    raise
                time.sleep(1 * (attempt + 1))  # 指数退避

    def process_request(self, user_input: str):
        # 实际项目中这里会有 NLU 组件
        if "天气" in user_input:
            city = extract_city(user_input)  # 假设的实体抽取函数
            return self.run_tool("weather", {"city": city})

通过对比测试（1000 次调用取平均），不同实现方式的性能表现：