☰
Agent工具调用引擎从零实现:注册表、参数解析与调度主循环
2026/10/26 14:18:05 网站建设 项目流程

我们内部的Agent项目,最早做工具调用就是最朴素的if/else:模型输出一句“帮我查一下订单状态”,代码这边就正则匹配关键词,然后硬编码调一个函数。五六个工具的时候还挺正常,等工具长到二十几个,分支开始失控——加一个工具要翻三个文件,还总出现工具匹配错的尴尬场面。后来我们把工具调用抽成了独立的Agent工具调用引擎,开发节奏才真正恢复正常。这篇文章就围绕这周的工作项W3,完整聊聊我实现这套引擎的整个过程:它怎么管理工具清单、怎么跟模型交互、怎么解析参数、怎么把执行结果回传给模型继续推理,以及实测中踩过的那些坑。适合正在做Agent应用、被工具调度搞到头大的开发者参考,也适合打算从零手写Agent框架的人看个思路。

1. 为什么Agent需要一套工具调用引擎:从if/else分发的失控说起

大部分人最开始接触Agent,都会走一条“野生实现”的弯路。模型是个大语言模型,它本身不能查数据库、不能发请求、不能改文件,但我们可以通过提示词告诉它“如果用户想查天气,就输出JSON”,然后在代码里接住这个JSON做分发。这套逻辑的核心就是:模型输出是信号,代码是执行器。工具少的时候,这种模式完全够用,甚至可以说是最快的原型方案。但一旦工具数量上去了,问题就变了。

1.1 分发逻辑膨胀:接口、参数、返回结构和业务逻辑全堆一起

我在重构之前,代码大概是这样的:先有一堆判断模型输出里有没有“查天气”“查订单”“发邮件”这类的关键词,然后各自调用函数,再把结果拼回字符串喂给模型。听着简单,但真实业务里工具之间往往是有关联的。查订单之前可能要查用户权限,发邮件之前要查收件人是否存在。当你把这些调用关系都写进if/else,每个工具实际上等于被“硬编码”进了控制流:改一个工具的返回结构,会影响所有相关判断分支。有一次我为了给查询工具加个分页参数,结果把识别关键词的正则也一起改崩了,因为提示词里我让模型输出分页信息,正则根本接不住。

这不是写代码能力的问题,是架构问题。Agent的工具调用,本质上应该是一个“声明式”的过程:你只需要告诉模型有哪些工具、每个工具接收什么参数,剩下的调度、校验、执行、回传,都交给引擎去做。这跟后端开发里把API路由抽出来、把Controller层和Service层分开是一个道理。

1.2 引擎究竟要解决哪几件事

我设计这套工具调用引擎时,给自己列了一个核心问题清单,后来发现它基本对应了Agent工具调用的所有关键环节:

  • 工具协议统一:每个工具都有一套标准描述,模型能理解、代码也能解析
  • 参数解析与校验:模型的输出是字符串JSON,不能直接塞进Python函数,必须做好解析、转类型、校验必填项
  • 调度执行:多个工具调用是串行还是并行,超时、重试、异常怎么处理
  • 结果回传:工具执行完,结果要以什么格式再喂回给模型,让模型能继续推理
  • 可观测性:一次工具调用到底调了哪个工具、传了什么参数、花了多少时间,出了问题要能复盘

想清楚这五件事之后,引擎的架构就呼之欲出了。这里想强调一个观念:Agent工具调用引擎,没有你想象的那么玄,它本质上就是一个“总机话务员”——模型的决策是电话那头的大脑,工具是各个分机,引擎负责把大脑的指令转成具体行动,再把行动结果转成大脑能听懂的语言反馈回去。抓住了这个核心,你不管是用LangChain、用现成的开源框架,还是自己手写,都不会跑偏。

2. 引擎的模块划分与一次工具调用的完整生命周期

有了上面的问题清单,接下来就是设计模块边界。我最终把引擎拆成了五个核心组件:工具注册中心、描述生成器、参数解析器、执行调度器、结果处理器。每个组件各干一件事,互不掺和。

2.1 五个核心组件各自的职责

工具注册中心是整个引擎的“通讯录”。业务方开发完一个工具后,调用注册中心的接口登记就可以了。注册中心存的是工具名、描述、参数Schema、可执行函数这四样东西。这里最关键的一点是:工具描述和参数Schema必须能够被“序列化”成模型能读的文本。也就是说,注册中心不只是给Python代码用的,它还要面向模型输出一份“工具菜单”。

