资讯详情 手写Tool Use:从提示词到自研执行器的完整实现
📅 2026/10/20 18:58:37
面试官抛出这个问题的时候我其实是有心理准备的因为在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*4810}} 使用规则 - 当你认为需要调用工具时只输出一个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: funknown 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_asciiFalse) }) 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( modeltext-assistant-pro, messagesmessages, toolstools, tool_choiceauto ) 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_asciiFalse) }) 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_workers4) 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(timeout10) 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_asciiFalse) }) 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(fmissing 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(fparameter {key} should be string, got {type(value)}) if expected_type number and not isinstance(value, (int, float)): raise TypeError(fparameter {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_workers1) as pool: future pool.submit(tool.run, **validated) result future.result(timeouttimeout) return result except TypeError as te: return {error: finvalid args: {str(te)}} except TimeoutError: return {error: tool execution timeout, tool: name} except Exception as exc: return {error: ftool execution failed: {str(exc)}}执行超时是不得不做的设计。有些外部API响应很慢比如第三方天气服务有时候会卡到30秒如果主线程一直等下去Agent的整体延迟会变得不可接受。我在执行器里统一设置15秒超时超时后返回一个错误结果给模型让模型决定是重试还是告知用户暂时无法获取。对话循环控制器里关键参数是最大循环轮数def run_agent(messages, max_iterations6): registry ToolRegistry() registry.register(WeatherTool) registry.register(CalculatorTool) # 也可以注册更多数据库查询、PDF解析、HTTP请求等 for iteration in range(max_iterations): response call_model( messagesmessages, toolsregistry.schemas(), tool_choiceauto ) 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_asciiFalse) }) 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文档自己从提示词方案写起再封装成注册中心加执行器跑通一个查询工具计算工具多轮调用的完整样例这个过程中踩过的坑会比刷几十道题都管用。