19.3 实现机制对比

19.3.1 Skills 的实现机制

1. 基于 LLM 的执行模型

Skills 的核心实现依赖于大语言模型(LLM)的理解和生成能力。

执行流程 : 用户请求 ↓ 自然语言解析(LLM) ↓ 意图识别 ↓ 参数提取 ↓ 技能选择 ↓ 技能执行 ↓ 结果生成(LLM) ↓ 返回用户

代码示例

    python
 
    class SkillExecutor:
        def __init__(self, llm_client):
            self.llm_client = llm_client
            self.skills = {}
 
        def register_skill(self, skill):
            """注册技能"""
            self.skills[skill.name] = skill
 
        def execute(self, user_input):
            """执行技能"""
            # 1. 使用 LLM 解析用户意图
            intent = self.parse_intent(user_input)
 
            # 2. 选择合适的技能
            skill = self.select_skill(intent)
 
            # 3. 提取参数
            parameters = self.extract_parameters(user_input, skill)
 
            # 4. 执行技能
            result = skill.execute(parameters)
 
            # 5. 使用 LLM 格式化结果
            formatted_result = self.format_result(result)
 
            return formatted_result
 
        def parse_intent(self, user_input):
            """解析用户意图"""
            prompt = f"""
            分析以下用户请求的意图:
            {user_input}
 
            返回意图类型和关键信息。
            """
            response = self.llm_client.complete(prompt)
            return parse_intent_response(response)

Skills 能够感知和利用上下文信息,提供更智能的响应。

上下文类型

    python
 
    class SkillContext:
        def __init__(self):
            self.conversation_history = []      # 对话历史
            self.project_context = {}           # 项目上下文
            self.user_preferences = {}          # 用户偏好
            self.environment_variables = {}     # 环境变量
            self.available_tools = []           # 可用工具
            self.available_plugins = []         # 可用插件
 
        def get_conversation_history(self, limit=10):
            """获取对话历史"""
            return self.conversation_history[-limit:]
 
        def get_project_info(self):
            """获取项目信息"""
            return self.project_context
 
        def get_user_preference(self, key):
            """获取用户偏好"""
            return self.user_preferences.get(key)
 
        def get_tool(self, name):
            """获取工具"""
            for tool in self.available_tools:
                if tool.name == name:
                    return tool
            return None

Skills 可以通过示例学习,提高执行准确性和一致性。

示例配置

    yaml
 
```python
    # skill.yaml
    name: code-reviewer
    version: 1.0.0
    description: 代码审查技能
 
    examples:
      - input: |
          请审查以下 Python 代码:
```python
          def calculate_sum(a, b):
              return a + b
        output: |

代码审查结果:

  • 代码结构清晰

  • 函数命名规范

  • 缺少类型注解

  • 建议添加文档字符串

  • input: | 审查这段 JavaScript 代码:

          function getData() {
              fetch('/api/data')
                  .then(response => response.json())
                  .then(data => console.log(data));
          }
        output: |

代码审查结果:

  • 缺少错误处理
  • 建议使用 async/await
  • 缺少加载状态处理
  • 建议添加类型检查
 
class ExampleBasedSkill(Skill):
def __init__(self, examples):
super().__init__()
self.examples = examples
def execute(self, parameters, context):
 
# 1. 查找相似示例
 
similar_examples = self.find_similar_examples(parameters["input"])
 
# 2. 构建提示
 
prompt = self.build_prompt(parameters["input"], similar_examples)
 
# 3. 使用 LLM 生成响应
 
response = context.llm_client.complete(prompt)
return SkillResult(success=True, data={"response": response})
def find_similar_examples(self, input_text):
"""查找相似示例"""
 
# 使用嵌入向量查找相似示例
 
input_embedding = self.get_embedding(input_text)
similarities = []
for example in self.examples:
example_embedding = self.get_embedding(example["input"])
similarity = cosine_similarity(input_embedding, example_embedding)
similarities.append((example, similarity))
 
# 返回最相似的 k 个示例
 
similarities.sort(key=lambda x: x[1], reverse=True)
return [example for example, _ in similarities[:3]]
 