描述生成器的任务,就是把注册中心里的工具信息转成标准的tool schema格式。目前主流的做法是使用OpenAI引入的function calling格式,很多兼容模型也都支持。我后面会详细展开这个格式。

参数解析器负责把模型返回的arguments字符串(JSON格式)解析成真实的Python函数调用参数。这里不光是json.loads那么简单,还得做类型转换、必填参数检查、枚举值校验。比如说工具定义的参数是整数,模型给了一个字符串的“3”,直接传进去就能把函数干崩,解析器要负责兜底。

执行调度器是引擎的“大脑执行中心”。它决定先调哪个工具、是否可以并行调用、超时是多少、失败要不要重试。对于一轮模型返回多个tool_call的情况,调度器尤其重要——有些工具之间没有依赖,完全可以并行执行来省时间。

结果处理器则负责把工具的执行结果(可能是对象、报错、空值)统一转换成模型友好的文本消息,并关联到对应的tool_call_id上。这个关联是很多新手容易漏的关键点,API层面要求每个工具结果必须带一个id,模型才知道这条结果对应的是哪个调用。

2.2 一次完整工具调用的生命周期

我把整个链路走一遍,大家就直观了。假设Agent收到用户问题:“帮我订一张明天上午从北京到上海的高铁票。”

第一步,引擎把系统提示词、历史对话和当前用户问题一起发给模型,同时附上工具菜单。模型判断这需要调用“订票工具”,于是在回复里返回一个tool_call,包含工具名和参数JSON。

第二步,引擎的解析器拿到这个tool_call,从注册中心查工具名,如果工具存在,就对arguments做解析和校验。缺了必填字段?引擎在这里直接拦截,不会进入执行阶段。

第三步,调度器执行工具函数。比如订票工具内部会调用12306的API。执行过程有超时控制,默认我这边设成15秒,超过就抛异常,由结果处理器转成“订票接口超时,请稍后重试”的提示。

第四步,结果处理器把执行结果打包成一条tool消息,带上tool_call_id,追加到消息列表里。

第五步,引擎带着更新后的消息列表再一次调用模型。模型看到工具执行结果,如果认为任务未完成,会继续发起新的工具调用;如果认为完成了,就直接输出最终答复,“车票已订好,车次是G102,明天早上9点发车”。

这整个流程就是Agent工具调用引擎的核心闭环。你可以把每一步拆开去优化,但骨架就是这个样子。我画过一张流程图,不过文字描述就足够理解了——核心就是“模型决策,引擎执行,结果反馈”三件事,循环往复,直到模型认为任务完成。

3. 工具注册表与协议描述:模型“看到”的工具长什么样

模型不像我们程序员,它能看到的不是Python函数源码,而是经过严格格式化的工具描述。这个格式化工作做得好不好,直接决定模型会不会正确调用工具。这一节要重点讲工具注册和协议描述。

3.1 为什么不直接传函数名,而要传JSON Schema

我刚做Agent时想过一个问题:既然模型那么聪明,直接把函数源码给它不就好了?实测效果很差。源码里充满了变量名、中间逻辑、依赖关系,模型根本没有能力在短时间内准确提炼出“这个工具是干什么的、需要什么参数”。真正的约定做法是只给模型一个精简的JSON Schema,业界默认格式长这样:

{ "type": "function", "function": { "name": "book_train_ticket", "description": "预订中国高铁车票。出发地和目的地必须是城市名,日期格式为YYYY-MM-DD。", "parameters": { "type": "object", "properties": { "from": {"type": "string", "description": "出发城市,如:北京"}, "to": {"type": "string", "description": "目的城市,如:上海"}, "date": {"type": "string", "description": "出发日期,格式YYYY-MM-DD"}, "seat_type": {"type": "string", "enum": ["商务座", "一等座", "二等座"], "description": "座位类型"} }, "required": ["from", "to", "date"] } } }

这个格式最大价值在于标准化。模型在预训练阶段见过大量类似格式的JSON,它知道该怎么填。字段名规范是硬要求:name就是唯一标识,description要写清楚用途和边界,parameters里的每个字段都要标注类型和说明,required列表告诉模型哪些参数不能省。

3.2 注册中心的最小实现

我这边用Python实现了一个很轻量的注册中心,核心代码大概长这样:

