MCP Client 与 Ollama / 本地模型

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

class OllamaConnector:
    def __init__(self, base_url: str = "http://localhost:11434/v1"):
        self.base_url = base_url
        self.client = httpx.AsyncClient(timeout=30.0)
    
    async def generate(self, prompt: str, model: str = "llama3", tools: Optional[List[Dict[str, Any]]] = None, **kwargs) -> Dict[str, Any]:
        """调用 Ollama 模型生成响应"""
        headers = {
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": model,
            "prompt": prompt,
            "stream": False,
            **kwargs
        }
        
        # 如果提供了工具定义，添加到 payload
        if tools:
            payload["tools"] = tools
            payload["tool_choice"] = "auto"
        
        try:
            response = await self.client.post(
                f"{self.base_url}/chat/completions",
                headers=headers,
                json=payload
            )
            response.raise_for_status()
            return response.json()
        except httpx.HTTPError as e:
            print(f"Ollama API 调用失败: {e}")
            raise
        except Exception as e:
            print(f"Ollama 生成失败: {e}")
            raise
    
    async def list_models(self) -> List[Dict[str, Any]]:
        """获取可用的 Ollama 模型列表"""
        try:
            response = await self.client.get(f"{self.base_url}/models")
            response.raise_for_status()
            return response.json().get("models", [])
        except Exception as e:
            print(f"获取 Ollama 模型列表失败: {e}")
            return []
    
    async def pull_model(self, model_name: str) -> Dict[str, Any]:
        """拉取 Ollama 模型"""
        try:
            response = await self.client.post(
                f"{self.base_url}/models/pull",
                json={"name": model_name},
                timeout=300.0  # 模型拉取可能需要较长时间
            )
            response.raise_for_status()
            return response.json()
        except Exception as e:
            print(f"拉取 Ollama 模型失败: {e}")
            raise
    
    async def delete_model(self, model_name: str) -> Dict[str, Any]:
        """删除 Ollama 模型"""
        try:
            response = await self.client.delete(
                f"{self.base_url}/models/{model_name}"
            )
            response.raise_for_status()
            return response.json()
        except Exception as e:
            print(f"删除 Ollama 模型失败: {e}")
            raise
    
    async def close(self):
        """关闭 HTTP 客户端"""
        await self.client.aclose()

from typing import Dict, List, Any, Optional
import asyncio
from datetime import datetime

class LocalModelManager:
    def __init__(self, ollama_connector):
        self.ollama_connector = ollama_connector
        self.models = {}
        self.model_status = {}
        self.refresh_interval = 300  # 秒
        self.is_refreshing = False
        
        # 启动定期刷新任务
        self._start_refresh_task()
    
    def _start_refresh_task(self):
        """启动定期刷新模型列表的任务"""
        async def refresh_task():
            while True:
                await self.refresh_models()
                await asyncio.sleep(self.refresh_interval)
        
        asyncio.create_task(refresh_task())
    
    async def refresh_models(self):
        """刷新可用模型列表"""
        if self.is_refreshing:
            return
        
        self.is_refreshing = True
        try:
            models = await self.ollama_connector.list_models()
            
            # 更新模型列表和状态
            self.models = {model["name"]: model for model in models}
            
            for model_name in self.models:
                if model_name not in self.model_status:
                    self.model_status[model_name] = {
                        "last_used": None,
                        "total_calls": 0,
                        "successful_calls": 0,
                        "avg_response_time": 0.0,
                        "status": "available"
                    }
            
            # 移除已不存在的模型状态
            for model_name in list(self.model_status.keys()):
                if model_name not in self.models:
                    del self.model_status[model_name]
        except Exception as e:
            print(f"刷新模型列表失败: {e}")
        finally:
            self.is_refreshing = False
    
    def get_available_models(self) -> List[str]:
        """获取可用的模型列表"""
        return list(self.models.keys())
    
    def get_model_info(self, model_name: str) -> Optional[Dict[str, Any]]:
        """获取模型信息"""
        return self.models.get(model_name)
    
    def update_model_status(self, model_name: str, success: bool, response_time: float):
        """更新模型状态"""
        if model_name in self.model_status:
            status = self.model_status[model_name]
            status["last_used"] = datetime.now().isoformat()
            status["total_calls"] += 1
            
            if success:
                status["successful_calls"] += 1
                
                # 更新平均响应时间
                status["avg_response_time"] = (
                    status["avg_response_time"] * (status["total_calls"] - 1) + response_time
                ) / status["total_calls"]
    
    def get_model_stats(self, model_name: str) -> Optional[Dict[str, Any]]:
        """获取模型统计信息"""
        if model_name not in self.model_status:
            return None
        
        status = self.model_status[model_name]
        model_info = self.models.get(model_name, {})
        
        return {
            **model_info,
            "status": status
        }
    
    def get_all_model_stats(self) -> Dict[str, Any]:
        """获取所有模型的统计信息"""
        stats = {}
        for model_name in self.models:
            stats[model_name] = self.get_model_stats(model_name)
        
        return stats

