记一次 OpenClaw 模型切换过程中的 API 兼容性问题排查

# adapter.py - API 适配层示例

class ModelAdapter:
    def __init__(self, provider, api_key, config):
        self.provider = provider
        self.api_key = api_key
        self.config = config
    
    def format_request(self, messages):
        """根据不同供应商格式化请求"""
        if self.provider == "minimax":
            return self._format_minimax_request(messages)
        elif self.provider == "other":
            return self._format_other_request(messages)
        else:
            raise ValueError(f"Unknown provider: {self.provider}")
    
    def parse_response(self, response):
        """根据不同供应商解析响应"""
        if self.provider == "minimax":
            return self._parse_minimax_response(response)
        elif self.provider == "other":
            return self._parse_other_response(response)
        else:
            raise ValueError(f"Unknown provider: {self.provider}")
    
    def _format_minimax_request(self, messages):
        # MiniMax 特定格式
        return {
            "model": self.config.get("model", "abab6.5s-chat"),
            "messages": messages
        }
    
    def _format_other_request(self, messages):
        # 其他供应商格式
        return {
            "model": self.config.get("model", "chat-model"),
            "prompt": messages[0]["content"] if messages else ""
        }

# 完整的适配器实现

class OpenClawModelAdapter:
    """OpenClaw 模型适配器"""
    
    SUPPORTED_PROVIDERS = ["minimax", "openai", "anthropic", "custom"]
    
    def __init__(self, provider_config):
        self.config = provider_config
        self.provider = provider_config.get("provider", "minimax")
        self._init_client()
    
    def _init_client(self):
        """初始化对应的客户端"""
        if self.provider == "minimax":
            self.client = MiniMaxClient(self.config)
        elif self.provider == "openai":
            self.client = OpenAIClient(self.config)
        else:
            raise ValueError(f"Unsupported provider: {self.provider}")
    
    def chat(self, messages, **kwargs):
        """统一的聊天接口"""
        # 统一的错误处理
        try:
            # 统一的请求处理
            request_data = self._prepare_request(messages, **kwargs)
            
            # 调用实际的客户端
            response = self.client.chat(request_data)
            
            # 统一的响应处理
            return self._parse_response(response)
        except Exception as e:
            # 统一的错误处理
            logger.error(f"Chat error: {e}")
            raise
    
    def _prepare_request(self, messages, **kwargs):
        """准备请求数据"""
        return {
            "messages": messages,
            "temperature": kwargs.get("temperature", 0.7),
            "max_tokens": kwargs.get("max_tokens", 2048),
            **kwargs
        }
    
    def _parse_response(self, response):
        """解析响应数据"""
        # 统一提取 content
        content = response.get("choices", [{}])[0].get("message", {}).get("content", "")
        
        return {
            "content": content,
            "usage": response.get("usage", {}),
            "model": response.get("model", "")
        }

# 渐进式切换示例

class GradualSwitcher:
    def __init__(self, primary_provider, backup_provider, switch_ratio=0.1):
        self.primary = primary_provider
        self.backup = backup_provider
        self.switch_ratio = switch_ratio
        self._init_counters()
    
    def _init_counters(self):
        # 统计切换次数
        self.primary_count = 0
        self.backup_count = 0
    
    def select_provider(self):
        """根据比例选择供应商"""
        import random
        if random.random() < self.switch_ratio:
            self.backup_count += 1
            return self.backup
        else:
            self.primary_count += 1
            return self.primary
    
    def get_stats(self):
        """获取切换统计"""
        return {
            "primary": self.primary_count,
            "backup": self.backup_count,
            "ratio": self.backup_count / (self.primary_count + self.backup_count)
        }