手写Tool Use：从面试原型到生产级智能体工具调用框架

1. 面试现场的真实压力：当“手写 Tool Use”成为临场能力试金石

“请手写一个 Tool Use 实现。”

这句话从面试官嘴里说出来的时候，我正端着一次性纸杯喝第三口温水——不是因为紧张，而是刚结束上一轮系统设计题，嗓子发干。但这句话一落，手指下意识在笔记本边缘划了两道浅痕。不是没准备过 Agent 架构、不是没调过 function calling 接口，可“手写”两个字像块冷铁压下来：不许 copy-paste，不许查文档，不许调 SDK，就用最基础的 Python（或你声称最熟的语言），把工具调用的骨架、路由逻辑、参数校验、错误兜底全摊开写清楚。

这根本不是考你会不会用 OpenAI 的tools字段，而是考你脑子里有没有一套可落地的 Tool Use 模型。它背后藏着三层隐性考察点：第一层是抽象能力——能否把“调用天气 API”“查数据库”“执行 shell 命令”这些异构操作，统一抽象成“工具注册-发现-解析-执行-返回”的闭环；第二层是工程边界感——知道哪些该由框架兜底（如序列化）、哪些必须自己控制（如参数类型强校验、超时熔断）；第三层是防御性思维——当用户输入“帮我把 /tmp 下所有 .log 文件压缩成 zip”，你写的代码能不能拦住路径遍历、空结果、权限拒绝、命令注入这四类典型风险。

我后来复盘过十几份真实面经，发现“手写 Tool Use”高频出现在三类岗位终面：大模型应用层工程师（偏落地）、智能体平台开发岗（偏基建）、以及某些对安全要求极高的垂类 Agent 项目组。他们不要你背诵 LlamaIndex 的 ToolNode 源码，但要你证明：当脱离 LangChain/LlamaIndex 这类胶水层后，你依然能徒手搭出一条干净、可控、可审计的工具调用链。关键词里没填内容？恰恰说明这是个“默认共识题”——就像考排序算法不会特意标出“快排/归并”，Tool Use 的核心要素早已沉淀为行业隐性标准：工具注册表、意图识别器、参数解析器、执行沙箱、结果标准化器。接下来我要拆解的三种方案，不是炫技，而是对应三种不同成熟度的工程判断：从最小可行原型，到生产级鲁棒实现，再到可扩展架构演进。每一种，我都带你在白板上真正“写”出来。

2. 方案一：极简原型版——37 行代码跑通核心闭环

先抛开日志、重试、监控这些“锦上添花”的东西。面试场景下，第一目标是5 分钟内让工具调用动起来，且逻辑无硬伤。我管这叫“心脏起搏器版本”——只要心跳（调用成功）和呼吸（返回结构）正常，其他都可以后续插拔。

核心就四步：注册工具 → 解析用户指令 → 匹配并执行 → 标准化返回。我们用 Python 写，不依赖任何第三方库，连json都只用内置模块。

