MCP Client 与云端模型混合使用

import asyncio
import random
from typing import Dict, Any, Optional, List
from dataclasses import dataclass, field

@dataclass
class ModelInfo:
    """模型信息"""
    name: str
    type: str  # local 或 cloud
    provider: str  # 模型提供商，如 ollama, openai, aws
    cost_per_token: float = 0.0  # 每 token 成本
    max_tokens: int = 4096  # 最大 token 数
    availability: float = 1.0  # 可用性，0-1
    priority: int = 0  # 优先级，值越高优先级越高

@dataclass
class ModelLoad:
    """模型负载信息"""
    model_name: str
    cpu_usage: float = 0.0  # CPU 使用率，0-1
    memory_usage: float = 0.0  # 内存使用率，0-1
    gpu_usage: float = 0.0  # GPU 使用率，0-1
    request_count: int = 0  # 当前请求数
    response_time: float = 0.0  # 平均响应时间，秒

class ModelRouter:
    def __init__(self):
        self.models: Dict[str, ModelInfo] = {}
        self.model_loads: Dict[str, ModelLoad] = {}
        self.route_config: Dict[str, Any] = {
            "default_strategy": "cost_priority",  # cost_priority, performance_priority, availability_priority
            "local_first": True,  # 是否优先使用本地模型
            "max_local_load": 0.8,  # 本地模型最大负载阈值
            "cost_threshold": 1.0,  # 成本阈值，超过则切换到更便宜的模型
            "response_time_threshold": 2.0,  # 响应时间阈值，秒
        }
    
    def register_model(self, model_info: ModelInfo):
        """注册模型"""
        self.models[model_info.name] = model_info
        self.model_loads[model_info.name] = ModelLoad(model_name=model_info.name)
    
    def update_model_load(self, model_name: str, load: ModelLoad):
        """更新模型负载信息"""
        if model_name in self.model_loads:
            self.model_loads[model_name] = load
    
    def update_route_config(self, config: Dict[str, Any]):
        """更新路由配置"""
        self.route_config.update(config)
    
    def calculate_model_score(self, model_name: str, task_type: str) -> float:
        """计算模型得分"""
        model = self.models.get(model_name)
        load = self.model_loads.get(model_name)
        
        if not model or not load:
            return 0.0
        
        # 基础得分
        score = 0.0
        
        # 1. 可用性得分（30%）
        availability_score = model.availability * 0.3
        
        # 2. 性能得分（30%）
        # 负载越低，性能得分越高
        load_score = (1.0 - max(load.cpu_usage, load.memory_usage, load.gpu_usage)) * 0.3
        
        # 3. 成本得分（20%）
        # 成本越低，得分越高
        # 归一化成本，假设最高成本为 0.01 美元/1K tokens
        cost_score = (1.0 - min(model.cost_per_token / 0.01, 1.0)) * 0.2
        
        # 4. 优先级得分（10%）
        priority_score = (model.priority / 10.0) * 0.1
        
        # 5. 任务匹配得分（10%）
        # 简单示例：根据任务类型调整得分
        task_score = 0.1
        if task_type == "coding" and "code" in model.name.lower():
            task_score = 0.2  # 编码任务优先选择代码模型
        elif task_type == "chat" and "chat" in model.name.lower():
            task_score = 0.2  # 聊天任务优先选择聊天模型
        
        # 计算总得分
        total_score = availability_score + load_score + cost_score + priority_score + task_score
        
        # 考虑本地模型优先
        if self.route_config.get("local_first") and model.type == "local":
            total_score *= 1.2  # 本地模型加分
        
        return total_score
    
    def select_best_model(self, task_type: str = "general") -> str:
        """选择最佳模型"""
        # 1. 获取所有可用模型
        available_models = [name for name, model in self.models.items() if model.availability > 0.8]
        
        if not available_models:
            raise ValueError("没有可用的模型")
        
        # 2. 根据策略选择模型
        strategy = self.route_config.get("default_strategy")
        
        if strategy == "cost_priority":
            # 成本优先策略：选择成本最低的模型
            return min(available_models, key=lambda x: self.models[x].cost_per_token)
        elif strategy == "performance_priority":
            # 性能优先策略：选择负载最低的模型
            return min(available_models, key=lambda x: max(self.model_loads[x].cpu_usage, self.model_loads[x].memory_usage, self.model_loads[x].gpu_usage))
        elif strategy == "availability_priority":
            # 可用性优先策略：选择可用性最高的模型
            return max(available_models, key=lambda x: self.models[x].availability)
        else:
            # 默认：综合得分策略
            return max(available_models, key=lambda x: self.calculate_model_score(x, task_type))
    
    async def route_request(self, prompt: str, task_type: str = "general", **kwargs) -> Dict[str, Any]:
        """路由请求到最佳模型"""
        # 1. 选择最佳模型
        best_model_name = self.select_best_model(task_type)
        best_model = self.models[best_model_name]
        
        print(f"路由请求到模型: {best_model_name} (类型: {best_model.type}, 提供商: {best_model.provider})")
        
        # 2. 模拟调用模型（实际实现中会调用对应的模型连接器）
        # 这里只是示例，实际实现中会根据模型类型调用不同的连接器
        result = {
            "model": best_model_name,
            "type": best_model.type,
            "provider": best_model.provider,
            "prompt": prompt,
            "response": f"这是模型 {best_model_name} 对 prompt '{prompt}' 的响应",
            "task_type": task_type,
            "**kwargs": kwargs
        }
        
        # 3. 更新模型负载（实际实现中会根据真实负载更新）
        # 这里只是模拟，实际实现中会从模型连接器获取真实负载
        load = self.model_loads[best_model_name]
        load.request_count += 1
        load.response_time = random.uniform(0.5, 3.0)  # 模拟响应时间
        
        return result

