Function Calling深度指南:让LLM精准调用工具的工程实践

张开发
2026/4/21 4:39:39 15 分钟阅读

分享文章

Function Calling深度指南:让LLM精准调用工具的工程实践
Function Calling是AI Agent的神经系统如果把AI Agent比作一个人Function Calling就是它的双手——让语言模型从说说而已变成真正执行。没有Function CallingLLM只是一个聪明的文字处理器。有了它LLM可以查数据库、调API、执行代码、控制系统。这是从AI助手到AI Agent的本质跨越。但Function Calling用起来简单用好却有相当多的工程细节。本文将从基础到生产级最佳实践全面拆解Function Calling的工程实现。—## 基础Function Calling的工作原理### 交互流程1. 开发者定义工具函数定义JSON │ ▼2. 用户发送请求 │ ▼3. LLM判断是否需要调用工具 ├── 不需要直接生成文本回答 └── 需要生成tool_calls包含函数名参数 │ ▼4. 应用层执行实际函数非LLM执行 │ ▼5. 将函数执行结果返回给LLM │ ▼6. LLM基于结果生成最终回答关键认知LLM本身不执行函数。它只是决定调用什么、传什么参数实际执行是你的代码。### 最简示例pythonfrom openai import OpenAIimport jsonclient OpenAI()# 第一步定义工具tools [ { type: function, function: { name: get_weather, description: 获取指定城市的当前天气。当用户询问天气时调用此函数。, parameters: { type: object, properties: { city: { type: string, description: 城市名称如北京、上海、广州 }, unit: { type: string, enum: [celsius, fahrenheit], description: 温度单位默认celsius } }, required: [city] } } }]# 第二步调用LLMmessages [{role: user, content: 北京今天天气怎么样}]response client.chat.completions.create( modelgpt-4o, messagesmessages, toolstools, tool_choiceauto # auto/none/required 或指定特定函数)# 第三步处理tool_callsmessage response.choices[0].messageif message.tool_calls: for tool_call in message.tool_calls: function_name tool_call.function.name arguments json.loads(tool_call.function.arguments) print(fLLM决定调用: {function_name}({arguments})) # 第四步实际执行函数 if function_name get_weather: result get_weather(arguments[city], arguments.get(unit, celsius)) # 第五步将结果返回LLM messages.append(message) # 添加LLM的tool_calls消息 messages.append({ role: tool, tool_call_id: tool_call.id, content: json.dumps(result, ensure_asciiFalse) }) # 第六步LLM生成最终回答 final_response client.chat.completions.create( modelgpt-4o, messagesmessages, toolstools ) print(final_response.choices[0].message.content)—## 工程实践一工具定义的最佳实践### 好的工具描述 vs 差的工具描述工具能不能被正确调用50%取决于描述写得好不好python# ❌ 差的工具定义bad_tool { name: search, description: 搜索, # 太模糊 parameters: { type: object, properties: { q: {type: string} # 参数名不清晰缺少描述 } }}# ✅ 好的工具定义good_tool { name: search_knowledge_base, description: 在公司内部知识库中搜索相关文档和FAQ。 适用场景 - 用户询问公司产品功能或使用方法 - 用户遇到技术问题需要查找解决方案 - 需要查找公司政策或流程说明 不适用一般性常识问题、需要实时数据的问题, parameters: { type: object, properties: { query: { type: string, description: 搜索关键词或问题描述尽量使用用户原始表述 }, category: { type: string, enum: [product, technical, policy, billing], description: 文档类别帮助缩小搜索范围 }, limit: { type: integer, description: 返回结果数量默认3最大10, default: 3, minimum: 1, maximum: 10 } }, required: [query] }}### 工具参数类型的完整示例pythoncomprehensive_tool { name: create_task, description: 创建项目任务, parameters: { type: object, properties: { # 字符串 title: {type: string, description: 任务标题50字以内}, # 枚举 priority: { type: string, enum: [low, medium, high, urgent], description: 优先级 }, # 数字 estimated_hours: { type: number, description: 预计工时小时, minimum: 0.5, maximum: 200 }, # 布尔 send_notification: { type: boolean, description: 是否发送通知给相关人员 }, # 数组 assignee_ids: { type: array, items: {type: string}, description: 负责人ID列表, maxItems: 5 }, # 嵌套对象 deadline: { type: object, properties: { date: {type: string, description: 截止日期格式YYYY-MM-DD}, flexible: {type: boolean, description: 是否可以延期} }, required: [date] } }, required: [title, priority] }}—## 工程实践二并行函数调用GPT-4o支持单次请求触发多个并行函数调用显著减少延迟python# 场景用户问帮我订下周三从北京到上海的机票同时查一下上海的天气# 这需要同时调用搜索航班 查天气async def handle_parallel_tool_calls(user_message: str): tools [search_flights_tool, get_weather_tool, book_hotel_tool] response await async_client.chat.completions.create( modelgpt-4o, messages[{role: user, content: user_message}], toolstools, tool_choiceauto ) message response.choices[0].message messages [{role: user, content: user_message}, message] if message.tool_calls: # 并行执行所有工具调用 import asyncio async def execute_tool(tool_call): fn_name tool_call.function.name args json.loads(tool_call.function.arguments) # 根据函数名分发执行 if fn_name search_flights: result await search_flights_async(**args) elif fn_name get_weather: result await get_weather_async(**args) elif fn_name book_hotel: result await book_hotel_async(**args) else: result {error: f未知函数: {fn_name}} return tool_call.id, fn_name, result # 并发执行不串行等待 results await asyncio.gather(*[ execute_tool(tc) for tc in message.tool_calls ]) # 将所有结果一起返回给LLM for tool_call_id, fn_name, result in results: messages.append({ role: tool, tool_call_id: tool_call_id, content: json.dumps(result, ensure_asciiFalse) }) # 最终回答 final await async_client.chat.completions.create( modelgpt-4o, messagesmessages ) return final.choices[0].message.content—## 工程实践三工具调用的安全控制Function Calling最大的风险是LLM被诱导调用危险操作pythonclass SafeToolExecutor: 安全的工具执行器 def __init__(self): # 定义每个工具的权限等级 self.tool_permissions { search_knowledge_base: read, # 只读安全 get_weather: external_api, # 外部API低风险 create_task: write, # 写操作中风险 delete_record: destructive, # 破坏性高风险 send_email: external_action, # 外部行为需确认 } # 高风险操作需要人工确认 self.require_confirmation {destructive, external_action} def execute(self, tool_name: str, arguments: dict, user_context: dict) - dict: permission self.tool_permissions.get(tool_name, unknown) # 检查工具是否存在 if permission unknown: return {error: f未知工具: {tool_name}} # 高风险操作拦截 if permission in self.require_confirmation: if not user_context.get(confirmed): return { status: requires_confirmation, message: f执行 {tool_name}({arguments}) 需要用户确认, confirmation_token: self._generate_token(tool_name, arguments) } # 参数验证 try: validated_args self._validate_arguments(tool_name, arguments) except ValueError as e: return {error: f参数验证失败: {e}} # 速率限制 if not self._check_rate_limit(user_context.get(user_id), tool_name): return {error: 调用频率超限请稍后重试} # 执行工具 try: result self._execute_tool(tool_name, validated_args) # 记录审计日志 self._audit_log(tool_name, arguments, result, user_context) return result except Exception as e: return {error: f执行失败: {str(e)}} def _generate_token(self, tool_name: str, arguments: dict) - str: import hashlib, time content f{tool_name}{arguments}{time.time()} return hashlib.md5(content.encode()).hexdigest()[:16]—## 工程实践四工具调用结果的质量控制pythondef enrich_tool_result(tool_name: str, result: any) - str: 标准化工具返回提升LLM理解质量 if isinstance(result, dict) and error in result: # 统一错误格式 return json.dumps({ status: error, error_type: result.get(error_type, general), message: result[error], suggestion: 请检查参数后重试或告知用户当前功能不可用 }, ensure_asciiFalse) if tool_name search_knowledge_base: if not result or len(result) 0: return json.dumps({ status: no_results, message: 知识库中没有找到相关内容, suggestion: 可以建议用户联系人工客服或重新描述问题 }, ensure_asciiFalse) # 格式化搜索结果让LLM更容易理解 formatted { status: success, count: len(result), results: [ { relevance: r.get(score, 0), title: r[title], content: r[content][:500], # 限制长度 source: r.get(url, 内部文档) } for r in result ] } return json.dumps(formatted, ensure_asciiFalse) return json.dumps(result, ensure_asciiFalse)—## 实战构建完整的工具调用Agentpythonclass ToolCallingAgent: 完整的工具调用Agent实现 def __init__(self, tools: list, max_iterations: int 10): self.client OpenAI() self.tools tools self.executor SafeToolExecutor() self.max_iterations max_iterations def run(self, user_message: str, user_context: dict None) - str: messages [ {role: system, content: 你是一个有能力调用工具的AI助手。}, {role: user, content: user_message} ] for iteration in range(self.max_iterations): response self.client.chat.completions.create( modelgpt-4o, messagesmessages, toolsself.tools, tool_choiceauto ) message response.choices[0].message messages.append(message) # 没有tool_calls直接返回最终答案 if not message.tool_calls: return message.content # 执行所有工具调用 for tool_call in message.tool_calls: fn_name tool_call.function.name args json.loads(tool_call.function.arguments) print(f[第{iteration1}轮] 调用工具: {fn_name}({args})) result self.executor.execute(fn_name, args, user_context or {}) enriched_result enrich_tool_result(fn_name, result) messages.append({ role: tool, tool_call_id: tool_call.id, content: enriched_result }) # 超过最大迭代次数强制结束 return 抱歉处理您的请求时遇到了复杂情况请尝试简化您的问题。—## 总结Function Calling是构建AI Agent的核心机制工程实践中要重点关注1.工具描述质量决定调用准确率好的description和参数说明值100行代码2.并行调用减少延迟合理设计工具粒度让LLM一次多调3.安全控制不可省高风险操作必须有确认机制防范Prompt Injection攻击4.统一结果格式标准化的工具返回格式让LLM更容易理解和决策5.设置迭代上限防止Agent进入无限循环生产环境必不可少掌握Function Calling是从会用LLM到能构建AI Agent的关键一步。

更多文章