1. 项目概述为什么我们需要一个结构化的 Skill 体系在构建一个功能日益复杂的智能体Agent时我们很快会遇到一个瓶颈功能膨胀与代码混乱。想象一下你最初设计的 Agent 只能回答天气问题后来你给它加上了日程管理、邮件发送、数据分析、甚至控制智能家居的能力。如果这些功能都以“if-else”或散落在各个角落的函数形式存在整个系统很快就会变成一团难以维护、无法扩展的“面条代码”。更糟糕的是当你想要复用某个功能或者为不同场景的 Agent 快速组合不同能力时你会发现无从下手。这正是“Skill 体系”要解决的核心问题。它不是一个炫技的概念而是来自一线工程实践的必然选择。简单来说Skill 就是将 Agent 的每一项独立能力如“查询天气”、“发送邮件”、“总结文档”进行标准化封装后的产物。一个设计良好的 Skill就像乐高积木的一块标准件有明确的接口输入/输出、独立的功能逻辑和自洽的错误处理。而 Skill Creator则是我们用来高效、规范地生产这些“乐高积木”的模具和流水线。我经历过从功能堆砌到体系化设计的完整过程。早期版本中每增加一个功能就要修改核心调度逻辑测试时牵一发而动全身。引入 Skill 体系后新功能的开发变成了“创建一个新的 Skill 模块”然后通过配置文件“声明”Agent 拥有此技能。系统的可维护性、可测试性和可扩展性得到了质的提升。今天我们就来深入拆解如何从零构建这套体系并打造一个强大的 Skill Creator 来支撑能力的快速迭代。2. Skill 体系的核心设计哲学与架构拆解2.1 定义 Skill不止是函数更是可复用的能力单元首先我们必须明确 Skill 与普通函数或类的区别。一个合格的 Skill 需要具备以下四个特征声明式接口Skill 必须对外明确声明它需要什么输入参数及其类型、格式以及它能提供什么输出结果及其结构。这通常通过一个标准的描述文件如skill_manifest.yaml或装饰器元数据来实现。例如一个“发送邮件”的 Skill会声明它需要recipient收件人字符串、subject主题字符串、body正文字符串三个必要参数。自包含性Skill 内部应封装完成其功能所需的所有逻辑、工具调用和外部服务依赖。理想情况下一个 Skill 不应对其他 Skill 或 Agent 的核心状态有强依赖。这保证了它的可移植性。统一的生命周期管理每个 Skill 应有标准的初始化init、执行run、清理cleanup等生命周期钩子。这便于系统进行资源管理、性能监控和错误恢复。上下文感知能力虽然强调自包含但 Skill 并非运行在真空中。它需要能够安全地访问和修改与当前任务相关的“会话上下文”Context例如用户的历史对话、当前任务的目标等。如何设计一个安全、高效的上下文共享机制是 Skill 体系的关键。基于这些特征一个典型的 Skill 基类设计可能如下所示以 Python 为例from abc import ABC, abstractmethod from typing import Any, Dict, Optional from pydantic import BaseModel class SkillInput(BaseModel): Skill 输入参数的基类使用 Pydantic 进行验证和解析。 pass class SkillOutput(BaseModel): Skill 输出结果的基类。 success: bool data: Optional[Any] None message: Optional[str] None class SkillContext: Skill 执行上下文用于在 Skill 间安全共享数据。 def __init__(self): self._store: Dict[str, Any] {} def set(self, key: str, value: Any): self._store[key] value def get(self, key: str, default: Any None) - Any: return self._store.get(key, default) class BaseSkill(ABC): 所有 Skill 的抽象基类。 name: str base_skill description: str A base skill. version: str 1.0.0 def __init__(self, context: Optional[SkillContext] None): self.context context or SkillContext() abstractmethod async def run(self, input_data: SkillInput) - SkillOutput: 执行 Skill 的核心逻辑。 pass async def initialize(self): Skill 初始化如加载模型、连接数据库。 pass async def cleanup(self): Skill 清理如关闭连接、释放资源。 pass这个设计通过抽象基类和强类型为所有 Skill 建立了统一的契约。SkillInput和SkillOutput利用 Pydantic 确保了数据进出的一致性并自动完成验证。SkillContext提供了一个简单的键值存储用于 Skill 间传递中间结果比如上一个 Skill 提取的“用户意图”可以被下一个 Skill 使用。2.2 Skill 体系的架构模式从集中注册到动态发现有了 Skill 的定义接下来需要思考如何组织和管理它们。常见的架构模式有两种模式一集中式注册表Registry这是最直观的方式。系统启动时所有 Skill 实例在一个中心化的SkillRegistry中注册。Agent 或调度器通过查询注册表来查找和调用 Skill。class SkillRegistry: _instance None _skills: Dict[str, BaseSkill] {} def __new__(cls): if cls._instance is None: cls._instance super().__new__(cls) return cls._instance def register(self, skill: BaseSkill): if skill.name in self._skills: raise ValueError(fSkill {skill.name} already registered.) self._skills[skill.name] skill print(fRegistered skill: {skill.name}) def get(self, skill_name: str) - Optional[BaseSkill]: return self._skills.get(skill_name) def list_all(self) - List[str]: return list(self._skills.keys()) # 使用示例 registry SkillRegistry() weather_skill WeatherSkill() registry.register(weather_skill)模式二动态发现与加载Discovery这种方式更适用于插件化系统。Skill 以独立模块如 Python 包的形式存在放在特定目录如skills/下。系统通过扫描该目录、导入模块并自动识别继承自BaseSkill的类来完成注册。import importlib import pkgutil from pathlib import Path def discover_skills(skills_dir: Path): for finder, name, ispkg in pkgutil.iter_modules([str(skills_dir)]): spec finder.find_spec(name) if spec and spec.loader: module importlib.import_module(spec.name) for attr_name in dir(module): attr getattr(module, attr_name) if (isinstance(attr, type) and issubclass(attr, BaseSkill) and attr ! BaseSkill): # 实例化并注册 skill_instance attr() SkillRegistry().register(skill_instance)选择建议在项目早期功能相对固定使用集中式注册表更简单可控。当 Skill 数量增多且希望第三方也能方便地扩展能力时动态发现模式的优势就体现出来了。在实际项目中我通常采用“混合模式”核心、高频的 Skill 在代码中显式注册以保证可靠性而实验性、可选的 Skill 则通过动态发现加载便于热插拔。2.3 Skill 间的协作与编排超越孤岛单个 Skill 能力有限真正的价值在于 Skill 之间的协作。这就需要一套编排Orchestration机制。最简单的编排是“线性链式调用”即一个 Skill 的输出作为下一个 Skill 的输入。更复杂的场景可能需要“条件分支”根据某个 Skill 的结果决定执行哪个分支或“并行执行”同时执行多个独立 Skill。我们可以引入一个WorkflowEngine或SkillOrchestrator来负责这件事。它解析一个用 YAML 或 DSL领域特定语言定义的工作流并负责调度执行。# workflow.yaml name: 处理用户查询 steps: - skill: intent_classifier inputs: query: {{user_input}} - skill: weather_query when: {{steps.intent_classifier.result.intent}} weather inputs: location: {{steps.intent_classifier.result.location}} - skill: calendar_check when: {{steps.intent_classifier.result.intent}} schedule inputs: date: {{steps.intent_classifier.result.date}} - skill: response_formatter inputs: raw_data: {{previous_step_result}}这个工作流定义清晰地描述了 Skill 的执行逻辑和依赖关系。WorkflowEngine的责任就是按顺序或并行执行这些步骤处理条件判断并管理步骤间的数据传递。这极大地提升了复杂任务处理的灵活性和可描述性。3. Skill Creator标准化生产流水线的构建当 Skill 的数量开始增长手动创建每个 Skill 的类、描述文件、测试用例就成了一件重复且容易出错的体力活。Skill Creator 的目标就是将这个过程自动化、标准化。3.1 Skill Creator 的核心功能设计一个完整的 Skill Creator 应该提供以下功能交互式脚手架生成通过命令行问答CLI或图形界面GUI收集新 Skill 的基本信息名称、描述、输入输出参数等自动生成符合项目规范的 Skill 代码骨架、配置文件、单元测试文件甚至基础的文档。模板引擎驱动代码生成不是硬编码而是基于模板。这样当项目的基础设施或规范变更时比如从argparse换成了click来解析 CLI只需更新模板所有新生成的 Skill 都会自动适应新规范。依赖管理与注入自动分析 Skill 声明的外部依赖如需要访问某个数据库或 API并在生成的代码中预留配置接口或在项目的全局依赖文件中添加相应条目。配置与清单Manifest生成自动生成或更新 Skill 的清单文件如skill_manifest.yaml其中包含 Skill 的元数据、输入输出模式Schema、版本、作者等信息。这个清单是系统发现、验证和调用 Skill 的重要依据。本地测试环境一键搭建生成一个docker-compose.yml或脚本能快速拉起该 Skill 所需的外部服务如测试数据库、Mock API 服务器方便开发者进行本地集成测试。3.2 实现一个基础的 CLI 版 Skill Creator下面我们实现一个简化但实用的命令行版 Skill Creator。它使用 Python 的click库和Jinja2模板引擎。首先定义项目模板结构。假设我们有一个templates/skill_template目录里面包含__init__.py.j2: Skill 包的初始化文件模板。skill.py.j2: Skill 主类文件模板。manifest.yaml.j2: Skill 清单文件模板。test_skill.py.j2: 单元测试文件模板。然后创建我们的 Creator 脚本skill_creator.pyimport click from pathlib import Path from jinja2 import Environment, FileSystemLoader import yaml TEMPLATES_DIR Path(__file__).parent / templates SKILLS_DIR Path(__file__).parent / skills click.group() def cli(): Skill 创建与管理工具。 pass cli.command() click.option(--name, promptSkill 名称英文小写加下划线如 send_email, helpSkill 的唯一标识名。) click.option(--desc, promptSkill 功能描述, help一句话描述这个 Skill 是做什么的。) click.option(--author, prompt开发者名称, defaultYour Name, helpSkill 的作者。) def new(name, desc, author): 创建一个新的 Skill 骨架。 skill_dir SKILLS_DIR / name if skill_dir.exists(): click.echo(f错误Skill 目录 {name} 已存在, errTrue) return # 收集输入输出参数简化示例实际可做成循环交互 inputs [] click.echo(现在定义 Skill 的输入参数输入空行结束) while True: param_name click.prompt(参数名, default, show_defaultFalse) if not param_name: break param_type click.prompt(参数类型 (e.g., str, int, List[str]), defaultstr) param_desc click.prompt(参数描述, default) inputs.append({name: param_name, type: param_type, description: param_desc}) # 准备模板渲染上下文 context { skill_name: name, skill_class_name: .join(word.capitalize() for word in name.split(_)) Skill, description: desc, author: author, inputs: inputs, version: 0.1.0 } # 设置 Jinja2 环境 env Environment(loaderFileSystemLoader(TEMPLATES_DIR / skill_template), trim_blocksTrue, lstrip_blocksTrue) env.filters[to_yaml] lambda x: yaml.dump(x, default_flow_styleFalse, allow_unicodeTrue) # 创建目录并渲染模板 skill_dir.mkdir(parentsTrue) (skill_dir / __init__.py).write_text() templates_to_render [ (skill.py.j2, skill_dir / skill.py), (manifest.yaml.j2, skill_dir / manifest.yaml), (test_skill.py.j2, skill_dir / test_skill.py), ] for template_name, output_path in templates_to_render: template env.get_template(template_name) content template.render(**context) output_path.write_text(content) click.echo(f已创建{output_path}) click.echo(f\n✅ Skill {name} 骨架创建成功) click.echo(f目录位于{skill_dir}) click.echo(下一步) click.echo( 1. 编辑 skill.py 实现你的核心逻辑。) click.echo( 2. 根据 manifest.yaml 检查输入输出定义。) click.echo( 3. 运行 pytest skills/{name}/ 进行测试。) if __name__ __main__: cli()对应的 Jinja2 模板示例 (skill.py.j2)# skills/{{ skill_name }}/skill.py from typing import Any, Dict, Optional from ..base import BaseSkill, SkillInput, SkillOutput, SkillContext class {{ skill_class_name }}Input(SkillInput): {{ skill_name }} 的输入参数定义。 {% for param in inputs %} {{ param.name }}: {{ param.type }} # {{ param.description }} {% endfor %} class {{ skill_class_name }}(BaseSkill): {{ description }} name {{ skill_name }} description {{ description }} version {{ version }} def __init__(self, context: Optional[SkillContext] None): super().__init__(context) # 在此初始化你的客户端、模型等资源 # self.client SomeClient() async def run(self, input_data: {{ skill_class_name }}Input) - SkillOutput: 执行 {{ skill_name }} 的核心逻辑。 请在此处实现你的功能。 try: # TODO: 实现你的业务逻辑 # 示例 result await self.client.do_something(**input_data.dict()) result {message: 功能待实现} return SkillOutput( successTrue, dataresult, message{{ skill_name }} 执行成功。 ) except Exception as e: # 务必捕获异常并返回明确的错误输出 return SkillOutput( successFalse, dataNone, messagef执行 {{ skill_name }} 时出错{str(e)} )运行python skill_creator.py new跟随提示操作一个结构清晰、符合规范的 Skill 骨架就生成了。这极大地提升了开发效率并保证了项目代码风格的一致性。3.3 清单Manifest文件Skill 的“身份证”上面模板中生成的manifest.yaml文件至关重要。它是系统了解一个 Skill 的权威来源应该包含# manifest.yaml name: send_email version: 0.1.0 description: 发送电子邮件到指定地址。 author: Your Name entry_point: skill:SendEmailSkill # 指向 Skill 类 input_schema: type: object required: - recipient - subject - body properties: recipient: type: string description: 收件人邮箱地址 subject: type: string description: 邮件主题 body: type: string description: 邮件正文 output_schema: type: object properties: message_id: type: string description: 邮件服务器返回的消息ID dependencies: - pydantic2.0 - aiosmtplib tags: - communication - notification这个清单文件可以被 Skill 加载器Loader读取用于验证在调用前校验输入参数是否符合input_schema。发现在 UI 或目录中展示所有可用的 Skill 及其功能。文档自动生成 API 文档。编排工作流引擎根据输入输出模式Schema自动连接不同的 Skill。4. Skill 的迭代、测试与部署实践4.1 版本控制与灰度发布Skill 作为独立的能力单元也应该有自己的版本生命周期。我们可以在manifest.yaml中定义版本号遵循语义化版本major.minor.patch。当 Skill 的逻辑发生不兼容的变更时升级主版本号新增功能但向下兼容时升级次版本号仅修复 bug 时升级修订号。在部署时可以利用这个版本号实现灰度发布。例如Agent 系统可以配置为某个 Skill 默认使用1.x的最新稳定版但可以将一小部分流量或特定用户路由到正在测试的2.0.0-beta.1版本。这允许我们在不影响主流用户的情况下安全地测试新功能。4.2 单元测试与集成测试策略为 Skill 编写测试是保证其质量的关键。我们的 Skill Creator 已经生成了测试骨架test_skill.py我们需要填充它。单元测试针对 Skill 内部的纯逻辑函数。Mock 掉所有外部依赖如网络请求、数据库调用。# test_send_email.py import pytest from unittest.mock import AsyncMock, patch from skills.send_email.skill import SendEmailSkill, SendEmailSkillInput pytest.mark.asyncio async def test_send_email_success(): 测试发送邮件成功场景。 skill SendEmailSkill() # Mock 掉 SMTP 客户端 with patch(skills.send_email.skill.AsyncSMTP) as mock_smtp: mock_instance AsyncMock() mock_smtp.return_value.__aenter__.return_value mock_instance mock_instance.sendmail.return_value {} input_data SendEmailSkillInput( recipienttestexample.com, subjectTest, bodyHello World ) result await skill.run(input_data) assert result.success is True assert 消息ID in result.message # 验证 mock 方法被以正确的参数调用 mock_instance.sendmail.assert_called_once()集成测试测试 Skill 与真实或测试环境中的外部服务的交互。这需要搭建一个包含所有依赖如测试用的邮件服务器、数据库的临时环境。使用docker-compose或pytest的fixture来管理这些环境生命周期是非常好的实践。4.3 性能监控与错误处理在生产环境中我们需要知道每个 Skill 的运行状况。可以在BaseSkill的run方法周围添加装饰器或使用中间件Middleware模式来统一注入监控逻辑。import time import functools from typing import Callable from prometheus_client import Counter, Histogram SKILL_EXECUTION_COUNT Counter(skill_execution_total, Total skill executions, [skill_name, status]) SKILL_EXECUTION_DURATION Histogram(skill_execution_duration_seconds, Skill execution duration, [skill_name]) def monitor_skill_execution(func: Callable): 监控 Skill 执行的装饰器。 functools.wraps(func) async def wrapper(self, input_data: SkillInput) - SkillOutput: start_time time.time() try: result await func(self, input_data) status success if result.success else failure SKILL_EXECUTION_COUNT.labels(skill_nameself.name, statusstatus).inc() return result except Exception as e: SKILL_EXECUTION_COUNT.labels(skill_nameself.name, statuserror).inc() # 重新抛出或返回统一的错误输出 return SkillOutput(successFalse, messagefUnhandled error: {str(e)}) finally: duration time.time() - start_time SKILL_EXECUTION_DURATION.labels(skill_nameself.name).observe(duration) return wrapper # 在 BaseSkill.run 方法上应用此装饰器这样我们就能在监控系统如 Grafana中看到每个 Skill 的调用次数、成功/失败率以及耗时分布便于快速定位性能瓶颈或故障 Skill。4.4 安全与权限考量Skill 可能执行敏感操作如发送邮件、访问数据库、调用付费 API。必须在设计层面加入权限控制。一种常见做法是在SkillContext中携带“用户身份”和“权限令牌”在每个 Skill 的run方法开始处进行校验。class SecureSkill(BaseSkill): required_permissions [send_email] async def run(self, input_data: SkillInput) - SkillOutput: # 从上下文中获取用户权限 user_permissions self.context.get(user_permissions, []) if not all(perm in user_permissions for perm in self.required_permissions): return SkillOutput( successFalse, message权限不足无法执行此操作。 ) # ... 执行业务逻辑更复杂的系统可能会引入基于角色的访问控制RBAC或属性基访问控制ABAC将权限检查抽象成独立的中间件在 Skill 被调用前统一拦截处理。5. 从理论到实践构建一个天气预报 Skill 的全流程让我们将上述所有概念串联起来实际构建一个“天气预报” Skill体验从创建到部署的完整流程。5.1 使用 Skill Creator 生成骨架运行我们之前写的 CLI 工具$ python skill_creator.py new Skill 名称英文小写加下划线如 send_email: get_weather Skill 功能描述: 根据城市名称查询实时天气和未来几天的预报。 开发者名称 [Your Name]: WeatherMaster 现在定义 Skill 的输入参数输入空行结束 参数名: city 参数类型 (e.g., str, int, List[str]) [str]: str 参数描述: 要查询天气的城市名称如“北京” 参数名: days 参数类型 (e.g., str, int, List[str]) [str]: int 参数描述: 预报的天数默认为1最多7天 参数名: ✅ Skill get_weather 骨架创建成功5.2 实现核心业务逻辑打开生成的skills/get_weather/skill.py文件填充run方法。我们假设使用一个免费的天气 API如 OpenWeatherMap。import aiohttp import os from typing import Any, Dict from ..base import BaseSkill, SkillInput, SkillOutput, SkillContext class GetWeatherSkillInput(SkillInput): city: str # 要查询天气的城市名称如“北京” days: int 1 # 预报的天数默认为1最多7天 class GetWeatherSkill(BaseSkill): name get_weather description 根据城市名称查询实时天气和未来几天的预报。 version 0.1.0 def __init__(self, context: SkillContext None): super().__init__(context) self.api_key os.getenv(WEATHER_API_KEY) # 从环境变量读取API密钥 if not self.api_key: raise ValueError(未设置 WEATHER_API_KEY 环境变量) self.base_url https://api.openweathermap.org/data/2.5 async def run(self, input_data: GetWeatherSkillInput) - SkillOutput: try: if input_data.days 1 or input_data.days 7: return SkillOutput(successFalse, message预报天数需在1-7天之间。) # 1. 获取实时天气 current_weather await self._fetch_current_weather(input_data.city) # 2. 获取天气预报 forecast await self._fetch_forecast(input_data.city, input_data.days) result { city: input_data.city, current: current_weather, forecast: forecast } return SkillOutput(successTrue, dataresult, messagef成功获取 {input_data.city} 的天气信息。) except aiohttp.ClientError as e: return SkillOutput(successFalse, messagef网络请求失败{str(e)}) except Exception as e: return SkillOutput(successFalse, messagef处理天气数据时出错{str(e)}) async def _fetch_current_weather(self, city: str) - Dict[str, Any]: async with aiohttp.ClientSession() as session: url f{self.base_url}/weather params {q: city, appid: self.api_key, units: metric, lang: zh_cn} async with session.get(url, paramsparams) as resp: resp.raise_for_status() data await resp.json() # 提取并格式化需要的信息 return { temp: data[main][temp], feels_like: data[main][feels_like], humidity: data[main][humidity], description: data[weather][0][description], icon: data[weather][0][icon] } async def _fetch_forecast(self, city: str, days: int) - list: # 类似地调用预报API这里省略具体实现 # 通常是调用 /forecast/daily 或处理 /forecast 的每3小时数据 pass5.3 编写与运行测试填充生成的测试文件并运行$ cd /path/to/your/project $ WEATHER_API_KEYtest_key pytest skills/get_weather/ -v确保所有测试通过特别是要覆盖网络错误、API 返回异常数据等边界情况。5.4 更新清单与注册检查并完善自动生成的manifest.yaml确保input_schema与代码中的GetWeatherSkillInput类定义一致。然后我们需要将这个 Skill 注册到系统中。如果使用动态发现只需确保skills/get_weather目录在扫描路径下即可。如果使用集中注册需要在 Agent 初始化代码中显式添加from skills.get_weather.skill import GetWeatherSkill registry SkillRegistry() registry.register(GetWeatherSkill())5.5 集成到 Agent 并验证最后在你的 Agent 主逻辑中就可以调用这个 Skill 了async def handle_user_query(query: str): # ... 假设通过某种方式如LLM解析出意图和参数 intent get_weather params {city: 北京, days: 2} if intent get_weather: skill registry.get(get_weather) if skill: input_data GetWeatherSkillInput(**params) result await skill.run(input_data) if result.success: # 将 result.data 格式化成对用户友好的回复 return format_weather_response(result.data) else: return f抱歉查询天气时出错{result.message} # ...至此一个具备完整生命周期、可测试、可监控、可复用的天气预报 Skill 就构建并集成完毕了。通过 Skill Creator我们标准化了创建流程通过清晰的架构和清单文件我们实现了能力的模块化和可管理性。这套体系不仅能用于聊天机器人同样适用于自动化流程、智能助手、后台任务调度等任何需要组合多种能力的复杂系统。 构建可扩展的智能体Skill体系：从设计哲学到工程实践
1. 项目概述为什么我们需要一个结构化的 Skill 体系在构建一个功能日益复杂的智能体Agent时我们很快会遇到一个瓶颈功能膨胀与代码混乱。想象一下你最初设计的 Agent 只能回答天气问题后来你给它加上了日程管理、邮件发送、数据分析、甚至控制智能家居的能力。如果这些功能都以“if-else”或散落在各个角落的函数形式存在整个系统很快就会变成一团难以维护、无法扩展的“面条代码”。更糟糕的是当你想要复用某个功能或者为不同场景的 Agent 快速组合不同能力时你会发现无从下手。这正是“Skill 体系”要解决的核心问题。它不是一个炫技的概念而是来自一线工程实践的必然选择。简单来说Skill 就是将 Agent 的每一项独立能力如“查询天气”、“发送邮件”、“总结文档”进行标准化封装后的产物。一个设计良好的 Skill就像乐高积木的一块标准件有明确的接口输入/输出、独立的功能逻辑和自洽的错误处理。而 Skill Creator则是我们用来高效、规范地生产这些“乐高积木”的模具和流水线。我经历过从功能堆砌到体系化设计的完整过程。早期版本中每增加一个功能就要修改核心调度逻辑测试时牵一发而动全身。引入 Skill 体系后新功能的开发变成了“创建一个新的 Skill 模块”然后通过配置文件“声明”Agent 拥有此技能。系统的可维护性、可测试性和可扩展性得到了质的提升。今天我们就来深入拆解如何从零构建这套体系并打造一个强大的 Skill Creator 来支撑能力的快速迭代。2. Skill 体系的核心设计哲学与架构拆解2.1 定义 Skill不止是函数更是可复用的能力单元首先我们必须明确 Skill 与普通函数或类的区别。一个合格的 Skill 需要具备以下四个特征声明式接口Skill 必须对外明确声明它需要什么输入参数及其类型、格式以及它能提供什么输出结果及其结构。这通常通过一个标准的描述文件如skill_manifest.yaml或装饰器元数据来实现。例如一个“发送邮件”的 Skill会声明它需要recipient收件人字符串、subject主题字符串、body正文字符串三个必要参数。自包含性Skill 内部应封装完成其功能所需的所有逻辑、工具调用和外部服务依赖。理想情况下一个 Skill 不应对其他 Skill 或 Agent 的核心状态有强依赖。这保证了它的可移植性。统一的生命周期管理每个 Skill 应有标准的初始化init、执行run、清理cleanup等生命周期钩子。这便于系统进行资源管理、性能监控和错误恢复。上下文感知能力虽然强调自包含但 Skill 并非运行在真空中。它需要能够安全地访问和修改与当前任务相关的“会话上下文”Context例如用户的历史对话、当前任务的目标等。如何设计一个安全、高效的上下文共享机制是 Skill 体系的关键。基于这些特征一个典型的 Skill 基类设计可能如下所示以 Python 为例from abc import ABC, abstractmethod from typing import Any, Dict, Optional from pydantic import BaseModel class SkillInput(BaseModel): Skill 输入参数的基类使用 Pydantic 进行验证和解析。 pass class SkillOutput(BaseModel): Skill 输出结果的基类。 success: bool data: Optional[Any] None message: Optional[str] None class SkillContext: Skill 执行上下文用于在 Skill 间安全共享数据。 def __init__(self): self._store: Dict[str, Any] {} def set(self, key: str, value: Any): self._store[key] value def get(self, key: str, default: Any None) - Any: return self._store.get(key, default) class BaseSkill(ABC): 所有 Skill 的抽象基类。 name: str base_skill description: str A base skill. version: str 1.0.0 def __init__(self, context: Optional[SkillContext] None): self.context context or SkillContext() abstractmethod async def run(self, input_data: SkillInput) - SkillOutput: 执行 Skill 的核心逻辑。 pass async def initialize(self): Skill 初始化如加载模型、连接数据库。 pass async def cleanup(self): Skill 清理如关闭连接、释放资源。 pass这个设计通过抽象基类和强类型为所有 Skill 建立了统一的契约。SkillInput和SkillOutput利用 Pydantic 确保了数据进出的一致性并自动完成验证。SkillContext提供了一个简单的键值存储用于 Skill 间传递中间结果比如上一个 Skill 提取的“用户意图”可以被下一个 Skill 使用。2.2 Skill 体系的架构模式从集中注册到动态发现有了 Skill 的定义接下来需要思考如何组织和管理它们。常见的架构模式有两种模式一集中式注册表Registry这是最直观的方式。系统启动时所有 Skill 实例在一个中心化的SkillRegistry中注册。Agent 或调度器通过查询注册表来查找和调用 Skill。class SkillRegistry: _instance None _skills: Dict[str, BaseSkill] {} def __new__(cls): if cls._instance is None: cls._instance super().__new__(cls) return cls._instance def register(self, skill: BaseSkill): if skill.name in self._skills: raise ValueError(fSkill {skill.name} already registered.) self._skills[skill.name] skill print(fRegistered skill: {skill.name}) def get(self, skill_name: str) - Optional[BaseSkill]: return self._skills.get(skill_name) def list_all(self) - List[str]: return list(self._skills.keys()) # 使用示例 registry SkillRegistry() weather_skill WeatherSkill() registry.register(weather_skill)模式二动态发现与加载Discovery这种方式更适用于插件化系统。Skill 以独立模块如 Python 包的形式存在放在特定目录如skills/下。系统通过扫描该目录、导入模块并自动识别继承自BaseSkill的类来完成注册。import importlib import pkgutil from pathlib import Path def discover_skills(skills_dir: Path): for finder, name, ispkg in pkgutil.iter_modules([str(skills_dir)]): spec finder.find_spec(name) if spec and spec.loader: module importlib.import_module(spec.name) for attr_name in dir(module): attr getattr(module, attr_name) if (isinstance(attr, type) and issubclass(attr, BaseSkill) and attr ! BaseSkill): # 实例化并注册 skill_instance attr() SkillRegistry().register(skill_instance)选择建议在项目早期功能相对固定使用集中式注册表更简单可控。当 Skill 数量增多且希望第三方也能方便地扩展能力时动态发现模式的优势就体现出来了。在实际项目中我通常采用“混合模式”核心、高频的 Skill 在代码中显式注册以保证可靠性而实验性、可选的 Skill 则通过动态发现加载便于热插拔。2.3 Skill 间的协作与编排超越孤岛单个 Skill 能力有限真正的价值在于 Skill 之间的协作。这就需要一套编排Orchestration机制。最简单的编排是“线性链式调用”即一个 Skill 的输出作为下一个 Skill 的输入。更复杂的场景可能需要“条件分支”根据某个 Skill 的结果决定执行哪个分支或“并行执行”同时执行多个独立 Skill。我们可以引入一个WorkflowEngine或SkillOrchestrator来负责这件事。它解析一个用 YAML 或 DSL领域特定语言定义的工作流并负责调度执行。# workflow.yaml name: 处理用户查询 steps: - skill: intent_classifier inputs: query: {{user_input}} - skill: weather_query when: {{steps.intent_classifier.result.intent}} weather inputs: location: {{steps.intent_classifier.result.location}} - skill: calendar_check when: {{steps.intent_classifier.result.intent}} schedule inputs: date: {{steps.intent_classifier.result.date}} - skill: response_formatter inputs: raw_data: {{previous_step_result}}这个工作流定义清晰地描述了 Skill 的执行逻辑和依赖关系。WorkflowEngine的责任就是按顺序或并行执行这些步骤处理条件判断并管理步骤间的数据传递。这极大地提升了复杂任务处理的灵活性和可描述性。3. Skill Creator标准化生产流水线的构建当 Skill 的数量开始增长手动创建每个 Skill 的类、描述文件、测试用例就成了一件重复且容易出错的体力活。Skill Creator 的目标就是将这个过程自动化、标准化。3.1 Skill Creator 的核心功能设计一个完整的 Skill Creator 应该提供以下功能交互式脚手架生成通过命令行问答CLI或图形界面GUI收集新 Skill 的基本信息名称、描述、输入输出参数等自动生成符合项目规范的 Skill 代码骨架、配置文件、单元测试文件甚至基础的文档。模板引擎驱动代码生成不是硬编码而是基于模板。这样当项目的基础设施或规范变更时比如从argparse换成了click来解析 CLI只需更新模板所有新生成的 Skill 都会自动适应新规范。依赖管理与注入自动分析 Skill 声明的外部依赖如需要访问某个数据库或 API并在生成的代码中预留配置接口或在项目的全局依赖文件中添加相应条目。配置与清单Manifest生成自动生成或更新 Skill 的清单文件如skill_manifest.yaml其中包含 Skill 的元数据、输入输出模式Schema、版本、作者等信息。这个清单是系统发现、验证和调用 Skill 的重要依据。本地测试环境一键搭建生成一个docker-compose.yml或脚本能快速拉起该 Skill 所需的外部服务如测试数据库、Mock API 服务器方便开发者进行本地集成测试。3.2 实现一个基础的 CLI 版 Skill Creator下面我们实现一个简化但实用的命令行版 Skill Creator。它使用 Python 的click库和Jinja2模板引擎。首先定义项目模板结构。假设我们有一个templates/skill_template目录里面包含__init__.py.j2: Skill 包的初始化文件模板。skill.py.j2: Skill 主类文件模板。manifest.yaml.j2: Skill 清单文件模板。test_skill.py.j2: 单元测试文件模板。然后创建我们的 Creator 脚本skill_creator.pyimport click from pathlib import Path from jinja2 import Environment, FileSystemLoader import yaml TEMPLATES_DIR Path(__file__).parent / templates SKILLS_DIR Path(__file__).parent / skills click.group() def cli(): Skill 创建与管理工具。 pass cli.command() click.option(--name, promptSkill 名称英文小写加下划线如 send_email, helpSkill 的唯一标识名。) click.option(--desc, promptSkill 功能描述, help一句话描述这个 Skill 是做什么的。) click.option(--author, prompt开发者名称, defaultYour Name, helpSkill 的作者。) def new(name, desc, author): 创建一个新的 Skill 骨架。 skill_dir SKILLS_DIR / name if skill_dir.exists(): click.echo(f错误Skill 目录 {name} 已存在, errTrue) return # 收集输入输出参数简化示例实际可做成循环交互 inputs [] click.echo(现在定义 Skill 的输入参数输入空行结束) while True: param_name click.prompt(参数名, default, show_defaultFalse) if not param_name: break param_type click.prompt(参数类型 (e.g., str, int, List[str]), defaultstr) param_desc click.prompt(参数描述, default) inputs.append({name: param_name, type: param_type, description: param_desc}) # 准备模板渲染上下文 context { skill_name: name, skill_class_name: .join(word.capitalize() for word in name.split(_)) Skill, description: desc, author: author, inputs: inputs, version: 0.1.0 } # 设置 Jinja2 环境 env Environment(loaderFileSystemLoader(TEMPLATES_DIR / skill_template), trim_blocksTrue, lstrip_blocksTrue) env.filters[to_yaml] lambda x: yaml.dump(x, default_flow_styleFalse, allow_unicodeTrue) # 创建目录并渲染模板 skill_dir.mkdir(parentsTrue) (skill_dir / __init__.py).write_text() templates_to_render [ (skill.py.j2, skill_dir / skill.py), (manifest.yaml.j2, skill_dir / manifest.yaml), (test_skill.py.j2, skill_dir / test_skill.py), ] for template_name, output_path in templates_to_render: template env.get_template(template_name) content template.render(**context) output_path.write_text(content) click.echo(f已创建{output_path}) click.echo(f\n✅ Skill {name} 骨架创建成功) click.echo(f目录位于{skill_dir}) click.echo(下一步) click.echo( 1. 编辑 skill.py 实现你的核心逻辑。) click.echo( 2. 根据 manifest.yaml 检查输入输出定义。) click.echo( 3. 运行 pytest skills/{name}/ 进行测试。) if __name__ __main__: cli()对应的 Jinja2 模板示例 (skill.py.j2)# skills/{{ skill_name }}/skill.py from typing import Any, Dict, Optional from ..base import BaseSkill, SkillInput, SkillOutput, SkillContext class {{ skill_class_name }}Input(SkillInput): {{ skill_name }} 的输入参数定义。 {% for param in inputs %} {{ param.name }}: {{ param.type }} # {{ param.description }} {% endfor %} class {{ skill_class_name }}(BaseSkill): {{ description }} name {{ skill_name }} description {{ description }} version {{ version }} def __init__(self, context: Optional[SkillContext] None): super().__init__(context) # 在此初始化你的客户端、模型等资源 # self.client SomeClient() async def run(self, input_data: {{ skill_class_name }}Input) - SkillOutput: 执行 {{ skill_name }} 的核心逻辑。 请在此处实现你的功能。 try: # TODO: 实现你的业务逻辑 # 示例 result await self.client.do_something(**input_data.dict()) result {message: 功能待实现} return SkillOutput( successTrue, dataresult, message{{ skill_name }} 执行成功。 ) except Exception as e: # 务必捕获异常并返回明确的错误输出 return SkillOutput( successFalse, dataNone, messagef执行 {{ skill_name }} 时出错{str(e)} )运行python skill_creator.py new跟随提示操作一个结构清晰、符合规范的 Skill 骨架就生成了。这极大地提升了开发效率并保证了项目代码风格的一致性。3.3 清单Manifest文件Skill 的“身份证”上面模板中生成的manifest.yaml文件至关重要。它是系统了解一个 Skill 的权威来源应该包含# manifest.yaml name: send_email version: 0.1.0 description: 发送电子邮件到指定地址。 author: Your Name entry_point: skill:SendEmailSkill # 指向 Skill 类 input_schema: type: object required: - recipient - subject - body properties: recipient: type: string description: 收件人邮箱地址 subject: type: string description: 邮件主题 body: type: string description: 邮件正文 output_schema: type: object properties: message_id: type: string description: 邮件服务器返回的消息ID dependencies: - pydantic2.0 - aiosmtplib tags: - communication - notification这个清单文件可以被 Skill 加载器Loader读取用于验证在调用前校验输入参数是否符合input_schema。发现在 UI 或目录中展示所有可用的 Skill 及其功能。文档自动生成 API 文档。编排工作流引擎根据输入输出模式Schema自动连接不同的 Skill。4. Skill 的迭代、测试与部署实践4.1 版本控制与灰度发布Skill 作为独立的能力单元也应该有自己的版本生命周期。我们可以在manifest.yaml中定义版本号遵循语义化版本major.minor.patch。当 Skill 的逻辑发生不兼容的变更时升级主版本号新增功能但向下兼容时升级次版本号仅修复 bug 时升级修订号。在部署时可以利用这个版本号实现灰度发布。例如Agent 系统可以配置为某个 Skill 默认使用1.x的最新稳定版但可以将一小部分流量或特定用户路由到正在测试的2.0.0-beta.1版本。这允许我们在不影响主流用户的情况下安全地测试新功能。4.2 单元测试与集成测试策略为 Skill 编写测试是保证其质量的关键。我们的 Skill Creator 已经生成了测试骨架test_skill.py我们需要填充它。单元测试针对 Skill 内部的纯逻辑函数。Mock 掉所有外部依赖如网络请求、数据库调用。# test_send_email.py import pytest from unittest.mock import AsyncMock, patch from skills.send_email.skill import SendEmailSkill, SendEmailSkillInput pytest.mark.asyncio async def test_send_email_success(): 测试发送邮件成功场景。 skill SendEmailSkill() # Mock 掉 SMTP 客户端 with patch(skills.send_email.skill.AsyncSMTP) as mock_smtp: mock_instance AsyncMock() mock_smtp.return_value.__aenter__.return_value mock_instance mock_instance.sendmail.return_value {} input_data SendEmailSkillInput( recipienttestexample.com, subjectTest, bodyHello World ) result await skill.run(input_data) assert result.success is True assert 消息ID in result.message # 验证 mock 方法被以正确的参数调用 mock_instance.sendmail.assert_called_once()集成测试测试 Skill 与真实或测试环境中的外部服务的交互。这需要搭建一个包含所有依赖如测试用的邮件服务器、数据库的临时环境。使用docker-compose或pytest的fixture来管理这些环境生命周期是非常好的实践。4.3 性能监控与错误处理在生产环境中我们需要知道每个 Skill 的运行状况。可以在BaseSkill的run方法周围添加装饰器或使用中间件Middleware模式来统一注入监控逻辑。import time import functools from typing import Callable from prometheus_client import Counter, Histogram SKILL_EXECUTION_COUNT Counter(skill_execution_total, Total skill executions, [skill_name, status]) SKILL_EXECUTION_DURATION Histogram(skill_execution_duration_seconds, Skill execution duration, [skill_name]) def monitor_skill_execution(func: Callable): 监控 Skill 执行的装饰器。 functools.wraps(func) async def wrapper(self, input_data: SkillInput) - SkillOutput: start_time time.time() try: result await func(self, input_data) status success if result.success else failure SKILL_EXECUTION_COUNT.labels(skill_nameself.name, statusstatus).inc() return result except Exception as e: SKILL_EXECUTION_COUNT.labels(skill_nameself.name, statuserror).inc() # 重新抛出或返回统一的错误输出 return SkillOutput(successFalse, messagefUnhandled error: {str(e)}) finally: duration time.time() - start_time SKILL_EXECUTION_DURATION.labels(skill_nameself.name).observe(duration) return wrapper # 在 BaseSkill.run 方法上应用此装饰器这样我们就能在监控系统如 Grafana中看到每个 Skill 的调用次数、成功/失败率以及耗时分布便于快速定位性能瓶颈或故障 Skill。4.4 安全与权限考量Skill 可能执行敏感操作如发送邮件、访问数据库、调用付费 API。必须在设计层面加入权限控制。一种常见做法是在SkillContext中携带“用户身份”和“权限令牌”在每个 Skill 的run方法开始处进行校验。class SecureSkill(BaseSkill): required_permissions [send_email] async def run(self, input_data: SkillInput) - SkillOutput: # 从上下文中获取用户权限 user_permissions self.context.get(user_permissions, []) if not all(perm in user_permissions for perm in self.required_permissions): return SkillOutput( successFalse, message权限不足无法执行此操作。 ) # ... 执行业务逻辑更复杂的系统可能会引入基于角色的访问控制RBAC或属性基访问控制ABAC将权限检查抽象成独立的中间件在 Skill 被调用前统一拦截处理。5. 从理论到实践构建一个天气预报 Skill 的全流程让我们将上述所有概念串联起来实际构建一个“天气预报” Skill体验从创建到部署的完整流程。5.1 使用 Skill Creator 生成骨架运行我们之前写的 CLI 工具$ python skill_creator.py new Skill 名称英文小写加下划线如 send_email: get_weather Skill 功能描述: 根据城市名称查询实时天气和未来几天的预报。 开发者名称 [Your Name]: WeatherMaster 现在定义 Skill 的输入参数输入空行结束 参数名: city 参数类型 (e.g., str, int, List[str]) [str]: str 参数描述: 要查询天气的城市名称如“北京” 参数名: days 参数类型 (e.g., str, int, List[str]) [str]: int 参数描述: 预报的天数默认为1最多7天 参数名: ✅ Skill get_weather 骨架创建成功5.2 实现核心业务逻辑打开生成的skills/get_weather/skill.py文件填充run方法。我们假设使用一个免费的天气 API如 OpenWeatherMap。import aiohttp import os from typing import Any, Dict from ..base import BaseSkill, SkillInput, SkillOutput, SkillContext class GetWeatherSkillInput(SkillInput): city: str # 要查询天气的城市名称如“北京” days: int 1 # 预报的天数默认为1最多7天 class GetWeatherSkill(BaseSkill): name get_weather description 根据城市名称查询实时天气和未来几天的预报。 version 0.1.0 def __init__(self, context: SkillContext None): super().__init__(context) self.api_key os.getenv(WEATHER_API_KEY) # 从环境变量读取API密钥 if not self.api_key: raise ValueError(未设置 WEATHER_API_KEY 环境变量) self.base_url https://api.openweathermap.org/data/2.5 async def run(self, input_data: GetWeatherSkillInput) - SkillOutput: try: if input_data.days 1 or input_data.days 7: return SkillOutput(successFalse, message预报天数需在1-7天之间。) # 1. 获取实时天气 current_weather await self._fetch_current_weather(input_data.city) # 2. 获取天气预报 forecast await self._fetch_forecast(input_data.city, input_data.days) result { city: input_data.city, current: current_weather, forecast: forecast } return SkillOutput(successTrue, dataresult, messagef成功获取 {input_data.city} 的天气信息。) except aiohttp.ClientError as e: return SkillOutput(successFalse, messagef网络请求失败{str(e)}) except Exception as e: return SkillOutput(successFalse, messagef处理天气数据时出错{str(e)}) async def _fetch_current_weather(self, city: str) - Dict[str, Any]: async with aiohttp.ClientSession() as session: url f{self.base_url}/weather params {q: city, appid: self.api_key, units: metric, lang: zh_cn} async with session.get(url, paramsparams) as resp: resp.raise_for_status() data await resp.json() # 提取并格式化需要的信息 return { temp: data[main][temp], feels_like: data[main][feels_like], humidity: data[main][humidity], description: data[weather][0][description], icon: data[weather][0][icon] } async def _fetch_forecast(self, city: str, days: int) - list: # 类似地调用预报API这里省略具体实现 # 通常是调用 /forecast/daily 或处理 /forecast 的每3小时数据 pass5.3 编写与运行测试填充生成的测试文件并运行$ cd /path/to/your/project $ WEATHER_API_KEYtest_key pytest skills/get_weather/ -v确保所有测试通过特别是要覆盖网络错误、API 返回异常数据等边界情况。5.4 更新清单与注册检查并完善自动生成的manifest.yaml确保input_schema与代码中的GetWeatherSkillInput类定义一致。然后我们需要将这个 Skill 注册到系统中。如果使用动态发现只需确保skills/get_weather目录在扫描路径下即可。如果使用集中注册需要在 Agent 初始化代码中显式添加from skills.get_weather.skill import GetWeatherSkill registry SkillRegistry() registry.register(GetWeatherSkill())5.5 集成到 Agent 并验证最后在你的 Agent 主逻辑中就可以调用这个 Skill 了async def handle_user_query(query: str): # ... 假设通过某种方式如LLM解析出意图和参数 intent get_weather params {city: 北京, days: 2} if intent get_weather: skill registry.get(get_weather) if skill: input_data GetWeatherSkillInput(**params) result await skill.run(input_data) if result.success: # 将 result.data 格式化成对用户友好的回复 return format_weather_response(result.data) else: return f抱歉查询天气时出错{result.message} # ...至此一个具备完整生命周期、可测试、可监控、可复用的天气预报 Skill 就构建并集成完毕了。通过 Skill Creator我们标准化了创建流程通过清晰的架构和清单文件我们实现了能力的模块化和可管理性。这套体系不仅能用于聊天机器人同样适用于自动化流程、智能助手、后台任务调度等任何需要组合多种能力的复杂系统。