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