import asyncio
import httpx
import json
from typing import Dict, Any, Optional, List

class CloudModelConnector:
    def __init__(self):
        self.providers = {
            "openai": {
                "base_url": "https://api.openai.com/v1",
                "api_key": None,
                "models": ["gpt-4o", "gpt-4-turbo", "gpt-3.5-turbo"]
            },
            "aws": {
                "base_url": "https://bedrock-runtime.us-east-1.amazonaws.com",
                "aws_access_key": None,
                "aws_secret_key": None,
                "models": ["anthropic.claude-3-opus-20240229-v1:0", "anthropic.claude-3-sonnet-20240229-v1:0"]
            },
            "azure": {
                "base_url": "https://{resource_name}.openai.azure.com/openai/deployments/{deployment_name}/chat/completions?api-version=2024-02-15-preview",
                "api_key": None,
                "resource_name": None,
                "deployment_name": None
            },
            "google": {
                "base_url": "https://generativelanguage.googleapis.com/v1/models/{model}:generateContent",
                "api_key": None,
                "models": ["gemini-1.5-pro", "gemini-1.5-flash"]
            }
        }
    
    def configure_provider(self, provider_name: str, config: Dict[str, Any]):
        """配置云端模型提供商"""
        if provider_name in self.providers:
            self.providers[provider_name].update(config)
    
    async def generate_openai(self, prompt: str, model: str, **kwargs) -> Dict[str, Any]:
        """调用 OpenAI API 生成响应"""
        provider_config = self.providers["openai"]
        
        if not provider_config["api_key"]:
            raise ValueError("OpenAI API 密钥未配置")
        
        headers = {
            "Authorization": f"Bearer {provider_config['api_key']}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": model,
            "messages": [
                {"role": "system", "content": "You are a helpful assistant."},
                {"role": "user", "content": prompt}
            ],
            "stream": False,
            **kwargs
        }
        
        async with httpx.AsyncClient(timeout=30.0) as client:
            response = await client.post(
                f"{provider_config['base_url']}/chat/completions",
                headers=headers,
                json=payload
            )
            response.raise_for_status()
            return response.json()
    
    async def generate_aws_bedrock(self, prompt: str, model: str, **kwargs) -> Dict[str, Any]:
        """调用 AWS Bedrock API 生成响应"""
        provider_config = self.providers["aws"]
        
        if not provider_config["aws_access_key"] or not provider_config["aws_secret_key"]:
            raise ValueError("AWS 访问密钥未配置")
        
        # 简化示例，实际实现需要使用 AWS SDK 进行签名
        # 这里使用模拟响应
        return {
            "model": model,
            "prompt": prompt,
            "response": f"这是 AWS Bedrock 模型 {model} 对 prompt '{prompt}' 的响应",
            "provider": "aws",
            "**kwargs": kwargs
        }
    
    async def generate_azure_openai(self, prompt: str, **kwargs) -> Dict[str, Any]:
        """调用 Azure OpenAI API 生成响应"""
        provider_config = self.providers["azure"]
        
        if not provider_config["api_key"] or not provider_config["resource_name"] or not provider_config["deployment_name"]:
            raise ValueError("Azure OpenAI 配置不完整")
        
        base_url = provider_config["base_url"].format(
            resource_name=provider_config["resource_name"],
            deployment_name=provider_config["deployment_name"]
        )
        
        headers = {
            "api-key": provider_config["api_key"],
            "Content-Type": "application/json"
        }
        
        payload = {
            "messages": [
                {"role": "system", "content": "You are a helpful assistant."},
                {"role": "user", "content": prompt}
            ],
            "stream": False,
            **kwargs
        }
        
        async with httpx.AsyncClient(timeout=30.0) as client:
            response = await client.post(
                base_url,
                headers=headers,
                json=payload
            )
            response.raise_for_status()
            return response.json()
    
    async def generate_google_gemini(self, prompt: str, model: str, **kwargs) -> Dict[str, Any]:
        """调用 Google Gemini API 生成响应"""
        provider_config = self.providers["google"]
        
        if not provider_config["api_key"]:
            raise ValueError("Google API 密钥未配置")
        
        base_url = provider_config["base_url"].format(model=model)
        
        payload = {
            "contents": [
                {
                    "parts": [
                        {"text": prompt}
                    ]
                }
            ],
            **kwargs
        }
        
        async with httpx.AsyncClient(timeout=30.0) as client:
            response = await client.post(
                f"{base_url}&key={provider_config['api_key']}",
                json=payload
            )
            response.raise_for_status()
            return response.json()
    
    async def generate(self, provider: str, prompt: str, model: Optional[str] = None, **kwargs) -> Dict[str, Any]:
        """通用生成方法"""
        if provider == "openai":
            if not model:
                model = "gpt-4o"
            return await self.generate_openai(prompt, model, **kwargs)
        elif provider == "aws":
            if not model:
                model = "anthropic.claude-3-opus-20240229-v1:0"
            return await self.generate_aws_bedrock(prompt, model, **kwargs)
        elif provider == "azure":
            return await self.generate_azure_openai(prompt, **kwargs)
        elif provider == "google":
            if not model:
                model = "gemini-1.5-pro"
            return await self.generate_google_gemini(prompt, model, **kwargs)
        else:
            raise ValueError(f"不支持的云端模型提供商: {provider}")

