百工具精度坍缩：选型衰减六栈实测

import openai
import json
from dataclasses import dataclass

@dataclass
class ToolSelectionResult:
    tool_name: str
    correct: bool
    position: int  # 工具在列表中的位置
    latency_ms: float

def measure_selection_decay(
    client: openai.OpenAI,
    model: str,
    tools: list[dict],
    test_queries: list[dict],
    positions: list[int] = [5, 15, 25, 35, 45]
):
    """测量工具在不同位置和数量下的选型精度"""
    results = {pos: {"correct": 0, "total": 0} for pos in positions}

    for query in test_queries:
        target_tool = query["expected_tool"]
        # 重组工具列表，将目标工具放到指定位置
        for pos in positions:
            reordered = reorder_tools(tools, target_tool, pos)
            resp = client.chat.completions.create(
                model=model,
                messages=[{"role": "user", "content": query["prompt"]}],
                tools=reordered,
                tool_choice="auto"
            )
            chosen = resp.choices[0].message.tool_calls[0].function.name
            results[pos]["total"] += 1
            results[pos]["correct"] += (chosen == target_tool)

    return {
        pos: r["correct"] / r["total"]
        for pos, r in results.items()
    }

# 策略1: 两阶段路由 —— 先选工具子集，再精确调用
def smart_tool_injection(user_query: str, all_tools: list[dict]) -> list[dict]:
    """先用便宜模型做工具预筛选，再用强模型精确调用"""
    # 阶段1: 用Haiku级模型做工具匹配（成本$0.0001）
    tool_summaries = [
        {"name": t["function"]["name"], "desc": t["function"]["description"][:50]}
        for t in all_tools
    ]
    routing_prompt = f"""用户意图: {user_query}
    可用工具摘要: {json.dumps(tool_summaries)}
    返回最相关的3个工具名称，JSON格式。"""

    selected = cheap_model_call(routing_prompt)  # ~500 token
    selected_tools = [t for t in all_tools if t["function"]["name"] in selected]

    # 阶段2: 只注入选中的3个工具Schema
    return selected_tools  # 从78K token降到~2K token

# 策略2: Schema压缩 —— 删除description中的冗余文本
def compress_schema(tools: list[dict]) -> list[dict]:
    """压缩工具Schema，保留关键字段"""
    compressed = []
    for t in tools:
        fn = t["function"]
        compressed.append({
            "type": "function",
            "function": {
                "name": fn["name"],
                # description压缩到1句话
                "description": fn["description"].split(".")[0] + ".",
                "parameters": {
                    "type": "object",
                    "properties": {
                        k: {"type": v["type"]}  # 只保留type，删除description
                        for k, v in fn["parameters"]["properties"].items()
                    },
                    "required": fn["parameters"].get("required", [])
                }
            }
        })
    return compressed  # 实测压缩率40-60%

from dataclasses import dataclass
from typing import Optional

@dataclass
class LatencyBudget:
    total_budget_ms: int = 10000  # 用户可接受的总延迟
    selection_ratio: float = 0.30
    execution_ratio: float = 0.45
    digestion_ratio: float = 0.25

    def allocate(self, num_tools: int, parallel_capable: bool = False):
        """根据工具数量和并行能力动态分配预算"""
        # 工具越多，选型分配越多（但有上限）
        selection_weight = min(0.50, 0.20 + num_tools * 0.005)
        # 并行执行可以压缩执行预算
        exec_weight = 0.40 if parallel_capable else 0.55
        digestion_weight = 1.0 - selection_weight - exec_weight

        return {
            "selection_ms": int(self.total_budget_ms * selection_weight),
            "execution_ms": int(self.total_budget_ms * exec_weight),
            "digestion_ms": int(self.total_budget_ms * digestion_weight),
            "max_parallel": 3 if parallel_capable else 1
        }

