共计 3448 个字符,预计需要花费 9 分钟才能阅读完成。
目录
为什么需要技能市场?
对新手开发者来说,直接开发一个完整的智能体往往面临两个核心痛点:

- 重复造轮子:许多基础功能(如天气查询、翻译等)需要大量开发时间,但这些功能在多个智能体中可能完全一致
- 技术门槛高:涉及自然语言理解、多轮对话管理等复杂逻辑时,初学者容易陷入技术细节而难以快速验证想法
技能市场的价值在于:
- 提供可复用的功能模块(即 ” 技能 ”)
- 标准化交互接口(输入 / 输出格式)
- 建立技能开发者和使用者的交易平台
技能市场架构解析
典型技能市场包含以下核心组件:
graph LR
A[技能仓库] --> B[调度引擎]
B --> C[计费系统]
C --> D[开发者控制台]
D --> E[数据分析平台]
- 技能仓库:存储所有已发布的技能及其元数据
- 调度引擎:根据用户请求动态调用合适的技能
- 计费系统:处理技能使用量统计和费用结算
- 开发者控制台:提供技能管理、测试和监控界面
- 数据分析平台:展示技能使用情况和性能指标
开发实战:天气预报技能
API 封装与错误处理
以下是用 Python 封装天气 API 的示例(使用 OpenWeatherMap API):
import requests
from typing import Dict, Optional
class WeatherAPI:
def __init__(self, api_key: str):
self.base_url = "https://api.openweathermap.org/data/2.5/weather"
self.api_key = api_key
def get_weather(self, city: str) -> Optional[Dict]:
"""
获取指定城市的天气数据
:param city: 城市名称(英文):return: 包含天气数据的字典,请求失败时返回 None
"""
try:
params = {
'q': city,
'appid': self.api_key,
'units': 'metric' # 使用摄氏度
}
response = requests.get(self.base_url, params=params, timeout=5)
response.raise_for_status() # 自动处理 HTTP 错误
return response.json()
except requests.exceptions.RequestException as e:
print(f"天气 API 请求失败: {e}")
return None
关键设计要点:
- 使用类型注解(Type Hints)明确输入输出类型
- 通过
raise_for_status()自动处理 HTTP 错误状态码 - 设置合理的超时时间(5 秒)
- 返回
None表示失败,避免抛出异常影响调用方
添加技能元数据
技能市场需要通过元数据了解你的技能功能。创建一个 skill_meta.json 文件:
{
"name": "weather_forecast",
"description": "提供全球城市当前天气信息,包括温度、湿度和天气状况",
"tags": ["weather", "forecast", "openweathermap"],
"input_schema": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "要查询的城市名称(英文)"
}
},
"required": ["city"]
},
"output_schema": {
"type": "object",
"properties": {"temperature": {"type": "number"},
"humidity": {"type": "number"},
"condition": {"type": "string"}
}
}
}
元数据的关键作用:
description和tags帮助用户发现你的技能input_schema定义调用时需要提供的参数output_schema明确返回数据的结构
测试与部署
单元测试要点
使用 pytest 编写测试用例,重点验证:
- 正常情况下的 API 响应
- 无效城市名的错误处理
- API 密钥错误时的行为
示例测试代码:
import pytest
from unittest.mock import patch
from weather_api import WeatherAPI
@patch('requests.get')
def test_get_weather_success(mock_get):
# 模拟 API 成功响应
mock_get.return_value.status_code = 200
mock_get.return_value.json.return_value = {'main': {'temp': 22, 'humidity': 65},
'weather': [{'main': 'Cloudy'}]
}
api = WeatherAPI("test_key")
result = api.get_weather("London")
assert result['temperature'] == 22
assert result['humidity'] == 65
assert result['condition'] == "Cloudy"
@patch('requests.get')
def test_get_weather_failure(mock_get):
# 模拟 API 失败响应
mock_get.return_value.raise_for_status.side_effect = \
requests.exceptions.HTTPError("404 Not Found")
api = WeatherAPI("test_key")
assert api.get_weather("InvalidCity") is None
性能基准测试
使用 locust 进行负载测试,重点关注:
- QPS(每秒查询数):技能能承受的最大请求频率
- P99 延迟:99% 的请求能在多少毫秒内完成
创建locustfile.py:
from locust import HttpUser, task
class WeatherSkillUser(HttpUser):
@task
def query_weather(self):
self.client.post("/execute", json={"city": "London"}, headers={"Authorization": "Bearer YOUR_SKILL_TOKEN"})
运行测试:
locust -f locustfile.py --headless -u 100 -r 10 -t 1m
参数说明:
-u 100:模拟 100 个并发用户-r 10:每秒启动 10 个用户-t 1m:测试持续 1 分钟
发布流程
典型的 CI/CD 流程:
- 开发者提交代码到 Git 仓库
- CI 系统(如 GitHub Actions)自动运行:
- 单元测试
- 代码风格检查(flake8)
- 构建 Docker 镜像
- 通过测试后自动部署到技能市场的测试环境
- 人工验证后点击 ” 发布 ” 按钮上线生产环境
避坑指南
新手常遇到的三个问题:
- 技能权限配置不当
- 问题现象:调用技能时返回 ”403 Forbidden”
-
解决方案:在开发者控制台正确设置技能的访问权限(公有 / 私有)
-
计费回调未实现
- 问题现象:技能能正常使用但开发者收不到分成
-
解决方案:实现
/callback接口处理市场的计费通知 -
输入验证缺失
- 问题现象:用户传入非法参数导致技能崩溃
- 解决方案:严格按照
input_schema校验输入数据
进阶建议
利用市场数据优化技能的三个方向:
- 使用频率分析:
- 哪些城市的查询最多?考虑缓存热门城市数据
-
一天中哪些时段流量高峰?合理调整自动扩缩容策略
-
错误日志监控:
- 建立错误告警(如 API 调用失败率 >1% 时通知)
-
分析常见错误类型并针对性优化
-
A/ B 测试:
- 尝试不同的技能描述文案,观察点击率变化
- 对比不同 API 供应商的性能和稳定性
动手实验
任务:基于以下模板开发一个日历查询技能
基础要求:
- 实现
/events接口,返回指定日期的节假日信息 - 使用公共 API(如 holidayapi.com)
- 添加完整的元数据和单元测试
扩展挑战:
- 添加缓存机制(相同日期不重复查询 API)
- 支持多国节假日查询
模板仓库:
git clone https://github.com/skill-market/calendar-skill-template.git
正文完