import asyncio
from typing import Dict, Any, Optional, List
from model_router import ModelRouter, ModelInfo
from cloud_model_connector import CloudModelConnector
from ollama_connector import OllamaConnector  # 假设已经实现了 Ollama 连接器

class HybridDeploymentManager:
    def __init__(self):
        # 1. 初始化模型路由器
        self.model_router = ModelRouter()
        
        # 2. 初始化云端模型连接器
        self.cloud_connector = CloudModelConnector()
        
        # 3. 初始化本地模型连接器
        self.local_connector = OllamaConnector()
        
        # 4. 配置信息
        self.config = {
            "default_task_type": "general",
            "auto_scaling": True,  # 是否自动扩展模型使用
            "scaling_threshold": 0.8,  # 扩展阈值
            "cloud_fallback_enabled": True,  # 是否启用云端回退
            "fallback_delay": 5.0,  # 回退延迟，秒
        }
        
        # 5. 状态信息
        self.state = {
            "is_cloud_fallback_active": False,
            "fallback_start_time": None,
            "total_requests": 0,
            "local_requests": 0,
            "cloud_requests": 0,
            "fallback_requests": 0,
        }
    
    def configure(self, config: Dict[str, Any]):
        """配置混合部署管理器"""
        self.config.update(config)
    
    async def register_models(self):
        """注册默认模型"""
        # 1. 注册本地模型
        local_models = await self.local_connector.list_models()
        for model in local_models:
            model_info = ModelInfo(
                name=model["name"],
                type="local",
                provider="ollama",
                cost_per_token=0.0,  # 本地模型成本为 0
                max_tokens=model.get("context_length", 4096),
                availability=1.0,
                priority=10,
            )
            self.model_router.register_model(model_info)
        
        # 2. 注册云端模型
        cloud_models = [
            ModelInfo(
                name="gpt-4o",
                type="cloud",
                provider="openai",
                cost_per_token=0.01,  # 0.01 美元/1K tokens
                max_tokens=128000,
                availability=0.99,
                priority=8,
            ),
            ModelInfo(
                name="gpt-3.5-turbo",
                type="cloud",
                provider="openai",
                cost_per_token=0.0015,  # 0.0015 美元/1K tokens
                max_tokens=16385,
                availability=0.995,
                priority=7,
            ),
            ModelInfo(
                name="claude-3-opus",
                type="cloud",
                provider="aws",
                cost_per_token=0.015,  # 0.015 美元/1K tokens
                max_tokens=200000,
                availability=0.99,
                priority=8,
            ),
            ModelInfo(
                name="claude-3-sonnet",
                type="cloud",
                provider="aws",
                cost_per_token=0.003,  # 0.003 美元/1K tokens
                max_tokens=200000,
                availability=0.995,
                priority=7,
            ),
            ModelInfo(
                name="gemini-1.5-pro",
                type="cloud",
                provider="google",
                cost_per_token=0.0075,  # 0.0075 美元/1K tokens
                max_tokens=1048576,
                availability=0.99,
                priority=8,
            ),
            ModelInfo(
                name="gemini-1.5-flash",
                type="cloud",
                provider="google",
                cost_per_token=0.00125,  # 0.00125 美元/1K tokens
                max_tokens=1048576,
                availability=0.995,
                priority=7,
            ),
        ]
        
        for model_info in cloud_models:
            self.model_router.register_model(model_info)
    
    async def generate(self, prompt: str, task_type: Optional[str] = None, **kwargs) -> Dict[str, Any]:
        """生成响应"""
        if not task_type:
            task_type = self.config["default_task_type"]
        
        self.state["total_requests"] += 1
        
        try:
            # 1. 使用模型路由器选择最佳模型
            route_result = await self.model_router.route_request(prompt, task_type, **kwargs)
            
            model_name = route_result["model"]
            model_type = route_result["type"]
            provider = route_result["provider"]
            
            # 2. 根据模型类型调用相应的连接器
            if model_type == "local":
                # 调用本地模型
                self.state["local_requests"] += 1
                result = await self.local_connector.generate(
                    prompt=prompt,
                    model=model_name,
                    **kwargs
                )
            else:
                # 调用云端模型
                self.state["cloud_requests"] += 1
                result = await self.cloud_connector.generate(
                    provider=provider,
                    prompt=prompt,
                    model=model_name,
                    **kwargs
                )
            
            # 3. 如果是回退状态，且本地模型恢复正常，则退出回退状态
            if self.state["is_cloud_fallback_active"]:
                # 检查本地模型负载是否恢复正常
                all_local_loads = [
                    load for name, load in self.model_router.model_loads.items() 
                    if self.model_router.models[name].type == "local"
                ]
                
                avg_load = sum(
                    max(load.cpu_usage, load.memory_usage, load.gpu_usage) 
                    for load in all_local_loads
                ) / len(all_local_loads) if all_local_loads else 0.0
                
                if avg_load < self.model_router.route_config["max_local_load"]:
                    self.state["is_cloud_fallback_active"] = False
                    self.state["fallback_start_time"] = None
                    print("退出云端回退状态，恢复正常模型选择")
            
            return {
                **result,
                "model_name": model_name,
                "model_type": model_type,
                "provider": provider,
                "task_type": task_type,
                "is_fallback": self.state["is_cloud_fallback_active"],
            }
        
        except Exception as e:
            print(f"模型调用失败: {e}")
            
            # 4. 处理失败情况，启用云端回退
            if self.config["cloud_fallback_enabled"] and not self.state["is_cloud_fallback_active"]:
                print("启用云端回退")
                self.state["is_cloud_fallback_active"] = True
                self.state["fallback_start_time"] = asyncio.get_event_loop().time()
                
                # 直接调用云端模型，不经过路由器
                self.state["cloud_requests"] += 1
                self.state["fallback_requests"] += 1
                
                # 使用默认云端模型
                result = await self.cloud_connector.generate(
                    provider="openai",
                    prompt=prompt,
                    model="gpt-3.5-turbo",
                    **kwargs
                )
                
                return {
                    **result,
                    "model_name": "gpt-3.5-turbo",
                    "model_type": "cloud",
                    "provider": "openai",
                    "task_type": task_type,
                    "is_fallback": True,
                    "error": str(e),
                }
            
            # 如果已经在回退状态，或者不允许回退，则抛出错误
            raise
    
    def get_state(self) -> Dict[str, Any]:
        """获取当前状态"""
        return self.state.copy()
    
    def get_model_stats(self) -> Dict[str, Any]:
        """获取模型统计信息"""
        stats = {}
        for model_name, model in self.model_router.models.items():
            load = self.model_router.model_loads[model_name]
            stats[model_name] = {
                "model": model.__dict__,
                "load": load.__dict__,
                "score": self.model_router.calculate_model_score(model_name, self.config["default_task_type"]),
            }
        return stats
    
    async def run_health_check(self):
        """运行健康检查"""
        while True:
            try:
                # 1. 检查本地模型健康状态
                await self.local_connector.list_models()
                
                # 2. 检查云端模型健康状态
                # 简化示例，实际实现需要检查每个云端模型提供商
                
                # 3. 更新模型可用性
                # 这里只是模拟，实际实现需要根据真实健康状态更新
                for model_name, model in self.model_router.models.items():
                    if model.type == "local":
                        # 本地模型可用性设为 1.0
                        model.availability = 1.0
                    else:
                        # 云端模型可用性设为 0.99-0.995
                        import random
                        model.availability = random.uniform(0.99, 0.995)
                
                print("健康检查完成")
            except Exception as e:
                print(f"健康检查失败: {e}")
            
            # 等待下一次健康检查
            await asyncio.sleep(30.0)  # 每 30 秒运行一次健康检查

