这一个多月我一直在手搓一个Agent项目前两周分别完成了大模型接口封装和上下文管理这周总算轮到最硬核的部分——实现Agent工具调用引擎。先摆结论工具调用这一层别太迷信框架的默认实现自己搭一遍看起来多花时间但你会对什么情况下会翻车建立完全不一样的直觉。这篇就把我这周的完整设计和踩坑过程拆开讲从工具注册、参数校验、执行分发到循环编排全部给到可直接抄作业的代码和思路适合正在做AI Agent开发、或者想从零理解Agent架构的朋友。1. 先想清楚工具调用引擎到底在解决什么问题1.1 模型没有手工具调用就是给模型装手很多人一上来就写循环调API但没想明白工具调用的本质。大模型本质上是一个文本生成器你问它帮我查一下当前订单的物流状态它脑子里根本没有查询订单数据库的能力它只能猜一个答案。要让Agent具备真实操作能力必须建立一个机制模型以结构化文本输出一个调用意图我想调用get_order_status这个工具参数是order_id12345然后由引擎真正执行这个函数把结果作为新的上下文喂回给模型让模型基于真实数据继续推理。这个过程在OpenAI的API里叫Function Calling在Anthropic里叫Tool Use国内各家模型也基本对齐了这套协议。不管名字怎么变核心流程都一样模型不直接执行代码它只负责决策和生成真正动手的是引擎。这个分工非常重要它决定了安全边界——模型永远不应该拿到直接执行任意代码的能力它只能在你预设好的工具清单里做选择。1.2 引擎要管的四件事把概念理清之后工具调用引擎的职责就非常明确了一共四件事工具描述与注册把你写好的Python函数翻译成模型能看懂的JSON Schema说明书让模型知道有什么工具、什么时候用、参数怎么传。调用请求解析模型吐出来的是文本形式的tool_call里面包含工具名和JSON字符串参数引擎要负责解析、校验、纠错。安全执行与结果回填真正去运行目标函数捕获异常、控制超时然后把结构化的执行结果组装成模型需要的消息格式。循环编排与终止决定模型说要调工具→执行→结果喂回去→模型再决策这个循环跑几轮、什么时候停、怎么防止死循环。听起来不复杂但每一环都有隐蔽的坑。比如模型偶尔会拼错工具名参数JSON会缺括号工具返回超长文本会撑爆上下文窗口模型还会在循环里反复调用同一个工具出不来。这些我在后面都会逐个说到。2. 整体设计为什么自己搭而不是直接套框架2.1 框架能帮你什么不能帮你什么市面上的Agent框架已经不少了LangChain、Dify、CrewAI各有各的拥趸。我前期也调研过一圈最后决定核心引擎自己写框架顶多当参考。原因很简单框架的抽象层级太高把工具注册、消息组装、模型调用全包在一起一旦出问题你根本不知道是哪一环断了。对比维度LangChain等框架自研引擎上手速度快装饰器一挂就完事慢得自己写注册表和循环调试透明度黑盒报错信息绕每行代码都清楚定制灵活性受限于框架设计哲学想怎么改怎么改生产环境依赖框架升级可能破坏行为完全可控学习价值学会用API真正理解Agent原理不是说框架不好你的业务场景复杂、团队人手紧用LangChain这类成熟方案完全合理。但如果你想深入Agent开发或者你接下来的项目要对接异构模型、特殊工具协议自研引擎省下来的调试成本会远超开发成本。我这周的项目W3就是要把引擎本身做扎实所以选择了完全自研。2.2 采用OpenAI Function Calling协议作为事实标准虽然我实际对接的模型既有国产的也有本地部署的开源模型但所有模型都兼容OpenAI的tools协议格式。这已经成了行业事实标准工具描述用JSON Schema模型响应里的tool_calls字段包含id、function.name和function.arguments。选择这个协议的好处是迁移成本极低。今天接的是GLM明天想换Qwen或者Llama只要它们都支持OpenAI兼容接口引擎代码一行都不用动。我见过有人自创一套工具协议结果换模型时适配层写了一百多行代码完全没有必要。站在巨人的肩膀上用行业通用协议这本身就是一种工程智慧。2.3 工具Schema设计说明书写得好不好直接决定模型智商工具描述是给模型看的说明书它的质量直接决定模型能不能在正确时机调用正确工具。一份标准的工具描述长这样{ type: function, function: { name: get_order_status, description: 根据订单ID查询订单当前状态包括待支付、已发货、已完成、已取消等。当用户询问订单进度、物流状态或发货情况时必须调用此工具。, parameters: { type: object, properties: { order_id: { type: string, description: 订单ID格式为纯数字字符串例如20240315001 } }, required: [order_id] } } }这里有三个设计要点。第一name必须是动词名词组合清晰表达行为意图get_order_status比order_tool好一万倍。第二description里要写清楚什么时候该用甚至加上触发场景的提示词比如当用户询问订单进度时必须调用此工具这比只写查询订单状态更能引导模型。第三parameters里的每个字段description也要写细模型参数推理很大程度上依赖这里。我实测下来的体感是工具描述写得好的Agent工具调用准确率能提升20个百分点以上。这不是玄学模型本质上是在做文本匹配描述越明确匹配越准。3. 核心实现注册、解析、执行一条龙3.1 工具注册用装饰器把函数变成说明书工具注册层的设计目标是写业务函数的人完全不需要关心工具协议只需要用装饰器声明一下自己的函数是可被Agent调用的工具即可。我用Python实现了一个极简的工具注册表# tool_registry.py from dataclasses import dataclass, field from typing import Callable, Dict, Any dataclass class ToolSpec: name: str description: str parameters: dict handler: Callable class ToolRegistry: def __init__(self): self._tools: Dict[str, ToolSpec] {} def register(self, name: str, description: str, parameters: dict): 装饰器把一个普通函数注册成Agent工具 def decorator(func: Callable) - Callable: self._tools[name] ToolSpec( namename, descriptiondescription, parametersparameters, handlerfunc, ) return func return decorator def get_openai_tools(self) - list: 生成发送给模型的tools参数 return [ { type: function, function: { name: spec.name, description: spec.description, parameters: spec.parameters, } } for spec in self._tools.values() ] def get_spec(self, name: str) - ToolSpec: return self._tools[name]注册侧的使用方式非常清爽registry ToolRegistry() registry.register( nameget_order_status, description根据订单ID查询订单当前状态当用户询问订单进度时必须调用此工具。, parameters{ type: object, properties: { order_id: { type: string, description: 订单ID格式为纯数字字符串 } }, required: [order_id] } ) def get_order_status(order_id: str) - dict: # 真实业务逻辑查数据库、调后端API return {order_id: order_id, status: 已发货, estimated_arrival: 2024-03-20}这样业务开发只需要关注这个函数怎么实现注册和协议适配全部由装饰器代劳。后面想增加一个工具就是写个函数加个装饰器的事完全不影响引擎主体代码。还有一点要注意注册表要保持全局唯一。多个Agent实例共享同一份工具注册表没问题但如果不同Agent需要不同的工具集合建议在注册表之上再包一层按Agent类型过滤的逻辑否则模型看到一堆跟自己无关的工具决策准确率会下降。3.2 参数解析模型输出的是文本不是程序模型返回的tool_calls里arguments是一个JSON字符串比如{order_id: 20240315001}。第一步是把它解析成Python字典这一步就开始了和模型的斗智斗勇。模型生成的JSON经常不完美我遇到过的典型问题包括多余的尾逗号、单引号混用、中文字符串没转义、输出被上下文截断只剩半截JSON。所以解析不能直接json.loads一把梭要做一层容错处理import json import re def parse_tool_arguments(arguments: str) - dict: 解析模型输出的arguments字符串。 先尝试标准解析失败后做轻量修复再解析。 if not arguments or not arguments.strip(): return {} try: return json.loads(arguments) except json.JSONDecodeError: pass # 轻量修复去除尾逗号常见于模型输出 fixed re.sub(r,\s*([}\]]), r\1, arguments) try: return json.loads(fixed) except json.JSONDecodeError: pass # 修复括号不闭合补全缺失的右大括号 depth 0 for ch in fixed: if ch {: depth 1 elif ch }: depth - 1 if depth 0: fixed } * depth try: return json.loads(fixed) except json.JSONDecodeError: raise ValueError(f无法解析工具参数: {arguments[:100]})解析完成后要进行参数校验。校验逻辑我分两层结构校验用jsonschema库按照注册时声明的parameters定义做完整校验检查required字段是否齐全、类型是否正确、枚举值是否合法。语义校准模型经常把数字类型的参数传成字符串123传给int字段或者把枚举值写成同义词。我在校验后加一个轻量转换层对能安全转换的类型做强制转换转换不了的再报错。校验失败时有一个关键的处理原则绝对不要把错误直接抛给用户或干脆崩溃。正确做法是把这个校验失败信息封装成一个错误结果返回给模型让模型自己看错误原因后修正参数重新调用。这其实就是后面要讲的反馈闭环这是Agent能不能自我纠错的关键。3.3 执行与结果回填让错误也变成模型的养料工具函数的执行本身不算难难在控制边界。我给每个工具调用都套了超时控制用的是concurrent.futures里的ThreadPoolExecutor加future.result(timeoutN)。为什么要这么干因为真实世界里工具函数可能是查数据库、调外部HTTP接口外部服务一旦卡死你的Agent循环就会永久阻塞。超时值我一般设置成5到10秒具体视工具类型而定重计算任务可以放宽到30秒。执行完之后不管成功还是失败都要把结果组装成role: tool的消息回填给模型def execute_tool(tool_call_id: str, fn_name: str, arguments: dict, registry: ToolRegistry) - dict: 执行工具调用并返回标准化的结果结构。 无论成功失败都返回结构化JSON让模型能读到错误信息。 spec registry.get_spec(fn_name) try: with ThreadPoolExecutor(max_workers1) as pool: future pool.submit(spec.handler, **arguments) result future.result(timeout10) return { status: success, data: result } except TimeoutError: return { status: error, error: {type: timeout, message: f工具 {fn_name} 执行超时10秒} } except Exception as e: return { status: error, error: {type: type(e).__name__, message: str(e)} }回传消息时要注意tool_call_id必须和模型返回的tool_calls里的id一一对应。模型靠这个id把工具结果和之前的调用请求关联起来如果id对不上一些模型会直接报错或者理解错乱。另外工具结果不要用纯文本堆进去最好统一序列化成紧凑JSON控制token消耗。还有一个容易忽略的点工具返回结果可能非常大。比如搜索工具返回了一大段网页内容直接塞进messages里几十轮后上下文早就爆了。我的策略是在回填前对超长结果做摘要或截断只保留前2000个字符加一个内容已截断的标记。上下文资源是有限的要把宝贵的token留给推理。4. 让循环跑起来编排、上下文与终止条件4.1 主循环模型决策、引擎执行、结果回填有了注册、解析、执行这三个基础模块引擎的主循环就可以串起来了。这个循环本质上就是一个计划-执行-观察的循环def run_agent(user_query: str, registry: ToolRegistry, max_steps: int 8) - str: Agent主循环 1. 发送消息列表给模型 2. 模型返回最终答案 或 工具调用请求 3. 若为工具调用请求执行工具并回填结果然后回到第1步 messages [{role: user, content: user_query}] for step in range(max_steps): response client.chat.completions.create( modelMODEL_NAME, messagesmessages, toolsregistry.get_openai_tools(), tool_choiceauto, ) assistant_msg response.choices[0].message # 情况一模型没有发起工具调用直接给出了最终答案 if not assistant_msg.tool_calls: return assistant_msg.content # 情况二模型要求调用一个或多个工具 messages.append(assistant_msg.model_dump()) for tool_call in assistant_msg.tool_calls: fn_name tool_call.function.name fn_args parse_tool_arguments(tool_call.function.arguments) # 工具不存在时不要崩溃把错误结果回传给模型 if not registry.has_tool(fn_name): result { status: error, error: {type: tool_not_found, message: f工具 {fn_name} 不存在可用工具列表请参考系统提示} } else: result execute_tool(tool_call.id, fn_name, fn_args, registry) messages.append({ role: tool, tool_call_id: tool_call.id, content: json.dumps(result, ensure_asciiFalse), }) # 每轮结束可以在这里做一轮上下文压缩见4.2 return 已达最大执行轮数任务可能尚未完成。请基于已有信息生成阶段性结论并告知用户。这里有一个设计细节很关键tool_choiceauto意味着让模型自己判断这轮要不要调工具、调哪个。也有场景需要强制调某个工具tool_choice{type: function, function: {name: ...}}我一般只在明确的意图识别场景下用强制模式通用对话里auto就够了。还有主循环里每轮都要把完整的tools描述发给模型是必要的因为模型是无状态的它不记得上轮看到了哪些工具。这也引出了上下文管理的问题。4.2 上下文管理messages膨胀是Agent的慢性病每执行一轮工具调用messages里就会多出两条消息模型侧的tool_calls消息 工具侧的结果消息。8轮下来轻轻松松多出十几条再加上原始工具结果可能很大很快就逼近上下文窗口上限。我当前的策略是分级压缩第一级结果截断。工具返回的内容在入messages前先截断到合理长度保留结构化关键字段。第二级历史压缩。维护一个窗口只保留最近4到6轮的完整消息更早的轮次压缩成一段摘要文本以system消息的形式注入。第三级总预算控制。在每次发送请求前估算messages的token总量超过阈值就先触发压缩再发送。压缩规则要小心工具调用的中间结果在后续推理里可能还要被引用粗暴删除会导致模型丢失线索。我的习惯是保留用户原始query 最近一轮完整交互 所有工具结果摘要中间过程的模型话语可以丢弃因为那些往往是推理碎碎念对最终答案贡献不大。4.3 终止条件比你想的更严格一点死循环是Agent最尴尬的翻车现场。模型可能反复调用同一个工具、调用结果不变、还继续调用看起来像个人在反复开关冰箱门。我设置了四道防线最大轮数限制。max_steps设成8这已经是比较宽松的值。几步内解决不了的任务模型继续绕圈大概率也解决不了。重复调用检测。维护一个哈希记录如果同一步内出现了相同的工具名参数组合调用直接拦截并返回提示让模型换一条路。结果相似度检测。连续两轮工具返回的结果一模一样说明模型在空转这时候我会注入一条提示你刚刚获取的信息没有变化请基于现有信息直接回答用户。敏感操作二次确认。对于删除、修改、支付这类有副作用的工具在执行前要求模型先输出确认意图由人工确认后才放行。这个在纯自动Agent里容易忽略但生产环境千万不能省。终止条件的设计原则是宁可误杀提前停止不能放任无限循环。用户最多失去一次不太完美的回答也比卡死强。4.4 并行工具调用的处理策略现在主流模型都支持在一次响应里返回多个tool_calls。这带来了一个执行顺序的问题多个工具之间如果有依赖关系A的结果是B的参数就必须串行执行如果没有依赖可以并行提升效率。我的引擎里实现了一个很朴素的方案默认并行执行无依赖工具但给每个工具的执行函数标记一个depends_on字段。在执行前做一次拓扑排序有依赖的按顺序跑无依赖的用线程池并发跑。并发时还要注意被调工具本身的线程安全性——如果是写同一个文件、操作同一个数据库事务建议还是串行否则会引入竞态问题。5. 实测与调试常见翻车现场实录5.1 模型幻觉出不存在的工具这周遇到最频繁的问题是模型编造工具名。比如我明明只注册了get_order_status模型却调用了get_order_info、check_order甚至get_order_pirce这种拼写错误的名字。一开始我在工具不存在时会直接抛异常结果整个Agent就崩了。后来想明白一个原则模型的错误应该是可观察、可纠正的而不是让系统崩溃的。正确做法我在主循环代码里已经写了把工具不存在作为一种错误结果回传给模型同时在错误信息里附上当前可用的工具列表。模型读到你调用的get_order_info不存在可用工具包括get_order_status、send_refund_request...之后绝大多数情况下会自动修正。这个反馈闭环让整个系统的容错能力上了两个台阶。5.2 工具返回大段原始文本模型开始胡言乱语这个坑特别隐蔽。我有个搜索工具返回的是网页原始HTML第一次接入时没做清理结果模型在后续推理里把HTML标签当成了正文信息回答得莫名其妙。工具返回的内容质量直接决定了模型后续推理的质量这就是典型的垃圾进垃圾出。解决方案是给每个工具的输出定义严格的JSON结构。以搜索工具为例返回格式统一为{ status: success, data: { total: 12, results: [ {title: ..., url: ..., snippet: ...} ] } }所有工具的输出都遵循status data的契约模型读起来一目了然也更容易判断这个工具调用到底成没成功。数据清洗和格式化这件事应该在工具内部做掉不要把脏数据丢给模型。5.3 工具卡死导致整个Agent卡死有一次接入一个汇率查询接口对方服务不稳定HTTP请求偶尔会挂住十几秒不返回。如果没做超时控制我的Agent每次遇到这种情况就整个卡住用户等半天没反应。超时机制加上之后又发现了新问题工具超时返回错误结果给模型模型可能会说我再试一次然后又超时形成超时循环。所以要给重试加一个计数同一个工具连续超时两次就直接放弃该工具让模型换别的途径或者直接给用户答复。真实的生产环境里外部依赖不可用是常态引擎要能优雅降级而不是对着一个坏掉的接口死磕。5.4 tool_call_id错乱导致的上下文混乱还有一个让我排查了半天的诡异问题模型偶尔会把工具结果和错误的历史记录关联起来明明这轮调用的结果它反应像是在聊上一轮的事。后来发现是消息列表里tool_call_id没有严格对应——我在并行执行多个工具时回填顺序乱掉了导致消息里的id和assistant消息里的tool_calls对不上。这里要提醒自己消息列表的顺序就是模型的记忆顺序任何一条消息的顺序错乱或id不匹配都会让模型产生幻觉。并行执行工具时尤其要小心先把所有结果收集齐再按模型返回的tool_calls顺序依次append到messages里不要执行完一个就立刻append。6. 踩坑之后我积累的几个核心认知这周的引擎做完最大的体会是工具调用引擎本质上是模型世界和现实世界之间的接口合同管理。它不负责让模型变聪明只负责让模型的意图被正确翻译、安全执行、结果如实反馈。只要这个合同清晰稳定即使模型的推理能力一般整个Agent系统的下限也会很高。第二个体会可能有点反直觉工具描述和错误反馈的重要性比模型本身的选择更重要。我在实验里换过一个能力更强的模型但工具描述写得很随意实际效果反而不如旧模型加详细描述的组合。工具的说明书质量决定了模型能不能在正确的时候做出正确的调用决策这一点在团队协作里也适用——给别人用的接口文档写清楚比什么都强。最后再分享一个后续的扩展方向工具调用引擎做到这个程度之后可以继续往前走两步。一步是工具的动态加载和按需注入不让模型每轮都面对全部工具而是根据当前对话的意图动态筛选相关工具既省token又提升准确率另一步是引入MCP这类开放的工具体系协议让引擎可以对接第三方提供的标准化工具而不是每个工具都自己写。等我把这两块跑通再来继续更新。