手写 AI Agent 工具调用系统：从零构建 Function Calling 执行引擎

一、为什么需要手写 Function Calling当你用 LangChain 或 Semantic Kernel 调用工具时有没有想过背后发生了什么# LangChain 的魔法 agent.run(查询北京的天气) # 然后... 奇迹般地调用了天气 API这个然后之间其实隐藏了一套完整的Function Calling 执行引擎。它要做的事远比表面看起来复杂函数注册——把 Python 函数描述成 LLM 能理解的 Schema意图识别——LLM 从用户问题中判断需要调用哪个函数参数提取——LLM 生成符合函数签名的 JSON 参数执行调用——执行函数并捕获结果结果回传——把函数结果送回 LLM 继续推理OpenAI 在 2023 年 6 月推出了 Function Calling API让 LLM 能主动请求调用预定义的函数。但很多人只用了皮毛——用 LangChain 的tool装饰器就完事了。一旦遇到复杂场景并行调用、链式调用、错误恢复第三方框架的抽象反而成了障碍。从零手写的好处- 理解每个环节的内部机制- 能针对自己的场景深度定制- 不依赖框架版本更新- 调试时知道问题出在哪一层本文的目标写一个完整可用的 Function Calling 执行引擎能注册函数、调用 LLM、执行函数、回传结果形成一个自动化的Agent 工具调用循环。二、核心架构Function Calling 的四个阶段一个完整的 Function Calling 流程分四个阶段用户输入 → [1. 函数选择] → [2. 参数生成] → [3. 函数执行] → [4. 结果处理] ↑ │ └──── 循环调用 ──────┘2.1 函数选择LLM 根据用户的输入和已注册的函数列表判断需要调用哪个函数。这需要把函数的签名、参数描述、返回值等信息以固定的 Schema 格式传给 LLM。OpenAI 的格式如下tools [ { type: function, function: { name: get_weather, description: 获取指定城市的天气信息, parameters: { type: object, properties: { city: { type: string, description: 城市名称 } }, required: [city] } } } ]2.2 参数生成LLM 返回的不是函数调用结果而是参数 JSON。如果 LLM 决定调用函数响应中会包含tool_calls字段{ tool_calls: [{ id: call_xxx, type: function, function: { name: get_weather, arguments: {\city\: \北京\} } }] }2.3 函数执行拿到参数后我们执行对应的 Python 函数获取返回值。2.4 结果处理把函数执行结果作为新的消息追加到对话上下文让 LLM 基于结果继续推理。这一步至关重要——函数结果必须和 LLM 的输出格式对齐否则模型会困惑。三、函数注册系统把 Python 函数变成 Tool Schema第一步是实现一个函数注册器能从 Python 函数自动生成 LLM 可理解的 Schema。3.1 基础函数描述器import inspect import json from typing import get_type_hints, Any class FunctionDescriptor: 把 Python 函数转成 OpenAI Tool Schema def __init__(self, fn): self.fn fn self.name fn.__name__ self.description (fn.__doc__ or ).strip() self.schema self._build_schema() def _build_schema(self): sig inspect.signature(self.fn) hints get_type_hints(self.fn) properties {} required [] for name, param in sig.parameters.items(): if name return: continue param_type hints.get(name, str) json_type self._to_json_type(param_type) prop {type: json_type} # 提取参数注释中的描述 if param.annotation is not inspect.Parameter.empty: ann param.annotation if hasattr(ann, __metadata__): prop[description] ann.__metadata__[0] properties[name] prop if param.default is inspect.Parameter.empty: required.append(name) return { type: function, function: { name: self.name, description: self.description, parameters: { type: object, properties: properties, required: required } } } def _to_json_type(self, py_type): mapping { str: string, int: integer, float: number, bool: boolean, list: array, dict: object } return mapping.get(py_type, string) def execute(self, arguments: dict) - Any: 执行函数并返回结果 return self.fn(**arguments)3.2 注册中心class ToolRegistry: 工具注册中心管理所有可调用的函数 def __init__(self): self._tools: dict[str, FunctionDescriptor] {} def register(self, fnNone, *, nameNone, descriptionNone): 注册一个函数为可用工具 def decorator(func): descriptor FunctionDescriptor(func) tool_name name or func.__name__ if description: descriptor.description description descriptor.schema[function][description] description self._tools[tool_name] descriptor return func return decorator(fn) if fn else decorator def get_schemas(self) - list[dict]: 返回所有工具的 OpenAI Tool Schema return [t.schema for t in self._tools.values()] def execute(self, name: str, arguments: dict) - Any: 按名称执行工具 if name not in self._tools: raise ValueError(f未知工具: {name}) return self._tools[name].execute(arguments) def list_tools(self) - list[str]: return list(self._tools.keys())3.3 注册实际函数registry ToolRegistry() registry.register def get_weather(city: str) - str: 获取指定城市的当前天气 # 实际项目中这里会调天气 API weather_data { 北京: 晴25°C湿度40%, 上海: 多云28°C湿度65%, 深圳: 阵雨30°C湿度80% } return weather_data.get(city, f{city}的天气数据暂不可用) registry.register def calculate(expression: str) - str: 计算数学表达式的结果 try: # 安全计算——只允许数字和运算符 allowed set(0123456789-*/.() ) if not all(c in allowed for c in expression): return 错误表达式包含不允许的字符 result eval(expression, {__builtins__: {}}, {}) return str(result) except Exception as e: return f计算错误: {str(e)} registry.register def search_knowledge(query: str) - str: 搜索知识库获取相关信息 # 模拟知识库搜索 knowledge { Python: Python 是一种高级编程语言由 Guido van Rossum 创建于 1991 年, AI Agent: AI Agent 是能自主感知环境并采取行动以实现目标的智能体, Function Calling: Function Calling 是 LLM 调用预定义函数的能力 } results [v for k, v in knowledge.items() if query in k or query in v] return \n.join(results) if results else f未找到与{query}相关的信息 # 验证 print(json.dumps(registry.get_schemas(), indent2, ensure_asciiFalse))看看生成的 Schema 长什么样[ { type: function, function: { name: get_weather, description: 获取指定城市的当前天气, parameters: { type: object, properties: { city: {type: string} }, required: [city] } } }, { type: function, function: { name: calculate, description: 计算数学表达式的结果, parameters: { type: object, properties: { expression: {type: string} }, required: [expression] } } } ]四、LLM 调用层对接 Function Calling API有了工具 Schema下一步就是和 LLM 对话。我们实现一个调用层用 OpenAI 兼容的 API 格式进行交互。import requests import json class LLMClient: 对接 LLM API 的客户端支持 Function Calling def __init__(self, api_key: str, base_url: str https://api.openai.com/v1, model: str gpt-4o): self.api_key api_key self.base_url base_url.rstrip(/) self.model model def chat(self, messages: list[dict], tools: list[dict] None) - dict: 发送聊天请求支持 Function Calling headers { Authorization: fBearer {self.api_key}, Content-Type: application/json } payload { model: self.model, messages: messages } if tools: payload[tools] tools payload[tool_choice] auto resp requests.post( f{self.base_url}/chat/completions, headersheaders, jsonpayload, timeout60 ) resp.raise_for_status() return resp.json() def parse_tool_calls(self, response: dict) - list[dict]: 从 LLM 响应中提取工具调用 choice response[choices][0] message choice[message] if tool_calls not in message or not message[tool_calls]: return [] calls [] for tc in message[tool_calls]: calls.append({ id: tc[id], name: tc[function][name], arguments: json.loads(tc[function][arguments]), type: tc[type] }) return calls def get_content(self, response: dict) - str: 获取 LLM 响应的文本内容 return response[choices][0][message].get(content, )这个客户端支持四种主流接口OpenAI原生兼容 OpenAI 接口的 API 服务DeepSeek兼容 OpenAI 接口Azure OpenAI只要你用的服务商兼容 OpenAI 的 /chat/completions 接口代码无需修改。想知道 DeepSeek 在 Function Calling 场景下的实战坑点和调优经验推荐翻翻《DeepSeek 实操》这本书。五、Agent 执行引擎把一切串起来核心部分——Agent 执行引擎。它协调 LLM 和工具之间的交互支持多轮工具调用。import time class FunctionCallingAgent: Function Calling 执行引擎 def __init__(self, llm: LLMClient, registry: ToolRegistry, max_iterations: int 10): self.llm llm self.registry registry self.max_iterations max_iterations self.history: list[dict] [] def run(self, user_input: str, verbose: bool True) - str: 运行 Agent处理用户输入 self.history [{role: user, content: user_input}] for step in range(self.max_iterations): if verbose: print(f\n{*50}) print(f 第 {step1} 轮调用) print(f{*50}) # 调用 LLM response self.llm.chat( messagesself.history, toolsself.registry.get_schemas() ) # 提取助手回复 assistant_msg response[choices][0][message] self.history.append(assistant_msg) content assistant_msg.get(content, ) if content and verbose: print(f LLM: {content[:200]}{... if len(content) 200 else }) # 检查是否有工具调用 tool_calls self.llm.parse_tool_calls(response) if not tool_calls: # 没有工具调用说明 LLM 已经生成了最终回答 if verbose: print(✅ LLM 给出最终回答无需调用工具) return content or assistant_msg.get(content, ) # 执行工具调用 if verbose: print(f 需要调用 {len(tool_calls)} 个工具) for tc in tool_calls: tool_name tc[name] tool_args tc[arguments] if verbose: print(f → 调用 {tool_name}({json.dumps(tool_args, ensure_asciiFalse)})) try: result self.registry.execute(tool_name, tool_args) result_str str(result) if not isinstance(result, str) else result if verbose: print(f ✅ 结果: {result_str[:100]}{... if len(result_str) 100 else }) except Exception as e: result_str f执行错误: {str(e)} if verbose: print(f ❌ 错误: {result_str}) # 工具结果追加到对话历史 self.history.append({ role: tool, tool_call_id: tc[id], content: result_str }) # 达到最大迭代次数返回最后一次的回复 if verbose: print(f⚠️ 达到最大迭代次数 {self.max_iterations}) last_msg self.history[-1] if last_msg[role] assistant: return last_msg.get(content, ) or str(last_msg) return 处理超时请重试或简化问题 def get_history(self) - list[dict]: return self.history5.1 测试完整流程# 使用兼容 OpenAI 接口的 API 服务 llm LLMClient( api_keyyour-api-key-here, base_urlhttps://api.siliconflow.cn/v1, modelQwen/Qwen2.5-7B-Instruct ) agent FunctionCallingAgent(llm, registry) # 测试 1单工具调用 result agent.run(北京的天气怎么样) print(f\n最终回答:\n{result}) # 重置历史 agent FunctionCallingAgent(llm, registry) # 测试 2多工具链式调用 result agent.run(搜索一下什么是 AI Agent顺便告诉我今天的天气) print(f\n最终回答:\n{result})六、高级特性并行工具调用OpenAI 的 Function Calling 支持一次返回多个tool_calls这意味着 LLM 可以同时请求多个并行的工具调用。我们的引擎已经内置了对多个 tool_calls 的支持但默认是串行执行的。实现并行执行from concurrent.futures import ThreadPoolExecutor, as_completed class ParallelFunctionCallingAgent(FunctionCallingAgent): 支持并行执行多个工具调用的 Agent def __init__(self, llm: LLMClient, registry: ToolRegistry, max_iterations: int 10, max_parallel: int 5): super().__init__(llm, registry, max_iterations) self.max_parallel max_parallel def _execute_tool_calls(self, tool_calls: list[dict], verbose: bool) - list[dict]: 并行执行多个工具调用 results [] with ThreadPoolExecutor(max_workersself.max_parallel) as executor: future_map {} for tc in tool_calls: future executor.submit( self._execute_single_tool, tc, verbose ) future_map[future] tc for future in as_completed(future_map): results.append(future.result()) return results def _execute_single_tool(self, tc: dict, verbose: bool) - dict: tool_name tc[name] tool_args tc[arguments] if verbose: print(f → 调用 {tool_name}({json.dumps(tool_args, ensure_asciiFalse)})) try: result self.registry.execute(tool_name, tool_args) result_str str(result) if not isinstance(result, str) else result except Exception as e: result_str f执行错误: {str(e)} return { tool_call_id: tc[id], content: result_str }并行执行的区别串行: 工具A → 等待 → 工具B → 等待 → 结果汇总 并行: 工具A ─┐ ├─→ 同时执行 → 结果汇总 工具B ─┘对于 I/O 密集型的工具调用调 API、查数据库、读文件并行执行可以显著降低总延迟。七、错误处理与重试机制生产环境中的 Function Calling 会遇到各种问题LLM 生成的参数格式错误——JSON 解析失败工具内部抛出异常——API 超时、数据库连接失败LLM 幻觉调用——调用了不存在的函数名无限循环——Agent 不断调用工具7.1 健壮的工具执行器import traceback class RobustToolExecutor: 带有错误处理、重试和超时的工具执行器 def __init__(self, registry: ToolRegistry, max_retries: int 2): self.registry registry self.max_retries max_retries def execute(self, name: str, arguments: dict, timeout: int 10) - dict: 执行工具返回标准化的结果 返回格式: {success: bool, result: str, error: str or None} # 1. 检查工具是否存在 if name not in self.registry.list_tools(): return { success: False, result: None, error: f未知工具 {name}可用工具: {, .join(self.registry.list_tools())} } # 2. 参数校验 try: tool self.registry._tools[name] # 检查必需参数 for param_name in tool.schema[function][parameters].get(required, []): if param_name not in arguments: return { success: False, result: None, error: f缺少必需参数: {param_name} } except Exception as e: return { success: False, result: None, error: f参数校验失败: {str(e)} } # 3. 带重试的执行 last_error None for attempt in range(self.max_retries 1): try: result self.registry.execute(name, arguments) return { success: True, result: result, error: None } except Exception as e: last_error str(e) if attempt self.max_retries: time.sleep(1 * (attempt 1)) # 退避等待 continue return { success: False, result: None, error: f执行失败 (重试{self.max_retries}次): {last_error} }7.2 循环检测class CyclicCallDetector: 检测工具调用的循环模式 def __init__(self, max_same_tool_calls: int 3): self.call_history: list[tuple[str, str]] [] self.max_same_tool_calls max_same_tool_calls def record_call(self, tool_name: str, arguments: dict): self.call_history.append((tool_name, json.dumps(arguments, sort_keysTrue))) def is_cyclic(self) - bool: 检测是否出现循环调用 if len(self.call_history) 2: return False # 检查最近 N 次调用是否完全相同 recent self.call_history[-self.max_same_tool_calls:] if len(recent) self.max_same_tool_calls: return False first recent[0] return all(c first for c in recent[1:])八、与主流框架对比特性我们的引擎LangChainAutoGen代码量~300行依赖整个框架依赖整个框架可定制性★★★★★★★★★★★学习成本低全可见高抽象层多中并行调用原生支持需额外配置支持错误恢复手动实现内置内置多 Agent不支持支持原生依赖requestslangchain-corepyautogen什么时候用自己的引擎- 项目只需要简单的工具调用- 想深入理解内部机制- 需要深度定制调用逻辑- 框架版本升级导致代码出问题什么时候用框架- 需要多 Agent 协作- 需要记忆、规划等高级能力- 团队对框架已有积累- 需要内置的监控和日志系统九、完整代码与使用示例下面是一个可以直接运行的完整示例import inspect import json import requests from typing import get_type_hints, Any # 1. 函数注册 class FunctionDescriptor: def __init__(self, fn): self.fn fn self.name fn.__name__ self.description (fn.__doc__ or ).strip() self.schema self._build_schema() def _build_schema(self): sig inspect.signature(self.fn) hints get_type_hints(self.fn) properties {} required [] for name, param in sig.parameters.items(): if name return: continue param_type hints.get(name, str) json_type {str: string, int: integer, float: number, bool: boolean, list: array, dict: object}.get(param_type, string) properties[name] {type: json_type} if param.default is inspect.Parameter.empty: required.append(name) return { type: function, function: { name: self.name, description: self.description, parameters: {type: object, properties: properties, required: required} } } def execute(self, arguments): return self.fn(**arguments) class ToolRegistry: def __init__(self): self._tools {} def register(self, fnNone, *, nameNone): def decorator(func): descriptor FunctionDescriptor(func) tool_name name or func.__name__ self._tools[tool_name] descriptor return func return decorator(fn) if fn else decorator def get_schemas(self): return [t.schema for t in self._tools.values()] def execute(self, name, arguments): if name not in self._tools: raise ValueError(f未知工具: {name}) return self._tools[name].execute(arguments) def list_tools(self): return list(self._tools.keys()) # 2. 注册工具 registry ToolRegistry() registry.register def get_weather(city: str) - str: 获取指定城市的当前天气 data {北京: 晴25°C, 上海: 多云28°C, 深圳: 阵雨30°C} return data.get(city, f{city}的天气数据暂不可用) registry.register def calculate(expression: str) - str: 计算数学表达式 allowed set(0123456789-*/.() ) if not all(c in allowed for c in expression): return 错误表达式包含不允许的字符 try: return str(eval(expression, {__builtins__: {}}, {})) except Exception as e: return f计算错误: {str(e)} # 3. LLM 客户端 class LLMClient: def __init__(self, api_key, base_urlhttps://api.openai.com/v1, modelgpt-4o): self.api_key api_key self.base_url base_url.rstrip(/) self.model model def chat(self, messages, toolsNone): headers {Authorization: fBearer {self.api_key}, Content-Type: application/json} payload {model: self.model, messages: messages} if tools: payload[tools] tools payload[tool_choice] auto resp requests.post(f{self.base_url}/chat/completions, headersheaders, jsonpayload, timeout60) resp.raise_for_status() return resp.json() def parse_tool_calls(self, response): message response[choices][0][message] if tool_calls not in message: return [] return [{id: tc[id], name: tc[function][name], arguments: json.loads(tc[function][arguments])} for tc in message[tool_calls]] # 4. Agent 执行引擎 class FunctionCallingAgent: def __init__(self, llm, registry, max_iterations10): self.llm llm self.registry registry self.max_iterations max_iterations def run(self, user_input, verboseTrue): history [{role: user, content: user_input}] for step in range(self.max_iterations): response self.llm.chat(history, self.registry.get_schemas()) msg response[choices][0][message] history.append(msg) tool_calls self.llm.parse_tool_calls(response) if not tool_calls: return msg.get(content, ) if verbose: print(f\n 第 {step1} 轮: 调用 {len(tool_calls)} 个工具) for tc in tool_calls: result self.registry.execute(tc[name], tc[arguments]) history.append({role: tool, tool_call_id: tc[id], content: str(result)}) if verbose: print(f → {tc[name]} → {str(result)[:60]}) return 达到最大迭代次数 # 5. 运行演示 if __name__ __main__: llm LLMClient( api_keyyour-api-key, base_urlhttps://api.siliconflow.cn/v1, modelQwen/Qwen2.5-7B-Instruct ) agent FunctionCallingAgent(llm, registry) result agent.run(计算 (2537)*2 等于多少) print(f\n最终回答: {result})实际跑一遍下来会发现Function Calling 没框架包装的那么玄乎。LLM 不执行代码它只是填参数表。真正的执行还是你注册的那些 Python 函数。几个关键心得- Schema 生成别偷懒type、description、required一个都不能少——LLM 很依赖这些信息做决策- 循环检测不能省模型偶尔会陷入工具死循环- 并行调用不是必须的但并发 IO 密集型 API 时提速很明显- 错误处理要前置——工具跑挂了别硬抛把错误消息送回 LLM 让它修正