4. 动态参数解析

Skills 能够动态解析用户输入中的参数,无需预先定义严格的参数结构。

参数解析示例

    python
 
```python
```python
    class DynamicParameterParser:
        def parse_parameters(self, user_input, skill_schema):

"""动态解析参数""" prompt = f""" 从以下用户输入中提取参数: 用户输入:{user_input}

技能参数模式: {json.dumps(skill_schema, indent=2)}

返回提取的参数 JSON。 """

response = self.llm_client.complete(prompt)

            return json.loads(response)
 

使用示例

parser = DynamicParameterParser(llm_client)

skill_schema = { "name": "code-generator", "parameters": { "language": {"type": "string", "description": "编程语言"}, "functionality": {"type": "string", "description": "功能描述"}, "style": {"type": "string", "description": "代码风格", "optional": True} } }

user_input = "用 Python 写一个快速排序算法,使用函数式编程风格" parameters = parser.parse_parameters(user_input, skill_schema)


## 结果:

## {

## "language": "Python",

## "functionality": "快速排序算法",

## "style": "函数式编程"

## }

```bash

    ### 1. 基于代码的执行模型

    插件通过直接执行代码来实现功能,具有确定性和高性能。

    > **执行流程**:
    用户请求
    ↓
    参数验证
    ↓
    方法调用
    ↓
    代码执行
    ↓
    结果处理
    ↓
    返回用户

代码示例

 
```python
    class PluginExecutor:
        def __init__(self):
            self.plugins = {}
 
        def register_plugin(self, plugin):

"""注册插件"""

            self.plugins[plugin.name] = plugin
 
        def execute(self, plugin_name, method_name, parameters):

"""执行插件方法"""

1. 获取插件

plugin = self.plugins.get(plugin_name)

            if not plugin:
                raise PluginNotFoundError(plugin_name)
 

2. 获取方法

method = getattr(plugin, method_name)

            if not method:
                raise MethodNotFoundError(method_name)
 

3. 验证参数

            self.validate_parameters(method, parameters)
 

4. 执行方法

            try:
                result = method(**parameters)
                return {
                    "success": True,
                    "data": result
                }
            except Exception as e:
                return {
                    "success": False,
                    "error": str(e)
                }
 
        def validate_parameters(self, method, parameters):

"""验证参数"""

            import inspect
            sig = inspect.signature(method)
 
            for param_name, param in sig.parameters.items():
                if param_name == "self":
                    continue
 
                if param.default == param.empty and param_name not in parameters:
                    raise MissingParameterError(param_name)
 

2. 类型安全机制

插件通过类型系统确保代码的健壮性和可维护性。

类型定义示例

    from typing import TypedDict, List, Optional
    from pydantic import BaseModel, validator
    class PaymentRequest(BaseModel):

"""支付请求""" amount: float currency: str payment_method: str card_number: Optional[str] = None expiry_date: Optional[str] = None cvv: Optional[str] = None @validator('amount')

    def amount_must_be_positive(cls, v):
    if v <= 0:
    raise ValueError('金额必须大于零')
    return v
    @validator('currency')
    def currency_must_be_valid(cls, v):
    valid_currencies = ['USD', 'EUR', 'CNY']
    if v not in valid_currencies:
    raise ValueError(f'货币必须是以下之一:{valid_currencies}')
    return v
    class PaymentResponse(BaseModel):

"""支付响应""" success: bool transaction_id: Optional[str] = None error_message: Optional[str] = None

    class PaymentPlugin(Plugin):
    def process_payment(self, request: PaymentRequest) -> PaymentResponse:

"""处理支付"""

类型自动验证

    try:

处理支付逻辑

transaction_id = self.call_payment_gateway(request)

    return PaymentResponse(
    success=True,
    transaction_id=transaction_id
    )
    except Exception as e:
    return PaymentResponse(
    success=False,
    error_message=str(e)
    )

### 3\. 依赖注入机制

插件通过依赖注入实现松耦合和可测试性。 **依赖注入示例** :

