第18讲：多平台 Skill 适配与迁移

掌握 Skill 在多平台间的适配与迁移技巧，实现"一次开发，多处部署"，最大化 Skill 的复用价值。

一、为什么要做多平台适配

1.1 平台差异带来的挑战

不同 AI 平台有各自的特点和限制：

平台	优势	限制	适用场景
Coze	生态完善，插件丰富	需要科学上网	海外用户，复杂功能
OpenClaw	国内访问稳定	生态相对较新	国内用户，快速上线
飞书 CLI	企业集成度高	依赖飞书生态	企业办公场景

1.2 多平台部署的价值

覆盖更广用户：不同用户习惯使用不同平台
分散风险：避免单一平台政策变化影响
功能互补：利用各平台优势实现最佳效果
品牌曝光：多平台展示增加 Skill 知名度

二、平台差异分析

2.1 核心差异对比

┌─────────────────────────────────────────────────────────────┐
│                    平台差异对比矩阵                          │
├──────────┬──────────┬──────────┬──────────┬────────────────┤
│   维度    │   Coze   │ OpenClaw │ 飞书 CLI │    应对策略     │
├──────────┼──────────┼──────────┼──────────┼────────────────┤
│ 开发语言  │  插件+LLM │ Python   │ Python   │ 核心逻辑抽象   │
│ 部署方式  │  云端     │ 云端/本地 │ 飞书服务器│ 适配层封装     │
│ 数据存储  │  变量     │ 数据库   │ 飞书存储  │ 统一存储接口   │
│ 触发方式  │  对话触发 │ API/定时 │ 命令/事件 │ 统一入口设计   │
│ 权限管理  │  平台控制 │ 自主控制 │ 飞书权限  │ 权限抽象层     │
│ 文件处理  │  插件支持 │ 本地文件 │ 云文档   │ 文件适配器     │
└──────────┴──────────┴──────────┴──────────┴────────────────┘

2.2 功能支持差异

Coze 特有功能：

丰富的官方插件市场
多模态交互（图片、语音）
工作流编排
Knowledge 知识库

OpenClaw 特有功能：

灵活的代码执行环境
本地部署能力
定时任务支持
自定义 API 接口

飞书 CLI 特有功能：

飞书生态深度集成
群聊机器人能力
审批、日程等办公功能
企业权限体系

三、适配架构设计

3.1 分层架构

┌─────────────────────────────────────────────────────────────┐
│                      平台适配层                              │
│  ┌─────────┐  ┌─────────┐  ┌─────────┐                     │
│  │  Coze   │  │OpenClaw │  │飞书 CLI │                     │
│  │ Adapter │  │ Adapter │  │ Adapter │                     │
│  └────┬────┘  └────┬────┘  └────┬────┘                     │
└───────┼────────────┼────────────┼───────────────────────────┘
        │            │            │
        └────────────┼────────────┘
                     │
┌────────────────────┼────────────────────────────────────────┐
│                    │                                         │
│  ┌─────────────────▼──────────────────┐                     │
│  │           业务逻辑层                │                     │
│  │  ┌────────┐ ┌────────┐ ┌────────┐ │                     │
│  │  │Intent  │ │ Action │ │  Tool  │ │                     │
│  │  │识别    │ │执行   │ │ 调用   │ │                     │
│  │  └────────┘ └────────┘ └────────┘ │                     │
│  └────────────────────────────────────┘                     │
│                                                             │
│  ┌────────────────────────────────────┐                     │
│  │           核心能力层                │                     │
│  │  ┌────────┐ ┌────────┐ ┌────────┐ │                     │
│  │  │ Excel  │ │  PDF   │ │  OCR   │ │                     │
│  │  │处理   │ │ 处理   │ │ 识别   │ │                     │
│  │  └────────┘ └────────┘ └────────┘ │                     │
│  └────────────────────────────────────┘                     │
│                                                             │
│  ┌────────────────────────────────────┐                     │
│  │           基础设施层                │                     │
│  │  ┌────────┐ ┌────────┐ ┌────────┐ │                     │
│  │  │ 存储   │ │ 日志   │ │ 配置   │ │                     │
│  │  └────────┘ └────────┘ └────────┘ │                     │
│  └────────────────────────────────────┘                     │
└─────────────────────────────────────────────────────────────┘

3.2 核心代码抽象

抽象基类设计：

from abc import ABC, abstractmethod
from typing import Dict, Any, List

class SkillAdapter(ABC):
    """Skill 适配器基类"""
    
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.skill_core = SkillCore()  # 核心业务逻辑
    
    @abstractmethod
    def handle_message(self, message: str, context: Dict) -> str:
        """处理用户消息"""
        pass
    
    @abstractmethod
    def handle_file(self, file_path: str, context: Dict) -> str:
        """处理文件上传"""
        pass
    
    @abstractmethod
    def send_response(self, response: str, attachments: List[str] = None):
        """发送响应"""
        pass

