16.3 Skills 生命週期管理

生命週期概述

Skills 的生命週期管理涵蓋了從建立到刪除的整個過程。本節將詳細介紹 Skills 的各個生命週期階段及其管理方法。

生命周期阶段

1. 创建阶段

1.1 Skill 创建流程

Skill 创建流程

步骤 1：需求分析

• 明確 Skill 的目標
• 確定輸入輸出
• 識別關鍵功能

步驟 2：設計 Skill

• 設計介面
• 規劃流程
• 選擇實現方式

步驟 3：實現 Skill

• 編寫程式碼
• 新增測試
• 編寫文件

步驟 4：註冊 Skill

• 新增到 Skill 庫
• 配置後設資料
• 設定許可權

步驟 5：測試驗證

• 功能測試
• 效能測試
• 使用者測試

#### 1.2 创建 API

    python


    class SkillManager:
        def __init__(self):
            self.skills = {}
            self.skill_registry = SkillRegistry()

        def create_skill(self, skill_definition):
            self.validate_definition(skill_definition)
            skill = self.instantiate_skill(skill_definition)
            skill_id = self.skill_registry.register(skill)
            self.save_skill(skill_id, skill)
            return skill_id

        def validate_definition(self, definition):
            required_fields = ["name", "version", "description"]
            for field in required_fields:
                if field not in definition:
                    raise ValueError(f"Missing required field: {field}")

        def instantiate_skill(self, definition):
            skill_type = definition.get("type", "natural_language")
            if skill_type == "natural_language":
                return NaturalLanguageSkill(definition)
            elif skill_type == "example_driven":
                return ExampleDrivenSkill(definition)
            elif skill_type == "programmatic":
                return ProgrammaticSkill(definition)
            else:
                raise ValueError(f"Unknown skill type: {skill_type}")

        def save_skill(self, skill_id, skill):
            skill_path = self.get_skill_path(skill_id)
            with open(skill_path, 'w') as f:
                json.dump(skill.to_dict(), f, indent=2)

### 2\. 激活阶段

#### 2.1 Skill 激活流程

## Skill 激活流程

### 步骤 1：加载 Skill

• 讀取 Skill 定義
• 解析配置
• 初始化 Skill

步驟 2：驗證依賴

• 檢查依賴項
• 驗證版本相容性
• 載入依賴

步驟 3：初始化 Skill

• 執行初始化邏輯
• 設定上下文
• 註冊事件監聽器

步驟 4：啟用 Skill

• 標記為活躍狀態
• 啟用功能
• 開始監聽事件

