面试官抛出这个问题的时候,我其实是有心理准备的,因为在AI应用开发这块,Tool Use(工具调用)基本是Agent系统绕不开的底座能力。但"手写"两个字还是让我意识到,他要的不是"你用过哪家平台的API",而是想看我能不能脱离文档、从底层把一条完整的调用链路讲清楚并写出来。我当时在共享编辑器里,用三种方案现场搭了一个可运行的Demo,从最朴素的提示词方案,到原生工具调用接口的方案,再到带状态和容错的自研执行器。整个过程大概四十分钟,最后面试官追问的几个问题也都落在这套链路的细节上。今天把这次现场手写的完整思路和代码整理出来,给准备面Agent方向或者想自己搭工具调用框架的朋友做个参考。
1. 面试官到底想考什么:Tool Use 手写的三层含义
很多人听到"手写Tool Use"第一反应是开始背某家API的格式。但面试官真正想验证的,不是你记住了几个字段,而是你有没有真正理解一条工具调用链路从发起、解析、执行到回填的完整闭环。我把它拆成三层。
1.1 表面问题只是一个开始
第一层是基础能力,也就是你是否知道Tool Use的基本流程:把工具的描述以结构化形式给到模型,模型根据用户问题和工具说明决定"我要调用哪个工具、传什么参数",然后返回一个机器可读的调用请求,应用侧解析并执行真实函数,再把执行结果作为上下文回传给模型,让模型继续回答。这个流程说起来简单,但每一步都有大量的细节。
第二层是工程意识。工具调用不是只有"成功"一条路。工具不存在怎么办?参数少传了怎么办?工具超时了怎么办?模型返回了非法JSON怎么办?多个工具并行请求怎么处理?这些都是在真实项目里一定会遇到的,面试官想看你能不能提前把这些坑堵住。
第三层是设计思维。同样是实现Tool Use,你可以用最笨的字符串拼接,也可以用标准化的函数定义协议,还可以抽象出一套可扩展的注册中心。三种做法代表了你对系统扩展性、可维护性的理解层次,也基本决定了你在一个成熟Agent项目里能承担多大的模块。
1.2 我心里瞬间列出的候选方案
在动手前,我快速在脑子里过了一遍可选的实现路径。与其盲目追求"高级",不如按场景把方案分层:
- 方案一:提示词注入工具描述,强制模型输出JSON,正则提取后执行。这是最不依赖平台功能、任何模型都能跑的办法,缺点是稳定性一般,适合轻量场景。
- 方案二:使用模型平台提供的原生工具调用能力,也就是业界常说的Function Calling风格接口。模型负责产出结构化调用参数,应用侧做好消息回填即可。这是目前最主流、最推荐的方式。
- 方案三:自研一个带注册中心、并发执行、超时控制、安全校验和循环上限的工具执行器。本质上是把方案二工程化,达到接近生产可用的程度。
我把这三个方案都写了出来,面试官明显更在意的是第三种的工程细节。下面一个一个说。
2. 方案一:提示词注入加 JSON 解出,最朴素也最考验基本功
我先写的是最基础的版本。这个方案不是最优,但能最快证明"我知道工具调用是怎么发生的"。
2.1 思路:把事情说清楚,让模型照着输出
核心思路非常直白:把工具的描述写进系统提示词,告诉模型"当你要调用工具时,请输出一个固定格式的JSON对象",然后应用侧把模型返回的文本里的JSON提取出来,解析出工具名和参数,执行后把结果拼回消息列表,再次调用模型。
这里最关键的是工具描述要写得"让模型看得懂"。我现场是这样组织的:
SYSTEM_PROMPT = """ 你是一个智能助手,你可以使用以下工具来帮助用户完成任务。 工具列表: 1. 名称: get_weather 功能: 查询指定城市的实时天气情况 参数说明: {"city": {"type": "string", "description": "城市名称"}} 2. 名称: calculate 功能: 执行数学计算,任何数学问题都必须使用该工具 参数说明: {"expression": {"type": "string", "description": "数学表达式,如 23*48+10"}} 使用规则: - 当你认为需要调用工具时,只输出一个JSON对象,不要输出任何其他文字 - JSON格式: {"name": "工具名称", "arguments": {"参数名": "参数值"}} - 一次只能调用一个工具 - 如果不需要调用工具,请直接用自然语言回答用户 """这个提示词有三点很讲究:第一,把"必须使用工具"的场景写死,比如数学计算,这是为了让模型不要自己硬算;第二,明确"不要输出其他文字",避免模型在JSON前后夹带解释,增大解析难度;第三,限定了"一次只能调用一个工具",因为文本格式天然不适合表达并行调用。
2.2 完整实现:工具描述、JSON提取、执行回填
模型返回的内容可能是一个纯文本,也可能用Markdown代码块包起来。为了稳妥,我用正则做两层提取,先剥代码块,再找大括号。
import json import re def parse_tool_call(text: str): # 先尝试提取代码块中的内容 code_match = re.search(r"```(?:json)?\s*(.*?)```", text, re.S) candidate = code_match.group(1).strip() if code_match else text.strip() # 再找第一个完整的 JSON 对象 obj_match = re.search(r"\{.*\}", candidate, re.S) if not obj_match: return None try: obj = json.loads(obj_match.group(0)) return obj except json.JSONDecodeError: return None工具执行侧就是一个简单的注册字典加分发函数:
def execute_tool(name: str, arguments: dict): if name == "get_weather": city = arguments.get("city", "") # 这里对接真实天气服务,返回结构化结果 return {"city": city, "weather": "晴", "temp": 26, "humidity": 40} if name == "calculate": expr = arguments.get("expression", "") # 注意:生产环境不要直接用 eval,这里只是演示 result = eval(expr, {"__builtins__": {}}, {}) return {"expression": expr, "result": result} return {"error": f"unknown tool: {name}"}真正的主循环是经典的"调用猜测-执行回填-再生成"过程:
def run_with_tools(user_input: str, max_rounds: int = 3): messages = [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": user_input}, ] for _ in range(max_rounds): resp = call_llm(messages) # 伪代码:调用模型 parsed = parse_tool_call(resp) if parsed is None: return resp # 模型认为不需要工具,直接回答 # 执行工具,把结果追加到消息里,让模型基于结果继续回答 tool_result = execute_tool(parsed["name"], parsed["arguments"]) messages.append({"role": "assistant", "content": resp}) messages.append({ "role": "user", "content": "工具调用结果: " + json.dumps(tool_result, ensure_ascii=False) }) return "达到最大轮数,已停止。" user_input = "帮我算一下 236*48 等于多少?" print(run_with_tools(user_input))我把执行结果拼接成一条新的user消息回传给模型,这是这个方案最核心的地方。模型并不会真的"执行"任何东西,它只是看到了一段"工具调用结果"文本,然后基于这段文本组织最终回答。
2.3 这套方案的硬伤
写完这个方案,我主动对面试官说:"这个方案能用,但我不推荐在生产环境用,原因有三个。"
第一,JSON格式不稳定。小参数模型可能输出{"name": "get_weather", "arguments": {city: "xx"}}这种不带引号的非法JSON,正则提取和json.loads都会失败。第二,指令遵从有概率失效,模型在一次回答里既想输出解释又想输出JSON,导致解析拿到残缺对象。第三,无法优雅表达并行调用,在一次希望查询多个城市天气的场景下,模型只能连续多次单工具调用,既慢又费token。
所以这个方案适合什么?适合模型本身不支持函数调用协议的时候做兜底,或者用在一些非关键路径的轻量场景。它最大的价值是让人理解工具调用的最小闭环原理。
3. 方案二:原生工具调用接口的标准循环
写完方案一,我接着写了方案二。这个方案我称之为"标准答案",因为目前主流平台都已经支持了结构化的工具调用协议,开发者只需要把工具列表按照合约传过去,模型会直接返回结构化的工具调用请求,不再需要自己操心JSON文本提取。
3.1 思路:让框架替你做结构化输出
平台的原生工具调用接口,本质上是把"工具描述"和"结果回填"都标准化了。我们不再需要在提示词里写一大段"请输出什么格式",而是把工具以Schema的形式显式传入。模型确认需要调用工具时,返回一个结构化的tool_calls数组,里面包含工具名、参数JSON字符串,以及一个唯一标识。应用侧执行完工具后,把结果按照role: "tool"的消息类型回传,并且用tool_call_id把结果和刚才的请求关联起来。
这个方案的前半段,是让模型侧的协议替我们解决结构化输出的难题;后半段,开发者要做的是维护好消息顺序和ID关联。
3.2 核心实现:tools 配置与 tool_calls 消息回填
先定义两个工具。这里我特意用统一的Schema格式,字段名和类型都要准确,因为模型会严格按照这套Schema产出参数。
tools = [ { "type": "function", "function": { "name": "get_weather", "description": "查询指定城市的实时天气,包含温度、湿度、天气状况", "parameters": { "type": "object", "properties": { "city": {"type": "string", "description": "城市名称"} }, "required": ["city"] } } }, { "type": "function", "function": { "name": "calculate", "description": "执行数学计算,返回计算结果", "parameters": { "type": "object", "properties": { "expression": {"type": "string", "description": "数学表达式"} }, "required": ["expression"] } } } ]然后写一个完整的多轮循环。注意,这里每一步都需要仔细处理消息类型。
def run_agent(messages): while True: response = call_model_stream( model="text-assistant-pro", messages=messages, tools=tools, tool_choice="auto" ) choice = response.choices[0] assistant_msg = choice.message # 先把模型自身输出的内容加入消息列表 messages.append({"role": "assistant", "content": assistant_msg.content or ""}) # 如果没有工具调用需求,直接返回最终答案 if not assistant_msg.tool_calls: return assistant_msg.content # 有工具调用需求:逐个执行 tool_messages = [] for tool_call in assistant_msg.tool_calls: func_name = tool_call.function.name args = json.loads(tool_call.function.arguments or "{}") result = execute_tool(func_name, args) # 关键:tool_call_id 必须对应上模型返回的 id tool_messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": json.dumps(result, ensure_ascii=False) }) messages.extend(tool_messages) # 循环继续,模型会看到工具结果,生成最终回答或再次调用工具这里有几个容易翻车的地方,我在写的时候特意向面试官指出了:
tool_call.function.arguments是一个JSON字符串,不是对象,必须先json.loads,很多人第一版代码会直接当字典用导致报错。- 模型的每一次回复都要追加到消息列表,工具执行结果也要追加,顺序不能乱。顺序乱了会导致多轮Agent上下文错位。
tool_call_id要保留,工具结果必须通过它和对应的工具调用请求关联。如果多个工具并行,每个结果的ID都要精确匹配。
3.3 并行工具调用的处理
我写的是一个for循环逐个执行,这样最稳。但真实场景里,用户可能会问"同时查一下A城市和B城市的天气,顺便把这两个数字算一下",模型可能一次返回多个tool_calls。逐个执行虽然正确,但明显浪费时间。
我在现场把执行部分改成了线程池并发:
from concurrent.futures import ThreadPoolExecutor, as_completed def resolve_tool_calls(assistant_msg): tool_messages = [] with ThreadPoolExecutor(max_workers=4) as executor: future_map = { executor.submit(execute_tool, tc.function.name, json.loads(tc.function.arguments or "{}")): tc for tc in assistant_msg.tool_calls } for future in as_completed(future_map): tc = future_map[future] try: result = future.result(timeout=10) except Exception as exc: result = {"error": str(exc), "tool": tc.function.name} tool_messages.append({ "role": "tool", "tool_call_id": tc.id, "content": json.dumps(result, ensure_ascii=False) }) return tool_messages改成并发后,效果立竿见影。假设两个天气接口每个耗时1秒,串行要2秒,并发只要1秒。但要注意:如果多个工具同时修改同一个全局资源,并发会引入竞态问题。所以我在现场强调了一个原则:"读类工具可以并行,写类工具必须串行"。这也是一个很加分的工程判断。
3.4 这套方案适合谁
方案二是目前绝大多数Agent项目的首选,因为它是平台原生协议,稳定性和兼容性都有保障。模型侧的解析是官方实现的,不需要我们维护"提示词-正则-容错"这一整套自研逻辑。
但方案二也有隐形成本:每轮请求都要把tools列表完整传给模型,工具描述占的token随着工具数量线性增长。假设你有20个工具,平均每个工具描述约200 token,这就占了4000 token,再加上对话历史和工具结果,一次请求可能直接顶到上下文窗口的60%。这种场景下,需要对工具做分组懒加载,或者在描述上精简化,这个我在后面避坑部分再说。
4. 方案三:自製带状态和容错的 Tool Executor
写完方案二,面试官点了点头,然后追问了一句:"如果工具越来越多,项目越做越大,你会怎么设计它的扩展性和稳定性?"我顺势写出了方案三,这也是我平时在真实Agent项目里采用的工程化结构。
4.1 架构:工具注册表加执行器再加安全层
我把整体拆成三个部分:
- 工具注册中心,负责维护工具名称到实现类的映射,同时负责生成协议要求的Schema列表。
- 工具执行器,负责参数校验、超时控制、并发调度、异常捕获和结果序列化。
- 对话循环控制器,负责Agent的主流程,包括循环上限、空工具调用保护、上下文窗口控制。
这样做的好处是逻辑解耦。新增一个工具时,我只需要实现一个Tool子类并注册,不需要改动主循环。老工具的参数变化,也只需要改它自己的Schema,执行器会统一处理。
4.2 核心代码:Tool 抽象、Registry、并发执行
先看Tool抽象基类:
class Tool: name = "" description = "" parameters = { "type": "object", "properties": {}, "required": [] } def validate_args(self, raw_args: dict): for key in self.parameters.get("required", []): if key not in raw_args: raise ValueError(f"missing required parameter: {key}") props = self.parameters.get("properties", {}) for key, value in raw_args.items(): if key not in props: continue expected_type = props[key].get("type") if expected_type == "string" and not isinstance(value, str): raise TypeError(f"parameter {key} should be string, got {type(value)}") if expected_type == "number" and not isinstance(value, (int, float)): raise TypeError(f"parameter {key} should be number") return raw_args def run(self, **kwargs): raise NotImplementedError参数校验看起来不起眼,但实际价值很大。模型虽然能结构化输出,但偶尔也会给错类型,比如把数字写成字符串。如果直接传给底层函数,可能触发奇怪的异常,或者更危险的是传给了命令执行类工具,产生安全隐患。
然后是注册表和执行器:
class ToolRegistry: def __init__(self): self._tools = {} def register(self, tool_cls: type): t = tool_cls() self._tools[t.name] = t def schemas(self): return [ { "type": "function", "function": { "name": t.name, "description": t.description, "parameters": t.parameters, } } for t in self._tools.values() ] def execute(self, name: str, raw_args: dict, timeout: int = 15): if name not in self._tools: return {"error": "tool not found", "tool": name} tool = self._tools[name] try: validated = tool.validate_args(raw_args) # 用线程池执行,带超时控制 with ThreadPoolExecutor(max_workers=1) as pool: future = pool.submit(tool.run, **validated) result = future.result(timeout=timeout) return result except TypeError as te: return {"error": f"invalid args: {str(te)}"} except TimeoutError: return {"error": "tool execution timeout", "tool": name} except Exception as exc: return {"error": f"tool execution failed: {str(exc)}"}执行超时是不得不做的设计。有些外部API响应很慢,比如第三方天气服务有时候会卡到30秒,如果主线程一直等下去,Agent的整体延迟会变得不可接受。我在执行器里统一设置15秒超时,超时后返回一个错误结果给模型,让模型决定是重试还是告知用户暂时无法获取。
对话循环控制器里,关键参数是最大循环轮数:
def run_agent(messages, max_iterations=6): registry = ToolRegistry() registry.register(WeatherTool) registry.register(CalculatorTool) # 也可以注册更多:数据库查询、PDF解析、HTTP请求等 for iteration in range(max_iterations): response = call_model( messages=messages, tools=registry.schemas(), tool_choice="auto" ) assistant_msg = response.choices[0].message messages.append({"role": "assistant", "content": assistant_msg.content or ""}) if not assistant_msg.tool_calls: return assistant_msg.content tool_messages = [] for tc in assistant_msg.tool_calls: args = json.loads(tc.function.arguments or "{}") result = registry.execute(tc.function.name, args) tool_messages.append({ "role": "tool", "tool_call_id": tc.id, "content": json.dumps(result, ensure_ascii=False) }) messages.extend(tool_messages) return "已达到最大迭代次数,任务未能完成。已停止继续调用工具。"max_iterations设6轮是实践经验。大多数问题3轮内就能解决,6轮足以应对查询-计算-再查询的嵌套场景。超过6轮,多半是模型在绕圈子甚至进入了死循环,继续下去只会在消耗token。
4.3 状态管理、重试与防死循环
方案三相比方案二还多了两个层面的设计。
一个是"需要状态"的工具怎么处理。比如一个数据库工具,第一步建立连接,第二步执行查询,第三步关闭连接。如果工具是无状态的,每次调用都新建连接,性能会很差。我在执行器里加了一个简单的会话上下文对象,挂在当前Agent运行实例上:
class AgentContext: def __init__(self): self.state = {} self.registry = ToolRegistry() context = AgentContext() class DbQueryTool(Tool): name = "db_query" description = "执行SQL查询" parameters = { "type": "object", "properties": { "sql": {"type": "string"} }, "required": ["sql"] } def run(self, sql: str): conn = context.state.get("db_conn") if conn is None: conn = create_connection() context.state["db_conn"] = conn cursor = conn.cursor() cursor.execute(sql) return {"rows": cursor.fetchall()}另一个是"系统提示词的防绕圈设计"。工具调用失败时,模型可能会尝试用同一个错误参数反复重试,浪费轮次。我会在系统提示词里加一条约束:
如果你尝试某个工具两次都得到错误结果,请停止重试,换一种策略,或者明确告诉用户当前无法完成该操作。这一条简单的话,实际能大幅减少死循环和token浪费。
5. 三套方案对比与面试复盘
面试官让我把三套方案放在一起对比,我直接打了个表,把复杂的东西摊开说。
5.1 对比表格
| 维度 | 方案一:提示词JSON方案 | 方案二:原生工具调用方案 | 方案三:自研执行器方案 |
|---|---|---|---|
| 实现复杂度 | 低,约100行代码 | 中,重点在消息回填 | 高,抽象层次多 |
| 稳定性和准确性 | 低,依赖模型指令遵从能力 | 高,平台协议保证结构化输出 | 最高,加了参数校验和容错 |
| 并行工具调用 | 不支持或极难 | 支持,需要自己改并发 | 原生支持,带超时控制 |
| 错误处理能力 | 弱,正则解析失败就崩 | 中,靠平台侧,执行错误需自处理 | 强,统一异常捕获和恢复 |
| 工具扩展性 | 差,改提示词加正则 | 中,加Schema即可 | 好,继承Tool类注册即可 |
| 资源消耗 | token占用少 | 工具Schema占token中等 | 略高,但有缓存和裁剪空间 |
| 生产可用度 | 仅限临时或轻量 | 可用 | 推荐,尤其工具多的时候 |
对比之后我总结了一句:方案一是"理解原理",方案二是"用对协议",方案三是"做好工程"。面试官其实就想看你能否把这三层说清楚,并且知道它们各自适配的场景。
5.2 我在面试中的讲述节奏
手写代码的过程中,我没有闷头写,而是边写边解释为什么要这么设计。比如写方案二时我会说"这里arguments是字符串必须先反序列化,这是个常见坑",写方案三时我会说"加个注册中心是为了让新工具接入的时候不用改主循环"。这种表达方式能让面试官看到你是在思考,而不仅仅是堆代码。
还有一点很重要:承认方案的局限性并不丢人。我在讲方案一的时候就直说了它不适合高并发高稳定场景,这反而建立了信任感,后面的方案二方案三才更有说服力。
6. 现场追问与避坑实录
代码写完后,面试官按惯例开始追问。这部分我觉得比手写代码本身更有价值,因为问题基本都是真实项目里会踩的坑。
6.1 追问一:多个工具并行返回怎么处理
面试官问的是:"如果一次返回了三个工具调用,其中两个是查询,一个是写操作,你会怎么做?"
我回答的思路是:大部分情况下,多个tool_calls可以先用线程池并发执行,然后把结果按tool_call_id回填。但要注意两点,第一,如果写操作依赖前面某个查询的结果,那说明本轮工具之间存在数据依赖,不能让模型在一个step里返回这些有依赖关系的调用,需要把子任务拆成多个step;第二,写操作类工具与读操作并发会有竞态,我一般会对写类工具加锁,或者配置为"必须串行执行"。面试官点头,说明这个点是他们真实考过的。
6.2 追问二:工具报错了要不要回传模型
面试官问:"如果工具执行抛异常,你是直接返回空字符串,还是怎么办?"
这题的陷阱是"返回空字符串"会让模型误以为结果为空,然后给出"未查到信息"的错误回答。正确做法是把异常信息作为内容回传,让模型基于错误事实做判断。比如天气接口超时,我回传{"error": "timeout", "tool": "get_weather"}给模型,模型可以说"天气服务暂时不可用,请稍后再试",也可以说"我换个来源查询"。实测下来,把错误透传给模型,Agent的鲁棒性会提升很多。
6.3 追问三:如何防止无限循环
我给出了三层防线:第一,主循环的max_iterations上限;第二,系统提示词里写清楚"工具多次失败就换策略或道歉退出";第三,在代码层面检测重复的tool_calls组合,如果模型连续两轮发出的工具调用请求完全一样,说明它在原地打转,直接中断。
第三点是我在一次生产事故里加上去的。那次Agent在循环调用一个查询工具,因为模型始终不满意查询结果,反复用同样的参数重试,直到把整轮额度耗光。加了重复检测之后,这类问题直接变成了"循环异常提前终止",补救措施是返回最终结果前把上下文里最有价值的信息总结出来。
6.4 一个容易漏掉的细节:参数校验与安全
面试官最后问了安全意识:"你这个执行器如果开放给外部工具接入,怎么保证安全?"
我补充了两个设计:第一,参数校验不仅是类型检查,还要做白名单和值域校验。比如一个工具只允许查询特定几个数据库,那入参里的库名必须在校验层强制限定。第二,敏感工具不能直接暴露给模型自由选择。我会在注册中心里给工具标记权限级别,模型生成的调用请求如果命中了"受限工具",执行器会返回"权限不足"的结果,而不是直接执行。涉及命令执行、文件删除一类的高危工具,甚至要二次人工确认。
这个细节说完,面试官明显更满意了。工程问题如果没有安全意识,就只停留在Demo水平,无法上生产。
整理下来我最大的感受是,Tool Use手写题其实考的不是代码量,而是你有没有把一条调用链路真正跑通过。面试官抛出的"三种方案",本质上是在看你对一个核心环节的理解深度和工程化程度。如果你也在准备这类面试,我建议别只看API文档,自己从提示词方案写起,再封装成注册中心加执行器,跑通一个"查询工具计算工具多轮调用"的完整样例,这个过程中踩过的坑,会比刷几十道题都管用。