Function Calling函数调用是让LLM从会聊天进化为能干活的关键能力。通过Function Calling模型可以决定何时调用外部工具、传递什么参数从而查询实时数据、执行代码、操作文件……几乎任何你能写成函数的能力都可以被LLM驱动。本文从协议原理到生产实践系统讲解Function Calling工程中的核心知识点与踩坑经验。Function Calling的工作原理理解原理是写好代码的基础。Function Calling的完整流程1.工具注册将函数的名称、描述和参数Schema告知模型2.模型决策模型根据用户请求决定是否调用工具以及调用哪个工具、传入什么参数3.工具执行应用代码实际执行被选中的函数4.结果回传将函数执行结果回传给模型5.最终生成模型根据函数结果生成最终回复这个流程中模型只做决策不执行代码。代码的实际执行权始终在你的应用手中这是一个重要的安全边界。python# 一个完整的Function Calling示例from openai import OpenAIimport jsonclient OpenAI()# 工具定义JSON Schema格式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] } } }]# 实际的工具实现def get_weather(city: str, unit: str celsius) - dict: # 实际中调用天气API return { city: city, temperature: 25, unit: unit, condition: 晴天, humidity: 60% }def run_conversation(user_message: str): messages [{role: user, content: user_message}] # 第一次调用模型决策是否使用工具 response client.chat.completions.create( modelgpt-4o, messagesmessages, toolstools, tool_choiceauto # 让模型自主决定 ) response_message response.choices[0].message # 检查模型是否决定调用工具 if response_message.tool_calls: messages.append(response_message) # 执行所有工具调用可能多个 for tool_call in response_message.tool_calls: function_name tool_call.function.name function_args json.loads(tool_call.function.arguments) # 路由到实际函数 if function_name get_weather: result get_weather(**function_args) else: result {error: f未知工具: {function_name}} # 将结果回传 messages.append({ role: tool, tool_call_id: tool_call.id, content: json.dumps(result, ensure_asciiFalse) }) # 第二次调用基于工具结果生成最终回复 final_response client.chat.completions.create( modelgpt-4o, messagesmessages ) return final_response.choices[0].message.content return response_message.contentresult run_conversation(上海今天天气怎么样)print(result)## 工具Schema设计的核心原则工具的Schema设计直接影响模型调用的准确率。这里有五个关键原则### 1. 描述要精准不要模糊python# ❌ 模糊的描述{ name: search, description: 搜索相关信息}# ✅ 精准的描述{ name: search_knowledge_base, description: 在公司内部知识库中搜索技术文档。仅用于查询公司产品、API文档和内部指南。不要用于查询实时行情、新闻等外部信息。}### 2. 参数类型和约束要明确python{ name: create_task, description: 在项目管理系统中创建任务, parameters: { type: object, properties: { title: { type: string, description: 任务标题50字以内, maxLength: 50 }, priority: { type: string, enum: [low, medium, high, urgent], description: 优先级low(低)、medium(中)、high(高)、urgent(紧急) }, due_date: { type: string, pattern: ^\\d{4}-\\d{2}-\\d{2}$, description: 截止日期格式YYYY-MM-DD如2026-05-31 }, assignee_id: { type: string, description: 负责人的用户ID不是用户名可以通过get_user_id工具获取 } }, required: [title, priority] }}### 3. 多工具时要消除歧义当有多个功能相似的工具时需要在描述中明确区分使用场景pythontools [ { function: { name: search_web, description: 搜索互联网上的公开信息适合最新新闻、实时行情、公开资料查询。不适合内部文档、公司数据。 } }, { function: { name: search_internal_docs, description: 搜索公司内部知识库适合API文档、技术规范、内部流程。不适合外部信息、实时数据。 } }]### 4. 设计原子性工具而非复合工具python# ❌ 复合工具功能太多参数复杂{ name: manage_calendar, description: 管理日历可以查看、创建、删除、修改事件}# ✅ 原子性工具每个工具做一件事tools [ {name: get_calendar_events, description: 查询指定日期范围内的日历事件}, {name: create_calendar_event, description: 创建新的日历事件}, {name: update_calendar_event, description: 修改已有日历事件}, {name: delete_calendar_event, description: 删除日历事件},]## 并行函数调用现代LLM支持在一次响应中同时调用多个工具利用好这个特性可以显著减少延迟pythonimport asyncioasync def handle_parallel_tool_calls(tool_calls: list) - list: 并行执行多个工具调用 async def execute_single(tool_call): function_name tool_call.function.name args json.loads(tool_call.function.arguments) try: if function_name get_stock_price: result await get_stock_price(**args) elif function_name get_company_info: result await get_company_info(**args) elif function_name get_news: result await get_news(**args) else: result {error: f未知工具: {function_name}} except Exception as e: result {error: str(e)} return { role: tool, tool_call_id: tool_call.id, content: json.dumps(result, ensure_asciiFalse) } # 并行执行所有工具调用 results await asyncio.gather(*[execute_single(tc) for tc in tool_calls]) return list(results)# 使用示例GPT-4o会自动识别可以并行的工具调用response client.chat.completions.create( modelgpt-4o, messages[{ role: user, content: 帮我查一下腾讯和阿里的当前股价以及它们最近的新闻 }], toolstools)# 模型可能同时调用 get_stock_price(腾讯)、get_stock_price(阿里)、get_news(腾讯)、get_news(阿里)if response.choices[0].message.tool_calls: tool_results await handle_parallel_tool_calls(response.choices[0].message.tool_calls)## 工具调用安全性Function Calling最大的风险是LLM决策 自动执行 潜在的高风险操作。特别是涉及数据写入、外部API调用、代码执行的工具需要严格的安全设计。### 分级权限控制pythonfrom enum import Enumclass ToolRiskLevel(Enum): READ_ONLY read_only # 只读操作无需确认 LOW_RISK low_risk # 低风险写入自动执行但记录日志 MEDIUM_RISK medium_risk # 中等风险需要用户确认 HIGH_RISK high_risk # 高风险操作需要显式授权TOOL_RISK_MAP { search_web: ToolRiskLevel.READ_ONLY, get_weather: ToolRiskLevel.READ_ONLY, create_task: ToolRiskLevel.LOW_RISK, send_email: ToolRiskLevel.MEDIUM_RISK, delete_records: ToolRiskLevel.HIGH_RISK, execute_code: ToolRiskLevel.HIGH_RISK,}class SafeToolExecutor: def __init__(self, user_approval_callback): self.approval_callback user_approval_callback async def execute(self, tool_name: str, args: dict) - dict: risk_level TOOL_RISK_MAP.get(tool_name, ToolRiskLevel.MEDIUM_RISK) if risk_level ToolRiskLevel.READ_ONLY: return await self._execute_directly(tool_name, args) elif risk_level ToolRiskLevel.LOW_RISK: result await self._execute_directly(tool_name, args) await self._audit_log(tool_name, args, result) return result elif risk_level in (ToolRiskLevel.MEDIUM_RISK, ToolRiskLevel.HIGH_RISK): # 暂停等待用户确认 approved await self.approval_callback({ tool: tool_name, args: args, risk_level: risk_level.value, message: fAI助手请求执行 [{tool_name}]参数{json.dumps(args, ensure_asciiFalse)} }) if not approved: return {error: 用户拒绝了此操作} return await self._execute_directly(tool_name, args) async def _execute_directly(self, tool_name: str, args: dict) - dict: # 实际工具执行逻辑 tool_func TOOL_REGISTRY[tool_name] return await tool_func(**args) async def _audit_log(self, tool_name: str, args: dict, result: dict): # 记录审计日志 pass### 参数验证与沙箱pythondef validate_tool_args(tool_name: str, args: dict) - tuple[bool, str]: 在执行工具前验证参数 if tool_name execute_code: code args.get(code, ) # 检查危险操作 dangerous_patterns [os.system, subprocess, eval(, exec(, __import__] for pattern in dangerous_patterns: if pattern in code: return False, f代码包含不允许的操作: {pattern} if tool_name delete_records: # 确保有明确的ID而非通配符删除 if * in str(args) or all in str(args).lower(): return False, 删除操作不允许使用通配符必须指定具体ID return True, OK## 流式Function Calling对于需要实时展示进度的场景可以使用流式APIpythonasync def stream_with_tools(user_message: str): 流式Function Calling支持实时展示工具调用过程 stream client.chat.completions.create( modelgpt-4o, messages[{role: user, content: user_message}], toolstools, streamTrue ) current_tool_calls {} async for chunk in stream: delta chunk.choices[0].delta # 处理工具调用块 if delta.tool_calls: for tc_chunk in delta.tool_calls: idx tc_chunk.index if idx not in current_tool_calls: current_tool_calls[idx] { id: tc_chunk.id, name: tc_chunk.function.name or , arguments: } if tc_chunk.function.arguments: current_tool_calls[idx][arguments] tc_chunk.function.arguments # 处理完成信号 if chunk.choices[0].finish_reason tool_calls: print(f\n[工具调用] 模型决定调用 {len(current_tool_calls)} 个工具) # 执行工具并继续对话...## 常见问题与解决方案| 问题 | 原因 | 解决方案 ||------|------|---------|| 模型总是调用工具即使不需要 | 工具描述过于宽泛 | 明确指定使用场景和不适用场景 || 参数格式经常不对 | Schema描述不够清晰 | 添加格式说明和示例值 || 模型选错工具 | 多个工具功能有重叠 | 重写描述强调使用边界 || 工具调用失败后模型卡住 | 没有处理错误返回 | 在工具结果中包含error字段模型会据此调整策略 || 多轮对话中重复调用工具 | 历史结果没有正确回传 | 确保tool角色消息正确插入到历史中 |## 总结Function Calling是构建实用AI Agent的基础能力。关键工程原则精心设计工具Schema描述要精准、参数要明确、合理处理并行调用减少延迟、严格控制执行安全分级权限参数验证以及完善的错误处理工具失败不应导致整个对话中断。掌握这些你的AI助手就能从聊天机器人进化为真正能做事的智能体。