#### 2.2 激活 API

    python


    class SkillManager:
        def activate_skill(self, skill_id):
            skill = self.load_skill(skill_id)
            self.verify_dependencies(skill)
            skill.initialize()
            skill.activate()
            self.update_skill_status(skill_id, "active")
            return skill

        def load_skill(self, skill_id):
            skill_path = self.get_skill_path(skill_id)
            with open(skill_path, 'r') as f:
                skill_data = json.load(f)
            skill_type = skill_data["type"]
            if skill_type == "natural_language":
                skill = NaturalLanguageSkill(skill_data)
            elif skill_type == "example_driven":
                skill = ExampleDrivenSkill(skill_data)
            elif skill_type == "programmatic":
                skill = ProgrammaticSkill(skill_data)
            else:
                raise ValueError(f"Unknown skill type: {skill_type}")
            return skill

        def verify_dependencies(self, skill):
            dependencies = skill.get_dependencies()
            for dep in dependencies:
                if not self.is_dependency_available(dep):
                    raise DependencyError(f"Dependency not available: {dep}")

        def is_dependency_available(self, dependency):
            pass

        def update_skill_status(self, skill_id, status):
            pass

    ### 3. 运行阶段

    #### 3.1 Skill 执行流程

    ## Skill 执行流程
    ### 步骤 1：接收请求
    - 接收用户请求
    - 解析参数
    - 验证输入
    ### 步骤 2：准备执行
    - 收集上下文
    - 准备资源
    - 初始化状态
    ### 步骤 3：执行逻辑
    - 执行 Skill 逻辑
    - 调用工具
    - 处理结果
    ### 步骤 4：返回结果
    - 格式化输出
    - 返回结果
    - 清理资源

    #### 3.2 执行 API

    ```python

    class SkillExecutor:
        def __init__(self):
            self.context_manager = ContextManager()
            self.resource_manager = ResourceManager()

```python
        def execute(self, skill, parameters):
            self.validate_parameters(skill, parameters)
            context = self.prepare_context(skill, parameters)
            resources = self.acquire_resources(skill, parameters)

            try:
                result = skill.execute(parameters, context)
                formatted_result = self.format_result(result, skill)
                return formatted_result
            finally:
                self.release_resources(resources)

        def validate_parameters(self, skill, parameters):
            schema = skill.get_parameter_schema()
            for param in schema.get("required", []):
                if param not in parameters:
                    raise ValueError(f"Missing required parameter: {param}")

        def prepare_context(self, skill, parameters):
            return self.context_manager.collect_context(skill, parameters)

        def acquire_resources(self, skill, parameters):
            return self.resource_manager.acquire(skill.get_resource_requirements())

        def format_result(self, result, skill):
            output_format = skill.get_output_format()
            if output_format == "json":
                return json.dumps(result)
            elif output_format == "markdown":
                return self.to_markdown(result)
            else:
                return result

        def release_resources(self, resources):
            self.resource_manager.release(resources)

    ### 4. 更新阶段

    #### 4.1 Skill 更新流程

    ## Skill 更新流程
    ### 步骤 1：检查更新
    - 检查新版本
    - 验证兼容性
    - 备份当前版本
    ### 步骤 2：下载更新
    - 下载新版本
    - 验证完整性
    - 解压更新
    ### 步骤 3：应用更新
    - 停用 Skill
    - 替换文件
    - 迁移配置
    ### 步骤 4：重新激活
    - 验证更新
    - 重新激活
    - 测试功能

    #### 4.2 更新 API

    ```python

    class SkillUpdater:
        def __init__(self):
            self.backup_manager = BackupManager()
            self.config_migrator = ConfigMigrator()

        def update_skill(self, skill_id, new_version):
            current_version = self.get_current_version(skill_id)
            if not self.should_update(current_version, new_version):
                return False

            backup = self.backup_manager.create_backup(skill_id)

            try:
                update_data = self.download_update(skill_id, new_version)
                self.validate_update(update_data)
                self.deactivate_skill(skill_id)
                self.apply_update(skill_id, update_data)
                self.migrate_config(skill_id, current_version, new_version)
                self.activate_skill(skill_id)
                self.verify_update(skill_id)
                return True
            except Exception as e:
                self.rollback_update(skill_id, backup)
                raise e

        def should_update(self, current_version, new_version):
            return self.compare_versions(current_version, new_version) < 0

        def compare_versions(self, v1, v2):
            v1_parts = [int(x) for x in v1.split(".")]
            v2_parts = [int(x) for x in v2.split(".")]

            for a, b in zip(v1_parts, v2_parts):
                if a < b:
                    return -1
                elif a > b:
                    return 1

            return 0

        def download_update(self, skill_id, version):

            # 下載更新

            pass

        def validate_update(self, update_data):

            # 驗證更新

            pass

        def apply_update(self, skill_id, update_data):

            # 應用更新

            pass

        def migrate_config(self, skill_id, from_version, to_version):

            # 遷移配置

            self.config_migrator.migrate(skill_id, from_version, to_version)

        def verify_update(self, skill_id):

            # 驗證更新

            pass

        def rollback_update(self, skill_id, backup):

            # 回滾更新

            self.backup_manager.restore_backup(skill_id, backup)

    ### 5. 停用階段

    #### 5.1 Skill 停用流程

    ## Skill 停用流程

    ### 步驟 1：停止執行


    - 停止新的請求
    - 完成正在執行的請求
    - 等待所有操作完成

    ### 步驟 2：清理資源


    - 釋放資源
    - 關閉連線
    - 清理快取

    ### 步驟 3：儲存狀態


    - 儲存當前狀態
    - 儲存執行歷史
    - 儲存配置

    ### 步驟 4：標記為停用


    - 更新狀態
    - 禁用功能
    - 停止監聽事件

    #### 5.2 停用 API

    ```python

    class SkillManager:
        def deactivate_skill(self, skill_id):
            self.stop_execution(skill_id)

            self.wait_for_completion(skill_id)

            self.cleanup_resources(skill_id)

            self.save_state(skill_id)

            self.update_skill_status(skill_id, "inactive")

        def stop_execution(self, skill_id):
            skill = self.get_skill(skill_id)
            skill.stop()

        def wait_for_completion(self, skill_id):
            skill = self.get_skill(skill_id)
            skill.wait_for_completion()

        def cleanup_resources(self, skill_id):
            skill = self.get_skill(skill_id)
            resources = skill.get_resources()
            for resource in resources:
                resource.release()

        def save_state(self, skill_id):
            skill = self.get_skill(skill_id)
            state = skill.get_state()
            state_path = self.get_state_path(skill_id)
            with open(state_path, 'w') as f:
                json.dump(state, f)

        def update_skill_status(self, skill_id, status):
            # 更新状态
            pass

    ### 6. 删除阶段

    #### 6.1 Skill 删除流程

    ## Skill 删除流程
    ### 步骤 1：确认删除
    - 显示删除警告
    - 确认删除操作
    - 检查依赖关系
    ### 步骤 2：停用 Skill
    - 停止执行
    - 清理资源
    - 保存状态
    ### 步骤 3：删除文件
    - 删除 Skill 文件
    - 删除配置文件
    - 删除状态文件
    ### 步骤 4：清理注册
    - 从注册表移除
    - 清理引用
    - 更新索引

    #### 6.2 删除 API

    ```python

    class SkillManager:
        def delete_skill(self, skill_id, force=False):

            # 確認刪除

            if not force and not self.confirm_deletion(skill_id):
                return False

            # 檢查依賴

            dependents = self.get_dependents(skill_id)
            if dependents and not force:
                raise DependencyError(f"Skill has dependents: {dependents}")

            # 停用 Skill

            self.deactivate_skill(skill_id)

            # 刪除檔案

            self.delete_skill_files(skill_id)

            # 清理註冊

            self.unregister_skill(skill_id)

            return True

        def confirm_deletion(self, skill_id):
            skill = self.get_skill(skill_id)
            print(f"Are you sure you want to delete skill '{skill.name}'?")
            response = input("Type 'yes' to confirm: ")
            return response.lower() == "yes"

        def get_dependents(self, skill_id):

            # 獲取依賴此 Skill 的其他 Skills

            pass

        def delete_skill_files(self, skill_id):
            skill_path = self.get_skill_path(skill_id)
            config_path = self.get_config_path(skill_id)
            state_path = self.get_state_path(skill_id)

            for path in [skill_path, config_path, state_path]:
                if os.path.exists(path):
                    os.remove(path)

        def unregister_skill(self, skill_id):
            self.skill_registry.unregister(skill_id)

    ## 狀態管理

    ### 1. 狀態定義

    ## Skill 狀態

    ### 狀態型別


    - CREATED: 已建立
    - ACTIVATING: 啟用中
    - ACTIVE: 活躍
    - DEACTIVATING: 停用中
    - INACTIVE: 停用
    - UPDATING: 更新中
    - ERROR: 錯誤
    - DELETED: 已刪除

    ### 狀態轉換

    CREATED → ACTIVATING → ACTIVE
    ACTIVE → DEACTIVATING → INACTIVE
    ACTIVE → UPDATING → ACTIVE
    ACTIVE → ERROR → INACTIVE
    INACTIVE → DELETED

### 2\. 狀態轉換

    python

class SkillStateMachine:
    def __init__(self):
        self.transitions = {
            "CREATED": ["ACTIVATING", "DELETED"],
            "ACTIVATING": ["ACTIVE", "ERROR"],
            "ACTIVE": ["DEACTIVATING", "UPDATING", "ERROR"],
            "DEACTIVATING": ["INACTIVE", "ERROR"],
            "INACTIVE": ["ACTIVATING", "DELETED"],
            "UPDATING": ["ACTIVE", "ERROR"],
            "ERROR": ["INACTIVE", "DELETED"],
            "DELETED": []
        }

    def can_transition(self, from_state, to_state):
        return to_state in self.transitions.get(from_state, [])

    def transition(self, skill_id, new_state):
        current_state = self.get_current_state(skill_id)

        if not self.can_transition(current_state, new_state):
            raise InvalidStateTransitionError(
                f"Cannot transition from {current_state} to {new_state}"
            )

        self.set_state(skill_id, new_state)

    def get_current_state(self, skill_id):
        # 获取当前状态
        pass

    def set_state(self, skill_id, state):
        # 设置状态
        pass

## 版本管理

### 1. 版本控制

```python