import json import re from typing import Dict, Any, Callable, Optional class SimpleToolUse: def __init__(self): self.tools: Dict[str, Dict[str, Any]] = {} def register(self, name: str, func: Callable, description: str, parameters: Dict[str, str]) -> None: """注册工具：名称、函数引用、描述、参数定义（仅类型提示）""" self.tools[name] = { "func": func, "description": description, "parameters": parameters # {"city": "string", "unit": "string"} } def parse_intent(self, user_input: str) -> Optional[Dict[str, Any]]: """用正则粗筛意图，实际项目中应替换为 LLM 分类或规则引擎""" # 示例：匹配 "查北京天气" -> {"tool": "get_weather", "args": {"city": "北京"}} match = re.search(r'查(.+?)天气', user_input) if match: return {"tool": "get_weather", "args": {"city": match.group(1).strip()}} return None def execute(self, user_input: str) -> Dict[str, Any]: """主执行入口""" intent = self.parse_intent(user_input) if not intent or intent["tool"] not in self.tools: return {"error": "未识别工具或指令", "raw_input": user_input} tool_def = self.tools[intent["tool"]] try: result = tool_def["func"](**intent["args"]) return { "tool_name": intent["tool"], "result": result, "status": "success" } except Exception as e: return { "tool_name": intent["tool"], "error": str(e), "status": "failed" } # 使用示例 def get_weather(city: str) -> str: return f"{city}今日晴，气温22-28℃，空气质量良" tool_use = SimpleToolUse() tool_use.register( name="get_weather", func=get_weather, description="查询指定城市的实时天气", parameters={"city": "string"} ) # 测试 print(json.dumps(tool_use.execute("查上海天气"), ensure_ascii=False, indent=2))

这段代码只有 37 行，但已覆盖 Tool Use 最核心的五个环节。重点看parse_intent方法——它用正则硬编码匹配，看似简陋，却是面试时最稳妥的选择。为什么不用 LLM 做意图识别？因为：第一，LLM 调用本身就需要网络和 token，脱离环境不可控；第二，面试官想看的是你对“工具调用”这个动作本身的建模能力，而非 NLP 能力。正则在这里是刻意为之的简化，它把问题域收窄到“如何把自然语言映射到工具名+参数”，而把更复杂的语义理解交给后续迭代。

提示：面试时写到这里，务必主动补一句：“实际项目中，parse_intent会替换为基于 LLM 的 function calling 或规则引擎，但底层的注册、执行、返回结构保持不变——这正是 Tool Use 框架的价值：解耦意图识别与工具执行。”

这个版本的致命短板是什么？参数校验为零。如果用户说“查北京和上海天气”，正则可能提取出乱码，get_weather函数收到非法参数直接崩溃。但这恰恰是面试官期待你指出的点——他要听你说：“下一步必须加参数 Schema 校验，比如用 Pydantic 定义工具参数模型”。这种“先跑通，再加固”的思路，比一上来就堆砌装饰器更体现工程节奏感。

3. 方案二：生产就绪版——参数强校验与执行沙箱的落地细节

当面试官点头说“不错，那如果参数传错了怎么办？”，就是切换到方案二的信号。这里不再容忍“靠人肉测试保证正确”，而是用可验证的约束替代经验主义。核心升级三点：参数类型与范围校验、执行超时控制、结果结构标准化。

我们引入pydantic.BaseModel作为参数契约，用concurrent.futures.TimeoutError捕获执行超时，并强制所有工具返回dict结构。关键不是用了什么库，而是为什么选这个解法。

先看参数校验的重构：

from pydantic import BaseModel, Field, ValidationError from typing import Dict, Any, Callable, Optional, Type import time from concurrent.futures import ThreadPoolExecutor, TimeoutError class ToolParamSchema(BaseModel): """所有工具参数必须继承此基类，实现标准化校验""" pass class WeatherParam(ToolParamSchema): city: str = Field(..., min_length=1, max_length=20, description="城市名称") unit: str = Field("celsius", pattern=r"^(celsius|fahrenheit)$", description="温度单位") class RobustToolUse: def __init__(self, timeout_seconds: float = 10.0): self.tools: Dict[str, Dict[str, Any]] = {} self.timeout = timeout_seconds def register(self, name: str, func: Callable, description: str, param_schema: Type[ToolParamSchema]) -> None: """注册时绑定参数 Schema 类""" self.tools[name] = { "func": func, "description": description, "param_schema": param_schema } def _validate_and_parse_params(self, raw_args: Dict[str, Any], schema: Type[ToolParamSchema]) -> Optional[ToolParamSchema]: """用 Pydantic 强校验参数，失败则返回 None""" try: return schema(**raw_args) except ValidationError as e: # 格式化错误信息，便于前端展示 errors = [{"loc": str(err["loc"]), "msg": err["msg"]} for err in e.errors()] return None, errors def _execute_with_timeout(self, func: Callable, **kwargs) -> Dict[str, Any]: """在独立线程中执行，超时则中断""" with ThreadPoolExecutor(max_workers=1) as executor: try: future = executor.submit(func, **kwargs) result = future.result(timeout=self.timeout) return {"result": result, "status": "success"} except TimeoutError: return {"error": f"工具执行超时（>{self.timeout}s）", "status": "timeout"} except Exception as e: return {"error": str(e), "status": "failed"} def execute(self, user_input: str) -> Dict[str, Any]: # ...（意图解析逻辑同方案一，此处省略）... intent = self.parse_intent(user_input) if not intent or intent["tool"] not in self.tools: return {"error": "未识别工具", "status": "failed"} tool_def = self.tools[intent["tool"]] # 步骤1：参数校验 validated_params, errors = self._validate_and_parse_params( intent["args"], tool_def["param_schema"] ) if errors: return {"error": "参数校验失败", "details": errors, "status": "validation_failed"} # 步骤2：带超时执行 exec_result = self._execute_with_timeout( tool_def["func"], **validated_params.dict() ) # 步骤3：标准化返回结构（强制 dict） if exec_result["status"] == "success": # 确保 result 是 dict，否则包装 if not isinstance(exec_result["result"], dict): exec_result["result"] = {"raw_output": exec_result["result"]} return { "tool_name": intent["tool"], "execution": exec_result, "timestamp": int(time.time()) }

这段代码的关键不在语法，而在每个设计选择背后的权衡。比如为什么用ThreadPoolExecutor而非signal.alarm？因为后者在 Windows 不可用，且对多线程不友好；为什么参数校验失败返回None, errors而非抛异常？因为异常会打断主流程，而结构化错误信息能让上层（如前端）精准提示用户“城市名不能超过20个字符”。

注意：面试时写到ThreadPoolExecutor，一定要解释：“我们不用asyncio是因为同步工具（如数据库驱动、requests）混用 async 可能引发事件循环混乱；用线程池是平衡兼容性与超时控制的务实选择。”

这个版本还埋了一个伏笔：exec_result["result"]被强制包装成dict。这是为方案三的“结果归一化”打基础——无论工具返回字符串、数字还是自定义对象，最终都转成键值对，方便 LLM 下一步消费。我在某次模拟面试中就因没提这点被追问：“如果天气 API 返回 XML，你怎么喂给 LLM？” 答案就是：在execute方法末尾加一层xml_to_dict()转换器，这就是架构演进的自然路径。

4. 方案三：可扩展架构版——插件化注册与运行时策略编排

当面试官问出“如果公司有 200 个工具，怎么管理？怎么灰度发布新工具？怎么限制某个工具每天最多调用 100 次？”，恭喜，你已进入方案三战场。这里考验的是系统级抽象能力：把 Tool Use 从“一段代码”升维成“一个可治理的平台”。

核心思想是分层解耦：

• 注册层：工具声明与元数据管理（谁提供、何时上线、负责人是谁）
• 路由层：动态匹配工具，支持权重、AB 测试、降级策略
• 执行层：隔离运行环境、资源配额、调用链路追踪

我们用“插件注册中心 + 策略路由器”实现。先看注册中心如何支持热加载：

import importlib import os from pathlib import Path from typing import Dict, Any, Callable, Optional, List class PluginRegistry: def __init__(self, plugins_dir: str = "./tools"): self.plugins_dir = Path(plugins_dir) self.tools: Dict[str, Dict[str, Any]] = {} self.load_all_plugins() def load_all_plugins(self) -> None: """扫描目录，自动加载所有 _plugin.py 文件""" for plugin_file in self.plugins_dir.glob("*_plugin.py"): try: module_name = f"tools.{plugin_file.stem}" spec = importlib.util.spec_from_file_location(module_name, plugin_file) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) # 插件必须提供 register_tools() 函数 if hasattr(module, "register_tools"): tools = module.register_tools() self.tools.update(tools) print(f"✅ 加载插件 {plugin_file.name}，注册 {len(tools)} 个工具") except Exception as e: print(f"❌ 加载插件 {plugin_file.name} 失败：{e}") def get_tool(self, name: str) -> Optional[Dict[str, Any]]: return self.tools.get(name) # 示例插件文件 ./tools/weather_plugin.py def get_weather(city: str) -> dict: return {"temperature": 25, "condition": "sunny"} def register_tools() -> Dict[str, Dict[str, Any]]: return { "get_weather": { "func": get_weather, "description": "获取城市天气", "param_schema": WeatherParam, "metadata": { "owner": "dev-weather-team", "rate_limit": {"calls_per_day": 100}, "tags": ["public", "low_risk"] } } }

