# 03.3 技能组合与复用 ## 核心问题 > 如何将复杂功能分解为可复用的技能单元？ > 如何设计技能接口以支持动态组合？ > 如何发现和加载新技能而不修改核心代码？ --- ## 概念讲解 ### 技能抽象的概念 **技能 (Skill)** 是Agent可执行的**独立功能单元**，具有明确的输入输出接口： ``` 技能 = 功能定义 + 接口契约 + 元数据 ┌─────────────────────────────────────────────────────────────┐ │ 技能的基本结构 │ ├─────────────────────────────────────────────────────────────┤ │ │ │ ┌─────────────────────────────────────────────────────┐ │ │ │ Skill Interface │ │ │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │ │ │ 名称 │ │ 描述 │ │ 参数 │ │ │ │ │ │ name │ │ description │ │ parameters │ │ │ │ │ └─────────────┘ └─────────────┘ └─────────────┘ │ │ │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │ │ │ 输入 │ │ 输出 │ │ 副作用 │ │ │ │ │ │ input │ │ output │ │ side_effects│ │ │ │ │ └─────────────┘ └─────────────┘ └─────────────┘ │ │ │ └─────────────────────────────────────────────────────┘ │ │ │ │ │ ↓ │ │ ┌─────────────────────────────────────────────────────┐ │ │ │ Implementation │ │ │ │ │ │ │ def execute(self, **kwargs) -> SkillResult: │ │ │ # 技能实现 │ │ │ pass │ │ │ └─────────────────────────────────────────────────────┘ │ │ │ └─────────────────────────────────────────────────────────────┘ ``` ### 技能组合模式 ``` 技能组合模式层次 1. 顺序组合 (Sequential) │ ├──→ 技能A输出 → 技能B输入 → 技能C输入 │ 适用：数据流水线处理 │ 2. 并行组合 (Parallel) │ ├──→ 技能A ──┐ │ 技能B ──┼──→ 合并结果 │ 技能C ──┘ │ 适用：独立任务并行执行 │ 3. 条件组合 (Conditional) │ ├──→ 条件判断 → 技能A或技能B │ 适用：分支处理逻辑 │ 4. 迭代组合 (Iterative) │ └──→ 技能A输出 → [循环] → 技能B 适用：增量式处理 ``` --- ## 设计原理 ### 技能接口设计 良好的技能接口是实现组合的基础： ```python from abc import ABC, abstractmethod from typing import Dict, Any, List, Optional, Type, Callable from dataclasses import dataclass, field from enum import Enum class SkillCategory(Enum): """技能分类""" DATA_PROCESSING = "data_processing" SPATIAL_ANALYSIS = "spatial_analysis" VISUALIZATION = "visualization" FILE_OPERATIONS = "file_operations" MODEL_EXECUTION = "model_execution" @dataclass class ParameterSpec: """参数规格""" name: str type: Type description: str required: bool = True default: Any = None constraints: Dict[str, Any] = field(default_factory=dict) @dataclass class SkillResult: """技能执行结果""" success: bool data: Any = None error: Optional[str] = None metadata: Dict[str, Any] = field(default_factory=dict) class Skill(ABC): """技能基类""" # 技能元数据 name: str = "" description: str = "" category: SkillCategory = SkillCategory.DATA_PROCESSING version: str = "1.0.0" parameters: List[ParameterSpec] = field(default_factory=list) @abstractmethod def execute(self, **kwargs) -> SkillResult: """执行技能""" pass def validate_input(self, **kwargs) -> tuple[bool, Optional[str]]: """验证输入参数""" for param in self.parameters: if param.required and param.name not in kwargs: return False, f"Missing required parameter: {param.name}" if param.name in kwargs: value = kwargs[param.name] if not isinstance(value, param.type): try: kwargs[param.name] = param.type(value) except (ValueError, TypeError): return False, f"Invalid type for {param.name}" return True, None def get_spec(self) -> Dict[str, Any]: """获取技能规格""" return { 'name': self.name, 'description': self.description, 'category': self.category.value, 'version': self.version, 'parameters': [ { 'name': p.name, 'type': p.type.__name__, 'description': p.description, 'required': p.required, 'default': p.default } for p in self.parameters ] } ``` ### 技能组合器 ```python class SkillComposer: """ 技能组合器：将多个技能组合成复合技能 """ def __init__(self): self.skills: Dict[str, Skill] = {} def register(self, skill: Skill) -> 'SkillComposer': """注册技能""" self.skills[skill.name] = skill return self def sequential(self, *skill_names: str) -> 'CompositeSkill': """ 顺序组合：前一个技能的输出传递给下一个 数据流: input → skill1 → skill2 → skill3 → output """ skills = [self.skills[name] for name in skill_names] return CompositeSkill( name=f"sequential_{'_'.join(skill_names)}", skills=skills, mode='sequential' ) def parallel(self, *skill_names: str, merge_func: Callable = None) -> 'CompositeSkill': """ 并行组合：所有技能并行执行，结果合并 数据流: input ──→ skill1 ──┐ ─→ skill2 ──┼──→ merge → output ─→ skill3 ──┘ """ skills = [self.skills[name] for name in skill_names] return CompositeSkill( name=f"parallel_{'_'.join(skill_names)}", skills=skills, mode='parallel', merge_func=merge_func ) def conditional(self, condition: Callable, true_skill: str, false_skill: str = None) -> 'CompositeSkill': """ 条件组合：根据条件选择技能执行 数据流: input → condition → skill_if_true / skill_if_false → output """ skills = [self.skills[true_skill]] if false_skill: skills.append(self.skills[false_skill]) return CompositeSkill( name=f"conditional_{true_skill}_{false_skill}", skills=skills, mode='conditional', condition=condition ) def loop(self, skill_name: str, until: Callable = None, max_iterations: int = 10) -> 'CompositeSkill': """ 迭代组合：循环执行技能直到满足条件 数据流: input → [skill → check] → output """ skill = self.skills[skill_name] return CompositeSkill( name=f"loop_{skill_name}", skills=[skill], mode='loop', until_condition=until, max_iterations=max_iterations ) class CompositeSkill(Skill): """复合技能：由多个子技能组合而成""" def __init__(self, name: str, skills: List[Skill], mode: str, **kwargs): self.name = name self.skills = skills self.mode = mode # sequential, parallel, conditional, loop self.condition = kwargs.get('condition') self.merge_func = kwargs.get('merge_func') self.until_condition = kwargs.get('until_condition') self.max_iterations = kwargs.get('max_iterations', 10) def execute(self, initial_input: Any = None, **kwargs) -> SkillResult: """执行复合技能""" if self.mode == 'sequential': return self._execute_sequential(initial_input, **kwargs) elif self.mode == 'parallel': return self._execute_parallel(initial_input, **kwargs) elif self.mode == 'conditional': return self._execute_conditional(initial_input, **kwargs) elif self.mode == 'loop': return self._execute_loop(initial_input, **kwargs) else: return SkillResult(success=False, error=f"Unknown mode: {self.mode}") def _execute_sequential(self, initial_input, **kwargs): """顺序执行""" current_input = initial_input results = [] for skill in self.skills: if isinstance(current_input, dict): result = skill.execute(**current_input) else: result = skill.execute(input=current_input) if not result.success: return SkillResult( success=False, error=f"Skill {skill.name} failed: {result.error}", metadata={'failed_at': skill.name} ) results.append(result) current_input = result.data return SkillResult( success=True, data=current_input, metadata={'sub_results': results} ) def _execute_parallel(self, initial_input, **kwargs): """并行执行""" import concurrent.futures results = [] with concurrent.futures.ThreadPoolExecutor() as executor: futures = { executor.submit(skill.execute, input=initial_input): skill for skill in self.skills } for future in concurrent.futures.as_completed(futures): skill = futures[future] try: result = future.result() results.append(result) except Exception as e: results.append(SkillResult( success=False, error=str(e), metadata={'skill': skill.name} )) # 合并结果 if self.merge_func: merged_data = self.merge_func(results) else: # 默认合并：收集所有成功的数据 merged_data = [r.data for r in results if r.success] return SkillResult( success=all(r.success for r in results), data=merged_data, metadata={'sub_results': results} ) def _execute_conditional(self, initial_input, **kwargs): """条件执行""" if self.condition and self.condition(initial_input): result = self.skills[0].execute(input=initial_input) elif len(self.skills) > 1: result = self.skills[1].execute(input=initial_input) else: result = SkillResult(success=True, data=initial_input) return result def _execute_loop(self, initial_input, **kwargs): """循环执行""" current_input = initial_input results = [] for i in range(self.max_iterations): result = self.skills[0].execute(input=current_input) if not result.success: return SkillResult( success=False, error=f"Iteration {i} failed: {result.error}" ) results.append(result) current_input = result.data # 检查终止条件 if self.until_condition and self.until_condition(result.data): break return SkillResult( success=True, data=current_input, metadata={'iterations': len(results), 'sub_results': results} ) ``` ### 动态技能发现与加载 ```python import importlib import importlib.util import inspect from pathlib import Path class SkillRegistry: """ 技能注册表：管理技能的发现、加载和注册 """ def __init__(self): self._skills: Dict[str, Type[Skill]] = {} self._categories: Dict[SkillCategory, List[str]] = { category: [] for category in SkillCategory } def register_class(self, skill_class: Type[Skill]) -> None: """注册技能类""" if not issubclass(skill_class, Skill): raise TypeError(f"{skill_class} must be a subclass of Skill") # 创建实例获取元数据 instance = skill_class() self._skills[instance.name] = skill_class self._categories[instance.category].append(instance.name) def get_skill(self, name: str) -> Optional[Skill]: """获取技能实例""" if name in self._skills: return self._skills[name]() return None def list_skills(self, category: SkillCategory = None) -> List[str]: """列出技能""" if category: return self._categories.get(category, []) return list(self._skills.keys()) def discover_from_directory(self, directory: Path) -> int: """ 从目录发现并加载技能 约定：技能文件以 _skill.py 结尾，包含继承自 Skill 的类 """ count = 0 for file_path in directory.rglob("*_skill.py"): try: # 动态导入模块 module_name = file_path.stem spec = importlib.util.spec_from_file_location(module_name, file_path) if spec and spec.loader: module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) # 查找 Skill 子类 for name, obj in inspect.getmembers(module, inspect.isclass): if (issubclass(obj, Skill) and obj != Skill and not obj.__module__.startswith('_')): self.register_class(obj) count += 1 except Exception as e: print(f"Failed to load {file_path}: {e}") return count def get_skill_spec(self, name: str) -> Optional[Dict]: """获取技能规格""" skill = self.get_skill(name) if skill: return skill.get_spec() return None # 全局技能注册表 registry = SkillRegistry() def register_skill(skill_class: Type[Skill]) -> Type[Skill]: """技能注册装饰器""" registry.register_class(skill_class) return skill_class ``` --- ## 代码示例 ### QGIS技能集成管理器 ```python """ QGIS技能集成管理器 演示如何为空间分析工具创建可组合的技能系统 """ import json from typing import Dict, Any, List, Optional from pathlib import Path # ==================== 基础空间分析技能 ==================== @register_skill class LoadVectorLayerSkill(Skill): """加载矢量图层的技能""" name = "load_vector_layer" description = "从文件加载矢量图层" category = SkillCategory.FILE_OPERATIONS version = "1.0.0" parameters = [ ParameterSpec("path", str, "文件路径", required=True), ParameterSpec("layer_name", str, "图层名称", required=False, default="layer"), ] def execute(self, **kwargs) -> SkillResult: valid, error = self.validate_input(**kwargs) if not valid: return SkillResult(success=False, error=error) path = kwargs['path'] layer_name = kwargs.get('layer_name', 'layer') # 模拟加载（实际会调用QGIS API） return SkillResult( success=True, data={ 'layer': layer_name, 'path': path, 'type': 'vector', 'feature_count': 1250, 'crs': 'EPSG:4326' }, metadata={'loaded_at': '2024-01-01T00:00:00'} ) @register_skill class BufferAnalysisSkill(Skill): """缓冲区分析技能""" name = "buffer_analysis" description = "对几何图形创建缓冲区" category = SkillCategory.SPATIAL_ANALYSIS version = "1.0.0" parameters = [ ParameterSpec("layer", str, "输入图层", required=True), ParameterSpec("distance", float, "缓冲距离", required=True), ParameterSpec("segments", int, "分段数", required=False, default=8), ] def execute(self, **kwargs) -> SkillResult: valid, error = self.validate_input(**kwargs) if not valid: return SkillResult(success=False, error=error) layer = kwargs['layer'] distance = kwargs['distance'] # 模拟缓冲区分析 return SkillResult( success=True, data={ 'output_layer': f"{layer}_buffer_{distance}m", 'input_layer': layer, 'distance': distance, 'area_ha': 542.3 } ) @register_skill class CalculateAreaSkill(Skill): """计算面积技能""" name = "calculate_area" description = "计算要素面积" category = SkillCategory.SPATIAL_ANALYSIS version = "1.0.0" parameters = [ ParameterSpec("layer", str, "输入图层", required=True), ParameterSpec("unit", str, "单位", required=False, default="ha"), ] def execute(self, **kwargs) -> SkillResult: valid, error = self.validate_input(**kwargs) if not valid: return SkillResult(success=False, error=error) return SkillResult( success=True, data={ 'layer': kwargs['layer'], 'unit': kwargs.get('unit', 'ha'), 'total_area': 1250.5, 'mean_area': 0.42, 'areas': [1.2, 0.8, 1.5, 0.3, 2.1] } ) @register_skill class ExportToGeoJSONSkill(Skill): """导出GeoJSON技能""" name = "export_geojson" description = "导出图层为GeoJSON格式" category = SkillCategory.FILE_OPERATIONS version = "1.0.0" parameters = [ ParameterSpec("layer", str, "输入图层", required=True), ParameterSpec("output_path", str, "输出路径", required=True), ] def execute(self, **kwargs) -> SkillResult: valid, error = self.validate_input(**kwargs) if not valid: return SkillResult(success=False, error=error) # 模拟导出 return SkillResult( success=True, data={ 'output_path': kwargs['output_path'], 'layer': kwargs['layer'], 'format': 'GeoJSON', 'size_kb': 245 } ) @register_skill class CreateHeatmapSkill(Skill): """创建热力图技能""" name = "create_heatmap" description = "创建点数据热力图" category = SkillCategory.VISUALIZATION version = "1.0.0" parameters = [ ParameterSpec("layer", str, "输入图层", required=True), ParameterSpec("radius", int, "影响半径", required=False, default=100), ParameterSpec("color_ramp", str, "颜色渐变", required=False, default="hot"), ] def execute(self, **kwargs) -> SkillResult: valid, error = self.validate_input(**kwargs) if not valid: return SkillResult(success=False, error=error) return SkillResult( success=True, data={ 'output_layer': f"{kwargs['layer']}_heatmap", 'radius': kwargs.get('radius', 100), 'color_ramp': kwargs.get('color_ramp', 'hot'), 'render_time_ms': 234 } ) @register_skill class CalculateStatisticsSkill(Skill): """统计计算技能""" name = "calculate_statistics" description = "计算字段统计值" category = SkillCategory.DATA_PROCESSING version = "1.0.0" parameters = [ ParameterSpec("layer", str, "输入图层", required=True), ParameterSpec("field", str, "统计字段", required=True), ] def execute(self, **kwargs) -> SkillResult: valid, error = self.validate_input(**kwargs) if not valid: return SkillResult(success=False, error=error) # 模拟统计计算 import random values = [random.uniform(0, 100) for _ in range(100)] return SkillResult( success=True, data={ 'field': kwargs['field'], 'count': len(values), 'mean': sum(values) / len(values), 'min': min(values), 'max': max(values), 'std': (sum((x - sum(values)/len(values))**2 for x in values) / len(values)) ** 0.5 } ) # ==================== 技能管理器 ==================== class QGISSkillManager: """QGIS技能管理器""" def __init__(self): self.composer = SkillComposer() self.registry = registry # 注册基础技能 for skill_name in self.registry.list_skills(): skill = self.registry.get_skill(skill_name) if skill: self.composer.register(skill) def list_available_skills(self) -> Dict[str, List[str]]: """列出所有可用技能""" skills_by_category = {} for category in SkillCategory: skills_by_category[category.value] = self.registry.list_skills(category) return skills_by_category def get_skill_info(self, skill_name: str) -> Optional[Dict]: """获取技能详细信息""" return self.registry.get_skill_spec(skill_name) def create_workflow(self, workflow_name: str, skill_names: List[str]) -> CompositeSkill: """创建工作流（顺序组合技能）""" return self.composer.sequential(*skill_names) def execute_skill(self, skill_name: str, **kwargs) -> SkillResult: """执行单个技能""" skill = self.registry.get_skill(skill_name) if skill: return skill.execute(**kwargs) return SkillResult(success=False, error=f"Skill not found: {skill_name}") def create_parallel_analysis(self, skill_names: List[str]) -> CompositeSkill: """创建并行分析工作流""" def default_merge(results): merged = {} for r in results: if r.success and isinstance(r.data, dict): merged.update(r.data) return merged return self.composer.parallel(*skill_names, merge_func=default_merge) # ==================== 预定义工作流 ==================== class CommonWorkflows: """常用分析工作流""" @staticmethod def impact_analysis(manager: QGISSkillManager) -> CompositeSkill: """ 影响范围分析工作流 流程：加载数据 → 缓冲区分析 → 导出结果 """ return manager.create_workflow( "impact_analysis", ["load_vector_layer", "buffer_analysis", "export_geojson"] ) @staticmethod def site_analysis(manager: QGISSkillManager) -> CompositeSkill: """ 场地分析工作流 流程：加载 → 计算面积 → 统计 → 可视化 """ return manager.create_workflow( "site_analysis", ["load_vector_layer", "calculate_area", "calculate_statistics"] ) @staticmethod def comprehensive_analysis(manager: QGISSkillManager) -> CompositeSkill: """ 综合分析工作流（并行+顺序） 流程：加载 → [缓冲区分析 | 统计计算] → 导出 """ load = manager.registry.get_skill("load_vector_layer") # 先加载数据 load_result = load.execute(path="sites.shp", layer_name="sites") # 然后并行执行多个分析 parallel_analysis = manager.create_parallel_analysis([ "buffer_analysis", "calculate_area", "calculate_statistics" ]) # 最后导出 return manager.composer.sequential( "buffer_analysis", "calculate_area", "export_geojson" ) # ==================== 演示程序 ==================== def demonstrate_skill_system(): """演示技能系统""" print("=" * 70) print("QGIS技能集成管理器演示") print("=" * 70) # 创建管理器 manager = QGISSkillManager() # 1. 列出所有可用技能 print("\n1. 可用技能列表:") print("-" * 70) skills_by_category = manager.list_available_skills() for category, skills in skills_by_category.items(): if skills: print(f"\n{category.upper()}:") for skill in skills: info = manager.get_skill_info(skill) print(f" - {skill}: {info['description']}" if info else f" - {skill}") # 2. 执行单个技能 print("\n2. 执行单个技能（加载矢量图层）:") print("-" * 70) result = manager.execute_skill( "load_vector_layer", path="data/sites.shp", layer_name="ecological_sites" ) print(f"成功: {result.success}") print(f"数据: {json.dumps(result.data, indent=2)}") # 3. 执行复合技能（顺序组合） print("\n3. 执行顺序复合技能（影响范围分析）:") print("-" * 70) workflow = manager.create_workflow( "impact_analysis", ["load_vector_layer", "buffer_analysis"] ) result = workflow.execute( path="data/sources.shp", layer_name="sources", distance=500 ) print(f"成功: {result.success}") print(f"最终数据: {json.dumps(result.data, indent=2)}") # 4. 执行并行技能 print("\n4. 执行并行复合技能（多个分析同时进行）:") print("-" * 70) parallel_workflow = manager.create_parallel_analysis([ "calculate_area", "calculate_statistics" ]) result = parallel_workflow.execute( layer="test_layer", field="area_ha" ) print(f"成功: {result.success}") print(f"合并数据: {json.dumps(result.data, indent=2)}") # 5. 创建自定义工作流 print("\n5. 创建自定义工作流（场地分析）:") print("-" * 70) custom_workflow = CommonWorkflows.site_analysis(manager) print(f"工作流名称: {custom_workflow.name}") print(f"包含技能: {[s.name for s in custom_workflow.skills]}") # 6. 条件组合示例 print("\n6. 条件组合示例（根据文件大小选择处理方式）:") print("-" * 70) def is_large_file(input_data): return input_data.get('feature_count', 0) > 1000 conditional_workflow = manager.composer.conditional( condition=is_large_file, true_skill="buffer_analysis", # 大文件用简单缓冲 # false_skill 可选 ) result = conditional_workflow.execute( layer="large_dataset", distance=100 ) print(f"条件结果: {result.success}") if __name__ == "__main__": demonstrate_skill_system() ``` --- ## 案例分析 ### Claude Code的技能系统 Claude Code使用**Frontmatter-based技能定义**，支持热加载和动态发现： ```python """ Claude Code风格的技能系统 技能定义在 .md 文件中，包含 YAML frontmatter """ # 示例：commit_skill.md """ --- name: commit description: Create git commits with staged changes category: git parameters: - name: message type: string description: Commit message required: true --- This skill creates git commits following best practices: 1. Runs git status and git diff first 2. Analyzes changes to draft commit message 3. Stages specific files (not all) 4. Creates commit with co-author tag """ class ClaudeCodeStyleSkill: """Claude Code风格的技能加载""" def __init__(self, skills_dir: Path): self.skills_dir = skills_dir self.skills = {} def load_skills(self): """从目录加载所有技能""" for md_file in self.skills_dir.glob("*.md"): skill = self._parse_skill_file(md_file) if skill: self.skills[skill['name']] = skill def _parse_skill_file(self, file_path: Path) -> Optional[Dict]: """解析技能文件""" content = file_path.read_text() # 解析 frontmatter if content.startswith('---'): parts = content.split('---', 2) if len(parts) >= 3: import yaml frontmatter = yaml.safe_load(parts[1]) body = parts[2] return { 'name': frontmatter.get('name'), 'description': frontmatter.get('description'), 'parameters': frontmatter.get('parameters', []), 'instructions': body.strip() } return None def get_skill_prompt(self, skill_name: str) -> str: """获取技能的执行提示""" if skill_name in self.skills: skill = self.skills[skill_name] return f""" Skill: {skill['name']} Description: {skill['description']} Parameters: {self._format_parameters(skill['parameters'])} Instructions: {skill['instructions']} """ return "" def _format_parameters(self, params: list) -> str: return "\n".join( f" - {p['name']}: {p['description']}" for p in params ) ``` ### ENAgent的技能组合 ENAgent将生态网络分析分解为可组合技能： ```python class ENAgentSkills: """ ENAgent技能集 将六阶段分析流程分解为可复用技能 """ # 数据处理技能 skills_data = [ "load_landcover", "load_elevation", "normalize_raster", "reclassify_landcover", ] # 空间分析技能 skills_spatial = [ "identify_core_areas", "calculate_resistance", "compute_mcr", "extract_corridors", ] # 可视化技能 skills_viz = [ "map_sources", "map_resistance_surface", "map_corridors", "export_report", ] def build_analysis_workflow(self, requirements: Dict) -> CompositeSkill: """ 根据需求构建分析工作流 Args: requirements: 包含分析需求的字典 """ selected_skills = [] # 数据准备阶段 if requirements.get('data_sources'): selected_skills.extend(self.skills_data) # 分析阶段 if requirements.get('identify_sources'): selected_skills.append("identify_core_areas") if requirements.get('build_resistance'): selected_skills.extend([ "calculate_resistance", "compute_mcr" ]) if requirements.get('extract_corridors'): selected_skills.append("extract_corridors") # 输出阶段 if requirements.get('visualize'): selected_skills.extend(self.skills_viz) return self.composer.sequential(*selected_skills) ``` --- ## 反思与延伸 ### 思考问题 1. **技能粒度**：技能应该多细粒度？太细会有什么问题？太粗会有什么问题？ 2. **接口设计**：如何设计技能接口以支持不同的数据格式？ 3. **版本兼容性**：技能升级时如何保持向后兼容？ 4. **技能发现**：如何让Agent自动发现和组合有用的技能？ ### 延伸阅读 - **"Design Patterns: Elements of Reusable Object-Oriented Software"** - Composite Pattern - **"Microservices Patterns"** (Richards) - 服务组合模式 - LangChain文档 - Tool/Agent composition --- ## 关键要点 1. **技能是Agent的功能单元**，具有明确定义的接口 2. **组合模式**包括顺序、并行、条件和迭代四种基本类型 3. **技能注册表**支持动态发现和加载新技能 4. **复合技能**可以像原子技能一样被使用和组合 5. **前后端分离**的设计使技能可以在不同上下文中复用 6. **元数据描述**使技能可被自动发现和组合