百工具路由迷失：从上下文爆炸到精准选型六层突围

{
  "name": "update_customer_record",
  "description": "Updates an existing customer record in the CRM system. Supports updating contact information, company details, subscription status, and custom fields. Requires valid customer_id. Use 'partial_update' mode for specific fields only.",
  "parameters": {
    "type": "object",
    "properties": {
      "customer_id": {"type": "string", "description": "Unique customer identifier"},
      "update_mode": {"type": "string", "enum": ["full", "partial"]},
      "contact_info": {"$ref": "#/definitions/ContactInfo"},
      "company_details": {"$ref": "#/definitions/CompanyDetails"}
    },
    "required": ["customer_id"]
  }
}

graph TD
    A[工具爆炸问题] --> B[上下文窗口溢出]
    A --> C[选型精度衰减]
    A --> D[调用延迟飙升]

    B --> B1[工具定义占30-50K token]
    B --> B2[工具结果回填10K+]
    B --> B3[推理空间被挤压]

    C --> C1[相似工具混淆]
    C --> C2[参数幻觉]
    C --> C3[工具遗漏]

    D --> D1[输入token翻倍]
    D --> D2[推理时间线性增长]
    D --> D3[API成本飙升]

    B3 --> E[模型拒绝推理或截断]
    C3 --> F[任务执行不完整]
    D3 --> G[单次调用成本翻3-5倍]

# 全量注入：所有工具直接塞入上下文
tools = load_all_tools()  # 加载全部50个工具
response = client.chat.completions.create(
    model="gpt-5",
    messages=[{"role": "system", "content": system_prompt}],
    tools=tools,  # 直接传入全部工具schema
)

# 静态分组：按业务域划分工具包
TOOL_GROUPS = {
    "crm": [get_customer, update_customer, delete_customer, search_customers],
    "billing": [create_invoice, get_payment, refund_payment],
    "support": [create_ticket, update_ticket, escalate_ticket],
}

def get_tools_for_intent(intent: str) -> list:
    """根据意图关键词选择工具组"""
    intent_lower = intent.lower()
    if any(kw in intent_lower for kw in ["customer", "client", "account"]):
        return TOOL_GROUPS["crm"]
    elif any(kw in intent_lower for kw in ["payment", "invoice", "bill", "refund"]):
        return TOOL_GROUPS["billing"]
    elif any(kw in intent_lower for kw in ["ticket", "support", "issue"]):
        return TOOL_GROUPS["support"]
    return []  # 无匹配时返回空

import numpy as np
from openai import OpenAI

client = OpenAI()

class SemanticToolRouter:
    def __init__(self, tools: list[dict]):
        self.tools = tools
        # 预计算所有工具的嵌入向量
        self.tool_embeddings = self._embed_tools()

    def _embed_tools(self) -> np.ndarray:
        """将工具名称+描述编码为向量"""
        texts = [
            f"{t['name']}: {t['description']}" 
            for t in self.tools
        ]
        resp = client.embeddings.create(
            model="text-embedding-3-large",
            input=texts
        )
        return np.array([d.embedding for d in resp.data])

    def select_tools(self, query: str, top_k: int = 10) -> list[dict]:
        """根据用户查询选择最相关的top_k个工具"""
        query_emb = np.array(
            client.embeddings.create(
                model="text-embedding-3-large",
                input=[query]
            ).data[0].embedding
        )
        # 余弦相似度排序
        similarities = np.dot(self.tool_embeddings, query_emb)
        top_indices = np.argsort(similarities)[-top_k:][::-1]
        return [self.tools[i] for i in top_indices]