from typing import Any, Callable import inspect ToolFn = Callable[..., Any] class Tool: def __init__(self, name: str, description: str, parameters: dict, fn: ToolFn): self.name = name self.description = description self.parameters = parameters self.fn = fn def to_schema(self) -> dict: return { "type": "function", "function": { "name": self.name, "description": self.description, "parameters": self.parameters, } } class ToolRegistry: def __init__(self): self._tools: dict[str, Tool] = {} def register(self, tool: Tool): if tool.name in self._tools: raise ValueError(f"Tool {tool.name} already registered") self._tools[tool.name] = tool def get(self, name: str) -> Tool | None: return self._tools.get(name) def schemas(self) -> list[dict]: return [tool.to_schema() for tool in self._tools.values()] def __contains__(self, name: str): return name in self._tools

注意一个细节:注册时同名工具要直接报错,而不是静默覆盖。我吃过这个亏。早期为了省事,允许后来的工具覆盖同名旧工具,结果某天排查发现模型在调用一个“意料之外版本”的工具,参数解析永远对不上,浪费了大半天。工具名就是契约,注册中心必须强约束。

有了注册中心之后,业务方接入新工具就变成了两三行代码的事情:

registry.register(Tool( name="get_weather", description="查询指定城市的当前天气状况,城市名用中文,如北京、上海", parameters={ "type": "object", "properties": { "city": {"type": "string", "description": "城市名"} }, "required": ["city"] }, fn=lambda city: get_weather_from_api(city) ))

3.3 工具分组与按需加载:工具菜单不是越长越好

工具注册多了以后,又出现一个新问题:每轮对话如果都把全部工具描述发给模型,Token消耗会非常惊人。一个工具的描述大概要200-500个Token,50个工具就是一两万个Token,这还没算模型的输出。于是我在注册中心外面加了一层“按需加载”的逻辑:根据用户问题的分类或者关键词命中,只挑一批候选工具给模型。这里可以用embedding做语义召回,也可以用规则做硬匹配,看你的业务复杂度决定。初期建议先用简单的规则匹配合集,控制工具数量在个位数到十几个之间,效果往往比暴力全量挂载还要好——模型不会因为工具列表太长而分散注意力。

4. 调度主循环:模型决策、参数转换与执行结果回传

模块注册好了,协议描述也通了,接下来就是整个引擎的主循环。这一节可以说是所有Agent的核心骨架,代码量不大,但每个细节都有讲究。

4.1 主循环伪代码与消息流

我直接用伪代码展示一下引擎的主循环,后面再拆细节:

for step in range(max_steps): response = llm.chat( messages=messages, tools=registry.schemas() ) if not response.has_tool_calls(): # 模型认为不需要调用工具,直接输出最终回答 return response.content # 模型可能要求同时调用多个工具 tool_calls = response.tool_calls for tool_call in tool_calls: tool = registry.get(tool_call.name) if tool is None: # 工具不存在时的容错处理,后面细讲 messages.append(tool_error(tool_call.id, f"工具{tool_call.name}不存在")) continue try: args = parse_arguments(tool_call.arguments, tool) result = executor.execute(tool.fn, args) except Exception as e: result = {"error": str(e)} messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": json.dumps(result, ensure_ascii=False) }) # 超过最大轮数 return "已达到最大推理步数,任务未能完成"

这个循环看起来直白,但有几个地方必须单独拿出来说。

4.2 参数转换:字符串JSON到Python实参的鸿沟

模型返回的arguments是一个字符串,比如这样:

{"from": "北京", "to": "上海", "date": "2024-06-15", "seat_type": "二等座"}

最朴素的转换方式是json.loads,然后一一校验。但实际项目里,这一步的返工率非常高。模型会把日期格式写错、把城市名写成英文、把枚举值写错,甚至会出现参数名和Schema对不上的情况。我的做法是:不完全信任json.loads的结果,而是再做一层基于参数Schema的清洗。

如果工具参数比较复杂,建议直接用Pydantic这类库做校验和类型转换。定义一个参数模型,让模型返回的JSON自动规范化:

from pydantic import BaseModel, ValidationError from typing import Literal class BookTicketArgs(BaseModel): from_city: str to_city: str date: str seat_type: Literal["商务座", "一等座", "二等座"] = "二等座"

解析时如果校验失败,不要直接抛异常给模型,而是把具体错误信息整理好,回传给它,告诉它哪里出了问题。比如“seat_type只允许商务座、一等座、二等座,你传了皇帝座”。模型看到之后一般会自己纠正,而不是直接摆烂。这个把错误转化为模型可读信息的能力,就是工具引擎和普通函数调用最大的区别。

