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

一、为什么需要手写 Function Calling？

当你用 LangChain 或 Semantic Kernel 调用工具时，有没有想过背后发生了什么？

# LangChain 的魔法 agent.run("查询北京的天气") # 然后... 奇迹般地调用了天气 API

这个"然后"之间，其实隐藏了一套完整的Function Calling 执行引擎。它要做的事远比表面看起来复杂：

1. 函数注册——把 Python 函数描述成 LLM 能理解的 Schema
2. 意图识别——LLM 从用户问题中判断需要调用哪个函数
3. 参数提取——LLM 生成符合函数签名的 JSON 参数
4. 执行调用——执行函数并捕获结果
5. 结果回传——把函数结果送回 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, fn=None, *, name=None, description=None): """注册一个函数为可用工具""" 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(), indent=2, ensure_ascii=False))

看看生成的 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": f"Bearer {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", headers=headers, json=payload, timeout=60 ) 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"🔄 第 {step+1} 轮调用") print(f"{'='*50}") # 调用 LLM response = self.llm.chat( messages=self.history, tools=self.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_ascii=False)})") 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.history

5.1 测试完整流程

# 使用兼容 OpenAI 接口的 API 服务 llm = LLMClient( api_key="your-api-key-here", base_url="https://api.siliconflow.cn/v1", model="Qwen/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_workers=self.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_ascii=False)})") 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 会遇到各种问题：

1. LLM 生成的参数格式错误——JSON 解析失败
2. 工具内部抛出异常——API 超时、数据库连接失败
3. LLM 幻觉调用——调用了不存在的函数名
4. 无限循环——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_keys=True))) 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:])

八、与主流框架对比

什么时候用自己的引擎？
- 项目只需要简单的工具调用
- 想深入理解内部机制
- 需要深度定制调用逻辑
- 框架版本升级导致代码出问题

什么时候用框架？
- 需要多 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, fn=None, *, name=None): 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_url="https://api.openai.com/v1", model="gpt-4o"): self.api_key = api_key self.base_url = base_url.rstrip("/") self.model = model def chat(self, messages, tools=None): headers = {"Authorization": f"Bearer {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", headers=headers, json=payload, timeout=60) 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_iterations=10): self.llm = llm self.registry = registry self.max_iterations = max_iterations def run(self, user_input, verbose=True): 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🔄 第 {step+1} 轮: 调用 {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_key="your-api-key", base_url="https://api.siliconflow.cn/v1", model="Qwen/Qwen2.5-7B-Instruct" ) agent = FunctionCallingAgent(llm, registry) result = agent.run("计算 (25+37)*2 等于多少？") print(f"\n最终回答: {result}")

实际跑一遍下来会发现，Function Calling 没框架包装的那么玄乎。LLM 不执行代码，它只是填参数表。真正的执行还是你注册的那些 Python 函数。

几个关键心得：
- Schema 生成别偷懒，type、description、required一个都不能少——LLM 很依赖这些信息做决策
- 循环检测不能省，模型偶尔会陷入工具死循环
- 并行调用不是必须的，但并发 IO 密集型 API 时提速很明显
- 错误处理要前置——工具跑挂了别硬抛，把错误消息送回 LLM 让它修正