from typing import Dict, List, Any, Optional
import random

class ModelRouter:
    def __init__(self, ollama_connector, local_model_manager, cloud_connector=None):
        self.ollama_connector = ollama_connector
        self.local_model_manager = local_model_manager
        self.cloud_connector = cloud_connector
        
        # 模型路由配置
        self.route_config = {
            "default_model": "llama3",
            "model_priorities": {},
            "load_balancing": "round_robin",  # round_robin, least_used, random
            "fallback_strategy": "cloud",  # cloud, error, retry
            "local_model_threshold": 0.8  # 本地模型使用率阈值
        }
        
        # 轮询计数器
        self.round_robin_counter = 0
    
    def set_route_config(self, config: Dict[str, Any]):
        """设置路由配置"""
        self.route_config.update(config)
    
    def get_best_model(self, task_type: str = "general") -> str:
        """根据任务类型和负载情况选择最佳模型"""
        available_models = self.local_model_manager.get_available_models()
        
        if not available_models:
            return self.route_config["default_model"]
        
        # 1. 基于优先级选择模型
        priorities = self.route_config.get("model_priorities", {})
        if task_type in priorities:
            task_models = priorities[task_type]
            # 过滤出可用的模型
            available_task_models = [model for model in task_models if model in available_models]
            if available_task_models:
                return available_task_models[0]
        
        # 2. 基于负载均衡策略选择模型
        load_balancing = self.route_config.get("load_balancing", "round_robin")
        
        if load_balancing == "round_robin":
            # 轮询策略
            self.round_robin_counter = (self.round_robin_counter + 1) % len(available_models)
            return available_models[self.round_robin_counter]
        elif load_balancing == "least_used":
            # 最少使用策略
            least_used_model = None
            least_usage = float("inf")
            
            for model in available_models:
                stats = self.local_model_manager.get_model_stats(model)
                if stats and stats["status"]["total_calls"] < least_usage:
                    least_usage = stats["status"]["total_calls"]
                    least_used_model = model
            
            return least_used_model or available_models[0]
        elif load_balancing == "random":
            # 随机策略
            return random.choice(available_models)
        else:
            # 默认使用第一个模型
            return available_models[0]
    
    async def route_request(self, prompt: str, task_type: str = "general", **kwargs) -> Dict[str, Any]:
        """路由请求到合适的模型"""
        # 1. 选择最佳模型
        best_model = self.get_best_model(task_type)
        
        start_time = asyncio.get_event_loop().time()
        success = False
        result = {}
        
        try:
            # 2. 调用本地模型
            result = await self.ollama_connector.generate(
                prompt=prompt,
                model=best_model,
                **kwargs
            )
            success = True
            return result
        except Exception as e:
            print(f"调用本地模型 {best_model} 失败: {e}")
            
            # 3. 处理失败情况
            fallback_strategy = self.route_config.get("fallback_strategy", "error")
            
            if fallback_strategy == "cloud" and self.cloud_connector:
                # 回退到云端模型
                print("回退到云端模型")
                return await self.cloud_connector.generate(prompt=prompt, **kwargs)
            elif fallback_strategy == "retry":
                # 重试调用
                print("重试调用本地模型")
                return await self.ollama_connector.generate(
                    prompt=prompt,
                    model=best_model,
                    **kwargs
                )
            else:
                # 抛出错误
                raise
        finally:
            # 4. 更新模型状态
            end_time = asyncio.get_event_loop().time()
            response_time = end_time - start_time
            self.local_model_manager.update_model_status(best_model, success, response_time)

# 示例：MCP Client 与 Ollama 集成的完整示例
import asyncio
from mcp_client.ollama_connector import OllamaConnector
from mcp_client.local_model_manager import LocalModelManager
from mcp_client.model_router import ModelRouter
from mcp_client.mcp_executor import MCPExecutor
from mcp_client.context_manager import ContextManager