```python

```python
    from dependency_injector import containers, providers

    class Container(containers.DeclarativeContainer):

"""依赖容器"""

配置

config = providers.Configuration()

数据库连接

database = providers.Singleton( DatabaseConnection, host=config.db.host, port=config.db.port, username=config.db.username, password=config.db.password )

缓存服务

cache = providers.Singleton( CacheService, host=config.cache.host, port=config.cache.port )

支付网关

payment_gateway = providers.Singleton( PaymentGateway, api_key=config.payment.api_key, environment=config.payment.environment )

支付插件

payment_plugin = providers.Factory( PaymentPlugin, database=database, cache=cache, payment_gateway=payment_gateway )

使用示例

container = Container() container.config.from_yaml('config.yaml')

payment_plugin = container.payment_plugin() result = payment_plugin.process_payment(payment_request)

4. 生命周期管理

插件具有完整的生命周期管理,包括初始化、启动、停止和销毁。

生命周期示例

    class PluginLifecycle:

"""插件生命周期"""

    def __init__(self):
    self.state = "uninitialized"
    self.dependencies = []
    def initialize(self, config):

"""初始化插件"""

    if self.state != "uninitialized":
    raise PluginError("插件已初始化")

加载配置

    self.config = config

初始化依赖

    self.initialize_dependencies()
    self.state = "initialized"
    logger.info(f"插件 {self.name} 初始化完成")
    def start(self):

"""启动插件"""

    if self.state != "initialized":
    raise PluginError("插件未初始化")

启动依赖

    self.start_dependencies()

执行启动逻辑

    self.on_start()
    self.state = "running"
    logger.info(f"插件 {self.name} 启动完成")
    def stop(self):

"""停止插件"""

    if self.state != "running":
    raise PluginError("插件未运行")

执行停止逻辑

    self.on_stop()

停止依赖

    self.stop_dependencies()
    self.state = "stopped"
    logger.info(f"插件 {self.name} 停止完成")
    def destroy(self):

"""销毁插件"""

    if self.state not in ["initialized", "stopped"]:
    raise PluginError("插件状态不正确")

销毁依赖

    self.destroy_dependencies()
    self.state = "destroyed"
    logger.info(f"插件 {self.name} 销毁完成")
    def on_start(self):

"""启动钩子""" pass

    def on_stop(self):

"""停止钩子""" pass


## 19.3.3 实现机制对比表

特性| Skills| 插件
---|---|---
执行方式| LLM 解释执行| 直接代码执行
参数解析| 动态解析| 静态验证
类型安全| 弱类型| 强类型
性能| 中等| 高
确定性| 低| 高
可预测性| 低| 高
调试难度| 高| 低
测试难度| 高| 低
灵活性| 高| 中
扩展性| 高| 中
依赖管理| 简单| 复杂
版本控制| 简单| 复杂
上下文感知| 强| 弱
自然语言支持| 强| 弱
错误处理| 困难| 容易
性能优化| 困难| 容易

## 19.3.4 技术栈对比

### Skills 技术栈

```yaml

    核心技术:
      - 大语言模型 (LLM)
      - 自然语言处理 (NLP)
      - 提示工程 (Prompt Engineering)
      - 嵌入向量 (Embeddings)

    辅助技术:
      - 向量数据库 (Vector Database)
      - 语义搜索 (Semantic Search)
      - 上下文管理 (Context Management)
      - 示例学习 (Few-shot Learning)

    开发工具:
      - LLM API (OpenAI, Anthropic, etc.)
      - 提示模板引擎
      - 嵌入模型
      - 向量存储

    ### 插件技术栈

    核心技术:
    - 编程语言 (Python, JavaScript, etc.)
    - 框架 (Flask, Express, etc.)
    - 数据库 (SQL, NoSQL)
    - API 设计
    辅助技术:
    - 类型系统 (TypeScript, Pydantic)
    - 依赖注入 (Dependency Injection)
    - 测试框架 (pytest, Jest)
    - 容器化 (Docker, Kubernetes)
    开发工具:
    - IDE (VS Code, IntelliJ)
    - 版本控制 (Git)
    - 包管理器 (npm, pip)
    - CI/CD (GitHub Actions, Jenkins)

19.3.5 性能对比

执行时间对比

 
    # Skills 执行时间
```python
    def execute_skill(user_input):
        start_time = time.time()
 

