Agent开发做到一定阶段,会遇到一个绕不开的问题:每加一个新能力,就要重写一大段Prompt,或者把工具函数写死进代码里,项目越滚越大,Agent却越来越笨。我搭建某个跨平台调度系统时就卡在了这里,后来把"能力"本身设计成一套可加载的技能系统,也就是项目标题里提到的agent-skills。这篇文章把整个设计思路和实现过程完整记录下来,包括注册表结构、参数声明、调度机制、反模式,以及踩过的几个比较深的坑,希望能给同样在做Agent工程的读者省点时间。
先说清楚这套技能系统到底在解决什么问题。那段时间我在做一个内部用的多Agent协作框架,基础模型已经接好,主流程也能跑通,但一旦涉及具体业务能力——比如查库存、算排期、发通知——就全是硬编码。每来一个新需求,我先写一个函数,再在系统提示词里补一段"当用户提到XX时,调用XX函数"。听起来很直接,实际用起来非常痛苦:提示词越来越长,模型越到后面越容易遗漏任务,函数边界稍微模糊一点,Agent就调用错。更麻烦的是,这些能力完全没法复用,换个项目就得把代码全部复制一遍,换模型后提示词的写法可能又要调。
后来我意识到问题出在"能力"的交付形态上。我需要的不是把能力塞进提示词,也不是把能力焊死在代码里,而是把能力变成一种Agent能自己发现、按需加载、用完即走的"技能"。整理清楚这个思路后,我用了大概两周时间重构项目,做了一个轻量的agent-skills系统,效果非常明显,所以决定把完整过程分享出来。
1. Skill与Prompt、Plugin的本质差异——它解决的是复用和自治问题
先把概念掰清楚。很多做Agent的朋友一开始都是靠Prompt来"教会"模型做某件事,比如在系统提示词里写"当用户想查天气时,调用weather_api(query=城市名)"。这种做法对一两个工具没问题,但对几十个工具来说完全失控。Prompt是线性文本,模型读它时只能被动理解,没有一个机制告诉模型"你现在有这么多能力,哪些是这个场景下能用的,哪些是应该避开的"。
Plugin的形态通常是把自己封装成固定的API入口,优点是边界清晰,缺点是太死板。插件面向的是硬编码调用场景,Agent拿到的是一堆互不相关的函数签名,它自己并没有能力判断"这个插件在什么上下文下可用、使用它的预期效果是什么、失败了该找谁"。更麻烦的是,传统插件的参数处理是代码侧做校验,模型侧对参数含义没有理解,经常出现传错参数类型、少传必填字段这类低级错误。
Skill的设计介于两者之间,它本质上是"带自我描述的代码单元"。每个技能包含四部分:一段让人和模型都能看懂的能力描述、一个声明参数结构的Schema、一段真正的执行逻辑、以及一组后台规则(比如权限要求、超时设置、回退方案)。这四个部分协同起来,效果完全不同。
我把这套东西想明白之后,最直观的感受是:Prompt是给模型看的话术,Plugin是给代码用的接口,Skill是给Agent用的能力。Skill和Plugin的根本区别在于"谁来做决策"。传统Plugin是代码决定一切,模型只能被动传参;Skill则把决策权交给模型,让它在上下文中看到技能描述后,自主判断该不该用、怎么用、什么时候停手。这种"看得见、用得对、收得回"的状态,才是Agent能力的合理形态。
下面用一张表把三者差异说清楚,这张表也是我给自己团队做方案评审时的内部资料:
| 维度 | Prompt内嵌 | Plugin硬编码 | Skill技能系统 |
|---|---|---|---|
| 能力发现 | 靠模型读文本 | 代码侧注册 | 技能描述自动聚合 |
| 参数校验 | 无,依赖模型自觉 | 代码强校验 | Schema语义校验 |
| 复用性 | 换项目就失效 | 可复用但改造重 | 天然可搬运 |
| 扩展成本 | 修改提示词 | 改代码重新部署 | 独立添加配置 |
| 容错 | 无 | 异常自己处理 | 有回退和降级策略 |
| 适用规模 | 3-5个工具 | 固定场景 | 大量能力持续增长 |
看完这张表就能明白,我为什么在当时那个多Agent协作项目里放弃前两种方案,把重心全部放到Skill的设计上。它转移了重心:不用花大量精力去调提示词让模型"记住"有什么工具,也不需要每次扩展能力都做一次重构部署。我要做的只是把新技能按规格写好,丢进技能目录,Agent在运行时会自动发现它。
2. 技能系统的骨架:注册、声明与调度的三份关键设计决策
技能系统听起来抽象,落到实现上其实就三件事:注册(能力怎么进系统)、声明(模型怎么理解能力)、调度(Agent怎么选中并执行能力)。我在重构时分别做了三个关键设计决策,这里逐个说清楚。
2.1 注册机制:文件名即身份,目录即生态
第一个决策是让"技能文件名"直接成为系统里的唯一标识。技能采用扁平化目录管理,每个技能是一个独立文件夹,内部包含SKILL.md(能力描述)、schema.json(参数声明)、run.py(执行逻辑)、meta.yaml(元数据)。Jinja或Python项目里大家习惯用包管理,但这里我不建议把技能做成Python包再import,那样会把技能与代码运行时强耦合,换语言或者做沙箱隔离时会很痛苦。
我的做法是运行时在启动阶段扫描技能目录,逐个读取SKILL.md和schema.json,统一注册进内存里的技能表。技能表的键就是目录名。每个技能目录里两份文件是必需的:SKILL.md负责给人看也给模型看,schema.json给模型提供参数约束。run.py不用在启动阶段加载,等真正执行时才动态拉取。
实际效果非常明显。有次团队里有人要加一个"汇率换算"技能,直接新建一个exchange_rate目录,写好三个文件放进去,主程序一行代码都没改,Agent第二天就能正常调用。这就是"目录即生态"的威力,也是技能系统之于硬编码Plugin的最大区别。
2.2 声明层设计:让模型看到的是语义,而不是函数签名
第二个决策,也是我认为最关键的一个,是让模型直接理解技能参数,而不仅仅是机械传参。很多Agent框架喜欢给模型展示"function calling"风格的列表,类似:{name: "exchange_rate", parameters: {from: "USD", to: "CNY", amount: 100}}。这看似直接,但实际运行中,模型经常觉得"amount"这个词太干瘪,不一定理解它的边界(比如是不是必须大于0,是不是支持小数)。
我参考了Function Calling里普遍使用的JSON Schema做法,把每个参数都描述成带语义的结构。以汇率换算举例,schema.json里会这样写:
{ "type": "object", "properties": { "from_currency": { "type": "string", "enum": ["USD", "CNY", "EUR", "JPY"], "description": "源币种,ISO 4217标准三位代码" }, "to_currency": { "type": "string", "enum": ["USD", "CNY", "EUR", "JPY"], "description": "目标币种,ISO 4217标准三位代码" }, "amount": { "type": "number", "minimum": 0.01, "description": "需要换算的金额,必须大于0" } }, "required": ["from_currency", "to_currency", "amount"], "description": "根据实时汇率将金额从源币种换算为目标币种" }这样的声明确实比简单传一个函数签名更容易让模型理解。每个参数都配了description、约束范围、枚举值,模型的幻觉空间被大幅压缩。团队后来也测过,同样一个汇率技能,写清楚参数描述后,错误传参的概率明显下降。这个决策表面上是写Schema,实际上是在降低模型的认知负担。
2.3 调度机制:双通道触发,避免"想不起来"和"乱调用"
第三个决策是关于触发方式。最初我以为让模型自主决定调用技能就够了,毕竟模型看到了技能描述。实际用了几天就发现两个问题:一是模型在复杂对话里会"想不起来"某个技能,尤其是技能数量超过30个时;二是有些技能不太适合让模型自主触发,比如"关闭系统""清理缓存"这类高权限操作,模型可能会在奇怪的时间点调用它们。
最终的调度机制是双通道:常规技能走"模型自主选择"通道,Agent在需要时自行从技能表中挑选并执行;关键技能走"系统绑定"通道,由主流程代码根据状态自动触发,模型没有权限主动调用这类技能。后者的典型场景是权限控制、资源清理、安全检查。
这套机制用一个简单的优先级规则协调:当某个业务节点上既有系统通道又有模型通道的技能候选时,系统通道优先执行,模型通道的技能会被暂时挂起。这样做有两个收益,一是关键操作不会被模型乱触发,二是模型也不用每次都被提醒"你还有一堆高权限技能",减少了思考噪音。
3. 核心实现:手搭一个可用的agent-skills骨架
聊完设计,这章直接上代码。我的实现基于一个内部事件循环框架,你可以换成任何偏好的异步框架,核心思路一致。我会把完整的目录结构、注册逻辑、执行逻辑、调用入口都写出来。
3.1 模块分工与目录结构
先看工程目录:
agent-skills/ ├── skills/ │ ├── exchange_rate/ │ │ ├── SKILL.md │ │ ├── schema.json │ │ ├── run.py │ │ └── meta.yaml │ ├── inventory_query/ │ │ ├── SKILL.md │ │ ├── schema.json │ │ ├── run.py │ │ └── meta.yaml │ └── ... ├── core/ │ ├── registry.py │ ├── executor.py │ ├── scheduler.py │ └── context.py ├── app.py └── requirements.txtcore目录里四个文件的分工:registry负责技能扫描与注册,executor负责执行技能run.py,scheduler负责遍历技能表决策最终调度走向,context负责向技能执行体传递全局上下文(会话状态、权限标识、用户偏好等)。这样拆分的好处是,业务逻辑与技能逻辑完全解耦,新技能只需要实现自己的业务,不需要理解Agent框架怎么工作。
3.2 注册表:扫描与加载的核心逻辑
registry是技能系统的地基。核心要求是三个:启动快、结构稳定、可增量更新。启动快的重点是尽量不加载run.py,启动时只解析SKILL.md和schema.json,执行时再动态载入。结构稳定的意思是技能表一旦初始化就是只读的,避免运行中被意外覆盖。可增量更新留给后续做动态热加载。
import json import logging from pathlib import Path logger = logging.getLogger(__name__) class SkillRegistry: def __init__(self, skills_dir: str): self.skills_dir = Path(skills_dir) self._skills = {} def load(self): if not self.skills_dir.exists(): raise FileNotFoundError(f"skills directory not found: {self.skills_dir}") for skill_dir in self.skills_dir.iterdir(): if not skill_dir.is_dir(): continue skill_id = skill_dir.name desc_file = skill_dir / "SKILL.md" schema_file = skill_dir / "schema.json" meta_file = skill_dir / "meta.yaml" if not (desc_file.exists() and schema_file.exists()): logger.warning("skill %s missing SKILL.md or schema.json, skipped", skill_id) continue self._skills[skill_id] = { "id": skill_id, "description": desc_file.read_text(encoding="utf-8"), "schema": json.loads(schema_file.read_text(encoding="utf-8")), "path": str(skill_dir), "meta": self._load_meta(meta_file), "loaded": False, "module": None, } logger.info("loaded %d skills", len(self._skills)) def _load_meta(self, meta_file: Path): if not meta_file.exists(): return {"mode": "auto"} try: import yaml return yaml.safe_load(meta_file.read_text(encoding="utf-8")) except Exception: return {"mode": "auto"} def get(self, skill_id: str): return self._skills.get(skill_id) def list_skills(self): return [ {"id": s["id"], "description": s["description"][:200], "schema": s["schema"]} for s in self._skills.values() ] def load_module(self, skill_id: str): skill = self._skills.get(skill_id) if not skill: raise KeyError(skill_id) if not skill["loaded"]: import importlib.util run_path = Path(skill["path"]) / "run.py" if not run_path.exists(): raise FileNotFoundError(f"run.py missing for skill {skill_id}") spec = importlib.util.spec_from_file_location(f"skill_{skill_id}", run_path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) skill["module"] = module skill["loaded"] = True return skill["module"]这段代码里我故意做了一个取舍:模块加载用了一次性缓存,同一技能生命周期内只import一次,避免重复加载消耗。但要注意,这种模式没法做动态更新,技能升级必须重启进程。想要热加载的话,需要把module缓存key加上版本号,这里不展开。另外,加载run.py时我沿用了importlib方式,方便以后做沙箱隔离时替换loader。
3.3 执行器:参数校验、超时控制与异常归一
executor负责真正调用run.py里的函数。为了防止技能执行拖垮整个Agent,必须在执行层做三件保护:参数校验(复用schema)、超时控制、异常归一。
import asyncio import time class SkillExecutor: def __init__(self, registry: SkillRegistry): self.registry = registry self.default_timeout = 20 # 秒 async def execute(self, skill_id: str, params: dict, context: dict): skill = self.registry.get(skill_id) if not skill: raise ValueError(f"unknown skill {skill_id}") normalized = self._validate_params(skill["schema"], params) module = self.registry.load_module(skill_id) run_func = getattr(module, "run", None) if not callable(run_func): raise TypeError(f"skill {skill_id} has no callable run()") timeout = skill.get("meta", {}).get("timeout", self.default_timeout) start = time.time() try: if asyncio.iscoroutinefunction(run_func): result = await asyncio.wait_for(run_func(normalized, context), timeout=timeout) else: result = await asyncio.wait_for(asyncio.to_thread(run_func, normalized, context), timeout=timeout) except asyncio.TimeoutError: return {"ok": False, "error": "timeout", "skill_id": skill_id} except Exception as exc: return {"ok": False, "error": str(exc), "skill_id": skill_id} elapsed_ms = (time.time() - start) * 1000 return {"ok": True, "skill_id": skill_id, "result": result, "elapsed_ms": elapsed_ms} def _validate_params(self, schema: dict, params: dict): from jsonschema import validate, ValidationError try: validate(instance=params, schema=schema) except ValidationError as exc: raise ValueError(f"param validation failed: {exc.message}") from exc return params这里有两个经验值得展开。第一个是jsonschema校验不能省。我一开始以为模型生成参数基本靠谱,没必要再做一次校验,结果连续出现把日期字符串传成时间戳的事。加了jsonschema之后,这类问题基本绝迹。第二个是同步函数也要异步化。很多技能写起来简单,比如直接调外部HTTP接口,但如果是同步阻塞实现,遇到网络抖动整个事件循环就卡死了。所以我用asyncio.to_thread把同步函数丢到线程池,再用wait_for控制总时长。
3.4 调度器与上下文传递
调度器是整个技能系统的"神经中枢",它负责接收当前会话状态,匹配可用技能,决定是否调用以及调用的先后顺序。我的调度器实现时借鉴了"意图路由"的思想,整体逻辑并不复杂:
class SkillScheduler: def __init__(self, registry: SkillRegistry, executor: SkillExecutor): self.registry = registry self.executor = executor self.system_skills = {"system_cleanup", "access_check", "quota_check"} async def route(self, agent_state: dict, context: dict): """agent_state 包含当前对话意图、已选技能列表、用户输入等""" intent_label = agent_state.get("intent", "") available = agent_state.get("available_skill_ids", self.registry.list_skills()) # 系统技能优先,且不可由模型直接触发 system_candidates = [] for skill_id in self.system_skills: if skill_id in agent_state.get("system_triggers", []): system_candidates.append(skill_id) # 模型自主候选 model_candidates = [ s["id"] for s in self.registry.list_skills() if s["id"] not in self.system_skills and intent_label in s["schema"].get("related_intents", []) ] if system_candidates: results = [] for skill_id in system_candidates: results.append(await self.executor.execute(skill_id, {}, context)) return {"selected": system_candidates, "results": results} if model_candidates: skill_id = model_candidates[0] params = agent_state.get("skill_params", {}) result = await self.executor.execute(skill_id, params, context) return {"selected": [skill_id], "results": [result]} return {"selected": [], "results": []}这段代码里我简化了很多业务细节,必要时请按自己业务的复杂性做扩展。核心想传达的是调度里"系统优先"和"模型候选"两个概念的落地方式。另外context参数要设计轻量,只传递必要的会话快照:用户身份、时区、权限级别、最近一次Agent消息。这样技能执行时不会因为拿不到上下文而问愚蠢的问题,也不会因为传太多上下文而泄露隐私。
4. 技能运行时的三个翻车现场:参数幻觉、语义冲突与上下文污染
这一章节我想重点分享踩坑过程,因为技能系统从"能用"到"好用"的差距,往往不在设计文档里,而在这些实际运行事故里。这三个坑我都遇到过,每一步的排查方式也和你们分享。
4.1 参数幻觉:模型编造了不存在的参数
第一次大规模测试时,我让Agent调用一个"库存查询"技能,Schema里清清楚楚写着参数是sku_code和warehouse_id。结果模型实际传参时,自己编了一个quantity参数进去,还给了个"查询数量"的值。系统严格按jsonschema校验,直接报ValidationError,Agent就卡住了。
排查过程很有意思。我先怀疑是不是SKILL.md写得太模糊,导致模型没理解参数边界。把SKILL.md改成更明确的话术,还加了"不要传入任何未在schema中声明的参数"后,问题缓解了几天。但很快又出现,而且更隐蔽:模型开始把参数名往Schema靠拢,却依然会传多余字段,或者把必填字段漏掉。
后来我才想明白根因,不是模型不理解,而是我让模型"背"了太多内容。主提示词里的工具描述、历史对话里的技能调用示例、系统对技能表的聚合摘要,三者信息冲突或交叉,模型无所适从时就倾向于"猜"。真正的解法不在提示词,而在架构层面:给模型展示技能信息时只展示技能描述的前两行和参数名列表,不展示完整Schema;参数校验失败时,不要直接告诉模型"校验失败",而是把错误信息回传给模型,让它自行修正后重试一次。
调整之后,参数幻觉发生概率下降了一个量级。这个坑给我的教训是:模型不是数据库,你不能让它在每个请求里都记住几十个技能的完整Schema。你要给它能用的索引,配合事后校验与修改回路,而不是期望它零失误。
4.2 语义冲突:两个技能"长得太像",模型选错
第二个坑是技能多了之后必然出现的语义冲突。我系统里有两个技能,一个叫"查天气",一个叫"查空气质量"。从命名上看很清楚,但从描述上看,两个SKILL.md的前缀几乎一样:"输入城市名,返回该城市的XX数据"。导致模型经常把"查空气质量"当成"查天气",返回了一堆AQI数据,用户显然不满意。
排查时,我先看了调度日志,发现模型选择的技能和用户意图确实不符。我最初的调整方案是改模块名,把"weather"改成"weather_forecast_current",把"air_quality"改成"air_quality_pm25_forecast"。结果更糟,名称变长了,模型反而更晕。
最后的有效方案是给SKILL.md增强区分度:在描述里明确"该技能返回的是天气现象(晴/雨/风),不包含污染物数据",同时在schema里增加"unit"枚举,让模型在输出前用更精确的约束来匹配用户需求。这样的语义限定词做出来后,选错率立刻下来了。这让我意识到,技能之间要有清晰的互斥约定,不是说把名字改得够长就万事大吉,而是让模型能从描述里感知到它们的边界和差异。
4.3 上下文污染:技能结果把主对话撑爆了
第三个坑是我自己设计上的疏漏。技能执行完之后,我简单地把结果字符串拼到系统上下文里,作为模型下一次推理的背景。少量技能还好,一旦某个技能处理了较大数据(比如查库存返回了上千行SKU),下一轮对话的输入长度就会暴涨,甚至直接爆掉上下文窗口,或者把注意力带偏。
这个坑排查起来不难,token统计能直接看出来,但根治没那么简单。我最终的方案是分层缓存上下文:技能执行结果默认为"私有数据",只存到短期记忆区,不自动拼进主对话;只有当调度器判断用户问题确实需要这些数据参与推理时,才通过一个轻量检索把相关片段注入系统上下文。
这样处理以后,主对话上下文长度稳定住了,Agent的响应速度也明显提升。做技能系统的人一定要时刻记住,技能是为了扩展Agent能力,而不是为了撑爆它的记忆。
| 坑 | 根因 | 修复方向 |
|---|---|---|
| 参数幻觉 | 模型需要背太多参数信息,信息冲突 | 精简暴露信息,失败回传模型自修 |
| 语义冲突 | 技能描述缺乏互斥边界 | 描述中加入"不包含什么"和精确约束 |
| 上下文污染 | 技能结果全量拼接主对话 | 结果分层缓存,按需检索注入 |
5. 从"单技能可用"到"系统好用":三个进阶策略
技能系统骨架搭好、坑也填了不少之后,我开始思考怎么让它真正有价值。这里分享三个策略,不算多高深,但都是我实际验证过效果的。
5.1 技能编排:让一个技能去调用另一个技能
单技能只能做原子操作,真实业务很少这么简单。例如用户想"对比上周和这周的销售额变化",系统至少需要调用两个技能:一个"拉周销售数据",一个"生成对比报告"。如果让模型硬写一份长Prompt把完整逻辑塞进来,那又会回到老路。
我的做法是允许技能描述中出现"依赖技能"字段。某个技能在执行过程中,可以显式地请求调度器调用另一个技能,把结果作为自己的输入。这本质上是把DAG(有向无环图)编排逻辑做进调度器。不需要复杂的流程引擎,只要在技能meta.yaml里声明depends_on,并在executor里加一个递归调用入口即可。
比如对比报告技能可以声明:
depends_on: - weekly_sales_data - weekly_sales_last_week执行时,executor会先调用依赖技能拿到原始数据,再进入主技能逻辑。这个策略让技能粒度可以保持很小,同时组合出复杂能力,非常适合团队协作时各自维护独立技能,由主流程统一编排。
5.2 技能结果缓存:降低重复计算和模型疲劳
技能一旦被频繁调用,缓存问题就躲不开。一个"查实时汇率"的技能,如果某个小时内稳定会话连续调用三次,每次都跑去请求外部API,既慢又没必要。所以我在executor外层加了一层简单缓存,key是skill_id加上参数哈希值,value是结果和过期时间。
import hashlib import time class SkillCache: def __init__(self, ttl: int = 300): self.ttl = ttl self._store = {} def get(self, skill_id: str, params: dict): key = self._make_key(skill_id, params) item = self._store.get(key) if not item: return None if time.time() - item["ts"] > self.ttl: self._store.pop(key, None) return None return item["value"] def set(self, skill_id: str, params: dict, value): key = self._make_key(skill_id, params) self._store[key] = {"value": value, "ts": time.time()} def _make_key(self, skill_id: str, params: dict): raw = skill_id + json.dumps(params, sort_keys=True) return hashlib.md5(raw.encode("utf-8")).hexdigest()用上缓存后,不仅响应变快,外部API费用也省了不少。但要小心,不是所有技能都适合缓存。涉及实时数据的技能,缓存TTL必须设得很短(比如5秒),或者干脆不缓存。涉及私有数据的技能,缓存key要带上用户ID,避免串数据。
5.3 技能监控与复盘:建立离线评估集
进阶策略里容易被忽略的是监控。技能系统真正上线后,运行频率与失败情况需要有一个面板或者日志体系,否则你会陷入"用户报障才查日志"的被动状态。
我搭了一套极简的技能运行统计表,记录每次调用的参数、耗时、状态码和错误信息。每周把这些日志抽样成评估集,用固定case去跑一遍,看技能选择的准确率、参数校验通过率、结果完整率。模型或技能有变更时,就跑一次回归对比,哪个技能的效果变差立刻就能发现。
这套做法让技能系统的迭代质量有了看得见的数字。有一次改了某个技能的描述,别的技能选择准确率掉了一截,靠回归对比抓到了问题,否则上线后用户可能莫名其妙觉得Agent变笨了。
6. 技能系统后续还能怎么生长
技能系统做到这个程度已经能支撑不少业务场景。接下来有几个方向值得继续扩展:一是加动态加载能力,让技能可以在不改动主程序的情况下,通过管理面板或文件监听实现热更新;二是做更细粒度的权限模型,按用户角色控制哪些技能可见、哪些技能可调用;三是把技能描述向量化,建立检索式技能发现机制,技能数量过千时依然能保证模型准确找到目标技能。
我用这个系统做项目最大的体会就是:Agent的能力扩展不该靠"缝缝补补写提示词"来维持,而应该像搭积木一样,随时加一块,随时拆一块,保持整体的生长性。第一次把新技能写进目录、系统自动识别的那一瞬间,确实很爽。希望这篇文章的思路和代码能给你一个足够具体的起点。