# 示例：混合部署管理器的使用
import asyncio
import json
from hybrid_deployment_manager import HybridDeploymentManager

async def main():
    # 1. 创建混合部署管理器
    manager = HybridDeploymentManager()
    
    # 2. 配置混合部署管理器
    manager.configure({
        "default_task_type": "general",
        "auto_scaling": True,
        "cloud_fallback_enabled": True,
        "fallback_delay": 5.0,
    })
    
    # 3. 注册模型
    await manager.register_models()
    
    # 4. 启动健康检查任务
    asyncio.create_task(manager.run_health_check())
    
    # 5. 示例 1：查看注册的模型
    print("=== 注册的模型 ===")
    model_stats = manager.get_model_stats()
    for model_name, stats in model_stats.items():
        print(f"- {model_name} (类型: {stats['model']['type']}, 提供商: {stats['model']['provider']}, 得分: {stats['score']:.2f})")
    
    # 6. 示例 2：生成响应
    print("\n=== 生成响应示例 ===")
    prompts = [
        "你好，能帮我介绍一下 MCP 协议吗？",
        "请写一个 Python 函数来计算斐波那契数列。",
        "解释一下量子计算的基本原理。",
        "请生成一个包含 10 个随机数的列表。",
        "请翻译 'Hello, world!' 到中文。",
    ]
    
    task_types = ["general", "coding", "general", "coding", "translation"]
    
    for i, (prompt, task_type) in enumerate(zip(prompts, task_types)):
        print(f"\n--- 请求 {i+1} ---")
        print(f"Prompt: {prompt}")
        print(f"Task Type: {task_type}")
        
        try:
            result = await manager.generate(
                prompt=prompt,
                task_type=task_type,
                max_tokens=500,
            )
            
            print(f"模型名称: {result.get('model_name', '未知')}")
            print(f"模型类型: {result.get('model_type', '未知')}")
            print(f"提供商: {result.get('provider', '未知')}")
            print(f"是否回退: {result.get('is_fallback', False)}")
            
            # 打印响应内容（根据不同模型提供商的响应格式调整）
            if result.get('model_type') == 'local':
                # 本地模型响应格式
                response_content = result.get('choices', [{}])[0].get('message', {}).get('content', '无响应')
            else:
                # 云端模型响应格式
                if 'choices' in result:
                    # OpenAI 格式
                    response_content = result['choices'][0]['message']['content']
                elif 'response' in result:
                    # 自定义格式
                    response_content = result['response']
                else:
                    response_content = str(result)
            
            print(f"响应: {response_content[:100]}...")
            
        except Exception as e:
            print(f"生成失败: {e}")
    
    # 7. 示例 3：查看状态信息
    print("\n=== 状态信息 ===")
    state = manager.get_state()
    print(json.dumps(state, indent=2, ensure_ascii=False))
    
    # 8. 示例 4：查看详细模型统计信息
    print("\n=== 详细模型统计信息 ===")
    model_stats = manager.get_model_stats()
    for model_name, stats in model_stats.items():
        print(f"\n- {model_name}:")
        print(f"  类型: {stats['model']['type']}")
        print(f"  提供商: {stats['model']['provider']}")
        print(f"  成本: {stats['model']['cost_per_token']} 美元/1K tokens")
        print(f"  最大 tokens: {stats['model']['max_tokens']}")
        print(f"  可用性: {stats['model']['availability']:.4f}")
        print(f"  得分: {stats['score']:.4f}")
        print(f"  CPU 使用率: {stats['load']['cpu_usage']:.4f}")
        print(f"  内存使用率: {stats['load']['memory_usage']:.4f}")
        print(f"  GPU 使用率: {stats['load']['gpu_usage']:.4f}")
        print(f"  请求数: {stats['load']['request_count']}")
        print(f"  平均响应时间: {stats['load']['response_time']:.4f} 秒")
    
    print("\n混合部署示例完成！")