class ToolCallDefense:
    """工具调用五层防御体系"""

    def __init__(self):
        self.validators = [
            self.L1_schema_check,
            self.L2_tool_selection_check,
            self.L3_param_value_check,
            self.L4_semantic_check,
            self.L5_cascade_check,
        ]

    def L1_schema_check(self, call, tool_def):
        """L1: JSON Schema校验"""
        try:
            jsonschema.validate(call.arguments, tool_def.parameters)
            return True, None
        except jsonschema.ValidationError as e:
            return False, f"Schema错误: {e.message}"

    def L4_semantic_check(self, call, tool_result, user_intent):
        """L4: 语义一致性校验（用轻量模型）"""
        check_prompt = f"""用户意图: {user_intent}
工具调用: {call.function.name}({call.arguments})
工具返回: {json.dumps(tool_result)[:500]}
返回结果是否真正回答了用户意图？回答YES或NO。"""
        result = cheap_model_call(check_prompt)
        return result.strip() == "YES", "语义不一致"

    def execute_with_defense(self, call, context):
        """带防御的工具执行"""
        for i, validator in enumerate(self.validators):
            passed, error = validator(call, context)
            if not passed:
                # 降级策略：L1-L2重试，L3-L5回退到更简单工具
                if i < 2:
                    return self.retry_with_correction(call, error)
                else:
                    return self.graceful_fallback(call, error)

def build_tool_dag(model, user_query: str, tools: list) -> dict:
    """让模型推断工具调用的DAG结构"""
    dag_prompt = f"""分析以下任务需要的工具调用及其依赖关系：
任务: {user_query}
可用工具: {[t["function"]["name"] for t in tools]}

输出JSON格式的DAG:
{{
  "nodes": [{{"id": "step1", "tool": "tool_name", "params": {{}}}}],
  "edges": [{{"from": "step1", "to": "step2"}}]}}
}}
"""
    dag = model_call(dag_prompt)
    return execute_dag_parallel(dag)  # 按拓扑序并行执行

class AdaptiveSchema:
    """根据上下文动态调整Schema粒度"""

    def get_schema(self, tool_name: str, context: dict) -> dict:
        relevance = self.estimate_relevance(tool_name, context)

        if relevance > 0.8:
            # 高度相关：完整Schema
            return self.full_schemas[tool_name]
        elif relevance > 0.4:
            # 可能相关：标准描述 + 必选参数
            return self.standard_schemas[tool_name]
        else:
            # 不太相关：仅名称和一句话描述
            return self.minimal_schemas[tool_name]

    def estimate_relevance(self, tool_name: str, context: dict) -> float:
        """估算工具与当前上下文的相关性"""
        # 简单版：基于历史调用频率 + 当前对话关键词匹配
        freq_score = context["call_history"].count(tool_name) / max(len(context["call_history"]), 1)
        keyword_score = len(set(self.tool_keywords[tool_name]) & set(context["keywords"])) / len(self.tool_keywords[tool_name])
        return 0.5 * freq_score + 0.5 * keyword_score

graph TB
    subgraph 用户意图层
        U[用户请求] --> INT[意图解析器]
    end

    subgraph 工具路由层
        INT -->|关键词+语义| TR[两阶段路由器]
        TR -->|阶段1: Haiku级预筛选| SUM[工具摘要匹配]
        SUM -->|Top-K候选| FULL[完整Schema注入]
    end

    subgraph 执行编排层
        FULL --> DAG[DAG依赖分析器]
        DAG --> PAR[并行执行器]
        PAR -->|串行队列| SEQ[串行执行器]
        PAR -->|并行组| CON[并发池]
    end

    subgraph 防御校验层
        SEQ & CON --> L1[L1:Schema校验]
        L1 --> L2[L2:工具选择校验]
        L2 --> L3[L3:参数值校验]
        L3 --> L4[L4:语义一致性]
        L4 --> L5[L5:级联传播检测]
    end

    subgraph 优化反馈层
        L5 --> LAT[延迟预算分配器]
        LAT --> TK[Token税监控]
        TK --> SC[自适应Schema]
        SC -.->|动态调整| FULL
    end

    L5 --> RESULT[最终响应]
    TK --> LOG[成本日志]

    style TR fill:#f9f,stroke:#333
    style DAG fill:#bbf,stroke:#333
    style L4 fill:#fbb,stroke:#333
    style SC fill:#bfb,stroke:#333