pengxiao
a90f7adfa1
将 Markdown 源文件移入 md/，LaTeX 工作目录保留在 latex/， Word 导出移入 word/；删除临时脚本、调试截图和空 stub。 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
33 KiB
33 KiB
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
适用：增量式处理
设计原理
技能接口设计
良好的技能接口是实现组合的基础：
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
]
}
技能组合器
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}
)
动态技能发现与加载
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技能集成管理器
"""
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技能定义，支持热加载和动态发现：
"""
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将生态网络分析分解为可组合技能：
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)
反思与延伸
思考问题
-
技能粒度：技能应该多细粒度？太细会有什么问题？太粗会有什么问题？
-
接口设计：如何设计技能接口以支持不同的数据格式？
-
版本兼容性：技能升级时如何保持向后兼容？
-
技能发现：如何让Agent自动发现和组合有用的技能？
延伸阅读
- "Design Patterns: Elements of Reusable Object-Oriented Software" - Composite Pattern
- "Microservices Patterns" (Richards) - 服务组合模式
- LangChain文档 - Tool/Agent composition
关键要点
- 技能是Agent的功能单元，具有明确定义的接口
- 组合模式包括顺序、并行、条件和迭代四种基本类型
- 技能注册表支持动态发现和加载新技能
- 复合技能可以像原子技能一样被使用和组合
- 前后端分离的设计使技能可以在不同上下文中复用
- 元数据描述使技能可被自动发现和组合