LLM 解析:~500ms

intent = parse_intent(user_input)

参数提取:~300ms

parameters = extract_parameters(user_input)

技能执行:~200ms

result = execute_skill_logic(parameters)

结果生成:~500ms

formatted_result = format_result(result)

end_time = time.time()

        return end_time - start_time  # ~1.5s
 

插件执行时间

    def execute_plugin(method_name, parameters):
        start_time = time.time()
 

参数验证:~10ms

validate_parameters(parameters)

方法执行:~50ms

result = execute_method(method_name, parameters)

结果处理:~10ms

formatted_result = process_result(result)

end_time = time.time()

        return end_time - start_time  # ~70ms
 

资源消耗对比

资源类型Skills插件
CPU 使用中等
内存使用
网络带宽
磁盘 I/O
GPU 使用
并发能力

19.3.6 可维护性对比

Skills 可维护性

优势

  • 代码量少,易于理解
  • 自然语言描述,易于修改
  • 无需编译,快速迭代

劣势

  • 依赖 LLM 行为,难以预测
  • 调试困难,难以定位问题
  • 版本控制复杂,提示变化影响大

插件可维护性

优势

  • 代码结构清晰,易于理解
  • 调试工具完善,易于定位问题
  • 版本控制成熟,变更可追踪

劣势

  • 代码量大,维护成本高
  • 需要编译和测试,迭代较慢
  • 依赖管理复杂

19.3.7 可扩展性对比

Skills 可扩展性

扩展 Skills:添加新功能

    class ExtendedSkill(Skill):
    def __init__(self):
    super().__init__()
    self.new_feature = None
    def add_feature(self, feature):

"""添加新功能"""

    self.new_feature = feature
    return True
    def execute(self, parameters, context):

使用新功能

    if self.new_feature:
    result = self.new_feature.process(parameters)
    else:
    result = super().execute(parameters, context)
    return result

### 插件可扩展性

```python

    # 扩展插件:添加新功能
```python
    class ExtendedPlugin(Plugin):
        def __init__(self):
            super().__init__()
            self.features = {}

        def register_feature(self, name, feature):

"""注册新功能"""

            self.features[name] = feature
            return True
 
        def execute_feature(self, name, parameters):

"""执行功能""" feature = self.features.get(name)

            if not feature:
                raise FeatureNotFoundError(name)
 
            return feature.execute(parameters)
 

19.3.8 安全性对比

Skills 安全性

安全挑战

  • 提示注入攻击
  • 敏感信息泄露
  • 不可预测的行为
  • 难以审计

安全措施

    class SecureSkill(Skill):
    def sanitize_input(self, user_input):

"""清理输入"""

移除敏感信息

sanitized = self.remove_sensitive_info(user_input)

检测注入攻击

    if self.detect_injection(sanitized):
    raise SecurityError("检测到注入攻击")
    return sanitized
    def sanitize_output(self, output):

"""清理输出"""

移除敏感信息

sanitized = self.remove_sensitive_info(output)

添加安全警告

    if self.contains_potentially_harmful_content(sanitized):
    sanitized = self.add_security_warning(sanitized)
    return sanitized

### 插件安全性

**安全优势** :

  * 代码静态分析
  * 类型安全
  * 访问控制
  * 审计日志 **安全措施** :

```python

```python
    class SecurePlugin(Plugin):
        def __init__(self):
            super().__init__()
            self.access_control = AccessControl()
            self.audit_logger = AuditLogger()

        def execute(self, method_name, parameters, user):

"""执行方法"""

检查访问权限

            if not self.access_control.check_permission(user, method_name):
                raise PermissionError("无权限访问")
 

记录审计日志

            self.audit_logger.log(user, method_name, parameters)
 

执行方法

result = super().execute(method_name, parameters)

            return result