if __name__ == "__main__":
    asyncio.run(main())

# MCP Client 与云端模型混合使用配置

hybrid_deployment:
  # 基本配置
  basic:
    default_task_type: "general"
    auto_scaling: true
    cloud_fallback_enabled: true
    fallback_delay: 5.0
    
  # 模型路由器配置
  model_router:
    default_strategy: "cost_priority"
    local_first: true
    max_local_load: 0.8
    cost_threshold: 1.0
    response_time_threshold: 2.0
    
  # 本地模型配置
  local_models:
    ollama:
      base_url: "http://localhost:11434/v1"
      timeout: 30.0
      retry_count: 3
      retry_delay: 1.0
    
  # 云端模型提供商配置
  cloud_providers:
    openai:
      base_url: "https://api.openai.com/v1"
      api_key: "${OPENAI_API_KEY}"
      models:
        - name: "gpt-4o"
          cost_per_token: 0.01
          priority: 8
        - name: "gpt-3.5-turbo"
          cost_per_token: 0.0015
          priority: 7
    
    aws:
      base_url: "https://bedrock-runtime.us-east-1.amazonaws.com"
      aws_access_key: "${AWS_ACCESS_KEY_ID}"
      aws_secret_key: "${AWS_SECRET_ACCESS_KEY}"
      models:
        - name: "anthropic.claude-3-opus-20240229-v1:0"
          cost_per_token: 0.015
          priority: 8
        - name: "anthropic.claude-3-sonnet-20240229-v1:0"
          cost_per_token: 0.003
          priority: 7
    
    azure:
      resource_name: "${AZURE_OPENAI_RESOURCE_NAME}"
      deployment_name: "${AZURE_OPENAI_DEPLOYMENT_NAME}"
      api_key: "${AZURE_OPENAI_API_KEY}"
      api_version: "2024-02-15-preview"
    
    google:
      base_url: "https://generativelanguage.googleapis.com/v1/models"
      api_key: "${GOOGLE_API_KEY}"
      models:
        - name: "gemini-1.5-pro"
          cost_per_token: 0.0075
          priority: 8
        - name: "gemini-1.5-flash"
          cost_per_token: 0.00125
          priority: 7
  
  # 监控配置
  monitoring:
    enabled: true
    metrics_port: 9090
    logging_level: "INFO"
    include_details: true
    alert_enabled: true
    alert_thresholds:
      avg_response_time: 2.0
      error_rate: 0.05
      cost_per_hour: 10.0
  
  # 成本配置
  cost:
    enabled: true
    budget_per_day: 100.0
    alert_enabled: true
    alert_threshold: 0.8
    currency: "USD"
  
  # 健康检查配置
  health_check:
    enabled: true
    interval: 30.0
    timeout: 10.0
    retry_count: 3
    fallback_on_failure: true