Agent技能模块化实战：解耦、注册表与稳定性设计

第一次尝试构建一个全能型Agent时我很快发现了一个尴尬的事实无论我把主循环写得多么巧妙真正决定好不好用的永远是那些挂在外面的小工具。我最早的那个Agent里塞了几十种能力从查天气到读PDF再到调用第三方API全堆在同一个文件里。功能确实是全能了可每次想改一个细节就得担心会不会把别的部分弄坏调一次参要跑半天测试。后来在社区里看到有人讨论“Agent Skills”这个概念我一下子意识到我一直缺的不是能力而是给能力分组、管理、隔离和编排的一套体系。如果你和我一样正在做一个面向多场景的AI助手、自动化机器人或者智能客服想把“一个能跑的Agent”进化成“一个稳定可靠的Agent系统”那这篇文章就是写给你的。我不会讲那种太空泛的架构理论而是从一个实践者踩坑的角度聊聊怎么把技能从Agent主体里拆出来怎么设计技能注册表怎么处理超时隔离以及怎么在生产环境里让技能持续进化。所有代码都是可以直接改改就用的所有结论都来自实际跑过的项目不是纸面推演。1. 技能与主循环解耦为什么通用Agent必然走向模块化1.1 单体Agent的失控从有用到不可维护先说说我那个反面教材式的第一版Agent。当时我为了做一个能自动处理会议纪要的助手给它塞了录音转文字、说话人分离、定制摘要模板、关键词提取、CRM系统录入六个功能。每个功能在单独测试时都表现良好可一旦放进同一个Agent循环里问题就来了一次会议纪要处理需要依次调用两三个技能而每个技能都依赖不同的配置参数和不同的内部状态。我花了整整两天去理清技能之间的调用关系最后发现光是一个“说话人分离”技能就和“录音转文字”共享了同一个临时目录结果两组并发任务直接互相覆盖了文件。这就是单体Agent的典型宿命能力越多耦合越重最终连一个Bug都修不动。如果你也遇到过类似的情况大概能理解我说的“失控”是什么意思——不是功能报错而是整个系统变得不可预测。直到我参考了开源社区里一些Agent框架的做法才真正意识到问题不在技能本身而在于我把技能直接焊死在了主循环里。Agent主循环本应该只做三件事理解用户意图、选择一个合适的技能、处理技能返回的结果。至于技能内部怎么实现、需要什么资源、怎样和外部系统交互都应该被隔离在技能自己的边界之内。1.2 技能的标准形态输入、输出与副作用在动手重构之前我给自己定了一个规矩任何技能都必须是一个标准的黑盒。所谓标准就是它对外只暴露三样东西——入参schema、出参schema、以及副作用声明。入参和出参很好理解就是一个技能能接受什么类型的数据、返回什么类型的数据。副作用声明是我在踩过坑之后才加上去的它用来描述这个技能是否会产生外部影响比如写文件、发请求、改数据库、启动子进程。举个例子一个“发送邮件”技能它的副作用就很明显会真的发一封邮件出去。一个“计算器”技能副作用基本为零。为什么声明副作用这么重要因为Agent主循环在并发调度技能时必须知道某个技能是否可以安全地并行执行。无副作用的技能可以同时跑很多个有副作用的技能就得加锁或者串行否则轻则数据错乱重则触发生产事故。我后来给所有技能统一做了一个接口约束核心部分大概是这样的dataclass class SkillResult: status: str # success, error, timeout output: dict # 结构化的返回数据 latency_ms: float # 耗时统计 dataclass class SkillManifest: name: str description: str input_schema: dict output_schema: dict has_side_effects: bool False timeout_ms: int 5000这套约束帮我解决了一个很现实的问题每当我想往Agent里加一个新能力不需要再去翻主循环的代码只需要写一个独立的函数填好manifest然后把技能注册进系统里就行。主循环永远是用同一套逻辑去调用技能不会因为新技能的增加而产生新的分支判断。1.3 解耦的价值扩展性、安全性与可测试性把技能从主循环里解耦出来带来的收益不是理论上的而是我可以直接量化感受得到的。第一是扩展性变好了。我想加一个新技能比如“航班查询”我不需要碰任何Agent核心代码只需要新建一个模块把查询逻辑包进去注册一下Agent立刻就能用了。这种体验就像给电脑插一个U盘插上就能认拔了也不影响别的设备。第二是安全性变好了。因为每个技能都声明了自己的副作用我在Agent主循环里加了一个简单的权限判断默认情况下所有技能都运行在一个受限的沙箱环境里只有明确声明了“需要外部网络”才能发出真实请求。这样即使某个技能内部有Bug它的破坏半径也是可控的不会感染到其他技能。第三是可测试性变好了。我以前测试一个Agent要串起整个流水线现在我可以单独对每个技能跑一套自动化测试甚至用mock数据来模拟外部依赖。回归测试的粒度一下子精细到了单个技能级别定位Bug的时间从半天缩短到了半小时以内。这里有一个很容易被忽略的小细节解耦之后你的项目目录结构也会跟着变得清晰。我现在是这么组织的agent-skills/ ├── core/ # Agent主循环、路由、调度 ├── skills/ # 具体技能实现 │ ├── email_sender/ │ │ └── skill.py │ ├── calculator/ │ │ └── skill.py │ └── ... ├── manifests/ # 技能的描述文件YAML/JSON └── tests/ └── skills/ # 每个技能单独一个测试文件这个结构我现在用了快半年每次往里加东西都觉得轻松很多。如果你还在被一个巨无霸Agent折磨我建议你先做一次简单的拆解把Agent里所有的函数按“能力”而不是“步骤”重新分组每个能力独立成模块再定义好输入输出。这个动作本身就已经迈出了模块化的第一步。2. 技能注册表以声明式配置驱动能力发现2.1 技能元数据设计让Agent“看懂”技能技能拆出来之后紧接着遇到的问题就是Agent主循环怎么知道有哪些技能可用又该在什么时候用哪个技能一开始我的做法是写一堆if-else比如“如果用户提到邮件关键词就调用send_email”。这招在技能不超过五个的时候还算好使等技能数量上了两位数if-else就变成了一场噩梦。每个关键词都要枚举重叠语义越来越多最终几乎没法维护。后来我换了个思路与其让Agent硬编码技能不如让Agent主动去查询一个注册表。这个注册表里存放着每个技能的元数据包括技能名称、描述、输入输出格式、适用场景的例子。Agent在收到用户请求后先从注册表里检索候选技能再根据用户的自然语言描述来匹配最合适的那个。这个元数据设计直接决定了技能匹配的准确率。我踩了很长时间的坑才总结出一套比较可靠的写法核心字段包括name技能的唯一标识最好是动词短语比如send_email、get_weather。description用一句话说清楚技能是干什么的尽量包含常见的同义词和上下位词比如“发送邮件/写邮件/给某人发消息”都写在描述里这样匹配率会明显上升。input_schema用JSON Schema描述输入参数包括每个参数的类型、是否必填、取值范围。output_schema定义返回结果的格式方便主循环以统一的方式处理。examples列出两到三个典型调用例子帮助大模型理解技能的使用场景。timeout_ms这个技能允许的最大执行时间。我是在项目里发现description写得越“啰嗦”匹配越准确。以前我写description: 发送邮件大模型经常在用户说“给经理发一封周报”时犹豫要不要用这个技能。后来我把描述改成description: 将指定内容通过电子邮件发送给一个或多个收件人支持正文、附件、抄送和密送适用于用户要求发信、写邮件、回复消息等场景识别率明显提升几乎不再有误判。2.2 路由决策LLM意图识别与规则兜底有了注册表接下来就是要决定怎么从注册表里选技能。我尝试过三种策略最终用的是“LLM主决策规则兜底”的组合模式。第一种是纯规则匹配。这个我已经说了技能多了以后完全不可用。第二种是纯LLM决策就是每次请求都问大模型“这里有十四个技能用户的需求是XXX你选哪个”。这种方法在技能较少时效果很好而且很灵活能处理各种口语化表达。但在技能数量很大、或者用户需求比较模糊时纯LLM决策有两个风险一是稳定性和可复现性差同样的话问两次可能选不同的技能二是延迟高每个请求都多一次大模型调用成本上不太划算。所以我的建议是第一层用关键词和向量检索缩小范围第二层再用LLM从Top K个候选中做最终确认。我的实现大致是这样一个流程把用户请求同时做两部分处理规则过滤器比如提取邮箱地址、日期、地点等实体和向量检索把用户请求embedding后和每个技能的描述做相似度检索取Top 3。如果规则过滤直接命中某个唯一技能比如请求里出现了明确的“发送邮件”意图就跳过LLM直接调用该技能。否则把Top 3的候选技能名称和描述拼成一个精简的prompt让LLM输出最终的选择和参数填充结果。如果LLM返回的结果不在候选列表里或者无法解析出合法的参数就返回“需求不明确”给用户让用户补充信息。这种分层路由的好处是大多数常见请求都能被规则层直接命中既快又省token只有那些语义模糊或者规则覆盖不到的请求才需要LLM出马。我在实际项目里测过大概有六成流量走规则层其余四成走LLM层整体准确率从原来纯规则的76%提升到了93%左右。2.3 一个最小可运行的注册表实现说了这么多理论我直接把一个可以跑起来的最小注册表实现贴出来。这个实现不算长但已经把核心逻辑都包含了技能加载、元数据校验、按描述检索、以及一个简单的路由决策接口。# registry.py import json import importlib from pathlib import Path from dataclasses import dataclass, asdict from typing import Dict, Optional class SkillRegistry: def __init__(self, manifests_dir: str): self._manifests_dir Path(manifests_dir) self._skills: Dict[str, dict] {} def load_manifests(self): 加载 manifests 目录下所有 JSON/YAML 描述文件 for file in self._manifests_dir.glob(*.json): manifest json.loads(file.read_text()) self._register(manifest) for file in self._manifests_dir.glob(*.yaml): import yaml manifest yaml.safe_load(file.read_text()) self._register(manifest) def _register(self, manifest: dict): name manifest.get(name) if not name: raise ValueError(技能缺少 name 字段) self._skills[name] manifest print(f[registry] 注册技能: {name}) def get_candidates(self, query: str, top_k: int 3) - list[dict]: 粗糙的检索先按关键词重叠打分后续可以换成向量检索。 这里只展示思路生产环境建议用 embedding 或 BM25。 query_tokens set(query.lower().split()) scored [] for manifest in self._skills.values(): desc_tokens set(manifest[description].lower().split()) overlap len(query_tokens desc_tokens) / max(len(query_tokens | desc_tokens), 1) scored.append((overlap, manifest)) scored.sort(keylambda x: x[0], reverseTrue) return [manifest for score, manifest in scored[:top_k]] # 使用示例 reg SkillRegistry(./manifests) reg.load_manifests() candidates reg.get_candidates(请发一封邮件给张三, top_k3) print(candidates)实际生产里我一般会把检索部分换成向量数据库因为纯粹的关键词重叠在中文场景下比较弱。但如果你只是在本地做个原型上面这个实现足够让你把整个流程跑通之后再平滑替换也不迟。3. 隔离、超时与降级别让一个坏技能拖垮整个Agent3.1 超时控制是技能的第一道保险我见过很多Agent项目技能调用直接用一个裸的函数调用没有超时、没有重试。这在Demo阶段没什么问题一旦上了生产环境一个外部API卡住整个Agent就跟着卡死用户等了几十秒得不到响应体验直接崩掉。给每个技能加超时是我做过的最便宜也最有效的改造。实现方式一般有两种一是用asyncio.wait_for如果技能是异步的二是用多进程/多线程加ThreadPoolExecutor如果技能是同步的。我倾向于让所有技能默认是异步函数然后用asyncio.wait_for统一包一层。下面这段代码是我在实际项目里用的技能执行器它做了三件事执行技能、记录延迟、在超时时抛出一个规范化的异常。# executor.py import asyncio from typing import Callable from dataclasses import dataclass, field import time dataclass class ExecutionStats: skill_name: str success: bool latency_ms: int error_type: Optional[str] None async def run_skill_with_timeout(skill_call: Callable, timeout_ms: int, *args, **kwargs) - ExecutionStats: start time.perf_counter() try: result await asyncio.wait_for(skill_call(*args, **kwargs), timeouttimeout_ms / 1000) elapsed_ms int((time.perf_counter() - start) * 1000) return ExecutionStats( skill_nameresult.get(skill_name, unknown), successTrue, latency_mselapsed_ms ) except asyncio.TimeoutError: elapsed_ms int((time.perf_counter() - start) * 1000) return ExecutionStats( skill_nametimeout, successFalse, latency_mselapsed_ms, error_typetimeout )超时时间怎么设置我的经验是第一版不要拍脑袋先看技能的P95延迟。方法很简单让技能在测试环境跑上一段时间统计一下耗时的分布然后把P95的1.5倍设为超时上限。比如一个技能P95是2.5秒那超时设为3.75秒比较合理。设得太短会让正常请求频繁失败设得太长又失去了兜底的意义。3.2 状态隔离与副作用管理除了超时状态隔离是另一个容易踩大坑的地方。我之前提到过两个技能共享临时目录导致互相覆盖的问题后来我彻底放弃了共享可变状态的思路改成了每个技能默认在自己独立的临时工作区里运行。实现起来不算复杂就是给每次技能调用都生成一个随机目录技能只能在这个目录里读写文件调用结束以后统一清理。如果技能需要跨调用保留状态那就不应该放在本地而是放到外部数据库或者对象存储里并且通过显式的参数传递。除了文件系统隔离还有环境变量和网络权限的隔离。我的做法是给每个技能声明一个permissions字段取值范围包括local_only、network、shell等。技能执行前执行器会做一次校验如果声明的权限和实际运行环境不匹配就拒绝启动。听起来有点繁琐但它能避免一个很实际的场景一个本身只该做计算的小技能因为依赖了某个内部库意外地访问了内部网络可能会把内网信息泄漏到日志里。另外一个容易被忽略的点是并发隔离。如果你的Agent是异步框架比如FastAPI多个用户可能同时触发同一个技能。如果这个技能内部包含非线程安全的全局变量就有可能在并发下产生不可预料的错误。我曾经在一个“报表生成”技能里用了一个模块级的变量保存Excel模板路径结果在高并发下出现了路径被篡改的问题。后来我把所有可变状态都收进技能的局部作用域再给需要串行访问的外部资源比如某台服务器的SSH连接加上一个asyncio.Lock这个问题才算彻底解决。3.3 降级策略与故障恢复最后再完备的隔离也没法保证技能永远不出错。所以必须在设计阶段就想清楚某个技能失败时Agent整体应该怎么办我自己的策略是三层降级第一层是有可替代技能时自动切换。比如“天气查询”技能挂了我可以切换到“通用网页搜索”技能把天气信息作为搜索结果返回给用户。虽然结果格式不一样但至少用户得到了可用信息。第二层是局部失败处理。比如一个技能要并发查询十个城市的天气其中两个失败了那我不会让整个请求失败而是把成功的那八个照常返回在结果里标注哪两个失败了并附上失败原因。这样用户体验是连贯的不会因为一个城市的数据缺失就什么都拿不到。第三层是兜底回复。如果所有候选技能都不可用Agent就返回一条明确的消息告诉用户“这个功能暂时不可用请稍后再试”而不是报一个技术性的Internal Server Error。这里有一个小技巧兜底回复最好也记录一条结构化日志留着后续做可用性统计。这三层降级策略我强烈建议你在写技能的时候就一并设计好而不是等到上线以后暴露了问题再去打补丁。因为你一旦上线每个小时都有真实用户在访问能留给你的修复窗口非常小提前把降级路径铺好等于给系统装了个缓冲垫至少不会摔得很难看。4. 从Demo到生产技能观测、评估与持续迭代4.1 技能的可观测性trace、metric与log技能模块化带来的另一个好处是可观测性变得清晰了。以前在单体Agent里我看一个请求的日志要把所有步骤的print翻一遍才能定位问题。现在每个技能都是独立单元我可以给每个技能调用自动埋点把关键信息统一上报到日志系统。我常用的字段包括技能名称、请求ID、输入参数摘要、输出结果摘要、耗时、状态成功/超时/失败、错误类型、重试次数等。每次技能调用这些信息都会生成一条结构化日志在分布式追踪系统里打一个span。这样当用户反馈一个问题时我可以从trace里直接看到是哪个技能出了问题、执行了多久、卡在哪一步。如果你用的框架还不支持distributed tracing也别急先手工在技能执行器里加一个日志钩子。我最初就是用一个简单的Python字典来聚合这些信息每天跑完以后导出一份CSV分析。量不大但已经足够发现问题了。这里我必须提醒一个很容易犯的错误不要在技能内部直接print日志更不要print完整输入参数。有些技能会接收敏感的输入比如邮件内容、手机号如果你直接把完整参数打到日志里不仅违反数据安全规范还可能在排查问题时把信息泄露给别人。正确做法是只打字段的哈希值或者脱敏后的摘要。4.2 技能质量评估从用例集到回归测试技能好不好用不能靠感觉得有数据。我在项目里维护了一套“技能回归测试用例集”每个技能至少对应三到五条测试用例覆盖正常请求、边界请求和异常请求。以“发送邮件”技能为例我会有这样几条用例正常场景合法的收件人和标题验证返回成功。非法参数收件人为空验证返回清晰的错误信息。外部依赖失败模拟SMTP服务不可用验证技能能否在超时前返回失败状态。副作用校验调用技能后确认没有产生意外文件或残留进程。这套用例集我放在项目根目录下每次对技能做改动后都会跑一遍pytest skills/tests。以前我改一个Agent功能要全链路手测现在只要技能层测试全过我就可以比较放心地把改动合入主分支。这个习惯帮我省下了大量手工回归的时间。另一个我觉得很值得做的事是给每个技能的返回结果打一个“用户满意度代理分”。比如对任务型技能我可以用“用户是否在收到结果后又发了一条追问”来粗评。如果用户每次都能满意地结束对话说明技能质量和预期匹配如果用户频繁在同一个技能之后追问“不对我要的是XX”那就说明技能的理解或者输出格式需要优化。我每个月会统计这个数据用来决定下个迭代周期的优先级。4.3 版本管理与灰度发布技能是独立的意味着它的版本迭代也应该独立。在我现在的系统里每个技能都有自己的版本号技能的注册表里记录着version字段。主Agent在调用技能时会带上版本信息这样如果我改动了一个技能我可以先在测试环境让一批测试用户用新版本其他用户继续用旧版本观察一段时间没有异常再全量切过去。灰度发布这个说法听起来很工程化但实现起来其实很简单。我的做法是在注册表里为同一个技能维护两个版本路由层根据请求的一个标记比如用户的账号ID取模决定走哪个版本。线上和线下同时跑我只要对比两个版本的耗时、成功率和用户反馈就能作出科学决策。这个机制特别适合那些需要调用外部API的技能。外部接口经常升级每次升级前我都无法确定它是否还保持原有行为所以我一定会让新旧两个版本并行跑一段时间。有一次第三方地图API偷偷改了返回字段结构我的新版本技能立刻和旧版本产生了差异正因为有灰度对比我在两小时内就发现了异常并回滚影响范围非常小。技能版本的管理还有一个深层好处它让整个系统的演化变成了可追溯的过程。每个技能的改变都能对应到一个版本号和一段发布记录出了问题可以精确回滚到上一个稳定版本而不是像以前那样改一版就得把整个系统重新部署。这种“小步快跑快速回滚”的方式是我目前在维护Agent技能库时最依赖的节奏。最后再说一个实操里的细节技能描述文件的版本变化会影响路由质量。如果你更新了某技能的描述一定要重新验证一下路由测试集别让新描述导致原本精准的技能匹配变得混乱。我有一次为了提高某个技能描述的分词效果把description里加了几句补充说明结果关键技能被更频繁地误选回归测试帮我抓了个正着。在实际做了快一年的Agent技能化改造后我想说再花哨的框架都不如一套清晰、隔离、可观测的技能体系来得踏实。每次往注册表里添加一个新技能看着它被主循环正常调度、被日志记录、被用例覆盖那种感觉就像在搭一个积木系统——每块积木都是独立的但组合起来却能爆发出完整的智能。