Agent技能的工程化落地:一个可插拔技能系统的设计与实践
这几年做AI Agent项目的人越来越多了,但一个很普遍的现象是:大家的demo跑得很欢,一旦往生产环境推,问题就全冒出来了。最典型的就是“工具越堆越多,调用越来越乱”——今天加一个天气查询,明天加一个数据库读写,后天塞一个数据分析脚本,最后Agent在几十个function之间反复横跳,该调用的不调用,不该调用的瞎调用,上下文还被一堆冗长描述挤爆了。
我最近在某个偏中大型的智能体项目里,专门花了两周时间把这块推倒重做,核心就是围绕"agent-skills"设计了一套可插拔的技能体系。这套体系解决的不只是“工具怎么塞给Agent”,而是把技能的定义、装载、调度、更新、下线、观测都收拢到一个统一框架里。今天想把整个设计和实现过程完整拆出来,包括踩过的坑、调整过的方案、最终跑稳的配置方式,给同样在做Agent落地的朋友一个可直接参考的样板。
1. 先从源头说起:Agent技能系统到底想解决什么问题
1.1 当场景越做越复杂,功能碎片化开始失控
我做Agent项目过程中比较早遇到的一个项目,它的场景是“让Agent帮运营团队处理日常数据报表”。一开始思路很直接,把需要的工具写成Python函数,然后通过function calling暴露给大模型。刚开始只有三五个函数,跑得还行。但运营侧需求膨胀得很快:
- 要查日报、周报、月报,还要查某个渠道的细分维度
- 要支持模糊搜索、精确匹配、时间范围筛选
- 要能做同环比的自动计算,还要把结果格式化成简报
- 后来甚至要求Agent可以主动订阅某类指标的异常波动
每个新需求加进来,就往代码里堆一个新的function。很快系统里就有了二十多个函数,命名风格也开始不统一——有的叫get_daily_report,有的叫query_channel_data_v3,还有的叫report_exception_alerts。大模型经常在多个功能相近的函数里做错误选择,比如用户问“昨天华东区的转化怎么样”,模型不去调query_channel_data_v3,反而去调get_daily_report,结果拿回来一坨不相关的数据。
这个问题的本质是:工具本身的代码实现和Agent对它能力范围的“认知”被耦合在一起了。模型只能靠函数名和description去猜测这个函数是干什么的,而函数一多,这种“猜测”就变得不可靠。我意识到必须换一种组织方式——不再面向函数编程,而是面向技能编程。
1.2 技能的全生命周期:从定义、注册、更新到下线
换成技能思维之后,我先重新梳理了需求。所谓“技能”,本质上是对一组功能性能力的封装,但这个封装不能只是代码层面的,它必须包含四个关键部分:
- 技能定义:描述这个技能是干什么的、适合什么场景、不适合什么场景、输入输出是什么
- 技能实现:真正可执行的代码逻辑,可以是一个函数、一组API调用、或者一串脚本
- 技能注册:把它登记到一个中心化的技能清单里,让调度器知道有这么一个技能存在
- 技能生命周期管理:包括版本的升级、停用、灰度、回滚,以及运行时的观测和告警
让我觉得这个思路值得坚持的,是它带来的一个连带好处:当每个能力都被抽成独立技能,Agent的调度就从“大海捞针式地在函数列表里找”变成了“在一个有结构的技能集里匹配”。同时,运维层面也可以单独对某项技能配置权限、频控、超时、重试策略,而不用动整体代码。
打个不太严谨但贴切的比方:之前的Agent像是拿着一本写得密密麻麻的通讯录,每次打电话都要从头翻到尾;技能系统相当于给通讯录加上了标签、分组和搜索索引——通讯录还是那本,但找人快多了,也不容易找错人。
1.3 做技能框架前,先理清边界:哪些东西算技能,哪些不算
有一个容易被忽略的问题,就是技能的边界到底划在哪里。我一开始踩过坑,几乎把什么功能都往技能里塞——结果技能系统和业务代码的耦合越来越深,维护反而更累了。
后面总结出几个划分原则:
- 技能不是业务流程:一次包含多步骤的审批流转,不该写成一个技能。技能应该服务于“给用户发一份审批通过通知”这样明确、单次可完成的操作;多步骤应该靠Agent的编排能力串联多个技能来完成。
- 技能需要可被自然语言描述:如果这个能力你自己都说不清楚在什么场景下用、怎么用,那模型大概率也搞不清。描述不清楚的技能,要么补充完善,要么说明粒度不对。
- 技能要有明确输入输出:输入参数尽量扁平化,输出结构化。如果一个技能的入参有七八个,而且互相依赖、需要内部状态才能执行,说明这个技能设计得抄了不该施的重。
这三条成了我后来判断“要不要做成技能”的金标准。
2. 技能框架的核心设计:数据结构、注册机制与调度策略
2.1 技能描述的数据结构:让大模型看一眼就懂
技能的“说明书”是整个系统里最核心的部分,直接决定了模型会不会选它、怎么调用它。我最终采用的是多字段描述+双层摘要的结构,每个技能的元数据字段大致如下:
| 字段 | 类型 | 说明 |
|---|---|---|
skill_id | string | 全局唯一技能ID,语义化命名,如channel_data.query |
name | string | 技能短名称,用于模型理解记忆 |
description | string | 技能用途,自然语言描述,150字以内 |
intents | list | 可能触发该技能的意图列表,如["查询渠道数据","渠道转化分析"] |
tags | list | 分类标签,如["数据", "报表", "运营"] |
parameters_schema | object | JSON Schema定义输入参数结构 |
output_schema | object | JSON Schema定义输出结构 |
permission_level | enum | 权限级别,如公共/受限/内部 |
cost_score | int | 调用成本评分,供调度参考 |
version | string | 技能版本号 |
intents字段是我强烈建议做的。一开始没有这个字段,模型只能靠description来“悟”,很多模糊意图就匹配偏了。加了intents之后,相当于给每个技能预置了多个触发场景的“锚点”,尤其是当用户的问法很口语化或省略时,命中率提升非常明显。
另一个细节是parameters_schema。很多团队用JSON Schema,但写得非常粗糙——只写了字段类型和是否必填,没有给枚举范围、字段含义、字段间约束。模型调用时就容易传一些离谱的参数。我后来要求所有技能都必须把枚举值、最小值、最大值、默认值、示例值写全,模型的参数正确率肉眼可见地提高了。
2.2 技能注册中心与动态加载机制
技能的注册和加载,我是用一套独立的注册中心来管理的。这个注册中心可以理解成技能系统的“大脑”,它掌握着当前环境里有多少技能可用、每个技能的版本是多少、状态是启用还是停用。
具体实现上,我采用的是注册表+插件化加载的组合方式:
- 每个技能是一个独立的模块,结构上可以是目录或单文件
- 模块内除代码实现外,必须附带一个
skill.yaml或skill.json作为技能元数据声明文件 - 框架启动时,扫描技能目录,解析所有声明文件,进行格式校验和ID冲突检测
- 校验通过后注册到内存注册表;失败技能单独标记,不影响其他技能加载
这样做的好处是,新增一个技能不需要改任何框架代码,只要往技能目录里丢一个符合规范的模块,重启(或触发热加载)后技能就自动注册了。技能做得多了之后,这种“无侵入式扩展”太重要了——你不会想每次加技能都把主程序改一遍的。
代码草图大致是这个样子:
class SkillRegistry: def __init__(self): self._skills = {} self._lock = asyncio.Lock() self._watcher = FileSystemWatcher(skill_dir, filter="skill.yaml") async def load_all(self): for skill_path in scan_skill_directories(): try: skill_meta = parse_yaml(skill_path / "skill.yaml") validate_metadata(skill_meta) module_loader = importlib.import_module(skill_meta.entry_point) skill = Skill( meta=skill_meta, handler=getattr(module_loader, "run"), ) await self._lock.acquire() self._skills[skill_meta.skill_id] = skill except Exception as e: log.skill_load_failed(skill_path, error=str(e)) def get(self, skill_id): ... def list_enabled(self): ...2.3 调度策略的取舍:一次全量投喂还是动态摘选
技能数量一旦多起来,还面临一个关键技术问题:把所有技能的说明都塞进系统提示词吗?如果技能只有5个,全量投喂没问题;但你有30个、50个技能,每个技能光description就有150字,再加上parameters_schema,一次就要吃掉好几千token。这类成本并不是一个小数目,而且会让模型的注意力被稀释。
这一步我试过两种路线:
方案A:全量投喂把注册表里所有启用的技能元数据都拼进系统提示词,让模型在完整集合里选择。优点是实现简单,模型能看到全局。缺点是上下文占用过高、调度响应慢,技能多了以后准确率反而下降。
方案B:动态摘选(两阶段检索)第一步,先把技能的name、description、intents、tags做成轻量索引(甚至可以构建一个向量索引做embedding检索),拿用户当前输入先去检索一遍,召回top K个候选技能。第二步,只把这几个候选技能的完整元数据(含完整参数schema)投给模型做最终选择。这样既控制了上下文长度,又保留了选择精度。
我最后落地的是方案B。第一阶段的检索我用的是关键词+向量混合检索:关键词做粗筛,向量做语义补充,然后把两个结果做加权融合,取top K。K值经过实测,取8~10比较合适——覆盖度够,token消耗也可控。
在微调完这一步之后,整体调度效果有了明显改善:上下文占用降到原来的1/4左右,技能选择的准确率从87%提升到96%上下,而且技能数量继续膨胀时,性能不会跟着劣化。
3. 实战落地:从零搭一套可用技能模块
3.1 第一步:搭框架基础设施,先把“空架子”立起来
我在新建的项目里部署这套框架时,环境的搭建大约花了半天。分成几块基础设施:
- 技能注册中心:承担元数据解析、加载、版本管理、状态管理
- 调度模块:负责用户意图到技能的匹配,以及参数生成和调用链编排
- 执行沙箱:逐一封装技能的安全执行环境
- 观测面板:收集调用日志、耗时、成功率、token消耗等数据
前面已经写过注册中心的核心代码,这里补充一下调度模块里最关键的“意图检索器”执行逻辑:
class SkillSelector: def __init__(self, registry, retriever): self._registry = registry self._retriever = retriever async def select(self, user_input: str, top_k: int = 8): all_skills = self._registry.list_enabled() retrievals = await asyncio.gather( self._retriever.keyword_search(user_input, skills=all_skills), self._retriever.vector_search(user_input, skills=all_skills), ) fused = rank_skills( keyword_score=retrievals[0], vector_score=retrievals[1], alpha=0.4, ) candidates = fused[:top_k] return candidates实际跑起来之后发现,光有选择器还不够——模型选定技能后要填参数,参数填不对等于白选。所以我额外做了一步“参数预校验”:模型生成参数后,先丢给JSON Schema校验器跑一遍,不合格的直接返回错误信息让模型重填。这把运行时参数错误率压到了一半以上。
3.2 核心实操:写第一个技能——渠道数据查询
在一个技能系统中,“数据查询类”技能是最常见、最基础,也最适合拿来练手的门类。下面以一个渠道数据查询技能为例,走一遍完整定义流程。
第一步,创建技能目录:
skills/ channel_data.query/ skill.yaml main.py tests/第二步,编写skill.yaml。我反复强调的一个原则是:description尽量用“为解决XX,在XX场景下,做XX”的句式,不要写成代码注释。针对该场景,我最初的版本是:
id: channel_data.query name: 渠道数据查询 description: 查询指定渠道、指定日期范围内的核心指标数据,支持按渠道分组、按指标筛选。 intents: - 查询渠道数据 - 看下各渠道的转化率 - 某个渠道最近几天的数据 tags: [数据分析, 渠道, 报表] parameters_schema: type: object properties: channel: type: string enum: [google, facebook, tiktok, organic] description: 渠道标识 start_date: type: string format: date end_date: type: string format: date metrics: type: array items: { type: string } description: 需要返回的指标列表 required: [channel, start_date, end_date] output_schema: type: object properties: rows: { type: array } summary: { type: object }第三步,编写执行函数。这里有个值得分享的问题:第一次给技能写执行函数时,我把数据源连接逻辑直接写死在技能里,后来发现这样一点都不好维护——数据源连接信息应该是全局配置,技能内部只该做“取数→处理→返回”。于是改成执行函数从外部注入数据源客户端,技能的代码变得更干净,也更容易测试。
最终技能执行函数的大致逻辑:
async def run(inputs: dict, ctx: SkillContext): channel = inputs["channel"] date_range = (inputs["start_date"], inputs["end_date"]) metrics = inputs.get("metrics", ["impressions", "clicks", "conversions"]) dataset = await ctx.database.query( table="channel_daily_stats", channel=channel, date_range=date_range, metrics=metrics, ) return { "rows": dataset.to_dicts(), "summary": { "total_clicks": sum(r["clicks"] for r in dataset), "total_conversions": sum(r["conversions"] for r in dataset), }, }3.3 再上一个台阶:流程类技能——自动化报告订阅
数据查询技能搞定后,我又加了一个流程类技能:指标异常订阅。这个技能的业务价值很明显——运营不希望天天手动查数,而是希望Agent在指标发生异常波动时主动发一条通知。
这类技能在设计上有一个特殊点:它不只是“查询后返回结果”,还要在后台保持运行,周期性地检查监控条件。因此,技能框架不能只支持“一次性调用”模式,还得支持周期任务和事件触发两种运行模式。
我扩充了技能元数据,加入execution_mode:
execution_mode: periodic schedule: cron: "0 9 * * *" timezone: "Asia/Shanghai"同时为这类技能增加了“触发策略”的概念:每个检查周期执行一次核心逻辑,若判定为“异常”,则调用技能自带的notify函数推送通知。它的执行逻辑大致是:
async def run(inputs, ctx): metric = inputs["metric"] threshold = inputs["threshold_change_pct"] current_val = await ctx.database.query_metric(metric) baseline = await ctx.database.query_baseline(metric, window=7) change_pct = (current_val - baseline) / baseline * 100 if abs(change_pct) >= threshold: await ctx.notifier.send( channel="slack", message=f"{metric} 当前值 {current_val:.2f},较上周同期变化 {change_pct:+.2f}%", ) return {"triggered": True, "change_pct": round(change_pct, 2)} return {"triggered": False}这一类技能跑起来之后,我从中学到了最关键的一课:技能在执行过程中必须用上下文对象ctx访问外部依赖,不要直接自己new各种客户端。把数据源、通知渠道、权限校验都收敛到上下文里,技能之间才不会互相踩踏,也才能做好单元测试。这也是这套框架从“能跑”通往“好用”的关键一步。
3.4 技能之间的组合:把原子技能编排成更大能力
单个技能做好之后,接下来就是组合。我这边比较常见的编排模式是“数据查询→格式转换→生成简报”这样的三段式流水线。一开始因为有多个技能节点,我只让模型逐个调用:
- 先查询渠道数据(
channel_data.query) - 再取业务KPI阈值(
kpi.thresholds.get) - 最后生成一份简报(
report.text.render)
但发现一个问题:模型在长链路里容易“丢步骤”,尤其当中间某一步返回空结果时,后面直接就断了。后来我把这种高频的固定编排用“技能链”的方式固化下来——定义一个channel_performance_brief作为组合技能,内部声明依赖哪些子技能、以什么顺序执行、中间结果如何传递。模型只需要选择这个组合技能,剩下的事情由框架内部编排完成。
这让我意识到,技能系统不能只做“原子技能”一个层级,一定要支持组合技能。组合技能的底层还是原子技能,但整体被视为一个可调度的单元,这样既保留了灵活性,又降低了模型在长链路编排上的心智负担。
4. 避坑日记:上下文膨胀、并发污染与安全边界
4.1 上下文膨胀问题:技能描述不是越长越好
这个问题前面提过,但这里想展开细说。最初我以为技能描述写得越细,模型越能理解。直到有一次,一个技能光description就写了500多字,外加一份特别精细的参数枚举,结果模型反而开始出现幻觉——它居然会“猜测”schema里没有的字段,并在调用时传参。
后面我反思了一下,大模型做function calling的时候,本质上是不是在做一个“意图–参数”的映射。信息过载会让这个映射变得不稳定。定下几条铁律之后,这个问题基本没再犯过:
- description限制在150字以内,只写核心用途和触发场景
- 参数schema必填字段不超过5个,其余能用默认值就给默认值
- 枚举值的数量控制在10个以内,超过就考虑拆技能或改用动态补充描述
4.2 并发场景下技能状态互相污染
另一个印象很深的线上事故,是一次大促活动期间,突然接到告警说Agent的响应大量超时。排查后发现,问题出在技能层:有个技能用了模块级全局变量保存临时状态,当多个会话并发调用同一个技能时,一个会话写入的状态把另一个会话覆盖了,导致后续查询返回错乱,甚至触发了重复扣减积分这类严重问题。
教训是:Agent技能绝不能设计成有状态的单例。所有会话相关的状态都必须放在调用上下文中,执行函数内部只能使用局部变量或上下文变量。为此我把技能执行器改造成了每次调用都生成一个独立的执行实例,相当于给每个会话都开了一个“工作台”,技能用的工具和数据都是一份独立的拷贝,从机制上杜绝了串号问题。
改造后的执行器核心逻辑大致是:
class SkillExecutor: async def execute(self, skill_id, inputs, session_id): skill = self._registry.get(skill_id) ctx = SkillContext(session_id=session_id, deps=self._build_deps()) instance = skill.create_instance() return await instance.run(inputs, ctx)4.3 安全边界设计:能力越大,权限越要清楚
技能系统还有一个绕不开的点:安全。Agent能调用的工具越多,潜在的风险就越大。特别是当一个技能可以触达生产数据库或对外发消息时,权限管控必须前置,不能靠模型自觉。
我的做法是定义一个全局的权限矩阵,每个技能对应一个权限等级:
| 权限级别 | 适用范围 | 示例技能 |
|---|---|---|
| L1 公共只读 | 所有用户可调用,只能查询脱敏数据 | 渠道数据查询、文档检索 |
| L2 受限读 | 需要登录用户且限定数据范围 | 用户画像读取 |
| L3 受限写 | 需要额外审批,且写操作需留痕 | 报表状态更新、通知发送 |
| L4 高危操作 | 默认禁止,需二次确认或人工审批 | 数据清除、配置变更 |
在调度流程中,每次模型选中技能之后、执行之前,都过一次权限校验。在高危操作场景下,系统会弹出一个二次确认的交互,只有用户明确了“确认执行”之后,技能才会真正被执行。
有一个细节我认为很有用:高危技能的执行日志要详细记录,包括触发方、完整参数、执行结果。出了问题是真能救命——没有这份日志,线上出了事故连怎么复盘都无从下手。它相当于给Agent的每一次“危险操作”都上了保险。
5. 拿来即用的关键配置与复盘清单
5.1 一份可以直接抄作业的技能框架配置模板
这一段我想把最终跑稳的配置直接整理出来,方便想动手试的朋友复制改改。整体的目录结构建议这样:
agent-skills/ ├── skills/ │ ├── channel_data.query/ │ │ ├── skill.yaml │ │ └── main.py │ ├── kpi.thresholds.get/ │ │ ├── skill.yaml │ │ └── main.py │ └── report.text.render/ │ ├── skill.yaml │ └── main.py ├── core/ │ ├── registry.py │ ├── selector.py │ ├── executor.py │ └── security.py ├── configs/ │ ├── registry.yaml │ └── permissions.yaml └── tests/ └── test_skills.py技术选型方面,我这边用的是Python异步框架,配合轻量级文件监听实现技能的动态加载。如果团队里有现成的服务治理体系,也可以在此基础上做扩展,但初期不用搞太重,先把主流程跑顺。
关于向量检索做候选召回的索引方式,我选择的是本地bge-small模型做embedding,加上少量调整取的效果已经够用。数据量在几千条用户意图样本以下时,不需要引大规模向量数据库直接起一个轻量索引就行——等到规模上来你再考虑引入外部方案完全来得及。
5.2 上线前必须过一遍的技能自检清单
为了避免上线出幺蛾子,我自己每次发布新版技能前都会过一遍检查清单,这里直接列出来:
- 技能元数据是否包含完整的description、intents、tags、参数schema、输出schema
- description是否控制在150字以内,参数是否少于5个必填项
- 技能执行是否无状态化,所有会话状态是否都通过上下文传递
- 是否有单元测试覆盖主要路径和异常路径
- 是否有权限级别标注,高危技能是否接入了审批逻辑
- 是否有超时配置和熔断配置,避免技能慢调用拖垮Agent主链路
- 技能加载失败时,系统是否能够降级或忽略,而不是整体崩溃
- 是否部署了观测探针,能记录每次调用的耗时、成功率、token消耗
这清单看着零碎,但哪一条没有守住,线上都可能变成事故。每次技能上线前,我都会跑一遍自动检查脚本,然后人工review一遍,速度虽然慢点,但稳。
5.3 几条我觉得非常重要的复盘经验
最后整体复盘一下,我想单独聊聊几条让我感触最深的经验,希望看到这里的人能少走弯路。
第一,技能的元数据比代码更需要用心设计。代码本身只要能跑就行,但元数据直接决定了模型能否正确选择并调用技能。它值得你花和写代码同样甚至更多的时间去打磨。尤其在描述里把触发词、边界条件写清楚,模型就不会做“错误联想”了。
第二,一定要把技能的加载做成无侵入的。前期我曾经把技能注册和主程序耦合在一起,只要加一个技能就要改主程序的初始化代码。这样在20个技能以内还行,再多就完全是灾难。改成目录扫描加自动注册之后,新增技能的操作成本就从“改代码”变成了“加文件”。
第三,技能调用链路的可观测性,值得在第一天就埋进去。每次调用、每个参数、每次报错,记录与复盘,这一习惯让我后来排查问题时省了大量时间。Agent系统的bug往往不是单一的代码问题,而是多个环节叠加的效果,没有日志你连切入点都找不到。
第四,不要试图让Agent一次把所有事情都干完。设计技能时如果发现一个技能回答要涉及三个完全独立的领域,那就该拆成三个技能,让Agent自己去编排。技能粒度太粗,模型的成功率必然下降,因为它的每一步都需要推理,而推理越多越容易出错。
最后
关于agent-skills这块,我目前还在持续迭代的方向,主要是技能的自动评测——就是想通过一套自动生成的测试用例集,在每次技能变更后自动验证它对真实用户意图的处理能力有没有退化。如果你也在做类似的事情,或者正在搭自己的技能系统,欢迎在评论区聊聊你的踩坑经历。我保证,这套东西的坑,至少还有一半是藏在文档和示例代码后方的,需要自己亲手踩一遍才能真正感受到。