插件机制解决了“200 个工具”的组织问题，但真正的难点在运行时策略。比如限流：不能简单用time.sleep()，而要对接分布式 Redis 计数器。我们设计一个轻量策略路由器：

import redis from functools import wraps class StrategyRouter: def __init__(self, redis_client: Optional[redis.Redis] = None): self.redis = redis_client or redis.Redis(host='localhost', db=0) def rate_limit(self, tool_name: str, limit: int, window_seconds: int = 86400): """装饰器：为工具添加 Redis 限流""" def decorator(func): @wraps(func) def wrapper(*args, **kwargs): key = f"tool:{tool_name}:calls" count = self.redis.incr(key) self.redis.expire(key, window_seconds) if count > limit: raise RuntimeError(f"工具 {tool_name} 调用超限（{limit}/{window_seconds}s）") return func(*args, **kwargs) return wrapper return decorator # 在插件中使用 router = StrategyRouter() @register_tools() def weather_plugin(): @router.rate_limit("get_weather", limit=100) def get_weather(city: str) -> dict: return {"temperature": 25, "condition": "sunny"} return {"get_weather": {...}}

这个设计的精妙之处在于：策略与工具逻辑物理分离。限流规则写在装饰器里，工具函数本身仍是纯业务逻辑。未来要加熔断、降级、审计日志，只需新增装饰器，无需修改get_weather一行代码。我在某次技术分享中听到一位平台工程师说：“我们最初把限流逻辑写死在工具函数里，结果改一个策略要发 50 个服务。后来改成装饰器模式，策略变更 5 分钟内全量生效。”