async def main():
    # 1. 创建 Ollama 连接器
    ollama_connector = OllamaConnector(base_url="http://localhost:11434/v1")
    
    # 2. 创建本地模型管理器
    local_model_manager = LocalModelManager(ollama_connector)
    
    # 3. 刷新模型列表
    await local_model_manager.refresh_models()
    
    # 4. 创建模型路由管理器
    model_router = ModelRouter(ollama_connector, local_model_manager)
    
    # 5. 创建 MCP 执行器
    mcp_executor = MCPExecutor(server_url="http://localhost:8000/mcp")
    
    # 6. 创建上下文管理器
    context_manager = ContextManager()
    
    # 7. 示例 1：列出可用模型
    print("=== 可用的 Ollama 模型 ===")
    available_models = local_model_manager.get_available_models()
    for model in available_models:
        model_info = local_model_manager.get_model_info(model)
        print(f"- {model}: {model_info.get('size', '未知大小')} bytes, 格式: {model_info.get('format', '未知格式')}")
    
    # 8. 示例 2：调用本地模型生成响应
    print("\n=== 调用本地模型生成响应 ===")
    prompt = "你好，我是用户，能帮我介绍一下 MCP 协议吗？"
    
    # 获取最佳模型
    best_model = model_router.get_best_model()
    print(f"使用模型: {best_model}")
    
    # 调用模型
    result = await ollama_connector.generate(prompt=prompt, model=best_model)
    print(f"模型响应: {result.get('choices', [{}])[0].get('message', {}).get('content', '无响应')}")
    
    # 9. 示例 3：使用模型路由调用模型
    print("\n=== 使用模型路由调用模型 ===")
    result = await model_router.route_request(prompt=prompt, task_type="general")
    print(f"路由响应: {result.get('choices', [{}])[0].get('message', {}).get('content', '无响应')}")
    
    # 10. 示例 4：调用本地模型进行工具调用
    print("\n=== 调用本地模型进行工具调用 ===")
    
    # 定义工具
    tools = [
        {
            "name": "weather_api",
            "description": "获取指定城市的天气信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称"
                    }
                },
                "required": ["city"]
            }
        }
    ]
    
    # 生成带工具调用指令的 Prompt
    tool_prompt = f"""你是一个助手，可以使用工具来帮助用户解决问题。

可用工具：
```json
{json.dumps(tools, indent=2)}

{
  "tool_call": {
    "name": "工具名称",
    "arguments": {
      "参数名": "参数值"
    }
  }
}

{
  "content": "直接回答内容"
}

# 调用模型进行工具调用
tool_result = await ollama_connector.generate(
    prompt=tool_prompt,
    model=best_model,
    tools=tools
)

print(f"工具调用结果: {json.dumps(tool_result, indent=2, ensure_ascii=False)}")

# 11. 关闭资源
await ollama_connector.close()

print("\nMCP Client 与 Ollama 集成示例完成！")

**代码解析**：
- 展示了 MCP Client 与 Ollama 集成的完整流程
- 包含了模型列表获取、模型调用、模型路由和工具调用等功能
- 演示了如何在实际应用中使用本地模型
- 提供了详细的日志输出，便于调试和理解

## 四、与主流方案深度对比

### 4.1 MCP v2.0 与其他本地模型集成方案的对比

| 对比维度 | MCP v2.0 | LangChain 本地模型集成 | LlamaIndex 本地模型集成 | 直接调用 Ollama API |
|---------|----------|------------------------|------------------------|---------------------|
| **标准化接口** | 统一的 API 接口，简化集成 | 支持多种本地模型，但接口不统一 | 主要针对文档检索，接口相对复杂 | 直接调用 Ollama 特定 API |
| **多模型支持** | 支持同时管理和使用多个本地模型 | 支持多种本地模型，但需要单独配置 | 支持多种本地模型，配置相对复杂 | 支持，但需要手动管理 |
| **混合部署** | 原生支持本地模型与云端模型的混合使用 | 支持，但需要额外配置 | 支持，但集成相对复杂 | 不支持，需要手动实现 |
| **负载均衡** | 内置多种负载均衡策略 | 缺乏内置负载均衡 | 缺乏内置负载均衡 | 不支持 |
| **故障回退** | 内置故障回退机制 | 需要手动实现 | 需要手动实现 | 不支持 |
| **性能优化** | 内置本地模型性能优化机制 | 支持部分优化，但需要额外配置 | 支持部分优化，但配置复杂 | 不支持 |
| **可视化支持** | 支持本地模型的可视化管理 | 缺乏内置可视化支持 | 缺乏内置可视化支持 | 依赖 Ollama 自身的可视化 |
| **易用性** | 提供简洁的 API 接口，易于集成 | 配置相对复杂 | 配置相对复杂 | 直接调用 API，需要处理更多细节 |
| **可扩展性** | 模块化设计，易于扩展新功能 | 模块化设计，易于扩展 | 模块化设计，易于扩展 | 相对固定，扩展困难 |
| **社区支持** | 正在快速发展的社区 | 成熟的社区支持 | 成熟的社区支持 | 依赖 Ollama 社区 |

### 4.2 主流本地模型管理工具的对比

| 工具名称 | 优势 | 劣势 | 适用场景 | MCP v2.0 支持 |
|---------|------|------|----------|---------------|
| **Ollama** | 简单易用，支持多种模型 | 功能相对简单，缺乏高级管理功能 | 个人开发者，小型团队 | 原生支持 |
| **LM Studio** | 可视化界面，支持模型量化 | 性能相对较低，资源消耗大 | 个人开发者，模型探索 | 可集成 |
| **text-generation-webui** | 功能丰富，支持多种模型和插件 | 部署复杂，资源消耗大 | 高级用户，研究人员 | 可集成 |
| **GPT4All** | 专注于隐私保护，支持本地运行 | 模型选择相对有限 | 隐私敏感场景 | 可集成 |
| **llama.cpp** | 高性能，支持多种硬件加速 | 配置复杂，缺乏易用的 API | 性能敏感场景 | 可集成 |

### 4.3 MCP v2.0 本地模型集成的优势分析

通过与其他方案和工具的对比，可以看出 MCP v2.0 本地模型集成的主要优势：

1. **标准化接口**：提供统一的 API 接口，简化本地模型集成
2. **多模型支持**：支持同时管理和使用多个本地模型
3. **混合部署**：原生支持本地模型与云端模型的混合使用
4. **负载均衡**：内置多种负载均衡策略，提高系统可靠性
5. **故障回退**：内置故障回退机制，确保系统可用性
6. **性能优化**：内置本地模型性能优化机制，提高运行效率
7. **易用性**：提供简洁的 API 接口，易于集成和使用
8. **可扩展性**：模块化设计，易于扩展新功能

这些优势使得 MCP v2.0 成为构建隐私优先、高性能、成本优化的 AI 工具调用系统的理想选择。

## 五、实际工程意义、潜在风险与局限性分析

### 5.1 MCP Client 与本地模型集成的工程实践

在实际工程实践中，MCP Client 与本地模型集成需要考虑以下几个方面：

1. **模型选择与部署**：
   - 根据任务需求选择合适的本地模型
   - 考虑模型大小和硬件资源限制
   - 选择合适的模型量化和优化策略

2. **性能优化**：
   - 对本地模型进行量化和蒸馏，减少资源消耗
   - 实现模型缓存，提高响应速度
   - 优化模型推理过程，提高吞吐量

3. **可靠性设计**：
   - 实现本地模型故障时的回退策略
   - 设计合理的重试机制
   - 监控本地模型的运行状态

4. **安全考虑**：
   - 确保本地模型的安全部署和使用
   - 实现模型访问权限控制
   - 监控模型的异常行为

5. **混合部署策略**：
   - 根据任务类型和优先级选择本地或云端模型
   - 实现本地模型和云端模型的无缝切换
   - 优化混合部署的成本和性能

### 5.2 潜在风险与挑战

MCP Client 与本地模型集成也面临一些潜在风险和挑战：

1. **硬件资源限制**：
   - 大型本地模型需要大量的硬件资源
   - 可能导致设备过热或性能下降
   - 需要考虑硬件资源的扩展和升级

2. **模型性能差异**：
   - 本地模型的性能可能不如云端模型
   - 可能影响系统的整体性能和用户体验
   - 需要进行充分的性能测试和评估

3. **模型管理复杂度**：
   - 管理多个本地模型增加了系统复杂度
   - 需要实现高效的模型管理和监控
   - 可能导致配置错误和管理混乱

4. **软件依赖问题**：
   - 本地模型依赖特定的软件环境
   - 可能导致版本冲突和兼容性问题
   - 需要实现可靠的依赖管理

### 5.3 局限性分析

MCP v2.0 本地模型集成目前仍存在一些局限性：

1. **模型兼容性**：不同本地模型的 API 接口和功能可能存在差异
2. **性能优化**：本地模型的性能优化仍需要进一步改进
3. **可视化管理**：本地模型的可视化管理工具仍需完善
4. **分布式支持**：分布式环境下的本地模型管理仍在开发中
5. **生态成熟度**：相关的工具和库仍在发展中，生态不够成熟

## 六、未来趋势展望与个人前瞻性预测

### 6.1 MCP Client 与本地模型集成的未来发展趋势

基于当前技术发展和社区动态，我预测 MCP Client 与本地模型集成将朝着以下方向发展：

1. **更强大的模型管理功能**：
   - 支持更多类型的本地模型
   - 实现更智能的模型选择和调度
   - 提供更丰富的模型监控和管理功能

2. **更高效的性能优化技术**：
   - 利用更先进的量化和蒸馏技术
   - 支持模型的动态加载和卸载
   - 实现模型的自动优化和调整

3. **更完善的混合部署策略**：
   - 实现更智能的本地/云端模型切换
   - 支持基于成本和性能的动态调整
   - 提供更丰富的混合部署配置选项

4. **更强大的可视化工具**：
   - 提供本地模型的可视化管理界面
   - 支持模型性能的实时监控和分析
   - 实现模型的可视化调试和优化

5. **更广泛的生态集成**：
   - 与更多本地模型管理工具集成
   - 支持更多类型的本地模型
   - 实现与企业级系统的深度集成

### 6.2 对 AI 工具生态的影响

MCP Client 与本地模型集成的发展将对 AI 工具生态产生深远影响：

1. **促进本地模型的普及**：简化本地模型的集成和使用，降低本地模型的使用门槛
2. **推动混合部署模式的发展**：结合本地模型和云端模型的优势，优化成本和性能
3. **提高 AI 系统的隐私保护能力**：确保敏感数据不离开本地环境，提高数据安全性
4. **降低 AI 系统的运营成本**：减少云端 API 调用成本，降低长期运营费用
5. **支持更多离线和边缘场景**：不依赖网络连接，适合离线或弱网环境

### 6.3 个人建议与行动指南

对于正在或计划使用 MCP Client 与本地模型集成的开发人员，我提出以下建议：

1. **从简单开始**：先从简单的本地模型集成入手，逐步扩展功能
2. **选择合适的模型**：根据任务需求和硬件资源选择合适的本地模型
3. **优化性能**：对本地模型进行量化和优化，提高运行效率
4. **实现可靠的回退机制**：设计合理的本地模型故障回退策略
5. **监控和管理**：实现本地模型的监控和管理，确保系统可靠性
6. **考虑混合部署**：结合本地模型和云端模型的优势，优化系统性能和成本
7. **关注生态发展**：持续关注本地模型生态的发展，及时更新和优化集成方案

## 参考链接：

- [MCP v2.0 官方规范](https://github.com/mcp-protocol/mcp-spec/blob/main/MCP_v2.0.md)
- [Ollama 官方文档](https://ollama.com/docs)
- [LangChain 本地模型集成文档](https://python.langchain.com/docs/modules/model_io/models/llms/local_llms/)
- [LlamaIndex 本地模型集成文档](https://docs.llamaindex.ai/en/stable/module_guides/models/llms/modules.html#local-llms)
- [llama.cpp 项目](https://github.com/ggerganov/llama.cpp)
- [LM Studio 官方网站](https://lmstudio.ai/)

## 附录（Appendix）：

### 附录 A：MCP Client 与 Ollama 集成配置示例

**完整配置文件（YAML 格式）**

```yaml
# MCP Client 与 Ollama 集成配置
ollama_integration:
  # Ollama 连接配置
  connection:
    base_url: "http://localhost:11434/v1"
    timeout: 30.0
    retry_count: 3
    retry_delay: 1.0
  
  # 本地模型管理配置
  model_manager:
    refresh_interval: 300  # 秒
    default_model: "llama3"
    auto_refresh: true
    cache_enabled: true
    cache_ttl: 3600  # 秒
  
  # 模型路由配置
  model_router:
    default_model: "llama3"
    load_balancing: "round_robin"  # round_robin, least_used, random
    fallback_strategy: "cloud"  # cloud, error, retry
    local_model_threshold: 0.8  # 本地模型使用率阈值
    model_priorities:
      general: ["llama3", "mistral", "gemma"]
      coding: ["codegemma", "llama3-code", "mistral-code"]
      chat: ["llama3", "mistral", "gemma"]
  
  # 性能优化配置
  performance:
    quantization: "int8"  # none, int8, int4
    caching: true
    batch_size: 1
    parallel_requests: 5
  
  # 监控配置
  monitoring:
    enabled: true
    metrics_port: 9090
    logging_level: "INFO"
    include_details: true
  
  # 安全配置
  security:
    enabled: false
    allowed_ips: ["127.0.0.1"]
    encryption_enabled: false
  
  # 混合部署配置
  hybrid_deployment:
    enabled: true
    cloud_fallback_threshold: 0.9  # 本地模型使用率阈值
    cloud_model: "gpt-4o"
    cloud_api_key: "${OPENAI_API_KEY}"