赛雷观影1. 跨平台Agent Skills开发的必要性
在当今AI技术快速发展的环境下，开发者面临着一个关键挑战：如何让智能体(Agent)的技能(Skills)能够在不同AI平台间无缝切换和复用。这个问题源于以下几个现实因素：
首先，各大AI平台如OpenAI、Claude、本地开源模型等，它们的API接口设计存在显著差异。这些差异体现在认证方式、请求参数格式、响应数据结构等多个方面。例如，OpenAI的ChatCompletion接口与Claude的Messages API在消息体结构上就有很大不同。
其次，不同平台对提示词的处理方式也各不相同。有些平台需要显式指定system角色，有些则采用更简单的user/assistant对话格式。这种不一致性使得为单一平台开发的技能很难直接迁移到其他平台。
更麻烦的是，当我们需要在项目中同时使用多个AI平台时（比如为了成本优化或功能互补），代码中会充斥着大量的平台特定逻辑。这不仅增加了维护难度，也使得核心业务逻辑被技术细节所淹没。
需要模型API调用？ 免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. 适配器模式深度解析
2.1 适配器模式的核心思想
适配器模式是一种结构型设计模式，它的核心目的是在不修改原有代码的情况下，使不兼容的接口能够协同工作。这种模式在现实世界中有很多类比，比如电源插头转换器，可以让不同国家标准的电器在异国他乡也能正常使用。
在软件设计中，适配器模式通过创建一个中间层（适配器）来解决接口不匹配的问题。这个适配器会封装目标接口的复杂性，并向客户端提供一个统一的、简化的接口。这样，客户端代码就不需要关心底层实现细节，只需与这个统一接口交互即可。
2.2 适配器在AI技能开发中的价值
在AI技能开发场景下，适配器模式带来了几个关键优势：
- 解耦：将平台特定的实现细节与业务逻辑分离，使两者可以独立演化
- 可扩展性：新增平台支持只需添加新的适配器，无需修改现有代码
- 可维护性：平台相关代码集中管理，问题定位和修复更加容易
- 可测试性：可以轻松创建测试适配器，方便单元测试和集成测试
从架构角度看，适配器模式帮助我们实现了"依赖倒置"原则——高层模块不依赖于低层模块，两者都依赖于抽象。这种架构特性对于构建健壮的AI应用系统至关重要。
3. 实战：构建跨平台提示词优化技能
3.1 需求分析与设计
让我们以一个具体的业务场景为例：开发一个能够跨平台工作的提示词优化技能。这个技能需要能够：
- 接收用户输入的原始提示词
- 根据配置或策略选择最合适的AI平台
- 调用相应平台的API进行提示词优化
- 返回优化后的提示词及相关元数据（如消耗的token数）
为了实现这个功能，我们需要设计以下几个核心组件：
- 统一技能接口(ISkill)：定义所有技能都必须实现的标准方法
- 平台适配器：针对每个平台实现具体的适配器类
- 技能调度器：负责选择合适的适配器并协调执行流程
3.2 代码实现详解
3.2.1 定义统一接口
首先，我们定义一个抽象基类来规范所有技能的行为：
python复制from abc import ABC, abstractmethod
from typing import Dict, Optional
class ISkill(ABC):
"""所有Agent技能必须实现的统一接口"""
@abstractmethod
def run(self, context: Dict) -> Dict:
"""执行技能的核心方法
Args:
context: 包含输入数据和执行环境的字典
- 必须包含input_text字段
- 可包含其他平台特定参数
Returns:
包含执行结果的字典
- 必须包含optimized_text字段
- 应包含platform和消耗统计信息
"""
pass
这个接口设计有几个关键考虑：
- 使用抽象基类确保所有实现类都提供必要的方法
- 输入输出都使用字典，保持足够的灵活性
- 明确定义了必填字段和可选字段
3.2.2 实现OpenAI适配器
接下来是OpenAI平台的适配器实现：
python复制import openai
from typing import Dict
class OpenAIPromptAdapter(ISkill):
"""OpenAI平台的提示词优化适配器"""
def __init__(self, api_key: str, model: str = "gpt-3.5-turbo"):
self.client = openai.OpenAI(api_key=api_key)
self.model = model
def run(self, context: Dict) -> Dict:
# 构造优化指令
optimization_prompt = f"""
你是一个专业的提示词优化专家。请根据以下要求优化给定的提示词：
原始提示词：{context['input_text']}
优化要求：
1. 明确指定角色和专业领域
2. 包含具体的约束条件和输出格式要求
3. 保持语义清晰且无歧义
4. 长度控制在100-200个token之间
请直接返回优化后的提示词，不需要额外解释。
"""
try:
# 调用OpenAI API
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": optimization_prompt}],
temperature=0.3,
max_tokens=300
)
# 解析并返回标准化结果
return {
"optimized_text": response.choices[0].message.content.strip(),
"platform": "OpenAI",
"model": self.model,
"tokens_used": response.usage.total_tokens,
"success": True
}
except Exception as e:
return {
"optimized_text": context['input_text'], # 失败时返回原提示词
"platform": "OpenAI",
"error": str(e),
"success": False
}
这个实现中有几个值得注意的点：
- 在优化指令中给出了具体的优化标准
- 包含了错误处理逻辑，保证健壮性
- 返回统一的成功/失败标识
- 记录了详细的元数据信息
3.2.3 实现Claude适配器
Claude平台的适配器实现会有一些差异：
python复制import anthropic
from typing import Dict
class ClaudePromptAdapter(ISkill):
"""Claude平台的提示词优化适配器"""
def __init__(self, api_key: str, model: str = "claude-3-haiku-20240307"):
self.client = anthropic.Anthropic(api_key=api_key)
self.model = model
def run(self, context: Dict) -> Dict:
# Claude需要显式指定system角色
system_prompt = """
你是一个专业的提示词优化专家。你的任务是根据给定的原始提示词，
生成一个更专业、更清晰的优化版本。优化后的提示词应该：
- 明确定义角色和专业领域
- 包含具体的约束条件
- 指定期望的输出格式
- 保持语义清晰无歧义
"""
try:
message = self.client.messages.create(
model=self.model,
system=system_prompt,
messages=[{
"role": "user",
"content": f"请优化以下提示词：{context['input_text']}\n只需返回优化后的提示词，不需要额外解释。"
}],
max_tokens=300,
temperature=0.2
)
return {
"optimized_text": message.content[0].text.strip(),
"platform": "Claude",
"model": self.model,
"input_tokens": message.usage.input_tokens,
"output_tokens": message.usage.output_tokens,
"success": True
}
except Exception as e:
return {
"optimized_text": context['input_text'],
"platform": "Claude",
"error": str(e),
"success": False
}
与OpenAI适配器相比，Claude适配器的主要区别在于：
- 使用了不同的客户端库(anthropic)
- 需要显式指定system角色
- API参数命名有所不同
- 返回的用量统计信息更详细
3.2.4 实现技能调度器
调度器负责管理多个适配器并实现智能路由：
python复制from typing import Dict, Optional
class PromptOptimizerSkill:
"""跨平台提示词优化技能"""
def __init__(self):
self.adapters = {} # 平台名称到适配器的映射
def register_adapter(self, platform: str, adapter: ISkill):
"""注册新的平台适配器"""
self.adapters[platform] = adapter
def execute(self, input_text: str, preferred_platform: Optional[str] = None) -> Dict:
"""执行提示词优化
Args:
input_text: 需要优化的原始提示词
preferred_platform: 指定平台名称，None表示自动选择
Returns:
标准化结果字典
"""
context = {"input_text": input_text}
# 确定使用哪个平台
platform = preferred_platform if preferred_platform else self._select_platform()
# 获取适配器实例
adapter = self.adapters.get(platform)
if not adapter:
return {
"optimized_text": input_text,
"error": f"Unsupported platform: {platform}",
"success": False
}
# 执行优化
return adapter.run(context)
def _select_platform(self) -> str:
"""选择最合适的平台
这里可以实现更复杂的路由逻辑，如：
- 基于成本
- 基于平台健康状态
- 基于性能历史数据
"""
# 简化实现：按注册顺序返回第一个可用平台
return next(iter(self.adapters.keys())) if self.adapters else None
调度器的关键设计点：
- 支持动态注册适配器，符合开闭原则
- 提供平台指定和自动选择两种模式
- 预留了扩展点用于实现更复杂的路由逻辑
- 保持了统一的错误处理模式
3.3 使用示例
下面是如何使用这个跨平台提示词优化技能的完整示例：
python复制def main():
# 初始化技能实例
optimizer = PromptOptimizerSkill()
# 注册各个平台的适配器
optimizer.register_adapter("openai", OpenAIPromptAdapter("your-openai-key"))
optimizer.register_adapter("claude", ClaudePromptAdapter("your-claude-key"))
# 测试提示词优化
test_prompt = "帮我写一篇关于机器学习的文章"
print("=== 使用OpenAI优化 ===")
result = optimizer.execute(test_prompt, "openai")
print(f"优化结果: {result['optimized_text']}")
print(f"使用平台: {result['platform']}")
print(f"消耗token: {result.get('tokens_used', 'N/A')}")
print("\n=== 使用Claude优化 ===")
result = optimizer.execute(test_prompt, "claude")
print(f"优化结果: {result['optimized_text']}")
print(f"使用平台: {result['platform']}")
print(f"输入token: {result.get('input_tokens', 'N/A')}")
print(f"输出token: {result.get('output_tokens', 'N/A')}")
print("\n=== 自动选择平台 ===")
result = optimizer.execute(test_prompt)
print(f"优化结果: {result['optimized_text']}")
print(f"自动选择平台: {result['platform']}")
if __name__ == "__main__":
main()
这个示例展示了：
- 如何初始化技能并注册适配器
- 如何指定平台执行优化
- 如何使用自动平台选择功能
- 如何访问统一的返回字段
4. 高级应用与优化
4.1 动态平台路由策略
在实际生产环境中，简单的平台选择策略往往不够用。我们可以扩展调度器，实现更智能的路由逻辑：
python复制class AdvancedPromptOptimizerSkill(PromptOptimizerSkill):
"""带智能路由的提示词优化技能"""
def __init__(self, config: Dict):
super().__init__()
self.config = config
self.platform_metrics = {} # 记录各平台性能指标
def _select_platform(self) -> str:
"""基于多种因素选择最佳平台"""
available_platforms = list(self.adapters.keys())
# 1. 检查显式配置的优先级
if 'platform_priority' in self.config:
for platform in self.config['platform_priority']:
if platform in available_platforms:
return platform
# 2. 基于成本考虑
if self.config.get('cost_aware', False):
low_cost_platform = min(
available_platforms,
key=lambda p: self._estimate_cost(p)
)
if self._estimate_cost(low_cost_platform) < self.config.get('cost_threshold', float('inf')):
return low_cost_platform
# 3. 基于性能历史
if self.platform_metrics:
best_platform = max(
available_platforms,
key=lambda p: self.platform_metrics.get(p, {}).get('success_rate', 0)
)
return best_platform
# 默认回退
return super()._select_platform()
def _estimate_cost(self, platform: str) -> float:
"""估算使用指定平台的预期成本"""
# 这里可以实现实际的成本计算逻辑
# 例如查询平台定价API或使用本地缓存数据
return 0.0 # 简化实现
这种高级路由策略可以考虑：
- 静态配置的优先级
- 成本因素
- 历史性能数据
- 平台健康状态
4.2 适配器模式的变体
除了基本的适配器模式，我们还可以考虑几种有用的变体：
-
缓存适配器：在适配器中加入缓存层，减少对平台API的调用
python复制class CachedOpenAIAdapter(OpenAIPromptAdapter): def __init__(self, *args, cache_ttl=3600, **kwargs): super().__init__(*args, **kwargs) self.cache = {} # 简单内存缓存 self.cache_ttl = cache_ttl def run(self, context: Dict) -> Dict: cache_key = hash(context['input_text']) if cache_key in self.cache: return self.cache[cache_key] result = super().run(context) if result['success']: self.cache[cache_key] = result return result -
熔断适配器：实现熔断机制，防止故障扩散
python复制class CircuitBreakerAdapter(ISkill): def __init__(self, wrapped_adapter: ISkill, failure_threshold=3, reset_timeout=60): self.wrapped = wrapped_adapter self.failure_count = 0 self.last_failure_time = 0 self.threshold = failure_threshold self.reset_timeout = reset_timeout def run(self, context: Dict) -> Dict: if self._is_circuit_open(): return { "optimized_text": context['input_text'], "error": "Circuit breaker is open", "success": False } try: result = self.wrapped.run(context) if result['success']: self._reset_counter() else: self._record_failure() return result except Exception as e: self._record_failure() return { "optimized_text": context['input_text'], "error": str(e), "success": False } def _is_circuit_open(self) -> bool: return (self.failure_count >= self.threshold and time.time() - self.last_failure_time < self.reset_timeout) -
日志适配器：记录详细的调用日志
python复制class LoggingAdapter(ISkill): def __init__(self, wrapped_adapter: ISkill, logger): self.wrapped = wrapped_adapter self.logger = logger def run(self, context: Dict) -> Dict: start_time = time.time() self.logger.info(f"Starting prompt optimization for: {context['input_text']}") result = self.wrapped.run(context) duration = time.time() - start_time log_data = { "input": context['input_text'], "output": result['optimized_text'], "platform": result.get('platform'), "success": result['success'], "duration": duration } if result['success']: self.logger.info("Prompt optimization succeeded", extra=log_data) else: self.logger.error("Prompt optimization failed", extra=log_data) return result
这些适配器可以组合使用，形成强大的处理链：
python复制# 创建带缓存、熔断和日志的OpenAI适配器
adapter = LoggingAdapter(
CircuitBreakerAdapter(
CachedOpenAIAdapter(api_key="your-key"),
failure_threshold=3
),
logger=get_logger()
)
4.3 性能优化技巧
在实际使用适配器模式时，有几个性能优化点值得注意：
-
连接池管理：为每个适配器维护可复用的API连接
python复制class OpenAIAdapterWithConnectionPool(OpenAIPromptAdapter): _client_pool = {} def __init__(self, api_key: str, model: str = "gpt-3.5-turbo"): if api_key not in self._client_pool: self._client_pool[api_key] = openai.OpenAI(api_key=api_key) self.client = self._client_pool[api_key] self.model = model -
批量处理：实现批量处理接口减少API调用次数
python复制def batch_run(self, contexts: List[Dict]) -> List[Dict]: """批量优化多个提示词""" # 实现平台特定的批量处理逻辑 pass -
异步支持：提供异步接口提高并发性能
python复制async def run_async(self, context: Dict) -> Dict: """异步执行提示词优化""" # 实现平台特定的异步调用 pass
5. 适配器模式的最佳实践
5.1 设计原则
在实现适配器模式时，遵循以下原则可以避免常见陷阱：
- 单一职责原则：每个适配器只负责一个平台的转换逻辑
- 接口隔离原则：保持统一接口精简，只包含必要方法
- 依赖倒置原则：高层模块不应该依赖低层模块，两者都应该依赖抽象
- 开闭原则：通过添加新适配器来扩展功能，而不是修改现有代码
5.2 常见问题与解决方案
问题1：平台差异过大，难以统一接口
解决方案：
- 在统一接口中只定义最核心的功能
- 允许适配器通过context参数接收平台特定配置
- 对于无法统一的功能，可以提供平台特定的扩展接口
问题2：适配器过多导致管理困难
解决方案：
- 使用工厂模式集中管理适配器创建
- 实现自动发现和注册机制
- 采用依赖注入框架管理适配器生命周期
问题3：性能开销
解决方案：
- 实现轻量级适配器，避免不必要的转换
- 考虑缓存常用转换结果
- 对于性能关键路径，可以提供绕过适配器的直接访问方式
5.3 测试策略
适配器模式的测试需要特别关注几个方面：
-
接口一致性测试：确保所有适配器都正确实现了统一接口
python复制def test_adapter_interface(adapter_factory): """验证适配器是否实现了所有必要方法""" adapter = adapter_factory() assert hasattr(adapter, 'run') assert callable(adapter.run) -
功能对等性测试：验证不同适配器对相同输入产生语义相似的输出
python复制def test_functional_equivalence(adapters): """测试不同适配器对相同输入产生相似输出""" test_input = "写一篇关于深度学习的文章" results = [adapter.run({"input_text": test_input}) for adapter in adapters] # 验证所有结果都是成功的 assert all(r['success'] for r in results) # 验证优化后的提示词都包含关键元素 for r in results: assert "深度学习" in r['optimized_text'] assert "文章" in r['optimized_text'] -
性能基准测试：比较不同适配器的性能特征
python复制def test_performance_benchmark(adapters): """测试各适配器的性能指标""" test_input = {"input_text": "解释神经网络的工作原理"} results = {} for adapter in adapters: start_time = time.perf_counter() result = adapter.run(test_input) duration = time.perf_counter() - start_time assert result['success'] results[adapter.__class__.__name__] = { "time": duration, "tokens": result.get('tokens_used', 0) } return results -
错误处理测试：验证适配器对异常情况的处理
python复制def test_error_handling(adapter): """测试适配器的错误处理能力""" # 测试无效输入 result = adapter.run({"input_text": ""}) assert not result['success'] assert "error" in result # 测试API错误模拟 with patch.object(adapter.client, 'chat') as mock: mock.side_effect = Exception("API error") result = adapter.run({"input_text": "valid input"}) assert not result['success'] assert "API error" in result['error']
6. 扩展应用场景
适配器模式在AI应用开发中还有许多其他应用场景：
6.1 多模型推理服务
python复制class ModelInferenceSkill(ISkill):
"""统一的多模型推理技能"""
def __init__(self):
self.adapters = {
"text-gen": TextGenerationAdapter(),
"image-gen": ImageGenerationAdapter(),
"audio-gen": AudioGenerationAdapter()
}
def run(self, context: Dict) -> Dict:
task_type = context.get('task_type', 'text-gen')
adapter = self.adapters.get(task_type)
if not adapter:
raise ValueError(f"Unsupported task type: {task_type}")
return adapter.run(context)
6.2 数据存储抽象
python复制class StorageAdapter(ABC):
"""统一存储接口"""
@abstractmethod
def save(self, key: str, data: Any) -> bool:
pass
@abstractmethod
def load(self, key: str) -> Any:
pass
class S3StorageAdapter(StorageAdapter):
"""AWS S3存储适配器"""
# 实现省略
class LocalStorageAdapter(StorageAdapter):
"""本地文件系统适配器"""
# 实现省略
class StorageSkill:
"""跨存储平台的统一技能"""
def __init__(self, default_adapter="local"):
self.adapters = {
"s3": S3StorageAdapter(),
"local": LocalStorageAdapter()
}
self.default_adapter = default_adapter
def execute(self, operation: str, key: str, data: Any = None) -> Any:
adapter = self.adapters.get(self.default_adapter)
if operation == "save":
return adapter.save(key, data)
elif operation == "load":
return adapter.load(key)
else:
raise ValueError(f"Unsupported operation: {operation}")
6.3 多模态数据处理
python复制class MultiModalAdapter(ISkill):
"""处理多种数据类型的适配器"""
def run(self, context: Dict) -> Dict:
input_data = context['input']
if isinstance(input_data, str):
return self._process_text(input_data)
elif isinstance(input_data, (PIL.Image.Image, np.ndarray)):
return self._process_image(input_data)
elif isinstance(input_data, (np.ndarray, torch.Tensor)):
return self._process_audio(input_data)
else:
raise ValueError(f"Unsupported input type: {type(input_data)}")
7. 架构演进与未来发展
随着AI技术的不断发展，适配器模式在Agent架构中的应用也在不断演进：
7.1 动态适配器加载
更先进的系统可以实现适配器的动态加载和热替换：
python复制class DynamicAdapterManager:
"""动态管理适配器实例"""
def __init__(self):
self.adapters = {}
self.adapter_dirs = ["adapters"]
def load_adapters(self):
"""动态加载所有适配器"""
for dir_path in self.adapter_dirs:
for file in os.listdir(dir_path):
if file.endswith("_adapter.py"):
module_name = file[:-3]
module = importlib.import_module(f"adapters.{module_name}")
for name, cls in inspect.getmembers(module, inspect.isclass):
if name.endswith("Adapter") and name != "ISkill":
platform = name[:-7].lower()
self.register_adapter(platform, cls)
def register_adapter(self, platform: str, adapter_class: Type[ISkill]):
"""注册适配器类（非实例）"""
self.adapters[platform] = adapter_class
def get_adapter(self, platform: str, **kwargs) -> ISkill:
"""获取适配器实例"""
if platform not in self.adapters:
raise ValueError(f"Unsupported platform: {platform}")
return self.adapters[platform](**kwargs)
7.2 自适应技能路由
未来的Agent系统可能会实现更智能的技能路由：
python复制class SmartSkillRouter:
"""基于多种因素自动选择最佳技能和平台"""
def __init__(self, skill_registry):
self.skills = skill_registry
self.routing_rules = self._load_routing_rules()
self.performance_metrics = defaultdict(dict)
def execute_skill(self, skill_name: str, context: Dict) -> Dict:
"""执行指定技能"""
# 1. 获取所有可用的技能实现
skill_implementations = self.skills.get_implementations(skill_name)
# 2. 应用路由规则筛选候选
candidates = self._apply_routing_rules(skill_name, context, skill_implementations)
# 3. 基于性能指标选择最佳实现
selected = self._select_best_implementation(candidates)
# 4. 执行并记录性能指标
start_time = time.time()
result = selected.run(context)
duration = time.time() - start_time
self._record_metrics(
skill_name,
selected.platform,
result['success'],
duration,
context
)
return result
def _apply_routing_rules(self, skill_name, context, implementations):
"""应用路由规则筛选候选实现"""
# 实现基于规则、成本、QoS等的筛选逻辑
pass
def _select_best_implementation(self, candidates):
"""基于性能指标选择最佳实现"""
# 实现基于历史性能数据的智能选择
pass
7.3 边缘计算集成
适配器模式也适用于边缘计算场景，实现云端和边缘的无缝切换：
python复制class EdgeCloudAdapter(ISkill):
"""智能切换云端和边缘计算的适配器"""
def __init__(self, cloud_adapter, edge_adapter):
self.cloud = cloud_adapter
self.edge = edge_adapter
self.offline_mode = False
def run(self, context: Dict) -> Dict:
# 根据网络状况、数据敏感性等因素决定使用哪个实现
if self._should_use_edge(context):
try:
return self.edge.run(context)
except Exception as e:
# 边缘计算失败时回退到云端
if not self.offline_mode:
return self.cloud.run(context)
raise
else:
return self.cloud.run(context)
def _should_use_edge(self, context: Dict) -> bool:
"""判断是否应该使用边缘计算"""
return (
self.offline_mode or
context.get('sensitive', False) or
not self._check_cloud_connection()
)
8. 经验总结与实用建议
在实际项目中应用适配器模式开发AI技能时，我总结了以下几点经验：
-
接口设计要前瞻：统一接口应该足够抽象，能够容纳未来可能出现的新平台特性。但也不应过度设计，保持简洁性很重要。
-
错误处理要全面：不同平台的错误类型和恢复策略可能大不相同。适配器应该能够捕获平台特定的异常，并转换为统一的错误表示。
-
性能监控不可少：为每个适配器添加详细的性能指标收集，包括响应时间、成功率、资源消耗等。这些数据对于后续的优化和路由决策至关重要。
-
配置要外部化：平台相关的参数（如API端点、认证信息）应该通过配置文件或环境变量管理，而不是硬编码在适配器中。
-
测试要全面：除了测试正常流程外，特别要重视对平台特定行为和边界条件的测试。可以考虑使用平台提供的模拟服务进行测试。
-
文档要详细：为每个适配器编写详细的文档，说明支持的平台版本、已知限制、特殊配置要求等。这对于团队协作和后期维护非常关键。
-
版本兼容要考虑：平台API可能会升级变化，适配器应该能够处理不同版本的API，或者在检测到不兼容时给出明确的错误提示。
-
资源清理要重视：特别是对于持有网络连接或文件句柄的适配器，要实现明确的资源清理机制，避免资源泄漏。
以下是一个综合了这些经验的适配器模板：
python复制class WellDesignedAdapter(ISkill):
"""良好设计的适配器模板"""
def __init__(self, config: Dict):
"""初始化适配器
Args:
config: 包含所有必要配置的字典
- api_key: 认证密钥
- endpoint: API端点(可选)
- timeout: 超时设置(可选)
"""
self.config = self._validate_config(config)
self.client = self._initialize_client()
self.metrics = {
"total_calls": 0,
"success_calls": 0,
"total_time": 0.0
}
def run(self, context: Dict) -> Dict:
"""执行技能处理"""
start_time = time.time()
self.metrics["total_calls"] += 1
try:
# 预处理输入
processed_input = self._preprocess(context)
# 调用平台API
raw_result = self._call_platform_api(processed_input)
# 后处理结果
standardized_result = self._postprocess(raw_result)
# 更新指标
self.metrics["success_calls"] += 1
self.metrics["total_time"] += time.time() - start_time
return standardized_result
except PlatformSpecificError as e:
# 转换平台特定错误
return {
"success": False,
"error": self._translate_error(e),
"error_code": e.code
}
except Exception as e:
# 捕获其他意外错误
return {
"success": False,
"error": f"Unexpected error: {str(e)}",
"error_code": "UNKNOWN"
}
def shutdown(self):
"""清理资源"""
if hasattr(self.client, 'close'):
self.client.close()
def get_metrics(self) -> Dict:
"""获取性能指标"""
return {
**self.metrics,
"success_rate": self.metrics["success_calls"] / self.metrics["total_calls"] if self.metrics["total_calls"] else 0,
"avg_time": self.metrics["total_time"] / self.metrics["total_calls"] if self.metrics["total_calls"] else 0
}
# 以下是平台特定的实现细节
def _validate_config(self, config: Dict) -> Dict:
"""验证配置有效性"""
if 'api_key' not in config:
raise ValueError("Missing required config: api_key")
return config
def _initialize_client(self):
"""初始化平台客户端"""
# 平台特定的初始化逻辑
pass
def _preprocess(self, context: Dict):
"""预处理输入数据"""
# 转换统一输入为平台所需格式
pass
def _call_platform_api(self, processed_input):
"""调用平台API"""
# 平台特定的调用逻辑
pass
def _postprocess(self, raw_result):
"""后处理平台响应"""
# 转换平台响应为统一格式
pass
def _translate_error(self, error):
"""转换平台错误为标准描述"""
# 平台特定的错误转换逻辑
pass