Function Calling的工程化封装：工具注册、参数校验与错误恢复

Function Calling的工程化封装工具注册、参数校验与错误恢复一、当Function Calling在真实场景中不断失败Function Calling是LLM应用从对话式AI升级为执行式Agent的关键能力。但在实际落地中大多数团队的实现停留在把API文档贴给LLM的阶段结果就是模型生成了格式错误的JSON参数导致工具调用失败工具执行超时或返回异常Agent陷入沉默或循环多工具调用场景下工具间的依赖关系未处理出现先调用了需要使用上一个工具结果的函数的情况调试困难——当用户反馈AI没有执行操作时无法确定是Prompt问题、模型幻觉还是工具本身的Bug一项对72个开源Agent项目的代码分析显示只有23%的项目包含完整的工具参数校验逻辑不到12%实现了工具调用的错误恢复机制。这意味着绝大多数Agent在工具调用失败时没有兜底策略。工程化的Function Calling封装不是简单地在Prompt里加上工具描述而是一整套围绕工具的注册、校验、执行和恢复体系。这套体系的价值不在于单次调用的成功而在于让Agent在出错时有可预期的行为。二、底层机制与原理剖析2.1 Function Calling的执行链路一次完整的Function Calling涉及四个角色用户请求、LLM推理、工具执行引擎、工具函数本身。其中工具执行引擎是工程化的核心——它负责将LLM生成的半结构化文本转换为可安全执行的函数调用。sequenceDiagram participant U as 用户请求 participant A as Agent核心 participant L as LLM participant E as 工具执行引擎 participant T1 as 工具A (搜索) participant T2 as 工具B (计算) U-A: 帮我查今天天气并换算华氏度 A-L: System Prompt 工具列表 用户请求 L--A: ToolCall: search_weather(city北京) A-E: 解析ToolCall E-E: 参数校验 (Pydantic) alt 校验通过 E-T1: 调用工具A T1--E: {temp_c: 30} E-E: 结果校验与格式化 E--A: 格式化结果 else 校验失败 E--A: 错误信息 修复建议 A-L: 带修复建议的重试Prompt end A-L: 工具结果 继续推理 L--A: ToolCall: convert_temp(celsius30, tofahrenheit) A-E: 调用工具B E-T2: 执行计算 T2--E: {temp_f: 86} E-E: 结果格式化并检查是否超最大调用次数 E--A: 结果 A-L: 生成最终回答 L--A: 北京今天30°C折合86°F A--U: 最终回复2.2 工具注册的双重Schema设计工具的Schema需要同时服务于两个消费者LLM需要自然语言的描述和执行引擎需要结构化的类型定义。这意味着每个工具至少需要两层定义。第一层是面向LLM的Function Description用自然语言描述工具的功能、参数含义、使用场景和限制。LLM通过这些描述来判断何时应该调用哪个工具。关键原则是描述越精确误调用越少——搜索最新新闻优于获取信息计算两个日期间的天数差优于处理日期。第二层是面向执行引擎的Parameter Schema使用JSON Schema或Pydantic Model严格定义参数类型、格式约束和默认值。执行引擎在收到LLM的工具调用请求后先用此Schema校验参数不通过时直接拦截并返回修复建议。2.3 错误恢复的三级策略工具调用失败时需要分级处理而非统一返回错误信息参数错误可自动修复如类型不匹配、必填参数缺失。策略是用错误信息重新构造Prompt让LLM修复参数后重试。工具执行失败可能可重试如网络超时、API限流。策略是指数退避重试最多3次。业务逻辑错误不可自动修复如查无结果、权限不足。策略是将错误信息返回给LLM让它决定是换一个工具还是告知用户。三、生产级代码实现与最佳实践3.1 工具的声明式注册与Schema定义# tool_registry.py — 工具的声明式注册引擎 # # 设计原则 # 1. 每个工具是独立的、可测试的函数 # 2. Schema声明代码与业务逻辑分离但近距共存 # 3. 支持工具的分组和权限控制 # 4. 注册时自动生成JSON Schema供LLM使用 from __future__ import annotations import inspect from typing import Any, Callable, Optional, TypeAlias from dataclasses import dataclass, field from pydantic import BaseModel, Field, create_model import json # Pydantic模型作为工具的输入参数Schema # 这比手写JSON Schema更类型安全且自动生成校验逻辑 class SearchWeatherParams(BaseModel): 天气查询工具的参数Schema。 description中的内容会被用作LLM的Function Description的一部分。 Field的description会直接出现在LLM的tool definition中。 city: str Field( ..., description城市名称如北京、上海。支持中文或拼音, min_length1, max_length50, ) date: Optional[str] Field( defaultNone, description查询日期格式YYYY-MM-DD。不填则查询今天, patternr^\d{4}-\d{2}-\d{2}$, ) class SendEmailParams(BaseModel): 发送邮件工具的参数Schema。 to: str Field( ..., description收件人邮箱地址, patternr^[a-zA-Z0-9._%-][a-zA-Z0-9.-]\.[a-zA-Z]{2,}$, ) subject: str Field( ..., description邮件主题, min_length1, max_length200, ) body: str Field( ..., description邮件正文内容, min_length1, ) dataclass class ToolDefinition: 工具的完整定义。 包含 - 函数引用实际执行的Python函数 - 参数模型Pydantic模型用于校验和生成Schema - LLM描述面向LLM的自然语言描述 - 元数据分组、权限、超时设置等 name: str description: str # 面向LLM的自然语言描述 func: Callable params_model: type[BaseModel] category: str general # 工具分类分组管理 timeout_seconds: float 30.0 max_retries: int 2 # 执行失败时的最大重试次数 requires_confirmation: bool False # 是否需要用户确认才执行 tags: list[str] field(default_factorylist) def to_openai_tool(self) - dict: 转换为OpenAI Function Calling格式。 这是生成LLM所需JSON的关键方法。 为什么在这里生成而非在调用处 ——确保LLM看到的描述和实际参数Schema严格一致。 schema self.params_model.model_json_schema() return { type: function, function: { name: self.name, description: self.description, parameters: { type: object, properties: schema.get(properties, {}), required: schema.get(required, []), additionalProperties: False, }, }, } class ToolRegistry: 工具的注册和管理中心。 使用方式 1. 先通过装饰器或手动调用 register() 注册工具 2. 调用 get_openai_tools() 获取所有工具的OpenAI格式 3. 调用 execute() 执行指定的工具 def __init__(self): self._tools: dict[str, ToolDefinition] {} def register(self, definition: ToolDefinition): 注册一个工具。 重复注册同名的工具会抛出异常——这是设计意图。 在Agent中每个工具名必须是唯一的以防止调用歧义。 if definition.name in self._tools: raise ValueError( f工具 {definition.name} 已注册。 f已有注册: {self._tools[definition.name].description[:50]}... ) self._tools[definition.name] definition def tool( self, name: str, description: str, params_model: type[BaseModel], category: str general, timeout_seconds: float 30.0, max_retries: int 2, ): 装饰器将函数注册为工具。 使用示例 registry.tool( namesearch_weather, description查询指定城市的天气信息, params_modelSearchWeatherParams, ) async def search_weather(params: SearchWeatherParams) - dict: ... def decorator(func: Callable): self.register(ToolDefinition( namename, descriptiondescription, funcfunc, params_modelparams_model, categorycategory, timeout_secondstimeout_seconds, max_retriesmax_retries, )) return func return decorator def get_openai_tools(self, categories: Optional[list[str]] None) - list[dict]: 获取所有注册工具的OpenAI Function Calling格式列表。 Args: categories: 可选筛选指定分类的工具。 为什么需要分类筛选 - 不同Agent角色可能需要不同的工具集 - 减少Token消耗不需要把所有工具都发给LLM tools self._tools.values() if categories: tools [t for t in tools if t.category in categories] return [t.to_openai_tool() for t in tools] def get_tool(self, name: str) - Optional[ToolDefinition]: 根据名称获取工具定义。 return self._tools.get(name) # 全局注册中心 registry ToolRegistry()3.2 工具执行引擎的完整实现# tool_executor.py — 工具执行引擎 # # 核心职责 # 1. 接收LLM生成的ToolCall解析并校验参数 # 2. 执行工具捕获所有异常 # 3. 对失败的工具调用实施分级恢复策略 # 4. 返回结构化结果包含执行状态和错误信息 import asyncio import time import logging from typing import Any, Optional from dataclasses import dataclass, field from enum import Enum from pydantic import ValidationError logger logging.getLogger(__name__) class ToolCallStatus(Enum): 工具调用的执行状态。 每种状态都对应不同的恢复策略 - SUCCESS: 正常返回结果 - VALIDATION_ERROR: 参数不符合Schema需LLM修复参数后重试 - EXECUTION_ERROR: 工具内部错误需引擎自动重试 - TIMEOUT: 工具执行超时需引擎自动重试 - BUSINESS_ERROR: 业务逻辑错误如查无结果直接返回信息给用户 SUCCESS success VALIDATION_ERROR validation_error EXECUTION_ERROR execution_error TIMEOUT timeout BUSINESS_ERROR business_error dataclass class ToolCallResult: 工具调用的结构化结果。 设计原则 - status明确标识结果类型调用方无需解析错误字符串 - error_message是人类可读的错误描述可直接展示给用户或LLM - retry_hint是可选的修复建议用来帮LLM修正参数 tool_name: str status: ToolCallStatus data: Any None error_message: str retry_hint: str # 提示LLM如何修正参数 execution_time_ms: float 0.0 retry_count: int 0 class ToolExecutor: 工具执行引擎。 执行流程 1. 找到对应的工具定义 2. 用Pydantic校验参数 3. 执行工具函数带超时和重试 4. 包装返回结果为ToolCallResult def __init__(self, registry: ToolRegistry): self.registry registry async def execute( self, tool_name: str, arguments: dict[str, Any], max_retries: Optional[int] None, ) - ToolCallResult: 执行一个工具调用。 Args: tool_name: 工具名称对应注册时的name arguments: LLM生成的参数字典 max_retries: 重试次数不传则使用工具定义中的值 为什么用async - 工具可能涉及IO操作API调用、数据库查询 - 多个工具调用时可以通过asyncio.gather并行执行 start_time time.monotonic() # 第1步查找工具 tool self.registry.get_tool(tool_name) if tool is None: return ToolCallResult( tool_nametool_name, statusToolCallStatus.EXECUTION_ERROR, error_messagef未注册的工具: {tool_name}, ) # 第2步参数校验 try: # Pydantic自动完成类型转换如字符串3转为整数3 # 和约束校验长度、正则、范围等 params tool.params_model(**arguments) except ValidationError as e: # 构造修复提示帮助LLM在下一次调用中修正参数 hints [] for error in e.errors(): field ..join(str(loc) for loc in error[loc]) hints.append(f参数 {field}: {error[msg]}) return ToolCallResult( tool_nametool_name, statusToolCallStatus.VALIDATION_ERROR, error_messagef参数校验失败: {; .join(hints)}, retry_hintf请修正以下参数: {; .join(hints)}, execution_time_ms(time.monotonic() - start_time) * 1000, ) # 第3步执行工具带重试逻辑 retries max_retries if max_retries is not None else tool.max_retries last_error: Optional[Exception] None for attempt in range(retries 1): try: # 使用asyncio.wait_for实现超时控制 # 防止工具无限等待如网络请求hang住 result await asyncio.wait_for( self._call_tool(tool.func, params), timeouttool.timeout_seconds, ) return ToolCallResult( tool_nametool_name, statusToolCallStatus.SUCCESS, dataresult, execution_time_ms(time.monotonic() - start_time) * 1000, retry_countattempt, ) except asyncio.TimeoutError: last_error TimeoutError( f工具 {tool_name} 执行超时 ({tool.timeout_seconds}s) ) logger.warning( fTool timeout (attempt {attempt 1}/{retries 1}): {tool_name} ) # 指数退避第1次等1秒第2次等2秒第3次等4秒 if attempt retries: await asyncio.sleep(2 ** attempt) except Exception as e: last_error e logger.warning( fTool execution error (attempt {attempt 1}/{retries 1}): f{tool_name} - {e} ) if attempt retries: await asyncio.sleep(2 ** attempt) # 所有重试都失败 return ToolCallResult( tool_nametool_name, statusToolCallStatus.EXECUTION_ERROR, error_messagef工具执行失败已重试{retries}次: {last_error}, execution_time_ms(time.monotonic() - start_time) * 1000, retry_countretries, ) async def _call_tool(self, func: Callable, params: BaseModel) - Any: 实际调用工具函数。 支持同步和异步函数。 检查是否为协程函数来决定调用方式。 if inspect.iscoroutinefunction(func): return await func(params) else: # 同步函数在线程池中执行避免阻塞事件循环 return await asyncio.get_event_loop().run_in_executor( None, func, params )3.3 Agent核心——多工具调用的编排# agent_core.py — Agent核心的工具调用循环 # # 实现ReAct模式的工具调用循环 # 1. 用户请求 → LLM推理 → 工具调用决策 # 2. 执行工具 → 结果返回LLM → 继续推理或生成最终回复 # 3. 循环直到LLM决定不再需要工具调用或达到最大迭代次数 import json from typing import Any, Optional from dataclasses import dataclass, field from openai import AsyncOpenAI dataclass class AgentConfig: Agent的配置参数。 max_tool_iterations: int 5 # 最大工具调用轮次防止无限循环 system_prompt: str 你是一个智能助手可以使用工具来完成用户的请求。 model: str gpt-4o-mini temperature: float 0.2 # Function Calling场景推荐较低温度 class Agent: 支持Function Calling的Agent实现。 核心循环 while iteration max_iterations: response llm.chat(messages, tools) if response has no tool_calls: return response.content # 最终回复 for each tool_call in response.tool_calls: result tool_executor.execute(tool_call) messages.append(tool_call_result) iteration 1 def __init__( self, client: AsyncOpenAI, tool_executor: ToolExecutor, config: AgentConfig, ): self.client client self.tool_executor tool_executor self.config config async def run(self, user_message: str) - str: 执行Agent的主循环。 messages: list[dict] [ {role: system, content: self.config.system_prompt}, {role: user, content: user_message}, ] tools tool_registry.get_openai_tools() for iteration in range(self.config.max_tool_iterations): # 调用LLM response await self.client.chat.completions.create( modelself.config.model, messagesmessages, toolstools if tools else None, temperatureself.config.temperature, ) choice response.choices[0] # 如果LLM决定不需要工具直接返回文本回复 if choice.finish_reason stop: return choice.message.content or # 处理工具调用 if choice.message.tool_calls: # 将Assistant的消息含tool_calls加入上下文 messages.append({ role: assistant, content: choice.message.content, tool_calls: [ { id: tc.id, type: function, function: { name: tc.function.name, arguments: tc.function.arguments, }, } for tc in choice.message.tool_calls ], }) # 执行每个工具调用 for tool_call in choice.message.tool_calls: func_name tool_call.function.name # 安全地解析JSON参数 # LLM可能生成格式有问题的JSON try: arguments json.loads(tool_call.function.arguments) except json.JSONDecodeError: messages.append({ role: tool, tool_call_id: tool_call.id, content: json.dumps({ error: 参数格式错误必须是有效的JSON, raw: tool_call.function.arguments[:200], }, ensure_asciiFalse), }) continue # 执行工具 result await self.tool_executor.execute(func_name, arguments) # 根据执行状态生成反馈消息 if result.status ToolCallStatus.VALIDATION_ERROR: content json.dumps({ error: result.error_message, hint: result.retry_hint, }, ensure_asciiFalse) elif result.status ToolCallStatus.EXECUTION_ERROR: content json.dumps({ error: result.error_message, }, ensure_asciiFalse) else: content json.dumps(result.data, ensure_asciiFalse) # 将工具结果加入消息历史 messages.append({ role: tool, tool_call_id: tool_call.id, content: content, }) # 达到最大迭代次数强制让LLM基于已有信息给出回复 messages.append({ role: user, content: 请基于目前已有的信息给出你的最佳回答不要再调用工具。, }) final await self.client.chat.completions.create( modelself.config.model, messagesmessages, temperatureself.config.temperature, ) return final.choices[0].message.content or 抱歉无法完成该请求。四、边界分析与架构权衡4.1 适用场景需要集成外部能力的LLM应用如搜索、计算、数据库查询、发送通知等多工具协作场景需要Agent自主决定调用顺序和工具组合生产环境部署需要完整的校验、重试、日志和监控4.2 不适用或需简化的场景单工具固定流程如始终是先搜索再总结用程序化的管道比Agent循环更可靠对延迟极度敏感的应用每次工具调用增加1至3秒延迟多轮调用可能累积到5至10秒安全性要求极高的场景如金融交易、医疗诊断不应让LLM自主决定工具调用4.3 关键设计权衡Pydantic vs 手写JSON Schema本方案选用Pydantic定义参数Schema。优点是类型安全、校验自动生成、与Python生态集成好。代价是启动时需要额外的模型解析时间。对于极简场景手写JSON Schema更轻量。同步重试 vs 异步通知工具执行失败时当前方案是同步重试最多3次。优点是简化了状态管理。对于需要更长时间恢复的失败如第三方API维护异步重试配合回调才是正确的方案。单个Agent vs 多Agent路由本方案将工具注册和执行耦合在一个Agent中。当工具数量超过20个时应该考虑引入工具路由——按领域将工具分组先用一个路由Agent选择工具组再由领域Agent执行具体工具。五、总结Function Calling的工程化需要工具注册、参数校验、执行引擎和错误恢复四个核心组件Pydantic Model作为参数Schema既能为LLM生成描述又能提供运行时校验工具执行失败应分三级处理参数错误提示LLM修复、执行异常自动重试、业务错误信息返回Agent循环应设置最大迭代次数推荐5轮防止无限循环消耗Token和延迟超过20个工具时应引入工具路由通过分组减少单次Prompt的Token开销对安全关键场景金融、医疗应禁止LLM自主工具调用改用程序化流程