1. 从“聊天”到“干活”:Skill 系统到底在解决什么问题
如果你最近半年一直在折腾 Agent 相关的东西,大概率会有一种强烈的割裂感:模型在对话框里能跟你聊哲学、写诗、解释量子纠缠,但一旦你让它“帮我把这周的销售数据拉出来,按区域汇总,生成一份带图表的周报”,它就开始装傻,或者给你一段看起来像那么回事、但根本跑不通的伪代码。
这个问题的本质,不是模型不够聪明,而是模型缺少一套可被调用、可被验证、可被复用的“技能”。LLM 本身是一个概率性的文本生成器,它擅长的是“理解和生成语言”,而不是“执行具体操作”。你让它直接输出一个 HTTP 请求去调接口,它可能会把 URL 拼错、把 Header 写反、把 JSON 格式搞乱。这不是它笨,是它压根就不该干这个。
Skill 技能系统要解决的,就是这个问题。它的核心思路很朴素:把“干活”的能力从模型内部剥离出来,封装成一个个独立的、可注册的、可被 Agent 调用的技能单元。模型只负责“决定调用哪个技能、传什么参数”,真正的执行逻辑由 Skill 自己完成。这样一来,Agent 就从“只会聊天的嘴炮”变成了“能动手干活的执行者”。
我最早接触这个概念是在做一些自动化工作流的时候。当时的需求很简单:让 Agent 帮我监控几个数据源,发现异常就发通知。一开始我试图用纯 Prompt 搞定,结果发现模型每次生成的判断逻辑都不一样,有时候漏判,有时候误报,根本没法稳定运行。后来我把“判断异常”这个逻辑抽出来,写成一个独立的 Skill,模型只负责在合适的时机触发它,整个系统的稳定性立刻上了一个台阶。
所以这篇文章,我想从实际落地的角度,把 Skill 技能系统的设计思路、核心细节、实操过程、常见坑点全部拆开讲一遍。不管你是刚接触 Agent 开发的新手,还是已经在做类似事情的同行,应该都能从中找到一些可以直接抄作业的东西。
提示:Skill 系统不是一个具体的框架或库,而是一种架构模式。不同团队、不同项目对它的实现方式可能完全不同,但核心思想是一致的——把执行能力从模型里拿出来,做成可管理的模块。
2. 整体架构设计:为什么要把“技能”单独抽出来
2.1 核心设计思路与方案选型
在动手写代码之前,先想清楚一件事:为什么不让模型直接生成代码或命令来执行?为什么要多一层 Skill 的封装?
我踩过的坑是这样的:早期我试过让模型直接输出 Shell 命令,然后我这边捕获执行。听起来很美好,但实际跑起来问题一大堆。模型有时候会生成rm -rf这种危险命令,有时候会把路径写错,有时候会在命令里夹杂一些莫名其妙的注释。更麻烦的是,你没法对模型生成的命令做有效的参数校验——因为每次生成的格式都不一样。
Skill 系统的设计思路,本质上是一种**“契约式”的能力封装**。每个 Skill 都有明确的输入参数定义、输出格式定义、以及执行逻辑。模型不需要知道 Skill 内部怎么实现,它只需要知道“这个 Skill 叫什么、需要什么参数、能干什么”。这就像你去餐厅点菜,你不需要知道厨师怎么炒菜,你只需要看菜单、点菜、等上菜。
具体来说,一个典型的 Skill 系统包含以下几个核心组件:
- Skill 注册中心:管理所有可用 Skill 的元信息,包括名称、描述、参数定义、返回值格式等。模型通过这个注册中心来“发现”有哪些技能可用。
- Skill 执行器:负责实际调用 Skill 的逻辑,包括参数校验、权限检查、执行、结果封装、错误处理等。
- Skill 描述协议:定义 Skill 如何向模型描述自己。通常是一段结构化的文本或 JSON Schema,告诉模型“我是谁、我能干什么、你需要给我什么”。
- 调用决策层:模型根据当前上下文和用户意图,决定是否调用 Skill、调用哪个 Skill、传什么参数。
这套架构的优势在于:执行逻辑和决策逻辑分离。模型只负责决策,执行由确定性代码完成。这样一来,系统的稳定性、可测试性、可维护性都大大提升。
2.2 为什么选择 HTTP + SSE 作为通信底座
在 Skill 系统的通信层设计上,我最终选择了 HTTP + SSE 的组合。这个选择不是拍脑袋决定的,而是经过了几轮对比。
先说说为什么不用 WebSocket。WebSocket 确实是全双工通信,理论上更适合实时交互场景。但实际用下来,WebSocket 的连接管理成本比较高,尤其是在需要横向扩展的时候,你需要额外维护连接状态、处理断线重连、做心跳检测。而且很多企业的网关和负载均衡器对 WebSocket 的支持并不友好,部署的时候容易出幺蛾子。
HTTP + SSE 的组合则简单得多。HTTP 负责请求-响应式的 Skill 调用,SSE 负责服务端向客户端推送执行过程中的事件流。SSE 本质上是基于 HTTP 的长连接,服务端可以持续向客户端发送文本事件,客户端通过EventSource接口接收。这个方案的好处是:
- 部署简单:SSE 就是普通的 HTTP 响应,不需要特殊的协议升级,现有的网关、负载均衡、CDN 都能直接支持。
- 调试方便:你可以直接用 curl 或者浏览器就能看到事件流,排查问题的时候非常直观。
- 连接复用:HTTP/1.1 的 Keep-Alive 和 HTTP/2 的多路复用都能有效减少连接建立的开销。
当然,SSE 也有它的局限性。它是单向的,只能服务端推客户端。但在 Skill 系统的场景下,这个限制其实不是问题——客户端发起 Skill 调用请求,服务端执行并推送执行进度和结果,这个方向正好是 SSE 擅长的。
注意:SSE 连接有一个常见的坑是空闲超时。很多网关默认 60 秒没有数据传输就会断开连接。如果你的 Skill 执行时间比较长,需要在服务端定期发送心跳事件(比如每 30 秒发一个 comment 行),保持连接活跃。
2.3 Skill 的粒度怎么控制
这是我在实际项目中纠结最久的问题:一个 Skill 到底应该做多少事?
粒度太粗,比如一个 Skill 叫“处理销售数据”,那它内部可能包含数据拉取、清洗、汇总、生成图表、发送邮件等一堆逻辑。这种 Skill 的问题是复用性差,换个场景就用不上了。而且模型很难判断什么时候该调用它,因为它的描述太模糊了。
粒度太细,比如一个 Skill 只做“把字符串转成大写”,那 Skill 的数量会爆炸,模型在选择的时候也会晕头转向。而且太细的 Skill 往往需要多个组合才能完成一个完整任务,增加了编排的复杂度。
我的经验是:一个 Skill 应该对应一个完整的、有明确业务含义的操作单元。比如“查询指定时间范围的订单数据”、“生成柱状图”、“发送邮件通知”这三个 Skill,粒度就比较合适。它们各自有明确的输入输出,可以独立测试,也可以组合使用。
具体判断标准可以参考这几条:
- 这个 Skill 能否用一句话说清楚它干什么?
- 这个 Skill 的输入参数是否超过 5 个?如果超过,可能需要拆分。
- 这个 Skill 是否会被多个不同的场景复用?如果只在一个地方用,可能不值得单独封装。
- 这个 Skill 的执行时间是否可控?如果可能跑几分钟甚至更久,需要考虑异步化和进度推送。
3. 核心细节解析:Skill 的描述、注册与调用
3.1 Skill 描述协议的设计要点
Skill 描述协议是模型“认识”Skill 的唯一途径。描述写得好不好,直接决定了模型能不能在正确的时机调用正确的 Skill。
我见过很多团队把 Skill 描述写成了一段技术文档,比如“该接口用于查询数据库中的订单表,支持按时间范围和状态过滤,返回 JSON 格式的结果”。这种描述对人来说很清晰,但对模型来说不够友好。模型需要的是意图导向的描述,而不是实现导向的描述。
一个好的 Skill 描述应该包含以下几个要素:
- 名称:简短、动词开头、见名知意。比如
query_orders、generate_chart、send_notification。 - 用途说明:用自然语言说明这个 Skill 解决什么问题,什么场景下应该使用它。比如“当用户需要查询订单数据时使用此技能”。
- 参数定义:每个参数的类型、是否必填、含义、示例值。参数描述要具体,避免模糊表述。
- 返回值说明:返回什么格式的数据,包含哪些字段。
- 使用限制:什么情况下不应该使用这个 Skill,或者有什么前置条件。
我通常会用一个结构化的 JSON 来描述 Skill,然后在系统提示词里把这个 JSON 转换成模型容易理解的格式。比如:
{ "name": "query_orders", "description": "查询指定时间范围内的订单数据,支持按状态和区域过滤。当用户需要获取订单列表、统计订单数量或金额时使用此技能。", "parameters": { "start_date": { "type": "string", "required": true, "description": "开始日期,格式 YYYY-MM-DD" }, "end_date": { "type": "string", "required": true, "description": "结束日期,格式 YYYY-MM-DD" }, "status": { "type": "string", "required": false, "description": "订单状态过滤,可选值:pending, completed, cancelled" } }, "returns": { "type": "array", "description": "订单列表,每条包含 order_id, amount, status, created_at 字段" } }这个 JSON 会被转换成模型能理解的工具描述格式。不同模型对工具描述的格式要求不一样,但核心信息是一样的。
实操心得:Skill 的 description 字段是最重要的。我通常会花 80% 的时间打磨 description,确保它既准确又容易被模型理解。一个技巧是,在 description 里加入“什么时候用”和“什么时候不用”的说明,能显著减少误调用。
3.2 Skill 注册与发现机制
Skill 注册中心的设计,决定了系统的扩展性和可维护性。最简单的做法是在代码里硬编码一个 Skill 列表,但这种方式在 Skill 数量增多之后会变得难以管理。
我目前采用的方案是基于配置文件的动态注册。每个 Skill 用一个独立的配置文件(YAML 或 JSON)描述,系统启动时扫描指定目录,加载所有 Skill 配置,注册到内存中的 Skill 注册表。这样新增一个 Skill 只需要加一个配置文件,不需要改代码。
配置文件的结构大概是这样:
name: query_orders version: 1.0.0 description: 查询指定时间范围内的订单数据... handler: handlers.order_query parameters: - name: start_date type: string required: true description: 开始日期,格式 YYYY-MM-DD - name: end_date type: string required: true description: 结束日期,格式 YYYY-MM-DD timeout: 30 retry: 2其中handler字段指向实际的执行函数。系统在加载配置时,会动态导入对应的模块,把函数注册到执行器中。
这种设计的好处是:
- 热插拔:新增或修改 Skill 不需要重启整个服务,只需要重新加载配置。
- 版本管理:每个 Skill 有独立的版本号,方便做灰度发布和回滚。
- 权限控制:可以在配置里指定哪些 Skill 对哪些用户或角色可见。
3.3 调用决策:模型怎么知道该用哪个 Skill
模型选择 Skill 的过程,本质上是一个意图匹配问题。用户说了一句话,模型需要判断:这句话对应的意图是什么?有没有现成的 Skill 可以满足这个意图?如果有多个,选哪个?
这个过程受几个因素影响:
- Skill 描述的质量:描述越清晰、越贴近用户的表达习惯,模型越容易匹配。
- 上下文信息:如果之前的对话已经涉及某个领域,模型会倾向于选择相关的 Skill。
- 参数完整性:如果用户没有提供 Skill 所需的必填参数,模型需要决定是追问用户,还是尝试从上下文推断。
我在实际项目中发现,模型在 Skill 选择上的准确率,很大程度上取决于 Skill 的数量和描述的区分度。当 Skill 数量少于 20 个时,只要描述写得不太差,模型的准确率通常能到 90% 以上。但当 Skill 数量超过 50 个时,准确率会明显下降,因为模型需要在更多的选项中做区分。
应对这个问题的策略有几个:
- 分组管理:把 Skill 按业务领域分组,模型先选组,再选具体 Skill。
- 动态筛选:根据当前对话上下文,只把相关的 Skill 暴露给模型,减少干扰项。
- Few-shot 示例:在提示词里加入一些“用户说了什么 → 应该调用哪个 Skill”的示例,帮助模型建立映射关系。
4. 实操过程:从零搭建一个可用的 Skill 系统
4.1 环境准备与基础框架搭建
我以 Python 技术栈为例,把整个搭建过程走一遍。其他语言栈的思路是一样的,只是具体实现方式不同。
首先明确依赖:
pip install fastapi uvicorn httpx pydantic pyyaml sse-starlettefastapi:Web 框架,提供 HTTP 接口和 SSE 支持。uvicorn:ASGI 服务器,用来跑 FastAPI 应用。httpx:异步 HTTP 客户端,Skill 内部如果需要调外部接口会用到。pydantic:数据校验,用来定义 Skill 的参数模型。pyyaml:解析 Skill 的 YAML 配置文件。sse-starlette:SSE 响应封装,简化事件流的发送。
项目目录结构大概是这样:
skill-system/ ├── main.py # 应用入口 ├── registry.py # Skill 注册中心 ├── executor.py # Skill 执行器 ├── models.py # 数据模型定义 ├── skills/ # Skill 配置目录 │ ├── query_orders.yaml │ ├── generate_chart.yaml │ └── send_notification.yaml ├── handlers/ # Skill 执行逻辑 │ ├── order_query.py │ ├── chart_gen.py │ └── notification.py └── config.yaml # 全局配置这个结构的好处是配置和执行逻辑分离,新增 Skill 的时候只需要在skills/下加一个 YAML 文件,在handlers/下加一个对应的 Python 模块。
4.2 Skill 注册中心的实现
注册中心的核心逻辑是:启动时扫描skills/目录,加载所有 YAML 配置,校验参数定义,把 Skill 元信息存入内存字典。
import yaml import importlib from pathlib import Path from models import SkillDefinition class SkillRegistry: def __init__(self, skills_dir: str): self.skills_dir = Path(skills_dir) self.skills: dict[str, SkillDefinition] = {} def load_all(self): for yaml_file in self.skills_dir.glob("*.yaml"): with open(yaml_file, "r", encoding="utf-8") as f: config = yaml.safe_load(f) skill = SkillDefinition(**config) self._validate_handler(skill) self.skills[skill.name] = skill def _validate_handler(self, skill: SkillDefinition): module_path, func_name = skill.handler.rsplit(".", 1) module = importlib.import_module(module_path) if not hasattr(module, func_name): raise ValueError(f"Handler {skill.handler} not found") def get(self, name: str) -> SkillDefinition | None: return self.skills.get(name) def list_all(self) -> list[SkillDefinition]: return list(self.skills.values())这段代码的关键点在于_validate_handler方法。它在加载阶段就检查 handler 是否存在,避免运行时才发现配置写错了。这个检查看起来简单,但能省掉很多调试时间。
4.3 Skill 执行器的实现
执行器负责接收调用请求,校验参数,执行 Skill,处理异常,返回结果。如果 Skill 执行时间较长,还需要通过 SSE 推送进度。
import asyncio import importlib from models import SkillCallRequest, SkillCallResult class SkillExecutor: def __init__(self, registry): self.registry = registry async def execute(self, request: SkillCallRequest) -> SkillCallResult: skill = self.registry.get(request.skill_name) if not skill: return SkillCallResult( success=False, error=f"Skill {request.skill_name} not found" ) # 参数校验 validation_error = self._validate_params(skill, request.params) if validation_error: return SkillCallResult(success=False, error=validation_error) # 动态加载 handler module_path, func_name = skill.handler.rsplit(".", 1) module = importlib.import_module(module_path) handler = getattr(module, func_name) # 执行,带超时和重试 for attempt in range(skill.retry + 1): try: result = await asyncio.wait_for( handler(**request.params), timeout=skill.timeout ) return SkillCallResult(success=True, data=result) except asyncio.TimeoutError: if attempt == skill.retry: return SkillCallResult( success=False, error=f"Skill timed out after {skill.timeout}s" ) except Exception as e: if attempt == skill.retry: return SkillCallResult(success=False, error=str(e)) return SkillCallResult(success=False, error="Unknown error") def _validate_params(self, skill, params: dict) -> str | None: for param in skill.parameters: if param.required and param.name not in params: return f"Missing required parameter: {param.name}" return None这里有几个设计决策值得说明:
- 超时控制:用
asyncio.wait_for给每个 Skill 设置执行超时,防止某个 Skill 卡死拖垮整个系统。 - 重试机制:对于可能因为网络抖动等原因失败的 Skill,配置重试次数。但要注意,不是所有 Skill 都适合重试——比如“发送邮件”这种有副作用的操作,重试可能导致重复发送。
- 参数校验前置:在调用 handler 之前先校验参数,避免无效调用。
4.4 SSE 进度推送的实现
对于执行时间较长的 Skill,同步等待结果体验很差。这时候需要用 SSE 把执行进度实时推送给客户端。
from sse_starlette.sse import EventSourceResponse import json async def execute_with_progress(request: SkillCallRequest): async def event_generator(): yield { "event": "start", "data": json.dumps({"skill": request.skill_name}) } # 模拟分步执行 steps = ["validating", "fetching_data", "processing", "finalizing"] for i, step in enumerate(steps): await asyncio.sleep(1) # 实际业务逻辑 yield { "event": "progress", "data": json.dumps({ "step": step, "percent": (i + 1) * 25 }) } result = await executor.execute(request) yield { "event": "complete", "data": json.dumps(result.dict()) } return EventSourceResponse(event_generator())客户端用EventSource接收事件:
const source = new EventSource('/skill/execute/stream?skill=query_orders'); source.addEventListener('progress', (e) => { const data = JSON.parse(e.data); console.log(`进度:${data.percent}% - ${data.step}`); }); source.addEventListener('complete', (e) => { const result = JSON.parse(e.data); console.log('执行完成', result); source.close(); });注意:SSE 的
EventSource默认只支持 GET 请求,且不能自定义 Header。如果你的 Skill 调用需要传复杂的参数或认证信息,可能需要用fetch+ReadableStream来手动处理 SSE 流。
4.5 一个完整 Skill 的示例:订单查询
以query_orders为例,看看一个完整的 Skill 从配置到执行是怎么串起来的。
配置文件skills/query_orders.yaml:
name: query_orders version: 1.0.0 description: | 查询指定时间范围内的订单数据,支持按状态过滤。 当用户需要获取订单列表、统计订单数量或金额时使用此技能。 如果用户没有指定时间范围,默认查询最近 7 天。 handler: handlers.order_query.query_orders parameters: - name: start_date type: string required: false description: 开始日期,格式 YYYY-MM-DD,默认为 7 天前 - name: end_date type: string required: false description: 结束日期,格式 YYYY-MM-DD,默认为今天 - name: status type: string required: false description: 订单状态过滤,可选值:pending, completed, cancelled timeout: 30 retry: 2执行逻辑handlers/order_query.py:
from datetime import datetime, timedelta async def query_orders( start_date: str = None, end_date: str = None, status: str = None ) -> list[dict]: if not end_date: end_date = datetime.now().strftime("%Y-%m-%d") if not start_date: start_date = (datetime.now() - timedelta(days=7)).strftime("%Y-%m-%d") # 实际项目中这里会查数据库或调接口 # 这里用模拟数据演示 orders = [ {"order_id": "ORD001", "amount": 299.00, "status": "completed", "created_at": "2024-01-15"}, {"order_id": "ORD002", "amount": 158.50, "status": "pending", "created_at": "2024-01-16"}, ] if status: orders = [o for o in orders if o["status"] == status] return orders这个 Skill 的设计有几个细节值得注意:
start_date和end_date设为非必填,并在 handler 里提供默认值。这样模型在用户没有明确指定时间范围时也能调用,降低了调用门槛。- description 里明确写了“如果用户没有指定时间范围,默认查询最近 7 天”,帮助模型理解默认行为。
- handler 是纯异步函数,方便在执行器里统一用
asyncio.wait_for做超时控制。
5. 常见问题与排查技巧实录
5.1 Skill 调用失败的高频原因
在实际运行中,Skill 调用失败的原因五花八门。我整理了一份速查表,覆盖了大部分常见情况:
| 问题现象 | 可能原因 | 排查方法 | 解决方案 |
|---|---|---|---|
| 模型不调用 Skill | Skill 描述不清晰 | 检查 description 是否说明了使用场景 | 补充“什么时候用”的说明 |
| 模型调用了错误的 Skill | 多个 Skill 描述相似 | 对比相似 Skill 的 description | 增加区分度,明确各自适用场景 |
| 参数校验失败 | 模型生成的参数格式不对 | 查看请求日志中的 params | 在 description 里给出参数示例 |
| 执行超时 | Skill 内部逻辑耗时过长 | 加日志看卡在哪一步 | 优化逻辑或改为异步+进度推送 |
| SSE 连接断开 | 网关空闲超时 | 查看网关配置 | 服务端定期发送心跳事件 |
| 返回结果模型不理解 | 返回值格式太复杂 | 检查返回的 JSON 结构 | 简化返回值,或在 description 里说明格式 |
5.2 SSE 连接断开的排查思路
SSE 连接断开是我遇到最多的线上问题之一。典型的表现是:客户端收到几个事件之后突然没了,或者报stream disconnected before completion: idle timeout waiting for SSE。
这个问题的根源通常不在 SSE 本身,而在中间的代理层。很多网关和负载均衡器对空闲连接有超时限制,默认可能是 60 秒。如果你的 Skill 执行时间超过这个限制,且中间没有数据传输,连接就会被断开。
排查步骤:
- 确认超时时间:查看网关或负载均衡器的空闲超时配置。
- 检查心跳:确认服务端是否在定期发送心跳事件。SSE 允许发送以
:开头的注释行作为心跳,不触发客户端事件。 - 测试直连:绕过网关直接连服务端,看是否还会断开。如果直连正常,说明问题在网关层。
- 调整配置:如果网关超时无法调整,缩短心跳间隔,确保在超时之前有数据传输。
心跳事件的实现很简单:
async def event_generator(): while True: yield {"comment": "heartbeat"} await asyncio.sleep(30)5.3 模型“幻觉调用”的应对策略
所谓“幻觉调用”,是指模型调用了一个根本不存在的 Skill,或者传入了 Skill 不支持的参数。这种情况在 Skill 数量较多时尤其常见。
我试过几种应对方式:
- 严格校验:在执行器里做严格的参数校验,不存在的 Skill 直接返回错误,不支持的参数直接拒绝。这是最后一道防线。
- 提示词约束:在系统提示词里明确列出所有可用 Skill 的名称,并强调“只能调用列表中的 Skill”。
- 错误反馈:当模型调用失败时,把错误信息返回给模型,让它重新决策。这招在大多数情况下有效,模型看到“Skill xxx not found”之后通常会换一个正确的。
- 限制重试次数:不要让模型无限重试,设置一个上限(比如 3 次),超过之后直接返回失败,避免死循环。
实操心得:我在提示词里加了一句“如果你不确定应该调用哪个 Skill,先向用户确认需求,不要猜测”。这句话显著降低了误调用率。模型在不确定的时候,追问用户比瞎猜要好得多。
5.4 性能优化的几个关键点
当 Skill 数量增多、调用频率上升之后,性能问题会逐渐暴露出来。我总结的几个优化点:
- HTTP 连接复用:Skill 内部如果需要调外部接口,用
httpx.AsyncClient并复用连接池,避免每次请求都新建连接。实测下来,连接复用能减少 30% 以上的延迟。 - Skill 元信息缓存:Skill 的描述信息不需要每次调用都重新生成,可以在启动时预计算并缓存。特别是当 Skill 描述需要转换成特定格式时,缓存能省不少 CPU。
- 并发执行:如果多个 Skill 之间没有依赖关系,可以并发执行。用
asyncio.gather把多个 Skill 调用并行化,整体耗时取决于最慢的那个。 - 结果缓存:对于查询类 Skill,如果同样的参数在短时间内被多次调用,可以加一层缓存。但要注意缓存失效策略,避免返回过期数据。
6. 进阶玩法:让 Skill 系统更智能
6.1 Skill 的组合与编排
单个 Skill 能做的事有限,真正的威力在于组合。比如“生成周报”这个任务,可以拆解为:查询订单数据 → 汇总统计 → 生成图表 → 发送邮件。这四个步骤分别对应四个 Skill,Agent 需要按顺序调用它们,并把前一个的输出作为后一个的输入。
这种编排能力,目前主要有两种实现方式:
一种是让模型自己编排。在提示词里告诉模型“你可以按顺序调用多个 Skill,前一个的结果会作为上下文传给后一个”。模型根据任务目标,自己决定调用顺序和参数传递。这种方式灵活,但稳定性依赖模型能力。
另一种是预定义工作流。把常见的组合固化成一个“超级 Skill”,内部按固定流程调用子 Skill。这种方式稳定,但灵活性差,适合高频、固定的场景。
我目前的策略是两者结合:高频场景用预定义工作流保证稳定性,长尾场景让模型自由编排。
6.2 Skill 的版本管理与灰度发布
当 Skill 系统服务于多个业务方时,版本管理就变得很重要。一个 Skill 的修改可能影响多个调用方,不能随便改。
我的做法是给每个 Skill 加版本号,注册中心同时保留多个版本。调用方可以指定版本,不指定则用最新稳定版。新版本先在小范围灰度,观察一段时间没问题再全量。
配置文件的命名可以体现版本:
skills/ ├── query_orders_v1.yaml ├── query_orders_v2.yaml └── generate_chart_v1.yaml注册中心加载时,按名称分组,保留所有版本。执行器根据请求中的版本号选择对应的 handler。
6.3 从 Skill 到“技能市场”的演进
当 Skill 数量积累到一定程度,一个自然的想法是:能不能让不同团队贡献 Skill,形成一个共享的技能市场?
这个想法听起来很美,但落地有几个挑战:
- 质量参差不齐:不同团队写的 Skill 质量差异很大,需要有一套审核机制。
- 依赖冲突:不同 Skill 可能依赖不同版本的库,需要做依赖隔离。
- 权限与安全:不是所有 Skill 都应该对所有调用方开放,需要细粒度的权限控制。
- 计费与配额:如果 Skill 调用了付费资源,需要有一套计费机制。
我目前的做法是在内部先跑通一个小规模的技能市场,只对特定团队开放,积累经验之后再考虑扩大范围。核心是先把审核流程和权限模型建立起来,技术实现反而是次要的。
6.4 监控与可观测性建设
Skill 系统上线之后,没有监控就是裸奔。我重点监控以下几个指标:
- 调用量:每个 Skill 的调用次数,按时间维度统计。突然的下降可能意味着模型不再选择这个 Skill,需要检查描述是否出了问题。
- 成功率:调用成功与失败的比例。成功率下降需要立即排查。
- 延迟分布:P50、P95、P99 延迟。P99 过高说明有长尾请求,需要优化。
- 参数分布:模型传入的参数分布。如果某个参数经常缺失或格式错误,说明 Skill 描述需要改进。
这些指标我通常用 Prometheus + Grafana 来采集和展示。每个 Skill 调用都会打点,包括 Skill 名称、耗时、成功与否、错误类型等标签。
日志方面,每次 Skill 调用都会记录完整的请求和响应,方便事后排查。但要注意脱敏,避免把敏感数据写进日志。
7. 我踩过的那些坑
7.1 不要低估参数校验的重要性
早期我觉得模型挺聪明的,参数应该不会传错。结果上线第一天就遇到了模型把日期格式传成2024/01/15而不是2024-01-15的情况。后来我在参数校验里加了格式检查,并且在 Skill 描述里明确写了格式要求,问题才解决。
现在的做法是:每个参数都有严格的类型和格式校验,校验失败时返回具体的错误信息,让模型知道哪里错了。
7.2 SSE 的 EventSource 不支持 POST
这是一个很容易踩的坑。浏览器的EventSourceAPI 只支持 GET 请求,不能设置请求体,也不能自定义 Header。如果你的 Skill 调用需要传复杂的参数,或者需要认证 Token,直接用EventSource是行不通的。
解决方案有两个:一是把参数放在 URL query string 里,但这样有长度限制且不安全;二是用fetch+ReadableStream手动处理 SSE 流。我目前用的是第二种方案,虽然代码复杂一点,但灵活性和安全性都更好。
7.3 模型对 Skill 描述的“过度解读”
有一次我写了一个 Skill 叫delete_record,描述是“删除指定 ID 的记录”。结果模型在用户只是说“我不想要这条数据了”的时候,就直接调用了这个 Skill。用户的本意可能只是“隐藏”而不是“删除”。
这件事给我的教训是:Skill 描述要明确边界。对于有副作用的操作,描述里要写清楚“此操作不可逆,仅在用户明确要求删除时使用”。后来我在所有危险操作的 Skill 描述里都加了类似的警告,误调用率明显下降。
7.4 超时设置不能一刀切
一开始我给所有 Skill 设置了统一的 30 秒超时。后来发现有些 Skill 确实需要更长时间(比如生成复杂报表),而有些 Skill 应该快速失败(比如简单的参数校验)。统一超时要么导致该快的慢,要么导致该慢的被误杀。
现在的做法是每个 Skill 在配置文件里单独设置超时时间,根据实际业务需求来定。查询类 Skill 通常 10-15 秒,生成类 Skill 可以到 60 秒,通知类 Skill 5 秒足够。
7.5 别忘了处理“模型不调用 Skill”的情况
有时候模型会忽略 Skill 的存在,直接用自然语言回答用户。这种情况在 Skill 描述不够突出时尤其常见。
我的应对方式是在系统提示词里加一段强制性的指令:“当用户的需求可以通过已有 Skill 满足时,必须调用对应的 Skill,不要直接回答。”同时,在 Skill 描述里用更强烈的语气,比如“必须使用此技能来获取订单数据,不要自行编造数据”。
这个问题的本质是模型倾向于用自己“擅长”的方式(生成文本)来解决问题,而不是调用外部工具。需要通过提示词不断强化“优先使用 Skill”的行为模式。
8. 后续可以继续深挖的方向
Skill 系统跑通之后,有几个方向我觉得值得继续投入:
Skill 的自动发现与推荐。目前 Skill 是手动注册的,未来可以基于用户的历史行为和当前上下文,自动推荐相关的 Skill。比如用户经常查询订单数据,系统可以主动把query_orders这个 Skill 放在更显眼的位置。
Skill 的自动化测试。每个 Skill 都应该有对应的测试用例,包括正常参数、边界参数、异常参数。我目前是手动写测试,未来可以考虑用模型自动生成测试用例,提高覆盖率。
跨语言的 Skill 调用。目前 Skill 都是 Python 实现的,但实际项目中可能有 Java、Go 等其他语言的服务。可以考虑用 gRPC 或 HTTP 做跨语言的 Skill 调用,让 Skill 系统不局限于单一技术栈。
Skill 执行的可解释性。当 Agent 调用了一系列 Skill 完成一个复杂任务时,用户往往想知道“它到底做了什么”。可以考虑记录完整的调用链路,生成一份可读的执行报告,提升系统的透明度和可信度。
这些方向我目前还在探索阶段,有些已经有了初步的原型,有些还停留在想法层面。等有更多实践经验之后,再单独写文章分享。