class HierarchicalToolRouter:
    def __init__(self):
        self.intent_classifier = self._load_classifier()
        self.toolkit_registry = self._load_toolkits()

    def route(self, query: str, available_tools: list) -> list[dict]:
        # 第一级：意图分类（用轻量模型）
        intent = self._classify_intent(query)

        # 第二级：选择工具包（粗筛）
        toolkit = self.toolkit_registry.get(intent)
        candidate_tools = toolkit.get_tools()  # 约15-20个

        # 第三级：精确匹配（用强模型+少量工具）
        selected = self._precise_select(query, candidate_tools, top_k=5)
        return selected

    def _classify_intent(self, query: str) -> str:
        """用小模型快速分类意图"""
        resp = client.chat.completions.create(
            model="gpt-4.1-mini",  # 轻量模型做粗筛
            messages=[{
                "role": "system",
                "content": "Classify the user intent into: crm, billing, support, analytics. Reply with one word only."
            }, {
                "role": "user", 
                "content": query
            }],
            max_tokens=5,
            temperature=0
        )
        return resp.choices[0].message.content.strip().lower()

AutoTool核心思路：
1. 构建工具依赖图：节点=工具，边=数据依赖
   get_user_id → get_customer_record → update_customer_record
2. 查询时在图上搜索可达路径
3. 只加载路径上的工具定义
4. 比全量注入减少60-80%的token开销

# AutoTool图路由简化实现
class AutoToolRouter:
    def __init__(self, tools: list[dict]):
        self.tool_graph = self._build_type_graph(tools)

    def _build_type_graph(self, tools):
        """基于输入输出类型构建工具依赖图"""
        graph = {}
        for tool in tools:
            output_types = self._extract_output_types(tool)
            input_types = self._extract_input_types(tool)
            graph[tool['name']] = {
                'outputs': output_types,
                'inputs': input_types,
                'downstream': []  # 类型匹配的下游工具
            }
        # 建立类型匹配的边
        for t1 in tools:
            for t2 in tools:
                if t1 == t2:
                    continue
                if self._types_match(
                    graph[t1['name']]['outputs'],
                    graph[t2['name']]['inputs']
                ):
                    graph[t1['name']]['downstream'].append(t2['name'])
        return graph

    def plan_tool_sequence(self, query: str, goal_type: str):
        """根据目标类型在图上搜索工具序列"""
        # BFS搜索从自然语言输入到目标类型的路径
        visited = set()
        queue = self._find_entry_tools(query)
        while queue:
            tool_name = queue.pop(0)
            if tool_name in visited:
                continue
            visited.add(tool_name)
            tool = self.tool_graph[tool_name]
            if goal_type in tool['outputs']:
                return self._reconstruct_path(tool_name)
            queue.extend(tool['downstream'])
        return []

class AdaptiveToolRouter:
    def __init__(self, tools: list[dict]):
        self.tools = tools
        self.call_history = []  # 调用历史
        self.routing_stats = {
            'brute_force': {'calls': 0, 'successes': 0},
            'semantic': {'calls': 0, 'successes': 0},
            'hierarchical': {'calls': 0, 'successes': 0},
        }
        self.semantic_router = SemanticToolRouter(tools)
        self.hierarchical_router = HierarchicalToolRouter()

    def route(self, query: str) -> list[dict]:
        tool_count = len(self.tools)
        complexity = self._estimate_complexity(query)

        # 自适应选择路由策略
        if tool_count <= 10:
            return self.tools  # 少量工具直接全量
        elif tool_count <= 30 and complexity == 'simple':
            return self.semantic_router.select_tools(query, top_k=8)
        elif tool_count > 30 or complexity == 'complex':
            return self.hierarchical_router.route(query, self.tools)
        else:
            return self.semantic_router.select_tools(query, top_k=12)

    def _estimate_complexity(self, query: str) -> str:
        """评估查询复杂度"""
        # 检查是否涉及多域
        domain_keywords = {
            'crm': ['customer', 'client', 'account'],
            'billing': ['payment', 'invoice', 'refund'],
            'support': ['ticket', 'issue', 'escalate'],
        }
        domains_hit = sum(
            1 for domain, kws in domain_keywords.items()
            if any(kw in query.lower() for kw in kws)
        )
        if domains_hit >= 2:
            return 'complex'
        elif domains_hit == 1:
            return 'moderate'
        return 'simple'

