19.3 实现机制对比
19.3.1 Skills 的实现机制
1. 基于 LLM 的执行模型
Skills 的核心实现依赖于大语言模型（LLM）的理解和生成能力。
执行流程 ： 用户请求 ↓ 自然语言解析（LLM） ↓ 意图识别 ↓ 参数提取 ↓ 技能选择 ↓ 技能执行 ↓ 结果生成（LLM） ↓ 返回用户
代码示例 ：
python
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)
### 2\. 上下文感知机制Skills 能够感知和利用上下文信息，提供更智能的响应。
上下文类型 ：
python
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
### 3\. 示例学习机制Skills 可以通过示例学习，提高执行准确性和一致性。
示例配置 ：
python
yaml
```yaml
```python
# skill.yaml
name: code-reviewer
version: 1.0.0
description: 代码审查技能
examples:
- input: |
请审查以下 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
```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
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 技术栈
bash
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 性能对比
### 执行时间对比
python
# Skills 执行时间
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
# 扩展插件：添加新功能
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
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