从零开始动手写一个 AI Skill,让大模型化身你的“全能助手”

01 一个让人失眠的夜晚
去年有个需求,直接让我熬了三个通宵。
业务方要求打造一个“智能客服”,但需求特别棘手——既要能查订单、处理退换货,又要根据用户历史行为做个性化推荐,还得实时对接库存系统。当时用的是某主流大模型 API,模型本身很聪明,可一碰到“查库存”这种需要实时数据的场景就开始胡编乱造。
那时候业内还没有特别成熟的 Agent 框架,尝试了 Function Calling,写了十几套 prompt 模板,甚至自己撸了一个简单的 RAG 流程。结果呢?代码越写越臃肿,prompt 越改越拧巴,每次新增一个能力就像重构整个系统。
直到接触到“Skill”这个概念。
说白了,Skill 就是把 AI 的某项“专项能力”打包成一个独立的插件。想让它查天气?装个 Weather Skill。想让它操作 Jira?装个 Jira Skill。想让它从数据库拉取实时报表?写个 SQL Skill 就完事了。
就像给手机装 App 一样,装上就能用,不用了就卸掉,核心系统干干净净。
今天就把这这块踩过的坑、总结的经验,一次性倒出来。
02 Skill 到底是什么?用一个例子说透
别被“Skill”这个花哨名字唬住,它的技术本质其实就三层:
一层接口:定义好输入输出规范(输入什么参数,输出什么格式)
一层执行逻辑:内部干活的代码(调 API、查数据库、跑算法都行)
一层注册机制:让 AI“知道”有这个能力存在,以及什么时候该调用它
画个简单的对比就懂了:
对比项
传统 Function Calling
Skill 插件
新增能力
改 prompt、改代码、重新部署
放一个 Skill 文件进去,热加载
复用性
每个项目重新写一遍
跨项目拷贝即用
调试难度
跟大段的系统代码混在一起
独立运行,日志隔离
团队协作
前后端、算法互相锁死
各写各的 Skill,最后组装
打个你肯定懂的比方:传统方式是给汽车发动机舱里硬塞新零件,Skill 是给车尾挂一个快拆拖斗。拖斗里装什么都可以,装错了摘下来换一个就行,发动机舱(核心系统)一根线都不用动。
03 手把手:10分钟写一个“查快递”Skill
光说不练假把式。下面用 Python 写一个真实的 Skill,让大模型能查询快递物流信息。
Step 1:定义 Skill 的“说明书”
每个 Skill 必须告诉 AI 三件事:我能干什么、我需要什么参数、我返回什么格式。
skill_express.py
from typing import Optional, Dict, Any
from pydantic import BaseModel, Field
1. 定义输入参数的结构
class ExpressInput(BaseModel):
tracking_number: str = Field(description="快递单号,支持中通/圆通/顺丰")
carrier: Optional[str] = Field(
default="auto",
description="快递公司编码,可选:zto/yto/sf,默认自动识别"
)
2. 定义输出的结构(让 AI 知道能拿到什么信息)
class ExpressOutput(BaseModel):
status: str = Field(description="物流状态:已揽收/运输中/派送中/已签收")
location: str = Field(description="最新位置")
timestamp: str = Field(description="更新时间")
details: list = Field(description="完整物流轨迹列表")
Step 2:写执行逻辑(真正的“干活代码”)
注意,这里为了演示,用了模拟数据。真实场景下,这里对接的是快递100、菜鸟裹裹或者顺丰开放平台的 API。
import httpx
import re
from datetime import datetime
class ExpressSkill:
"""快递查询技能"""
def __init__(self):
# 真实场景这里放 API 密钥
self.api_key = "your_kuaidi100_api_key"
self.base_url = "https://api.kuaidi100.com/query"
def execute(self, input: ExpressInput) -> ExpressOutput:
# 自动识别快递公司(真实场景用正则或算法)
carrier = input.carrier
if carrier == "auto":
carrier = self._detect_carrier(input.tracking_number)
# 调用快递查询 API(这里用 Mock 演示)
# 实际开发中换成:requests.get(self.base_url, params={...})
raw_data = self._mock_query(input.tracking_number, carrier)
# 数据清洗、格式转换
return ExpressOutput(
status=raw_data["status"],
location=raw_data["location"],
timestamp=datetime.now().isoformat(),
details=raw_data["traces"]
)
def _detect_carrier(self, tn: str) -> str:
# 简单识别逻辑:顺丰12位纯数字,中通12位数字字母混合...
if re.match(r'^\d{12}$', tn):
return "sf"
elif re.match(r'^[A-Za-z0-9]{12}$', tn):
return "zto"
else:
return "yto"
def _mock_query(self, tn: str, carrier: str) -> dict:
# 模拟数据,真实场景替换为 API 调用
return {
"status": "派送中",
"location": "广州市天河区体育西路营业点",
"traces": [
{"time": "2026-07-06 08:30", "content": "快递员正在派送中,预计今日送达"},
{"time": "2026-07-05 22:15", "content": "到达广州转运中心"},
{"time": "2026-07-05 14:20", "content": "已从深圳发出"},
]
}
Step 3:注册到 AI(最关键的一步)
这一步决定了 AI 在什么时候、以什么方式调用你的 Skill。用 LangChain 的 Tool 装饰器来注册:
from langchain.tools import tool
from langchain.schema import Tool
方式一:用装饰器快速注册
@tool
def query_express(tracking_number: str, carrier: str = "auto") -> str:
"""
查询快递物流信息。
tracking_number: 快递单号
carrier: 快递公司编码(zto/yto/sf),不填则自动识别
"""
skill = ExpressSkill()
result = skill.execute(ExpressInput(
tracking_number=tracking_number,
carrier=carrier
))
return f"物流状态:{result.status} 最新位置:{result.location} 轨迹详情:" + "".join(
[f" {d['time']} {d['content']}" for d in result.details]
)
方式二:手动注册为 Tool(更灵活,适合复杂场景)
express_tool = Tool(
name="快递查询",
func=lambda tn, c="auto": query_express(tn, c),
description="用于查询快递物流信息,输入单号和可选快递公司编码"
)
Step 4:接入大模型(让 AI“学会”用这个 Skill)
from langchain.agents import initialize_agent, AgentType
from langchain_openai import ChatOpenAI
初始化模型
llm = ChatOpenAI(model="gpt-4", temperature=0)
把 Skill 塞进 Agent 的工具箱
tools = [query_express] # 可以继续添加其他 Skill
agent = initialize_agent(
tools=tools,
llm=llm,
agent=AgentType.OPENAI_FUNCTIONS,
verbose=True # 调试时打开,能看清 AI 的思考过程
)
测试
response = agent.invoke({
"input": "帮我查一下单号SF1234567890的快递到哪了?"
})
print(response["output"])
运行结果:
物流状态:派送中
最新位置:广州市天河区体育西路营业点
轨迹详情:
2026-07-06 08:30 快递员正在派送中,预计今日送达
2026-07-05 22:15 到达广州转运中心
2026-07-05 14:20 已从深圳发出
看到没?AI 自动识别了“查快递”的意图,提取了单号,调用了 Skill,然后把结果整理成可读的文字。全程没改一行核心系统代码。
04 进阶:Skill 设计的 3 个“要命”细节
上面那个 Demo 能跑通,但离“生产级”还差得远。线上环境被这 3 个问题坑过不止一次,最好提前知道。
细节 1:超时控制 + 熔断
Skill 调用外部 API 如果挂了,整个 Agent 会跟着卡死。必须加超时:
import asyncio
from functools import wraps
def with_timeout(seconds=10):
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
try:
return await asyncio.wait_for(
func(*args, **kwargs),
timeout=seconds
)
except asyncio.TimeoutError:
return "【系统提示】该 Skill 响应超时,请稍后重试或尝试其他方式"
return wrapper
return decorator
用法
@with_timeout(5)
async def query_express_async(tn: str):
# 异步请求快递 API
pass
细节 2:输入校验的“防呆设计”
用户输入的快递单号千奇百怪,空格、换行、中文括号全都有。吃过亏,用户输入“SF 1234567890”中间有个空格,正则直接匹配失败。
def sanitize_tracking_number(raw: str) -> str:
"""清洗用户输入,只保留数字和字母"""
return re.sub(r'[^A-Za-z0-9]', '', raw.strip())
还有,Skill 里一定要有 fallback。识别不出快递公司就用“中通”兜底,查不到数据就返回明确的“未查询到”而不是抛异常。
细节 3:并发安全
如果你的 AI 同时处理多个用户的请求,Skill 内部如果有共享状态(比如缓存、数据库连接池),必须注意线程安全。
import threading
class ExpressSkill:
_cache = {}
_cache_lock = threading.Lock()
def execute(self, input: ExpressInput):
# 读缓存要加锁
with self._cache_lock:
if input.tracking_number in self._cache:
return self._cache[input.tracking_number]
# 查询逻辑...
# 写缓存也要加锁
with self._cache_lock:
self._cache[input.tracking_number] = result
return result
05 团队协作:Skill 的“集市模式”
团队现在有 6 个人,分别维护不同的 Skill。前端写“页面操作 Skill”,后端写“数据查询 Skill”,算法组写“推荐 Skill”。大家互不干扰,最后通过一个统一的 Skill Registry(注册中心)聚合起来。
目录结构长这样:
skills/
├── registry.yaml # 所有 Skill 的注册清单
├── express/
│ ├── skill.py
│ ├── config.yaml
│ └── tests/
├── jira/
│ ├── skill.py
│ └── README.md
├── sql_analytics/
│ ├── skill.py
│ └── queries/
└── weather/
├── skill.py
└── requirements.txt # 每个 Skill 可以有自己的依赖
每个 Skill 独立维护自己的版本、依赖、测试用例。集成测试只测“调用接口是否正常”,不管内部实现。
这是最像“App Store”的地方——各个 Skill 独立开发、独立发布,主系统只认接口契约。
06 写在最后:Skill 不是银弹,但它是方向
老实说,Skill 这套模式不是没有代价。它增加了系统整体的复杂度——你得维护注册中心、做热加载、处理版本兼容。如果你的业务只有两三个 AI 能力,硬编码比搞 Skill 架构快得多。
但当 AI 能力超过 5 个、团队超过 3 个人的时候,Skill 带来的解耦收益会指数级增长。系统里跑了 17 个 Skill,涵盖客服、运维、数据分析三个领域,新增一个能力从“改核心代码 + 全量回归”变成了“写 Skill + 注册”,周期从 2 周压缩到 2 天。
如果正在项目中引入 AI 能力,建议是:
前 3 个能力硬编码,先跑通业务,理解真实需求
到第 4-5 个能力时,回头抽象出 Skill 接口,把前 3 个也迁过来
第 6 个以后,所有新能力一律以 Skill 形式接入
这样既不会过度设计,又能平滑过渡到插件化架构。
最后,送给大家写在团队 Wiki 首页的一句话:
“Skill 化的本质,是把 AI 从一个‘什么都懂但什么都做不精’的泛泛之辈,变成一个‘随时可以请外援’的团队指挥官。”
下次有机会,再写一篇《Skill 的异步调度与分布式部署实战》,讲讲怎么把 Skill 跑在 K8s 上、以及如何用 RabbitMQ 做 Skill 的任务队列。