class SkillCore:
    """Skill 核心业务逻辑（平台无关）"""
    
    def __init__(self):
        self.intent_classifier = IntentClassifier()
        self.action_executor = ActionExecutor()
        self.tool_manager = ToolManager()
    
    def process(self, user_input: str, context: Dict) -> Dict:
        """处理用户输入"""
        # 1. 意图识别
        intent = self.intent_classifier.classify(user_input)
        
        # 2. 参数提取
        params = self.extract_params(user_input, intent)
        
        # 3. 执行动作
        result = self.action_executor.execute(intent, params, context)
        
        return {
            'intent': intent,
            'result': result,
            'response': self.generate_response(result)
        }
    
    def extract_params(self, text: str, intent: str) -> Dict:
        """提取参数"""
        # 使用 LLM 或规则提取参数
        pass
    
    def generate_response(self, result: Dict) -> str:
        """生成响应"""
        # 根据结果生成自然语言响应
        pass

四、各平台适配实现

4.1 Coze 适配器

class CozeAdapter(SkillAdapter):
    """Coze 平台适配器"""
    
    def __init__(self, config: Dict):
        super().__init__(config)
        self.bot_id = config['bot_id']
        self.api_token = config['api_token']
    
    def handle_message(self, message: str, context: Dict) -> str:
        """处理 Coze 消息"""
        # 调用核心业务逻辑
        result = self.skill_core.process(message, context)
        
        # 适配 Coze 响应格式
        return self.format_coze_response(result)
    
    def handle_file(self, file_url: str, context: Dict) -> str:
        """处理 Coze 文件"""
        # 下载文件
        local_path = self.download_file(file_url)
        
        # 调用文件处理逻辑
        result = self.skill_core.process_file(local_path, context)
        
        return self.format_coze_response(result)
    
    def send_response(self, response: str, attachments: List[str] = None):
        """发送 Coze 响应"""
        # Coze 通过工作流返回，这里主要是格式化
        return {
            'text': response,
            'files': attachments or []
        }
    
    def format_coze_response(self, result: Dict) -> str:
        """格式化为 Coze 响应"""
        # Coze 支持 Markdown 格式
        return result['response']
    
    def download_file(self, file_url: str) -> str:
        """下载 Coze 文件"""
        import requests
        
        response = requests.get(file_url, headers={
            'Authorization': f'Bearer {self.api_token}'
        })
        
        local_path = f"/tmp/{file_url.split('/')[-1]}"
        with open(local_path, 'wb') as f:
            f.write(response.content)
        
        return local_path

4.2 OpenClaw 适配器

from openclaw import Skill, Tool

class OpenClawAdapter(SkillAdapter):
    """OpenClaw 平台适配器"""
    
    def __init__(self, config: Dict):
        super().__init__(config)
        self.skill = self._create_skill()
    
    def _create_skill(self) -> Skill:
        """创建 OpenClaw Skill"""
        
        class MySkill(Skill):
            name = self.config.get('name', 'MySkill')
            description = self.config.get('description', '')
            
            @Tool
            def chat(self, message: str) -> str:
                """对话接口"""
                return self.handle_message(message, {})
            
            @Tool
            def process_file(self, file_path: str) -> str:
                """文件处理接口"""
                return self.handle_file(file_path, {})
        
        return MySkill()
    
    def handle_message(self, message: str, context: Dict) -> str:
        """处理 OpenClaw 消息"""
        result = self.skill_core.process(message, context)
        return result['response']
    
    def handle_file(self, file_path: str, context: Dict) -> str:
        """处理 OpenClaw 文件"""
        result = self.skill_core.process_file(file_path, context)
        return result['response']
    
    def send_response(self, response: str, attachments: List[str] = None):
        """OpenClaw 通过 Tool 返回"""
        return response

4.3 飞书 CLI 适配器

from lark_cli import Skill, Message