4.3 多个工具调用的串行与并行策略

模型经常一次返回多个tool_call,比如用户问“北京和上海的天气怎么样”,模型可能会同时发起两个get_weather调用。这时候调度器就要做决策。我的建议是优先并行执行:因为很多工具之间没有依赖,并行能大幅缩短总响应时间。

并行实现要注意的是线程安全。Python里最简单的方式是用concurrent.futures.ThreadPoolExecutor,但要给每个工具的执行加上超时控制,防止某个工具长时间挂起拖垮整个Agent会话。另外一个经验是:如果你用异步框架,比如FastAPI,主循环里的工具执行尽量用async/await来做并发,避免线程切换开销。

import asyncio async def execute_tool_calls(tool_calls, registry): async def run_one(tool_call): tool = registry.get(tool_call.name) args = parse_arguments(tool_call.arguments, tool) loop = asyncio.get_event_loop() return await asyncio.wait_for( loop.run_in_executor(None, tool.fn, **args), timeout=15 ) return await asyncio.gather(*(run_one(tc) for tc in tool_calls))

这里的run_in_executor是为了把同步阻塞的函数放到线程池,不阻塞事件循环。不过如果工具本身就是异步函数,建议直接用async def定义,引擎要兼容这两类工具。

4.4 执行结果回传:role=“tool”消息与tool_call_id

模型规定了工具结果必须以role为“tool”的消息回传,并且带一个tool_call_id,用来关联是哪个工具调用。这个id是模型在tool_call里生成的一串唯一标识,我们原样带上就行。忘了带id或者带错id,很多模型会报上下文错误或者当作无效消息,这是我踩过的坑。此时顺手把工具结果统一转成JSON字符串,保证消息结构一致。

还有一点要特别提醒:工具执行结果不能太长。有些工具会把数据库里的几十条记录直接返回,塞进上下文之后,Token直接爆掉,还严重干扰模型对后续问题的判断。正确做法是只回传摘要。比如查询订单列表,不要回传完整列表,而是回传“共查到8条订单,第一条:订单号123,状态已发货;第二条……” 这个抽象能力,是工具引擎和业务方需要反复磨合的地方。

4.5 最隐蔽的坑:没有限制最大迭代轮数

主循环如果不设max_steps,会出大问题。模型在拿到工具结果之后,有时候会因为一次失败而反复重试,或者进入自我对话的循环,一次任务跑出几十轮工具调用,Token消耗巨大,用户等半天。我在第一版就没设,结果有个测试用例跑了32轮才结束,费用吓人。现在默认max_steps是5到8,到顶就直接让模型输出一个坦诚的结果:“经过多次尝试仍未成功,建议检查工具配置或稍后再试。” 这比盲目追求“一定要成功”要健康得多。

5. 工具引擎上线前必须处理的稳定性与安全细节

引擎能跑通主线任务,和能稳定跑在线上,中间隔着一整条“坑路”。这一节的价值在于帮大家少走弯路,我把实测中最痛的几个点分享出来。

5.1 工具描述的Token成本与劣币驱逐良币现象

工具菜单越长,模型每次请求的输入Token就越多。如果接入的工具描述写得啰嗦,比如有人把函数内部实现飞页复制进description里,整个工具菜单会迅速膨胀。更麻烦的是,模型面对超长菜单时,对后面工具的注意力显著下降,排在前面的工具被调用的概率远大于后面的。这会导致“劣币驱逐良币”——不是模型选对工具,而是工具排在前面就占便宜。

我做了两件事:一是规定每个工具的description不超过100个汉字,必须用一句话说清楚用途和边界,其余细节全部放在parameters字段的description里;二是对工具做“重要性分级”,核心工具排前面,辅助工具排后面,定期根据调用频率调整顺序。你也可以考虑用语义检索动态选择工具,但这个比较重,小规模项目先别上。

5.2 模型“编造”工具名和幻觉参数

即使工具菜单很清晰,模型偶尔还是会返回一个不存在的工具名。原因通常是模型把用户问题里的词联想成了工具名。比如你有“get_weather”工具,用户说“顺便帮我查一下明天的降雨概率”,模型居然返回一个“get_rain_probability”,但这工具压根不存在。

容错策略有两个:第一,解析层做名称归一化,比如把工具名里的下划线去掉、全部转小写后再比对,可以收敛一部分错误;第二,如果还是没有匹配,就往消息列表里回传一条错误信息:“工具get_rain_probability不存在,可选工具有get_weather、get_air_quality……”,让模型自己看到可选清单,重新决策。实测下来,第二招的恢复率很高,因为模型看到错误会被“拉回现实”。

