Function Calling是AI Agent的神经系统
如果把AI Agent比作一个人，Function Calling就是它的双手——让语言模型从"说说而已"变成"真正执行"。没有Function Calling，LLM只是一个聪明的文字处理器。有了它，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( model="gpt-4o", messages=messages, tools=tools, tool_choice="auto" # 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(f"LLM决定调用: {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_ascii=False) }) # 第六步：LLM生成最终回答 final_response = client.chat.completions.create( model="gpt-4o", messages=messages, tools=tools ) 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( model="gpt-4o", messages=[{"role": "user", "content": user_message}], tools=tools, tool_choice="auto" ) 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_ascii=False) }) # 最终回答 final = await async_client.chat.completions.create( model="gpt-4o", messages=messages ) 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_ascii=False) if tool_name == "search_knowledge_base": if not result or len(result) == 0: return json.dumps({ "status": "no_results", "message": "知识库中没有找到相关内容", "suggestion": "可以建议用户联系人工客服或重新描述问题" }, ensure_ascii=False) # 格式化搜索结果，让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_ascii=False) return json.dumps(result, ensure_ascii=False)—## 实战：构建完整的工具调用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( model="gpt-4o", messages=messages, tools=self.tools, tool_choice="auto" ) 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"[第{iteration+1}轮] 调用工具: {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"的关键一步。
