Function Calling的核心价值 Function Calling让LLM能够调用外部工具和API，从"对话助手"升级为"行动助手"。但实践中的挑战在于：如何让模型可靠地选择正确的工具、生成正确的参数、处理工具执行失败的情况。
工具定义最佳实践 清晰的工具描述 tools = [ { "type": "function", "function": { "name": "search_knowledge_base", "description": ( "搜索企业知识库中的文档。当用户询问公司政策、产品信息、" "技术文档等内部知识时使用此工具。\n" "注意：不要用于搜索公开互联网信息。" ), "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "搜索关键词，使用自然语言描述要查找的内容" }, "department": { "type": "string", "enum": ["engineering", "sales", "hr", "finance"], "description": "限定搜索的部门范围，不指定则搜索全部" }, "limit": { "type": "integer", "description": "返回结果数量，默认5", "default": 5, "minimum": 1, "maximum": 20 } }, "required": ["query"] } } } ] 工具选择的引导 # 好的描述：明确什么场景用，什么场景不用 "description": "获取当前天气信息。当用户询问天气状况、温度、降水概率时使用。不要用于历史天气查询。" # 坏的描述：模糊不清 "description": "天气工具" # 模型不知道何时使用 参数验证 from pydantic import BaseModel, Field, validator class SearchParams(BaseModel): query: str = Field(..., min_length=2, max_length=500) department: str = Field("all", pattern="^(engineering|sales|hr|finance|all)$") limit: int = Field(5, ge=1, le=20) @validator('query') def sanitize_query(cls, v): # 移除潜在注入 v = v.replace('\n', ' ').replace('\r', ' ') return v.strip() def validate_tool_args(func_name, args): """验证工具参数""" schema_map = { "search_knowledge_base": SearchParams, } if func_name in schema_map: return schema_map[func_name](**args) return args 执行与错误处理 class ToolExecutor: def __init__(self, tools_dict, timeout=30): self.tools = tools_dict self.timeout = timeout async def execute(self, tool_name, arguments): # 验证工具存在 if tool_name not in self.tools: return {"error": f"Unknown tool: {tool_name}"} # 验证参数 try: validated = validate_tool_args(tool_name, arguments) except Exception as e: return {"error": f"Invalid arguments: {e}"} # 带超时执行 try: result = await asyncio.wait_for( self.tools[tool_name](**validated.dict()), timeout=self.timeout ) return {"result": result} except asyncio.TimeoutError: return {"error": f"Tool execution timed out after {self.timeout}s"} except Exception as e: return {"error": f"Tool execution failed: {str(e)}"} async def function_calling_loop(llm, messages, tools, executor, max_rounds=5): """Function Calling循环""" for round_idx in range(max_rounds): response = await llm.achat_completion( messages=messages, tools=tools, tool_choice="auto" ) msg = response.choices[0].message messages.append(msg) # 如果模型没有调用工具，返回最终回答 if not msg.tool_calls: return msg.content # 执行所有工具调用 for tool_call in msg.tool_calls: result = await executor.execute( tool_call.function.name, json.loads(tool_call.function.arguments) ) messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": json.dumps(result, ensure_ascii=False) }) return "达到最大工具调用轮次限制。" 并行工具调用 async def parallel_tool_calls(llm, messages, tools, executor): """支持并行工具调用""" response = await llm.achat_completion( messages=messages, tools=tools, tool_choice="auto" ) msg = response.choices[0].message messages.append(msg) if msg.tool_calls: # 并行执行所有工具调用 tasks = [] for tool_call in msg.tool_calls: task = executor.execute( tool_call.function.name, json.loads(tool_call.function.arguments) ) tasks.append((tool_call.id, task)) results = await asyncio.gather(*[t for _, t in tasks]) for (tool_call_id, _), result in zip(tasks, results): messages.append({ "role": "tool", "tool_call_id": tool_call_id, "content": json.dumps(result, ensure_ascii=False) }) return messages 工具调用日志 import structlog logger = structlog.get_logger() class LoggingToolExecutor: def __init__(self, executor): self.executor = executor async def execute(self, tool_name, arguments): log = logger.bind(tool=tool_name) log.info("tool_call_start", args=arguments) start = time.time() result = await self.executor.execute(tool_name, arguments) duration = time.time() - start if "error" in result: log.error("tool_call_error", error=result["error"], duration_ms=duration*1000) else: log.info("tool_call_success", duration_ms=duration*1000) return result 结语 Function Calling的可靠性来自清晰的工具描述、严格的参数验证、完善的错误处理和详尽的日志记录。将这些实践标准化，可以显著提升LLM Agent在生产环境中的稳定性。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。
...