class FeishuAdapter(SkillAdapter):
    """飞书 CLI 适配器"""
    
    def __init__(self, config: Dict):
        super().__init__(config)
        self.app_id = config['app_id']
        self.app_secret = config['app_secret']
    
    def handle_message(self, message: Message, context: Dict) -> str:
        """处理飞书消息"""
        # 提取文本内容
        text = message.text
        
        # 添加飞书特有的上下文
        context['user_id'] = message.user_id
        context['chat_id'] = message.chat_id
        context['message_id'] = message.message_id
        
        result = self.skill_core.process(text, context)
        
        return self.format_feishu_response(result)
    
    def handle_file(self, file_key: str, context: Dict) -> str:
        """处理飞书文件"""
        # 下载飞书文件
        local_path = self.download_feishu_file(file_key)
        
        result = self.skill_core.process_file(local_path, context)
        
        return self.format_feishu_response(result)
    
    def send_response(self, response: str, attachments: List[str] = None):
        """发送飞书响应"""
        # 飞书支持富文本、卡片等多种格式
        return {
            'msg_type': 'text',
            'content': {
                'text': response
            }
        }
    
    def format_feishu_response(self, result: Dict) -> Dict:
        """格式化为飞书响应"""
        # 可以使用飞书卡片格式增强展示
        return {
            'msg_type': 'interactive',
            'card': {
                'header': {
                    'title': {
                        'tag': 'plain_text',
                        'content': '处理结果'
                    }
                },
                'elements': [
                    {
                        'tag': 'div',
                        'text': {
                            'tag': 'lark_md',
                            'content': result['response']
                        }
                    }
                ]
            }
        }
    
    def download_feishu_file(self, file_key: str) -> str:
        """下载飞书文件"""
        # 使用飞书 API 下载文件
        pass

五、数据持久化适配

5.1 统一存储接口

class StorageAdapter(ABC):
    """存储适配器基类"""
    
    @abstractmethod
    def save(self, key: str, data: Any):
        """保存数据"""
        pass
    
    @abstractmethod
    def load(self, key: str) -> Any:
        """加载数据"""
        pass
    
    @abstractmethod
    def delete(self, key: str):
        """删除数据"""
        pass

class CozeStorage(StorageAdapter):
    """Coze 变量存储"""
    
    def __init__(self):
        self.variables = {}
    
    def save(self, key: str, data: Any):
        """Coze 通过工作流变量存储"""
        import json
        self.variables[key] = json.dumps(data)
    
    def load(self, key: str) -> Any:
        import json
        data = self.variables.get(key)
        return json.loads(data) if data else None
    
    def delete(self, key: str):
        if key in self.variables:
            del self.variables[key]

class OpenClawStorage(StorageAdapter):
    """OpenClaw 数据库存储"""
    
    def __init__(self, db_path: str):
        import sqlite3
        self.conn = sqlite3.connect(db_path)
        self._init_table()
    
    def _init_table(self):
        self.conn.execute('''
            CREATE TABLE IF NOT EXISTS skill_data (
                key TEXT PRIMARY KEY,
                value TEXT,
                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        self.conn.commit()
    
    def save(self, key: str, data: Any):
        import json
        value = json.dumps(data)
        self.conn.execute(
            'REPLACE INTO skill_data (key, value) VALUES (?, ?)',
            (key, value)
        )
        self.conn.commit()
    
    def load(self, key: str) -> Any:
        import json
        cursor = self.conn.execute(
            'SELECT value FROM skill_data WHERE key = ?',
            (key,)
        )
        row = cursor.fetchone()
        return json.loads(row[0]) if row else None
    
    def delete(self, key: str):
        self.conn.execute('DELETE FROM skill_data WHERE key = ?', (key,))
        self.conn.commit()

class FeishuStorage(StorageAdapter):
    """飞书云文档/数据存储"""
    
    def __init__(self, app_id: str, app_secret: str):
        self.app_id = app_id
        self.app_secret = app_secret
        # 初始化飞书客户端
    
    def save(self, key: str, data: Any):
        # 保存到飞书云文档或数据存储
        pass
    
    def load(self, key: str) -> Any:
        # 从飞书加载数据
        pass
    
    def delete(self, key: str):
        pass

六、迁移实战

6.1 迁移检查清单

□ 功能兼容性检查
  □ 核心功能在新平台是否支持
  □ 依赖的插件/API 是否可用
  □ 文件处理方式是否需要调整

□ 数据迁移
  □ 用户数据导出
  □ 配置信息迁移
  □ 历史记录处理

□ 测试验证
  □ 单元测试通过
  □ 集成测试通过
  □ 用户验收测试

□ 部署上线
  □ 新平台 Skill 创建
  □ 配置参数设置
  □ 监控告警配置

6.2 迁移示例

从 Coze 迁移到 OpenClaw：

# 1. 导出 Coze 配置
coze_config = {
    'bot_id': 'xxx',
    'prompt': '...',
    'plugins': ['plugin1', 'plugin2']
}

# 2. 转换配置
openclaw_config = {
    'name': 'Excel智能助手',
    'description': '自动化处理 Excel 文件',
    'python_dependencies': [
        'pandas',
        'openpyxl',
        'xlrd'
    ]
}

# 3. 重写代码逻辑
# Coze 版本（使用插件）
# → OpenClaw 版本（使用 Python 代码）

# 4. 测试对比
# 确保输出结果一致