class SkillVersionManager:
    def __init__(self):
        self.versions = {}

        def create_version(self, skill_id, version_number, changes):
            version = {
                "id": self.generate_version_id(),
                "skill_id": skill_id,
                "version": version_number,
                "created_at": datetime.now(),
                "changes": changes,
                "checksum": self.calculate_checksum(skill_id)
            }
            if skill_id not in self.versions:
                self.versions[skill_id] = []
            self.versions[skill_id].append(version)
            return version["id"]

        def get_version(self, skill_id, version_number):
            versions = self.versions.get(skill_id, [])
            for version in versions:
                if version["version"] == version_number:
                    return version
            return None

        def get_latest_version(self, skill_id):
            versions = self.versions.get(skill_id, [])
            if not versions:
                return None
            return versions[-1]

        def list_versions(self, skill_id):
            return self.versions.get(skill_id, [])

        def rollback_to_version(self, skill_id, version_number):
            version = self.get_version(skill_id, version_number)
            if not version:
                raise VersionNotFoundError(version_number)
            self.restore_version(skill_id, version)

        def generate_version_id(self):
            return str(uuid.uuid4())

        def calculate_checksum(self, skill_id):
            pass

        def restore_version(self, skill_id, version):
            pass

### 2\. 变更日志

    python

    class ChangeLog:
        def __init__(self):
            self.entries = []

        def add_entry(self, skill_id, version, changes):
            entry = {
                "id": self.generate_entry_id(),
                "skill_id": skill_id,
                "version": version,
                "timestamp": datetime.now(),
                "changes": changes
            }
            self.entries.append(entry)
            return entry["id"]

        def get_entries(self, skill_id):
            return [e for e in self.entries if e["skill_id"] == skill_id]

        def get_entry(self, entry_id):
            for entry in self.entries:
                if entry["id"] == entry_id:
                    return entry
            return None

        def generate_changelog(self, skill_id, from_version, to_version):
            entries = self.get_entries(skill_id)
            changelog = []

            for entry in entries:
                version = entry["version"]
                if self.compare_versions(from_version, version) <= 0 and \
                   self.compare_versions(version, to_version) <= 0:
                    changelog.append(entry)

            return changelog

        def generate_entry_id(self):
            return str(uuid.uuid4())

        def compare_versions(self, v1, v2):
            v1_parts = [int(x) for x in v1.split(".")]
            v2_parts = [int(x) for x in v2.split(".")]

            for a, b in zip(v1_parts, v2_parts):
                if a < b:
                    return -1
                elif a > b:
                    return 1

            return 0

    ## 依赖管理

    ### 1. 依赖解析

    ```python

    class DependencyResolver:
        def __init__(self):
            self.dependency_graph = {}

        def add_dependency(self, skill_id, depends_on):
            if skill_id not in self.dependency_graph:
                self.dependency_graph[skill_id] = []
            if depends_on not in self.dependency_graph[skill_id]:
                self.dependency_graph[skill_id].append(depends_on)

        def resolve_dependencies(self, skill_id):
            visited = set()
            result = []

            def visit(node):
                if node in visited:
                    return
                visited.add(node)
                for dep in self.dependency_graph.get(node, []):
                    visit(dep)
                result.append(node)

            visit(skill_id)
            return result

        def check_circular_dependencies(self):
            visited = set()
            rec_stack = set()

            def dfs(node):
                visited.add(node)
                rec_stack.add(node)
                for dep in self.dependency_graph.get(node, []):
                    if dep not in visited:
                        if dfs(dep):
                            return True
                    elif dep in rec_stack:
                        return True
                rec_stack.remove(node)
                return False

            for node in self.dependency_graph:
                if node not in visited:
                    if dfs(node):
                        return True
            return False

### 2\. 依賴安裝

    python

class DependencyInstaller:
    def __init__(self):
        self.resolver = DependencyResolver()

    def install_dependencies(self, skill_id):
        # 解析依赖
        dependencies = self.resolver.resolve_dependencies(skill_id)

        # 安装依赖
        for dep_id in dependencies:
            if dep_id != skill_id:
                self.install_skill(dep_id)

    def install_skill(self, skill_id):
        # 安装 Skill
        pass

    def uninstall_dependencies(self, skill_id):
        # 查找依赖此 Skill 的其他 Skills
        dependents = self.get_dependents(skill_id)

        # 如果没有其他依赖，可以卸载
        if not dependents:
            self.uninstall_skill(skill_id)

    def get_dependents(self, skill_id):
        # 获取依赖此 Skill 的其他 Skills
        pass

总结

Skills 的生命週期管理是一個複雜而重要的過程，涉及建立、啟用、執行、更新、停用和刪除等多個階段。透過有效的生命週期管理，可以確保 Skills 的穩定執行和持續改進。

在下一節中，我們將探討 Skills 的許可權控制，瞭解如何管理和限制 Skills 的訪問許可權。