资讯详情

从工具函数到技能系统:打造高可靠AI Agent的工程实践

📅 2026/10/8 5:02:27 | 华诺云谱 👁 阅读
从工具函数到技能系统:打造高可靠AI Agent的工程实践
1. 为什么单纯的工具函数列表会在Agent规模变大时先崩溃先把话说在前面所有做AI Agent的项目最初都是从给模型挂几个函数开始的。我做过好几个从零到一的应用最开始的形式都差不多——一个Python字典里面每个key对应一个方法把函数名和描述塞给LLM让它决定什么时候调用、传什么参数。在小规模情景下确实管用三五个工具模型基本不会选错。但只要你把一个Agent真正推向业务工具数量滚到二三十个问题就一个接一个地冒出来。首先是Prompt越来越难写。你不可能把所有工具的完整JSON Schema都塞进系统提示词里Token会爆模型注意力会被稀释调用准确率肉眼可见地往下掉。其次是参数问题。多个工具之间存在隐式依赖比如先查用户部门再查该部门负责人这类流程如果全靠模型自由发挥它要么传错参数要么漏掉前置步骤。最要命的是故障隔离——某个外部API一时抖动异常直接穿透到Agent主循环整个对话就僵在那里User还在等模型已经死了。这就是我后来接触agent-skills这个概念、并重新组织整个Agent工程结构的契机。所谓Skills本质是给Agent的可执行能力做一个独立、显式、可管理的抽象层。它不是简单的函数列表每个Skill都是一等公民有自己的元数据面向模型与面向系统的描述、自己的参数契约、自己的失败策略和验证手段。系统不再把几十个函数平铺给模型而是通过一层注册表做动态检索和组合——模型先发现一个技能再决定是否调用调用后还能拿到结构化结果。这个抽层的价值在Agent规模增长和跨团队协作阶段会体现得非常明显。这篇文章就把我这套技能体系的设计和落地过程完整拆一遍。适用对象是那些已经做过基本工具调用的开发者以及正在纠结Agent代码越写越乱、模型越来越不听话的团队。会重点聊技能注册表、参数契约、执行引擎的可靠性和测试评估方案——全是当时踩坑之后觉得要是早点有人讲这个就好了的东西。2. 先搞懂技能层与裸函数清单的本质区别很多团队在引入Skill概念时容易犯一个错给每个函数套个类、起个名字就管自己叫技能系统了。实际上技能层要解决的核心问题有四个任何一个没做到你做的只是穿了马甲的函数字典。2.1 可发现性让模型在合适时机想到某个技能裸函数清单里模型对工具的感知完全依赖系统提示词里的文本描述。技能层则把发现这件事显式化每个技能带一组标签tags、能力描述description、适用条件triggers系统可以动态决定当前回合暴露哪些技能给模型。我在实际项目中验证过一种做法很有效维护一个技能索引Skill Index按语义相似度和历史调用频率做粗排每轮选择Top-K个技能注入上下文。这个策略大大降低了Prompt体积也能避免冷门技能永远躺在清单里没人用。2.2 参数自治每个技能自解释模型不需要记住全部细节传统工具调用的另一个痛点参数Schema复杂嵌套对象、枚举类型、条件必填模型在长上下文中很容易整出格式错误的JSON。技能层改变了游戏规则——每个技能自己声明入参并同时提供一份面向模型的参数说明书。核心思路是机器校验归机器校验模型理解归模型理解两者解耦。比如天气查询技能的机器Schema可能是{city: string, date: string}但面向模型的说明会写成city为城市中文名date为YYYY-MM-DD格式缺省表示今天。我在自己的框架里给每个技能增加了一个humanized_params字段实测模型参数正确率提升十几个百分点——记住模型不是读不懂Schema而是读太长太冷的Schema容易忽略关键信息。2.3 执行隔离跨技能故障不拖垮Agent主循环这是裸函数和技能层最实质的差别。裸函数异常会直接向上抛打穿调用链技能层则在边界上做统一拦截。任何技能执行都走同一个引擎引擎负责捕获所有Exception把它包装成结构化错误结果交还给Agent的决策循环而不是让进程崩溃、会话卡死。我甚至给技能执行加了独立的resource_limit配置——内存超限、执行超时、网络不可达都有默认处理和可降级路径。你可以把它理解为给每项能力一个沙箱沙箱里怎么折腾都不会炸到主程序。2.4 能力组合技能之间可以通过编排完成复杂任务技能层最终的目标不是让模型一次只用一个工具而是允许技能编排。举例来说生成周报这个技能内部可能依赖读取本周代码提交汇总合入请求拉取团队成员的日报三个子技能。组合发生在技能描述层而不是强行写死在应用的业务逻辑里。模块化带来的直接好处是一个技能可以被多个高层的业务技能复用——这背后就是Agent技能库的价值所在技能越多组合出复杂行为的能力越强而且每一条调用链都清晰可追踪。3. 技能注册表设计先定义清晰元数据再谈调度和扩展技能层的核心引擎就是注册表Registry。它承担的事情概括起来有三件登记、检索、体检。登记是在启动阶段把准备好的技能加载进内存检索是提供按语义、标签、描述查找的技能地址体检则是加载时发现的元数据和代码错误比如参数Schema不合法、依赖技能不存在等。3.1 技能的元数据结构如何让系统和模型各取所需我在项目中采用的Skill元数据字段如下经过几轮线上应用验证这套结构能稳定覆盖大多数业务场景字段类型用途说明namestring技能唯一标识命名风格统一为动词开头如query_orderversionstring技能版本号采用语义化版本当参数变更时递增主版本summarystring给模型看的一句话描述控制在30字内要能回答什么时候用descriptionstring较长的模型可读描述写明前置条件、执行后果、典型使用场景tagsstring[]给系统检索用的关键词可包含业务分类、能力类型等paramsobject机器可读的JSON Schema进行类型和格式校验humanized_paramsstring面向模型的中文参数说明描述每个字段的含义、格式、缺省行为dependenciesstring[]依赖的其他技能名注册时会做依赖检查fallbackstring降级技能的name主技能不可用时代替执行timeoutint该技能允许的最大执行秒数默认300多数人忽略的是description的写法。这块有真实经验可讲不要写根据输入参数查询数据库而应该写当用户询问订单状态、物流信息时使用输入为订单号若订单号缺失请先调用extract_order_id技能解析对话内容。模型是一种意图匹配器description越贴近用户的真实表达命中率越高。3.2 注册加载机制静态声明优于运行时扫描技能系统做到后面大家会发现加载机制决定了这个系统好不好维护。我自己第一版用的是启动时扫描目录、根据register_skill装饰器自动注册。开发初期很爽加一个新技能只要写个文件就行但项目推广到多人协作后代码里到处散布装饰器一个技能删了注册语句却忘了清理直接导致运行时出现幽灵技能定位问题浪费了大量时间。现在这套采用的是静态声明表单独维护一个skills.yaml列出所有启用的技能及其依赖、入口类名。Registry只认这个清单不在清单里的代码一律不加载。好处非常明显启用/停用一个技能改动一行配置即可新成员入门只需看这个文件对它做语法检查就能避免启动时找不到装饰器这种低级事故。3.3 依赖解析与冲突检测宁可启动失败不要运行期爆炸依赖解析是注册表最容易出问题的环节出现最多的三类报错是依赖技能不存在、循环依赖、同一技能名被重复注册。我的做法是把依赖检测做成启动阶段的硬校验——任何一个技能声明了自己的依赖启动时就必须把所有下游能力加载完毕否则直接拒绝启动。相关经验是建立一张有向图做拓扑排序按依赖顺序初始化技能实例。一个业务技能依赖多个底层技能时底层的实例先被构建并注入到上层。这套机制稳定跑过多个项目从未出现过运行期技能不存在的崩溃。关于冲突检测代码里一个重要约定上面也说过技能名在注册表内全局唯一重复注册直接启动失败。宁可启动时花几秒做检查也不让它变成一个半小时后才能发现的问题。4. 从零撸一个轻量技能系统注册、检索、执行全链路前面讲了很多设计原则这一章给一份可以直接抄作业的最小实现。整个系统跑在一个Python工程里核心文件就三个skill.py技能基类、registry.py注册与检索、executor.py执行引擎。整体设计目标是把技能代码和调度逻辑彻底刨开让队友写技能的时候不需要关心谁在调用它。4.1 Skill基类约定接口保护实现先定义技能基类所有业务技能都继承并能统一被Executor加载# skill.py from abc import ABC, abstractmethod from typing import Any, Dict, Optional class BaseSkill(ABC): name: str version: str 1.0.0 summary: str description: str tags: list[str] [] params: Dict[str, Any] {} humanized_params: str dependencies: list[str] [] timeout: int 300 def __init__(self, registryNone): self.registry registry abstractmethod async def execute(self, **kwargs) - Any: 技能主执行逻辑注意这里的kwargs是经过校验之后的实参 pass async def validate(self, **kwargs) - Optional[str]: 可选的自定义前置校验返回错误字符串或None return None async def fallback_execute(self, **kwargs) - Any: 降级逻辑默认类同execute可被子类覆盖实现降级策略 return await self.execute(**kwargs)这里有几个细节值得注意dependencies要写子技能名一旦声明Registry会在启动时把依赖技能实例注入到一个全局上下文表技能内部通过self.registry.get_skill(dep_name)获取依赖。注入时间选在初始化不要放到每次调用去查能省很多函数调用开销。4.2 Registry注册、拓扑构建、语义检索接下来是注册表。关键是它不仅要能按名字取技能还要让上层调度能按意图取技能。我实现了一个简单的基于关键词/标签匹配的检索器没有引入重量级的向量数据库业界常用方案是先按标签精确匹配做粗筛再对description做关键词打分做精排。对小规模技能库够用了将来技能破百再换向量检索引擎。# registry.py import yaml from typing import Any, Dict, List, Optional class SkillRegistry: def __init__(self): self._skills: Dict[str, BaseSkill] {} self._tags_index: Dict[str, List[str]] {} def load_from_config(self, config_path: str): 从skills.yaml静态声明表加载技能 with open(config_path, r, encodingutf-8) as f: config yaml.safe_load(f) for entry in config[skills]: module __import__(entry[module], fromlist[]) skill_cls getattr(module, entry[class_name]) skill_instance skill_cls(registryself) self.register(skill_instance) def register(self, skill: BaseSkill) - None: if skill.name in self._skills: raise RuntimeError(fDuplicate skill name: {skill.name}) self._skills[skill.name] skill for tag in skill.tags: self._tags_index.setdefault(tag, []).append(skill.name) def get_skill(self, name: str) - Optional[BaseSkill]: return self._skills.get(name) def search(self, query: str, top_k: int 10) - List[BaseSkill]: 简易意图检索先标签粗筛再走评分精排 candidates set() tokens query.lower().split() for token in tokens: candidates.update(self._tags_index.get(token, [])) if not candidates: candidates set(self._skills.keys()) scored [] for name in candidates: skill self._skills[name] score self._score_skill(skill, tokens) scored.append((score, skill)) scored.sort(keylambda x: -x[0]) return [s for _, s in scored[:top_k]] def _score_skill(self, skill: BaseSkill, tokens: List[str]) - int: score 0 desc f{skill.summary} {skill.description} {skill.name} for token in tokens: if token in skill.tags: score 3 if token in skill.name: score 2 if token in desc.lower(): score 1 return score def validate_dependencies(self) - None: 拓扑检查依赖必须存在且无环 visiting set() visited set() for name in self._skills: self._dfs(name, visiting, visited) def _dfs(self, name: str, visiting: set, visited: set) - None: if name in visiting: raise RuntimeError(fCircular dependency detected: {name}) if name in visited: return visiting.add(name) skill self._skills[name] for dep in skill.dependencies: if dep not in self._skills: raise RuntimeError(fSkill {name} depends on missing skill {dep}) self._dfs(dep, visiting, visited) visiting.remove(name) visited.add(name)写这段时反复提醒自己的一句话是检索逻辑和调用逻辑一定要分开。实际项目后来加入了权重打分、context rerank等高级特性但核心思想就是这个——先粗筛再精排别让Agent每轮都把全部技能读一遍。4.3 Executor统一入口、参数校验、超时控制执行引擎是技能边界。对外暴露一个方法execute_skill(name, **kwargs)它在内部完成参数校验、依赖解析、超时控制、异常包装最后把结构化结果交还给调用方无论是LLM调度循环还是手动调试。# executor.py import asyncio, logging from typing import Any class SkillExecutor: def __init__(self, registry: SkillRegistry): self.registry registry self.logger logging.getLogger(skill_executor) async def execute_skill(self, name: str, **kwargs) - Dict[str, Any]: start asyncio.get_event_loop().time() skill self.registry.get_skill(name) if skill is None: return {ok: False, skill: name, error: SKILL_NOT_FOUND, result: None} # 1. 参数Struct校验 try: validated self._validate_params(skill.params, kwargs) except Exception as e: return self._failure(name, INVALID_PARAMS, str(e), start) # 2. 业务前置校验 try: pre_error await skill.validate(**validated) if pre_error: return self._failure(name, VALIDATE_FAILED, pre_error, start) except Exception as e: return self._failure(name, VALIDATE_EXCEPTION, str(e), start) # 3. 超时控制的执行管线 try: result await asyncio.wait_for(skill.execute(**validated), timeoutskill.timeout) return {ok: True, skill: name, result: result, latency: self._elapsed(start)} except asyncio.TimeoutError: # 4. 超时降级 self.logger.warning(fSkill {name} timeout, try fallback) try: fb_result await asyncio.wait_for(skill.fallback_execute(**validated), timeoutskill.timeout) return {ok: True, skill: name, result: fb_result, fallback: True, latency: self._elapsed(start)} except Exception as e: return self._failure(name, TIMEOUT_FALLBACK_FAILED, str(e), start) except Exception as e: return self._failure(name, EXECUTION_FAILED, str(e), start) def _validate_params(self, schema: Dict[str, Any], kwargs: Dict[str, Any]) - Dict[str, Any]: # 实现轻量JSON Schema校验这里略写可直接用jsonschema库 return kwargs def _failure(self, name: str, code: str, msg: str, start: float) - Dict[str, Any]: return {ok: False, skill: name, error: code, error_message: msg, latency: self._elapsed(start)} def _elapsed(self, start: float) - float: return round(asyncio.get_event_loop().time() - start, 3)这里最关键的一条设计决策是不管底层发生了什么Executor永远返回一个字典绝不让上层感知到Exception级别的事件。这样Agent的调度循环只需判断ok字段真假无论什么原因失败都能拿到可读的错误码和错误消息再喂回给LLM做下一步决策。4.4 一个完整的业务场景示例订单查询技能拿一个贴近日常的场景做演示——企业内部订单查询Agent。它的技能有parse_order_context从聊天文本中抽取订单号、query_order_by_id查询订单详情、get_customer_info查询下单客户信息、compose_order_reply把查询结果整理成客户可读的回复。技能query_order_by_id可以这样写# skills/order_skills.py from skill import BaseSkill class QueryOrderById(BaseSkill): name query_order_by_id summary 根据订单号查询订单详情 description 当用户询问订单状态、物流信息、支付金额时使用输入为订单号如果对话中没有订单号不要直接调用本技能先调用 parse_order_context 提取订单号 tags [order, query, 物流] params {order_id: {type: string, required: True}} humanized_params order_id 为订单号字符串例如 OD20240601XXXX async def execute(self, order_id: str, **kwargs): # 在这里写查数据库或调用内部订单服务的逻辑 return {order_id: order_id, status: shipped, amount: 199.00}写技能description这段我反复打磨过很多版本。一个建议是用当……时使用如果……不要调用句式这话对模型有极强的引导力正例和反例都给了比单纯正向描述命中率高很多。5. 执行引擎的可靠性细节超时、重试、降级与错误信息的误区技能系统只要上线执行引擎的可靠性就决定了整个Agent的稳定性。单纯在技能内部写try/except属于基本功但有几层细节经常被跳过。5.1 超时与取消asyncio.wait_for不一定够Executer里的asyncio.wait_for只对协程有效。如果你在技能里调用了同步阻塞的第三方库线程会被真正卡住取消协程没有意义。实测中我踩过一个大坑某个技能内部用了同步的requests.post外网接口偶发卡顿30秒asyncio.wait_for把协程取消了但线程池里的线程还在阻塞最终把整个事件循环的线程池拖垮。解决方式把同步阻塞调用扔到独立线程池执行并给线程池设置最大工作线程数。业界常用asyncio.to_thread包一层它基于ThreadPoolExecutor至少要保证阻塞操作不会占用事件循环线程import asyncio async def blocking_skill_execute(): # 同步阻塞的第三方调用 result await asyncio.to_thread(requests.post, https://api.internal.example.com/query, timeout5) return result.json()对于CPU密集的调用同样问题事件循环一旦被占满其他所有任务的超时统计都会失真。比较好的方案是干脆把重负载技能做成独立的子进程或服务技能层通过RPC通信但这属于大工程一般等体量起来了再升级也来得及。5.2 重试策略什么时候该重试什么时候立刻返回无脑重试是一个常见但危险的误区。外部API瞬断可以重试参数校验失败重试一万次也没意义。我的分类策略很直接错误类型动作理由SKILL_NOT_FOUND不重试技能名错误重试只会浪费TokenINVALID_PARAMS不重试回传错误消息给LLM说明消息本身描述得不清楚需要让LLM重新理解并补参EXTERNAL_TIMEOUT最多重试1次且走降级技能连续超时的服务大概率在过载继续重试会放大压力NETWORK_UNAVAILABLE立即走降级服务级故障不是一个技能能解决的重试逻辑放在技能内部还是Executor统一做我倾向放在技能内部——每个技能最了解自己的依赖容错策略。技能内部有_call_with_retry这样的工具函数主动性更强。5.3 降级设计给定一个说法而不是一个错误跌代链设计的教训降级不是返回None那么简单。模型在拿到一个空结果之后往往会自己脑补内容导致用户看到幻觉。所以降级返回的内容应该是面向Agent决策的、一段带原因的说明文字让模型知道自己能做什么、不能做什么。举例如果一个订单查询技能超时了降级返回内容可以写订单服务暂时不可用本次无法获取订单详情。请告知用户系统正在维护并向用户致歉。这是给LLM的控制指令而不是给用户的。我实测中这种降级即提示词的方式比让模型自己去猜回复要好得多用户满意度也不会被技术故障拖垮。5.4 结构化错误消息让LLM能自救错误信息设计有一个很重要的原则错误消息不只是给人看的更是给LLM看的。你要把错误分类、给修复建议、给可调用技能的提示都浓缩到一段短文本里。比如INVALID_PARAMS错误如果Executor直接回传{error: order_id required}LLM大概率不知道该怎么办。而回传{error: 缺少订单号请提取用户对话中的订单号如果对话中没有明确订单号调用 parse_order_context 从上下文中提取或主动向用户询问订单号}LLM能立刻进入正确修复路径。这是我在线上对比过的一组数据优化错误描述之后Agent单轮恢复率从61%升到82%本质就是让模型有路可走而不是让它撞了墙还要自己凿墙。6. 技能的测试与评估没有评测体系技能越加越虚技能库的增长速度快于代码库的增长速度。我见过最大的一个Agent项目有140多个技能如果没有对应的测试与评估体系上线一次基本等于拆盲盒——最好的自保方式就是给技能建两条生命线单元测试和命中率评测。6.1 单元测试只测执行器不依赖LLM技能模块本身尽量用纯函数写业务逻辑方便做单测。测试的重点永远放在三块参数校验逻辑、核心业务处理逻辑、异常与超时后的返回结构。样例# tests/test_query_order.py import pytest async def test_query_order_success(executor): result await executor.execute_skill(query_order_by_id, order_idOD20240601XXXX) assert result[ok] is True assert result[result][status] shipped async def test_query_order_missing_param(executor): result await executor.execute_skill(query_order_by_id) assert result[ok] is False assert result[error] INVALID_PARAMS这种测试的最大价值是给重构兜底。技能内部从直连数据库换成调用微服务只要测试用例还在就能很快判断行为没有偏移。另外建议把测试用例的输入参数固定写死不要依赖实时外部服务否则测试会变得不稳定你就不愿意跑了。6.2 命中率评测用技能调用样例集衡量description质量一个技能描述写得再漂亮到了真实文本里该不该被调用需要一个客观标准。我的做法是维护一个评测集每条数据是真实用户语句 期望调用的技能名 期望参数例如{query: 帮我查一下昨天那单的物流, expect_skill: query_order_by_id, expect_params: {order_id: OD20240601XXXX}}评测流程跑一遍注册表检索和LLM调度统计两个指标命中率该调用的被调用了和误伤率不该调用的被调用了。别看不起这两个指标它们直接反映description写得怎么样。我把真实线上日志导出来人工标了200条样例跑完测试发现有两个技能description写得太泛在查物流和查订单场景互相误伤随后做了一轮描述精修命中率从70%拉到91%。6.3 长尾调优的实操方法给模型喂负例这里分享一个可复用的调优套路改技能描述的时候除了优化正向用例的描述一定要收集几个最容易误用的样例把反例直接写进去。例如在query_order_by_id的description末尾加一句注意如果用户在询问退换货进度请调用query_return_request不要调用本技能。很多团队忽略负例的价值结果描述越写越长模型反而越容易选错。我自己的经验是每个技能的description控制在80到150字之间其中至少有一句负例超过200字的描述命中率反而会有下降趋势。7. 技能体系上线之后我踩过的坑和坚持至今的几条约定最后按惯例记录几条经验都是真实项目中反复踩过、最后形成了约定俗成规则的第一技能命名要动词开头、全局唯一且不要用缩写。queryOrder01这种命名在调试日志里根本没法看。命名风格统一为动作_对象比如send_email_digest、ban_user。这不仅能提升检索算法的评分也方便多人协作时描述与代码对得上号。第二技能返回值必须可序列化。这句话重要到要加粗。无论你的技能内部返回什么对象——pandas DataFrame、自定义ORM类、生成器——在Executor返回给Agent调度循环之前必须转成JSON可序列化的结构。否则一旦接入LLM序列化坑会让你半夜爬起来加班。通用的处理方式是在基类写一个_serialize_result子类按需重写。第三技能描述变更要走评审跟代码变更一样的流程。一个小经验description被质量低劣的外包团队改了往往是直接复制粘贴堆一段功能性描述一本正经地说本技能用于查询数据库中的订单信息并返回结果这种东西对模型的引导性为零。我把description的评审标准写进了团队的技能贡献文档不以开发者的视角描述技能做了什么而以用户的视角描述用户在什么时候会需要它。第四始终保留技能版本的灰度开关。新技能上线先是20%流量观察几天误伤率和调用频次稳定后再全量。注册表里每个技能带enabled字段运行时通过配置中心动态启停不需要重新部署。这套技能体系在三个项目上应用了最大的一个线上Agent每天处理几千次技能调用关键技能命中率维持在90%以上崩溃事件几乎归零。回头去看技能层最大的意义不是更高大上的架构而是给了团队一套可以讨论、可以迭代、可以测试的共同语言——每个人都知道一个技能该有什么字段、该写什么描述、该怎么验证协作效率自然就上来了。如果你现在的Agent代码开始出现工具越加越乱、模型选择越来越不可控的味道找个周末按这套思路把技能层搭起来后续维护成本的下限会非常可观。
📝

华诺云谱内容团队

资深建站顾问 · 行业研究员

10年+企业数字化服务经验,专注智能建站、SEO优化与品牌营销,持续输出建站技巧、行业洞察与营销干货,已帮助5000+企业实现数字化增长。

你可能需要的服务

订阅华诺云谱资讯周报

每周一封,精选建站技巧、SEO与营销干货,直达邮箱。已有 8,000+ 企业主订阅,助你少走弯路。

↑