作为一名在AI领域摸爬滚打多年的开发者,我深知新手入门AI Skill开发时的困惑。很多人一上来就急着写代码,结果发现跑不起来,或者写出来的Skill难以维护和扩展。今天,我想分享一套经过实战检验的AI Skill开发方法,用餐厅运营的类比帮你快速理解核心概念。
想象一下你去餐厅吃饭的场景,这个过程中涉及三个关键角色:服务员、厨师和传菜员。在AI Skill开发中,这三个角色分别对应着:
这种类比之所以有效,是因为餐厅运营和AI Skill开发都强调"分工明确"。就像餐厅里不会让厨师同时负责点单和上菜一样,AI Skill开发也需要清晰的职责划分。
我见过太多新手开发者的Skill项目,结构混乱得像一锅粥:有的Skill目录叫"weather",有的叫"query_weather";有的把函数放在main.py,有的放在handler.py。这种不一致性会导致:
这就是为什么我们要从一开始就建立严格的规范。就像连锁餐厅的标准化操作流程一样,统一的Skill结构能让你的开发效率提升数倍。
让我们先来看一个标准的单个Skill目录结构,以查询天气功能为例:
code复制weather-query/ # 技能根目录(功能-动作格式)
├── SKILL.md # 技能说明文档
└── scripts/ # 执行代码目录
└── skills.py # 技能实现函数
这个结构看似简单,但每个部分都有其特定用途:
提示:即使你目前只开发一个Skill,也应该遵循这个结构。等到需要添加第二个Skill时,你会感谢现在的自己。
当你的项目包含多个Skill时,目录结构需要升级为:
code复制ai-skills/ # 项目根目录
├── common/ # 公共组件
│ └── harness.py # 统一调度器
├── weather-query/ # 天气查询技能
│ ├── SKILL.md
│ └── scripts/
│ └── skills.py
└── add-calculator/ # 加法计算技能
├── SKILL.md
└── scripts/
└── skills.py
这种结构的优势在于:
你可能觉得文件名无关紧要,但在我参与过的大型AI项目中,不一致的命名导致了无数问题。我们的规范要求:
这种严格的命名约定虽然初期需要适应,但长期来看能显著降低维护成本。当你的项目有几十个Skill时,统一的命名能让你快速定位到任何功能。
让我们从最简单的查询天气功能开始,我会带你一步步完成实现。
首先创建weather-query/SKILL.md文件,内容如下:
markdown复制# 技能名称:查询天气
功能:用户输入城市名,返回该城市的实时天气(模拟真实接口,可直接替换为真实API)
参数:
- city:城市名称(字符串类型,比如北京、上海、成都),必填项,不可为空
调用格式(AI输出格式,必须严格遵循,所有Skill统一此格式):
{
"name": "查询天气",
"params": {
"city": "北京"
}
}
备注:
1. 若未传入city参数,AI需提示用户"请输入要查询的城市名称"
2. 若传入的城市名称无效,函数会返回提示信息
这个文件有几个关键点:
接下来是weather-query/scripts/skills.py文件:
python复制def query_weather(city):
"""
查询城市天气(模拟接口)
:param city: 城市名称(字符串)
:return: 天气信息(字符串)
"""
# 参数校验
if not isinstance(city, str) or not city.strip():
return "请输入有效的城市名称"
# 模拟天气查询(实际项目中替换为真实API调用)
weather_data = {
"北京": "晴天,20℃,微风",
"上海": "多云,22℃,东南风3级",
"广州": "阵雨,25℃,南风2级"
}
return weather_data.get(city, f"暂未收录{city}的天气信息")
这个实现包含了几个最佳实践:
创建weather-query/scripts/test.py进行测试:
python复制from skills import query_weather
def test_query_weather():
# 测试正常情况
print(query_weather("北京")) # 预期:晴天,20℃,微风
print(query_weather("上海")) # 预期:多云,22℃,东南风3级
# 测试异常情况
print(query_weather("")) # 预期:请输入有效的城市名称
print(query_weather("杭州")) # 预期:暂未收录杭州的天气信息
if __name__ == "__main__":
test_query_weather()
运行测试确保功能正常:
bash复制cd weather-query/scripts
python test.py
现在你已经完成了第一个Skill,第二个就简单多了。我们快速实现一个加法计算功能。
创建add-calculator/SKILL.md:
markdown复制# 技能名称:加法计算
功能:计算两个数字的和,支持整数、小数计算
参数:
- a:第一个数字(整数/小数),必填项
- b:第二个数字(整数/小数),必填项
调用格式(AI输出格式,必须严格遵循,和所有Skill统一):
{
"name": "加法计算",
"params": {
"a": 10,
"b": 20
}
}
备注:若传入的参数不是数字,函数会返回"参数错误,请传入有效的数字(整数或小数)"
注意这个SKILL.md的结构与天气查询完全一致,只是内容不同。
add-calculator/scripts/skills.py内容:
python复制def add(a, b):
"""
计算两个数字的和
:param a: 第一个数字(整数/小数)
:param b: 第二个数字(整数/小数)
:return: 加法计算结果(字符串)
"""
# 参数校验
if not isinstance(a, (int, float)) or not isinstance(b, (int, float)):
return "参数错误,请传入有效的数字(整数或小数)"
# 返回格式化结果
return f"{a} + {b} = {a + b}"
这个实现展示了数值处理的技巧:
创建add-calculator/scripts/test.py:
python复制from skills import add
def test_add():
# 测试整数
print(add(10, 20)) # 预期:10 + 20 = 30
# 测试小数
print(add(1.5, 2.5)) # 预期:1.5 + 2.5 = 4.0
# 测试错误输入
print(add("10", 20)) # 预期:参数错误...
if __name__ == "__main__":
test_add()
运行测试:
bash复制cd add-calculator/scripts
python test.py
在开发多个Skill后,你会发现每个Skill都需要:
统一调度器解决了这些问题,它相当于一个"总控中心",负责:
创建ai-skills/common/harness.py:
python复制import logging
from importlib import import_module
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class SkillHarness:
def __init__(self):
self.skill_map = {}
self._load_skills()
def _load_skills(self):
"""动态加载所有Skill"""
skills = [
"weather_query.scripts.skills",
"add_calculator.scripts.skills"
]
for skill_path in skills:
try:
module = import_module(skill_path)
skill_name = skill_path.split(".")[0].replace("_", " ")
# 假设每个模块只有一个主要函数
func = getattr(module, next(
f for f in dir(module)
if not f.startswith("_")
))
self.skill_map[skill_name] = func
logger.info(f"成功加载技能: {skill_name}")
except Exception as e:
logger.error(f"加载技能{skill_path}失败: {str(e)}")
def execute(self, skill_name, params):
"""
执行指定技能
:param skill_name: 技能名称(如"weather query")
:param params: 参数字典
:return: (success, result)
"""
if skill_name not in self.skill_map:
return False, f"未找到技能: {skill_name}"
try:
result = self.skill_map[skill_name](**params)
return True, result
except Exception as e:
logger.exception(f"执行技能{skill_name}出错")
return False, f"技能执行错误: {str(e)}"
# 示例使用
if __name__ == "__main__":
harness = SkillHarness()
# 测试天气查询
success, result = harness.execute(
"weather query",
{"city": "北京"}
)
print(f"天气查询结果: {result}")
# 测试加法计算
success, result = harness.execute(
"add calculator",
{"a": 10, "b": 20}
)
print(f"加法计算结果: {result}")
这个调度器有几个关键设计:
在实际项目中,你还可以为调度器添加更多功能:
这些扩展都能在调度器中统一实现,而不需要修改各个Skill的代码。
首先确保你的项目结构如下:
code复制ai-skills/
├── common/
│ └── harness.py
├── weather-query/
│ ├── SKILL.md
│ └── scripts/
│ ├── skills.py
│ └── test.py
└── add-calculator/
├── SKILL.md
└── scripts/
├── skills.py
└── test.py
创建ai-skills/main.py作为应用入口:
python复制from common.harness import SkillHarness
import json
def main():
# 初始化调度器
harness = SkillHarness()
# 模拟用户交互
while True:
print("\n可用技能:")
print("1. 查询天气")
print("2. 加法计算")
print("0. 退出")
choice = input("请选择功能编号: ")
if choice == "0":
break
elif choice == "1":
city = input("请输入城市名称: ")
success, result = harness.execute(
"weather query",
{"city": city}
)
elif choice == "2":
try:
a = float(input("请输入第一个数字: "))
b = float(input("请输入第二个数字: "))
success, result = harness.execute(
"add calculator",
{"a": a, "b": b}
)
except ValueError:
result = "输入错误,请输入有效数字"
success = False
else:
result = "无效选择"
success = False
print("\n结果:")
print(result if success else f"错误: {result}")
if __name__ == "__main__":
main()
启动程序:
bash复制cd ai-skills
python main.py
你将看到交互式菜单,可以测试两个Skill的功能。这个简单的CLI界面展示了如何将多个Skill集成到一个统一系统中。
在开发AI Skill过程中,我遇到过各种问题,以下是典型场景及解决方法:
问题1:Skill无法被调度器识别
问题2:参数传递错误
问题3:新增Skill后系统崩溃
当Skill数量增多时,需要考虑性能问题:
完善的测试是保证Skill质量的关键:
掌握了基础框架后,你可以考虑以下扩展方向:
之前的天气查询使用的是模拟数据,现在我们来对接真实API。以和风天气为例:
python复制import requests
def query_weather(city, api_key="YOUR_API_KEY"):
"""
查询真实天气(使用和风天气API)
:param city: 城市名称
:param api_key: 和风天气API Key
:return: 天气信息字符串
"""
# 1. 获取城市LocationID
geo_url = f"https://geoapi.qweather.com/v2/city/lookup?location={city}&key={api_key}"
try:
geo_resp = requests.get(geo_url).json()
if geo_resp["code"] != "200":
return f"获取城市信息失败: {geo_resp.get('message', '未知错误')}"
location_id = geo_resp["location"][0]["id"]
except Exception as e:
return f"查询城市ID出错: {str(e)}"
# 2. 获取实时天气
weather_url = f"https://devapi.qweather.com/v7/weather/now?location={location_id}&key={api_key}"
try:
weather_resp = requests.get(weather_url).json()
if weather_resp["code"] != "200":
return f"获取天气失败: {weather_resp.get('message', '未知错误')}"
now = weather_resp["now"]
return (
f"{city}当前天气:{now['text']},"
f"温度{now['temp']}℃,"
f"湿度{now['humidity']}%,"
f"风向{now['windDir']},"
f"风力{now['windScale']}级"
)
except Exception as e:
return f"查询天气出错: {str(e)}"
这个改进版本:
某些Skill可能需要权限控制,可以在调度器中实现:
python复制class SkillHarness:
def __init__(self):
self.skill_map = {}
self.skill_permissions = {
"weather query": ["user", "admin"],
"add calculator": ["user", "admin"],
"admin only skill": ["admin"]
}
self._load_skills()
def execute(self, skill_name, params, user_role="user"):
"""添加user_role参数"""
if skill_name not in self.skill_map:
return False, f"未找到技能: {skill_name}"
# 检查权限
if user_role not in self.skill_permissions.get(skill_name, []):
return False, "无权使用此技能"
try:
result = self.skill_map[skill_name](**params)
return True, result
except Exception as e:
logger.exception(f"执行技能{skill_name}出错")
return False, f"技能执行错误: {str(e)}"
对于耗时的Skill,可以使用异步模式:
python复制import asyncio
class AsyncSkillHarness:
async def execute(self, skill_name, params):
if skill_name not in self.skill_map:
return False, f"未找到技能: {skill_name}"
try:
# 假设skill函数是async的
result = await self.skill_map[skill_name](**params)
return True, result
except Exception as e:
logger.exception(f"执行技能{skill_name}出错")
return False, f"技能执行错误: {str(e)}"
使用Prometheus等工具监控Skill使用情况:
python复制from prometheus_client import Counter, Summary
# 定义指标
SKILL_CALL_TOTAL = Counter(
'skill_calls_total',
'Total number of skill calls',
['skill_name']
)
SKILL_DURATION = Summary(
'skill_duration_seconds',
'Time spent processing skill calls',
['skill_name']
)
class MonitoredSkillHarness(SkillHarness):
def execute(self, skill_name, params):
SKILL_CALL_TOTAL.labels(skill_name).inc()
with SKILL_DURATION.labels(skill_name).time():
return super().execute(skill_name, params)
经过多个AI Skill项目的实践,我总结了以下最佳实践:
记住,好的AI Skill系统就像运转良好的餐厅:每个角色各司其职,流程标准化,扩展性强。当你需要新增功能时,就像在菜单上添加新菜品一样简单。