手搓AI Agent工具调用引擎：从注册到循环编排的完整实践

这一个多月我一直在手搓一个Agent项目，前两周分别完成了大模型接口封装和上下文管理，这周总算轮到最硬核的部分——实现Agent工具调用引擎。先摆结论：工具调用这一层，别太迷信框架的默认实现，自己搭一遍看起来多花时间，但你会对"什么情况下会翻车"建立完全不一样的直觉。这篇就把我这周的完整设计和踩坑过程拆开讲，从工具注册、参数校验、执行分发到循环编排，全部给到可直接抄作业的代码和思路，适合正在做AI Agent开发、或者想从零理解Agent架构的朋友。

1. 先想清楚：工具调用引擎到底在解决什么问题

1.1 模型没有"手"，工具调用就是给模型装手

很多人一上来就写循环调API，但没想明白工具调用的本质。大模型本质上是一个文本生成器，你问它"帮我查一下当前订单的物流状态"，它脑子里根本没有查询订单数据库的能力，它只能"猜"一个答案。要让Agent具备真实操作能力，必须建立一个机制：模型以结构化文本输出一个"调用意图"（我想调用get_order_status这个工具，参数是order_id=12345），然后由引擎真正执行这个函数，把结果作为新的上下文喂回给模型，让模型基于真实数据继续推理。

这个过程在OpenAI的API里叫Function Calling，在Anthropic里叫Tool Use，国内各家模型也基本对齐了这套协议。不管名字怎么变，核心流程都一样：模型不直接执行代码，它只负责"决策"和"生成"，真正动手的是引擎。这个分工非常重要，它决定了安全边界——模型永远不应该拿到直接执行任意代码的能力，它只能在你预设好的工具清单里做选择。

1.2 引擎要管的四件事

把概念理清之后，工具调用引擎的职责就非常明确了，一共四件事：

• 工具描述与注册：把你写好的Python函数翻译成模型能看懂的JSON Schema说明书，让模型知道"有什么工具、什么时候用、参数怎么传"。
• 调用请求解析：模型吐出来的是文本形式的tool_call，里面包含工具名和JSON字符串参数，引擎要负责解析、校验、纠错。
• 安全执行与结果回填：真正去运行目标函数，捕获异常、控制超时，然后把结构化的执行结果组装成模型需要的消息格式。
• 循环编排与终止：决定"模型说要调工具→执行→结果喂回去→模型再决策"这个循环跑几轮、什么时候停、怎么防止死循环。

听起来不复杂，但每一环都有隐蔽的坑。比如模型偶尔会拼错工具名，参数JSON会缺括号，工具返回超长文本会撑爆上下文窗口，模型还会在循环里反复调用同一个工具出不来。这些我在后面都会逐个说到。

2. 整体设计：为什么自己搭而不是直接套框架

2.1 框架能帮你什么，不能帮你什么

市面上的Agent框架已经不少了，LangChain、Dify、CrewAI各有各的拥趸。我前期也调研过一圈，最后决定核心引擎自己写，框架顶多当参考。原因很简单：框架的抽象层级太高，把工具注册、消息组装、模型调用全包在一起，一旦出问题你根本不知道是哪一环断了。

不是说框架不好，你的业务场景复杂、团队人手紧，用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( name=name, description=description, parameters=parameters, handler=func, ) 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( name="get_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(timeout=N)。为什么要这么干？因为真实世界里工具函数可能是查数据库、调外部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_workers=1) as pool: future = pool.submit(spec.handler, **arguments) result = future.result(timeout=10) 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( model=MODEL_NAME, messages=messages, tools=registry.get_openai_tools(), tool_choice="auto", ) 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_ascii=False), }) # 每轮结束可以在这里做一轮上下文压缩（见4.2） return "已达最大执行轮数，任务可能尚未完成。请基于已有信息生成阶段性结论并告知用户。"

这里有一个设计细节很关键：tool_choice="auto"意味着让模型自己判断"这轮要不要调工具、调哪个"。也有场景需要强制调某个工具（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这类开放的工具体系协议，让引擎可以对接第三方提供的标准化工具，而不是每个工具都自己写。等我把这两块跑通，再来继续更新。