提示：面试时提到 Redis，一定要补充：“生产环境会用连接池和故障转移，这里为简洁省略。关键是策略的可插拔性——即使换成 etcd 或数据库计数，装饰器接口不变。”

方案三的终极价值，是让 Tool Use 从“功能”变成“能力”。当新业务方要接入支付工具，他们只需提交一个_plugin.py文件，填写元数据，平台自动完成注册、限流、监控埋点。这才是大厂面试官想看到的：你不仅会写代码，更懂如何构建可持续演进的系统。

5. 三种方案的实战取舍：什么场景该用哪一种？

写完三个方案，面试官常会问：“如果让你现在立刻上线一个客服机器人，用哪个？” 这不是考你背答案，而是考你根据约束条件做技术决策的能力。我把判断逻辑拆解成一张决策表，这是我在多个项目中踩坑后总结的：

举个真实案例：某教育 SaaS 公司要做“作业批改助手”，初期只接入 3 个工具（查错字、算公式、评作文）。CTO 明确要求“两周上线，不许影响现有客服系统”。我力推方案二——用 Pydantic 校验确保学生上传的 PDF 路径不被注入，用线程池超时防止 OCR 服务卡死。上线后发现作文评分工具偶发超时，我们只改了timeout_seconds=30这一个参数就解决，没动任何业务逻辑。如果是方案一，就得重写整个执行链；如果是方案三，光搭 Redis 就要三天。

再看反例：某金融客户要求“所有工具调用必须留审计日志，且能按用户 ID 追溯”。这直接否决方案一和二——它们没有统一的调用上下文。必须上方案三，在StrategyRouter.execute()中注入trace_id和user_id，所有插件通过context.get("user_id")获取，日志格式由中心统一定义。这时候多花的 3 天架构时间，换来的是合规审计零风险。

所以，回答“用哪个方案”，永远不是选“最牛的”，而是选“刚刚好够用的”。我在简历里写“主导 Tool Use 框架从 0 到 1”，其实就包含这三次演进：第一次用方案一跑通 demo 拿到预算，第二次用方案二支撑首期用户，第三次用方案三接入 12 个业务线。每次升级，都是需求倒逼架构，而不是为了炫技提前铺路。

6. 面试之外的真相：Tool Use 不是终点，而是 Agent 的起点

写到这里，可能有人觉得：“不就是个工具调用吗？至于搞这么复杂？” 但如果你真做过 Agent 项目，就会明白：Tool Use 是整个智能体系统的承重墙。它崩了，LLM 再强也输出垃圾；它稳了，LLM 才能专注做它最擅长的事——推理与规划。

我参与过一个医疗咨询 Agent 项目，初期用方案一快速验证，结果用户问“我昨天吃的阿司匹林和今天开的布洛芬能一起吃吗”，意图解析器把“阿司匹林”误判为城市名，调用了天气工具，返回“阿司匹林今日多云”——这种荒诞错误，根源不在 LLM，而在 Tool Use 的边界模糊。后来我们强制所有工具返回{"type": "drug_interaction", "data": {...}}这种带类型的结构，LLM 的 system prompt 明确要求：“只处理 type 为 drug_interaction 的结果，其他一律忽略”。这就是方案二带来的收益：用结构化约束，把 LLM 从“全能选手”降级为“专业裁判”。

更深层的影响在可观测性。方案三的插件中心天然生成调用拓扑图：哪个工具调用最频繁？哪个错误率最高？哪个用户在滥用翻译工具？这些数据直接驱动产品迭代。我们曾发现 70% 的“查药品说明书”请求来自同一 IP，分析日志后确认是竞品爬虫——于是给该工具加了验证码，同时优化了缓存策略。这种决策，没有方案三的元数据和监控，根本无从谈起。

所以，当面试官让你手写 Tool Use，他其实在问：你是否理解，真正的 AI 应用不是拼模型参数，而是拼系统韧性？不是比谁 prompt 写得巧，而是比谁能把不确定的 LLM，装进确定的工程框架里。我最后想分享一个心得：在白板上写代码时，别只盯着语法正确。每写一行，都问问自己——

• 这行代码在百万 QPS 下会不会成为瓶颈？
• 这个异常被捕获后，用户看到的是友好的提示，还是 500 错误？
• 如果明天要加一个“调用企业微信发通知”的工具，现有架构需要改几处？

答案越具体，你的工程师成色就越足。毕竟，能手写 Tool Use 的人很多，但能说清“为什么这样写”的人，永远稀缺。