MCP Client 如何选择 Tool

from typing import Dict, List, Any, Tuple
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np

class SimilarityCalculator:
    def __init__(self):
        self.vectorizer = TfidfVectorizer(stop_words='english')
        self.tool_vectors = None
        self.tool_names = []
    
    def fit(self, tools: List[Dict[str, Any]]):
        """训练相似度模型，生成工具向量"""
        self.tool_names = [tool['name'] for tool in tools]
        
        # 提取工具描述和参数信息作为特征
        tool_descriptions = []
        for tool in tools:
            desc = tool.get('description', '')
            
            # 提取参数描述
            params = tool.get('parameters', {})
            if isinstance(params, dict):
                properties = params.get('properties', {})
                for param_name, param_schema in properties.items():
                    param_desc = param_schema.get('description', '')
                    desc += f" {param_name}: {param_desc}"
            
            tool_descriptions.append(desc)
        
        # 生成 TF-IDF 向量
        self.tool_vectors = self.vectorizer.fit_transform(tool_descriptions)
    
    def calculate_similarity(self, user_request: str) -> List[Tuple[str, float]]:
        """计算用户请求与每个工具的相似度"""
        if self.tool_vectors is None:
            return []
        
        # 生成用户请求的向量
        request_vector = self.vectorizer.transform([user_request])
        
        # 计算余弦相似度
        similarities = cosine_similarity(request_vector, self.tool_vectors)[0]
        
        # 生成相似度列表
        similarity_list = []
        for tool_name, similarity in zip(self.tool_names, similarities):
            similarity_list.append((tool_name, float(similarity)))
        
        # 按相似度降序排序
        similarity_list.sort(key=lambda x: x[1], reverse=True)
        
        return similarity_list
    
    def get_top_tools(self, user_request: str, top_k: int = 3, threshold: float = 0.1) -> List[Tuple[str, float]]:
        """获取相似度最高的前 k 个工具，过滤掉低于阈值的结果"""
        similarity_list = self.calculate_similarity(user_request)
        
        # 过滤掉低于阈值的结果
        filtered_list = [(tool, sim) for tool, sim in similarity_list if sim >= threshold]
        
        # 返回前 k 个结果
        return filtered_list[:top_k]

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