5.3 工具执行的超时、重试与异常隔离

工具执行一定会遇到超时和异常。我的经验是:给每个工具定义一个超时时间,默认15秒;对于网络型工具,比如调外部API,可以考虑最多重试1次到2次,但一定要指数退避,不然对方服务抖动时你的Agent会变成请求风暴。异常处理上,所有异常都必须在调度器层捕获,不能让异常直接冒出主循环,否则整个对话就中断了。

日志和可观测性是我早期严重忽略的部分。直到线上一次故障无法回溯,我才痛下决心,给每次工具调用加上一个trace_id,完整记录:什么对话、模型发起了什么tool_call、参数是什么、执行耗时多少、结果摘要是什么、回传给模型后模型下一步干了什么。有了这条链路,至少能快速定位是工具本身问题还是模型决策问题。这个真建议提前做,越早越好。

5.4 安全边界与权限控制

工具调用引擎如果什么工具都开放、什么参数都执行,那就是在给外部用户开后门。我给你几个底线建议:

  • 工具白名单机制:任何工具必须先注册、审批后才可被模型调用,运行时不允许动态注册
  • 参数校验要到业务层:模型说“删除用户id=1”,你至少要确认当前会话是否有这个权限
  • 危险操作二次确认:删除、转账、发邮件这类工具,执行前要回到模型的“拟确认消息”,让用户明确点击确认后再真正执行
  • 敏感信息隔离:工具返回的数据如果包含手机号、身份证号,回传模型前要做脱敏,模型不需要知道完整号码也能完成对话
  • 限制工具单次返回数据大小:防止某个工具一次性把全库数据拉进上下文

这一节的内容可能偏工程和运维,但Agent工具引擎一旦接上真实业务系统,安全就是生死线,不能省。

6. 用回归测试和评估指标验收工具调用引擎

引擎开发到一定程度,不能只靠“我觉得它工作正常”来判断。Agent太随机了,同一个问题跑十次,可能九次成功一次失败。没有评估体系,你根本无法迭代。

6.1 模拟环境的构造与测试用例沉淀

我当时的做法是准备一个测试工具集,覆盖正常场景、参数缺失场景、工具不存在场景和超时场景。测试不调用真实API,而是用Mock数据,保证每次执行结果可控。测试用例沉淀成JSON快照,存历史对话快照和期望路径。每次改引擎逻辑后,跑一遍快照回归,观察解析结果和决策路径有没有变化。

这里要提醒一个点:Agent的回归测试和普通单元测试不一样,不能断言最终文本精确一致。正确的做法是断言结构,比如:

  • 最终是否成功调用了目标工具
  • 调用序列是否正确,比如“先查库存,再下单”
  • 对错误情况的处理是否符合预期
  • 没有出现未注册的工具调用

能写出这四类断言,回归测试就有效果了。

6.2 三个最关键的评估指标

我在这套引擎上长期观测的指标有三个,推荐你也用起来:

  • 工具调用成功率:模型发起工具调用后,解析、校验、执行全链路成功的比例
  • 参数合法率:发起调用的参数第一次就通过校验的比例,这个指标能反应工具描述写得好不好
  • 平均工具轮数:完成一个用户任务需要多少轮工具调用,轮数太高说明工具设计或者描述有问题

这三个指标配合trace日志,基本能定位引擎的大部分问题。比如参数合法率低,大概率是工具描述写得含糊;平均轮数高,可能是工具粒度太细,把多个小工具合并成一个组合工具反而更有效。

6.3 从单用户到多Agent共享引擎的演进

最后聊一个方向。工具引擎如果被多个Agent实例同时使用,库存、并发、幂等就会变成新课题。比如两个Agent同时调用“扣减库存”会发生什么?此时引擎层面要加分布式锁或用数据库事务保证一致性。另外,工具的执行权限应该绑定到会话级身份,不同用户能调用的工具集合可能不一样。等做到这一步,你的工具调用引擎就不再是“玩具”,而是具备了一定平台属性。

我个人觉得,工具调用引擎其实是整个Agent系统里最值得花时间打磨的部分。模型本身大家都在用同一个,拼的就是谁能把模型和真实世界的连接做得又稳又快。这篇文章里的模块划分和实现思路,很多是从线上踩坑里总结出来的,希望能给正在搭工具引擎的人一个可以参考的起点。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询