# 工具token预算计算器
def calculate_tool_budget(tools: list[dict], model_context: int = 128000) -> dict:
    import tiktoken
    enc = tiktoken.encoding_for_model("gpt-4")

    total_tokens = 0
    for tool in tools:
        schema_str = json.dumps(tool)
        total_tokens += len(enc.encode(schema_str))

    budget_ratio = total_tokens / model_context
    recommendation = "OK" if budget_ratio < 0.10 else (
        "ROUTING_NEEDED" if budget_ratio < 0.25 else "CRITICAL"
    )

    return {
        "tool_count": len(tools),
        "total_schema_tokens": total_tokens,
        "budget_ratio": f"{budget_ratio:.1%}",
        "recommendation": recommendation,
        "max_safe_tools": int(0.10 * model_context / (total_tokens / len(tools)))
    }

# 优化前：冗长描述（280 tokens）
{
    "name": "search_knowledge_base",
    "description": "Searches the internal knowledge base for articles, documents, and FAQs related to the given query. Supports full-text search and semantic search. Returns up to 10 results with relevance scores. Use this tool when the user asks questions about company policies, product documentation, or technical procedures.",
    "parameters": { ... }
}

# 优化后：精简描述（95 tokens，减少66%）
{
    "name": "search_kb",
    "description": "Search internal KB for articles/docs/FAQs. Returns top 10 with relevance. Use for: company policies, product docs, tech procedures.",
    "parameters": { ... }
}

def compress_tool_result(result: str, max_tokens: int = 500) -> str:
    if len(result.split()) < max_tokens:
        return result
    resp = client.chat.completions.create(
        model="gpt-4.1-mini",
        messages=[{
            "role": "system",
            "content": f"Summarize in under {max_tokens} tokens. Keep key data points."
        }, {"role": "user", "content": result}],
        max_tokens=max_tokens
    )
    return resp.choices[0].message.content

def extract_relevant_fields(result: dict, needed_fields: list[str]) -> dict:
    return {k: result.get(k) for k in needed_fields if k in result}

graph LR
    A[用户查询] --> B[路由层]
    B --> C{路由决策}
    C -->|正确| D[精准工具集]
    C -->|错误| E[错误工具集]
    D --> F[任务成功]
    E --> G[任务失败/重试]

    H[可观测性指标] --> H1[路由命中率<br>选中的工具被实际调用]
    H --> H2[路由召回率<br>所需工具是否被选中]
    H --> H3[上下文利用率<br>工具定义/token预算]
    H --> H4[端到端延迟<br>路由+推理+执行]

# MCP动态工具发现 + 语义路由
class MCPDynamicRouter:
    def __init__(self, mcp_client):
        self.mcp = mcp_client
        self.tool_cache = {}
        self.last_refresh = 0
        self.semantic_router = None

    async def route(self, query: str, top_k: int = 8):
        # 1. 动态发现所有可用工具
        if self._needs_refresh():
            servers = await self.mcp.list_servers()
            all_tools = []
            for server in servers:
                tools = await self.mcp.list_tools(server)
                all_tools.extend(tools)
            self.semantic_router = SemanticToolRouter(all_tools)

        # 2. 语义路由选择最相关的工具
        selected = self.semantic_router.select_tools(query, top_k)

        # 3. 只将选中的工具schema注入上下文
        return selected

# 企业MCP注册中心 + RBAC路由
class EnterpriseToolRouter:
    def __init__(self, registry_url: str, user_context: dict):
        self.registry = registry_url
        self.user = user_context
        self.permission_cache = {}

    async def get_permitted_tools(self) -> list[dict]:
        """根据用户角色过滤可见工具"""
        role = self.user.get("role", "viewer")
        all_tools = await self._fetch_registry()

        permitted = []
        for tool in all_tools:
            required_role = tool.get("metadata", {}).get("min_role", "viewer")
            if self._role_level(role) >= self._role_level(required_role):
                permitted.append(tool)
        return permitted

    def _role_level(self, role: str) -> int:
        return {"viewer": 0, "editor": 1, "admin": 2}.get(role, 0)