class AIToolSelector:
    def __init__(self, llm_connector):
        self.llm_connector = llm_connector
    
    def generate_selection_prompt(self, user_request: str, tools: List[Dict[str, Any]]) -> str:
        """生成用于工具选择的 Prompt"""
        # 构建工具描述部分
        tools_desc = "可用工具：\n\n"
        for i, tool in enumerate(tools):
            tools_desc += f"{i+1}. **{tool['name']}**\n"
            tools_desc += f"   描述：{tool['description']}\n"
            
            if isinstance(tool['parameters'], dict):
                properties = tool['parameters'].get('properties', {})
                required = tool['parameters'].get('required', [])
                
                tools_desc += "   参数：\n"
                for param_name, param_schema in properties.items():
                    required_mark = "*" if param_name in required else ""
                    param_type = param_schema.get('type', 'string')
                    param_desc = param_schema.get('description', '')
                    tools_desc += f"   - {param_name}{required_mark}: {param_type} - {param_desc}\n"
        
        # 构建最终 Prompt
        prompt = f"""请根据用户请求，从以下可用工具中选择最适合的工具。

{tools_desc}

用户请求：
{user_request}

请输出 JSON 格式的结果，包含以下字段：
- selected_tool: 选择的工具名称
- confidence: 选择的置信度（0-1 之间的小数）
- reason: 选择该工具的原因
- alternatives: 其他可能适合的工具列表（最多 2 个）

输出示例：
```json
{
  "selected_tool": "web_search",
  "confidence": 0.95,
  "reason": "用户请求需要搜索最新的天气信息，web_search 工具最适合此需求。",
  "alternatives": ["weather_api"]
}

    return prompt

async def select_tool(self, user_request: str, tools: List[Dict[str, Any]]) -> Dict[str, Any]:
    """使用 LLM 进行工具选择"""
    # 生成选择 Prompt
    prompt = self.generate_selection_prompt(user_request, tools)
    
    # 调用 LLM 进行工具选择
    response = await self.llm_connector.generate(prompt=prompt)
    
    # 解析 LLM 响应
    if "choices" in response and len(response["choices"]) > 0:
        content = response["choices"][0].get("message", {}).get("content", "")
        
        # 提取 JSON 部分
        if "```json" in content:
            json_str = content.split("```json")[1].split("```")[0]
        else:
            json_str = content
        
        try:
            selection_result = json.loads(json_str)
            return selection_result
        except json.JSONDecodeError:
            # 如果解析失败，返回默认结果
            return {
                "selected_tool": tools[0]["name"] if tools else "",
                "confidence": 0.5,
                "reason": "无法解析 LLM 响应，使用默认工具",
                "alternatives": [tool["name"] for tool in tools[1:3]] if len(tools) > 1 else []
            }
    
    # 如果 LLM 响应格式不符合预期，返回默认结果
    return {
        "selected_tool": tools[0]["name"] if tools else "",
        "confidence": 0.5,
        "reason": "LLM 响应格式不符合预期，使用默认工具",
        "alternatives": [tool["name"] for tool in tools[1:3]] if len(tools) > 1 else []
    }

**代码解析**：
- 利用 LLM 进行工具选择，生成结构化的选择结果
- 支持置信度评分和备选工具列表
- 实现了健壮的响应解析和错误处理
- 可扩展支持不同的 LLM 模型

#### 3.3.3 多策略融合的工具选择实现

多策略融合将多种选择策略结合起来，生成更准确的工具选择结果。

**代码示例 3：多策略融合的工具选择实现**

```python
from typing import Dict, List, Any, Tuple

class MultiStrategyToolSelector:
    def __init__(self, similarity_calculator, ai_selector, history_manager, capability_validator):
        self.similarity_calculator = similarity_calculator
        self.ai_selector = ai_selector
        self.history_manager = history_manager
        self.capability_validator = capability_validator
        
        # 策略权重配置
        self.strategy_weights = {
            "similarity": 0.4,
            "ai_selection": 0.3,
            "history": 0.2,
            "capability": 0.1
        }
    
    async def select_tool(self, user_request: str, tools: List[Dict[str, Any]], context: List[Dict[str, Any]]) -> Dict[str, Any]:
        """融合多种策略进行工具选择"""
        # 1. 初始化工具评分字典
        tool_scores = {}
        for tool in tools:
            tool_scores[tool["name"]] = 0.0
        
        # 2. 相似度匹配策略
        similarity_results = self.similarity_calculator.get_top_tools(user_request, top_k=10)
        for tool_name, similarity in similarity_results:
            if tool_name in tool_scores:
                tool_scores[tool_name] += similarity * self.strategy_weights["similarity"]
        
        # 3. AI 辅助选择策略
        ai_result = await self.ai_selector.select_tool(user_request, tools)
        selected_tool = ai_result.get("selected_tool", "")
        confidence = ai_result.get("confidence", 0.0)
        
        if selected_tool in tool_scores:
            tool_scores[selected_tool] += confidence * self.strategy_weights["ai_selection"]
        
        for alt_tool in ai_result.get("alternatives", []):
            if alt_tool in tool_scores:
                tool_scores[alt_tool] += confidence * 0.5 * self.strategy_weights["ai_selection"]
        
        # 4. 历史数据策略
        history_data = self.history_manager.get_tool_history(user_request)
        for tool_name, history_score in history_data.items():
            if tool_name in tool_scores:
                tool_scores[tool_name] += history_score * self.strategy_weights["history"]
        
        # 5. 能力验证策略
        for tool in tools:
            tool_name = tool["name"]
            capability_score = await self.capability_validator.validate_capability(tool)
            tool_scores[tool_name] += capability_score * self.strategy_weights["capability"]
        
        # 6. 生成最终评分结果
        scored_tools = sorted(tool_scores.items(), key=lambda x: x[1], reverse=True)
        
        # 7. 处理歧义情况
        if len(scored_tools) >= 2 and abs(scored_tools[0][1] - scored_tools[1][1]) < 0.1:
            # 存在歧义，需要进一步处理
            top_tools = scored_tools[:2]
            return {
                "selected_tool": top_tools[0][0],
                "confidence": top_tools[0][1],
                "reason": f"存在工具选择歧义，根据综合评分选择 {top_tools[0][0]}，备选工具 {top_tools[1][0]}",
                "alternatives": [tool[0] for tool in top_tools[1:]],
                "ambiguous": True,
                "scores": tool_scores
            }
        
        # 8. 返回最终选择结果
        if scored_tools:
            return {
                "selected_tool": scored_tools[0][0],
                "confidence": scored_tools[0][1],
                "reason": f"根据综合评分选择 {scored_tools[0][0]}",
                "alternatives": [tool[0] for tool in scored_tools[1:3]],
                "ambiguous": False,
                "scores": tool_scores
            }
        
        # 9. 没有可用工具的情况
        return {
            "selected_tool": "",
            "confidence": 0.0,
            "reason": "没有找到适合的工具",
            "alternatives": [],
            "ambiguous": False,
            "scores": {}
        }
    
    def update_strategy_weights(self, new_weights: Dict[str, float]):
        """更新策略权重配置"""
        self.strategy_weights.update(new_weights)
        
        # 确保权重总和为 1
        total_weight = sum(self.strategy_weights.values())
        if total_weight != 1.0:
            for key in self.strategy_weights:
                self.strategy_weights[key] /= total_weight

from typing import Dict, List, Any, Tuple
import json
import hashlib
from datetime import datetime, timedelta

class ToolHistoryManager:
    def __init__(self, max_history_size: int = 10000):
        self.max_history_size = max_history_size
        self.history = []
        self.tool_statistics = {}
    
    def record_tool_call(self, user_request: str, tool_name: str, success: bool, execution_time: float, result: Any):
        """记录工具调用历史"""
        # 生成请求哈希，用于后续分析
        request_hash = hashlib.sha256(user_request.encode()).hexdigest()
        
        # 构建历史记录
        history_item = {
            "timestamp": datetime.now().isoformat(),
            "request": user_request,
            "request_hash": request_hash,
            "tool_name": tool_name,
            "success": success,
            "execution_time": execution_time,
            "result": str(result)[:1000]  # 限制结果长度
        }
        
        # 添加到历史列表
        self.history.append(history_item)
        
        # 裁剪历史记录，保持最大大小
        if len(self.history) > self.max_history_size:
            self.history = self.history[-self.max_history_size:]
        
        # 更新工具统计信息
        if tool_name not in self.tool_statistics:
            self.tool_statistics[tool_name] = {
                "total_calls": 0,
                "success_calls": 0,
                "avg_execution_time": 0.0,
                "last_used": datetime.now().isoformat()
            }
        
        stats = self.tool_statistics[tool_name]
        stats["total_calls"] += 1
        
        if success:
            stats["success_calls"] += 1
        
        # 更新平均执行时间
        stats["avg_execution_time"] = (
            stats["avg_execution_time"] * (stats["total_calls"] - 1) + execution_time
        ) / stats["total_calls"]
        
        stats["last_used"] = datetime.now().isoformat()
    
    def get_tool_history(self, user_request: str) -> Dict[str, float]:
        """获取与当前请求相关的工具历史数据"""
        # 生成请求哈希
        request_hash = hashlib.sha256(user_request.encode()).hexdigest()
        
        # 查找相似请求的历史记录
        relevant_history = []
        for item in self.history:
            # 简单的相似度判断：检查请求哈希或请求包含相同关键词
            if item["request_hash"] == request_hash or any(word in item["request"] for word in user_request.split()[:5]):
                relevant_history.append(item)
        
        # 计算工具历史评分
        tool_history_scores = {}
        for item in relevant_history:
            tool_name = item["tool_name"]
            success = item["success"]
            
            # 基于成功率和时间衰减的评分
            time_diff = datetime.now() - datetime.fromisoformat(item["timestamp"])
            time_decay = 1.0 / (1.0 + time_diff.total_seconds() / 3600)  # 小时级衰减
            
            score = (1.0 if success else 0.5) * time_decay
            
            if tool_name in tool_history_scores:
                tool_history_scores[tool_name] += score
            else:
                tool_history_scores[tool_name] = score
        
        # 归一化评分
        if tool_history_scores:
            max_score = max(tool_history_scores.values())
            for tool_name in tool_history_scores:
                tool_history_scores[tool_name] /= max_score
        
        return tool_history_scores
    
    def get_tool_statistics(self) -> Dict[str, Any]:
        """获取所有工具的统计信息"""
        return self.tool_statistics.copy()
    
    def clear_old_history(self, days: int = 30):
        """清除指定天数之前的历史记录"""
        cutoff_time = datetime.now() - timedelta(days=days)
        self.history = [item for item in self.history if datetime.fromisoformat(item["timestamp"]) >= cutoff_time]
    
    def save_history(self, file_path: str):
        """保存历史记录到文件"""
        with open(file_path, "w", encoding="utf-8") as f:
            json.dump({
                "history": self.history,
                "tool_statistics": self.tool_statistics
            }, f, ensure_ascii=False, indent=2)
    
    def load_history(self, file_path: str):
        """从文件加载历史记录"""
        try:
            with open(file_path, "r", encoding="utf-8") as f:
                data = json.load(f)
                self.history = data.get("history", [])
                self.tool_statistics = data.get("tool_statistics", {})
        except Exception as e:
            print(f"加载历史记录失败: {e}")

# MCP Client 工具选择配置
tool_selection:
  # 选择策略配置
  strategies:
    similarity: true
    ai_selection: true
    history: true
    capability: true
  
  # 策略权重配置
  weights:
    similarity: 0.4
    ai_selection: 0.3
    history: 0.2
    capability: 0.1
  
  # 相似度匹配配置
  similarity:
    algorithm: "tfidf_cosine"
    top_k: 10
    threshold: 0.1
    vectorizer_params:
      max_features: 10000
      stop_words: "english"
  
  # AI 辅助选择配置
  ai_selection:
    model: "gpt-4o"
    temperature: 0.1
    timeout: 10.0
    fallback_strategy: "similarity"
  
  # 历史数据配置
  history:
    max_size: 10000
    time_decay: 3600  # 小时
    relevant_threshold: 0.5
  
  # 能力验证配置
  capability:
    validate_availability: true
    validate_version: true
    validate_parameters: false
  
  # 歧义处理配置
  ambiguity:
    threshold: 0.1
    resolution_strategy: "ask_user"  # ask_user, pick_highest, random
  
  # 缓存配置
  cache:
    enabled: true
    ttl: 3600  # 秒
    max_size: 1000
  
  # 日志配置
  logging:
    enabled: true
    level: "INFO"
    include_details: true

 ext{similarity}(Q, D_i) = os(	heta) = rac{athbf{v}_q dot athbf{v}_i}{|athbf{v}_q| 	imes |athbf{v}_i|}