graph TD
    A[工具调用请求] --> B{失效模式识别}
    B -->|函数不存在| C[工具选择幻觉]
    B -->|参数类型错误| D[参数类型漂移]
    B -->|Schema不合规| E[Schema合规裂缝]
    B -->|多轮偏移| F[工具选择偏移]
    B -->|Token爆炸| G[Token成本暗耗]
    B -->|并行错误| H[并行依赖死锁]

    C --> I[纠偏层1: 约束解码]
    D --> I
    E --> J[纠偏层2: Schema运行时校验]
    F --> K[纠偏层3: 上下文压缩与工具过滤]
    G --> K
    H --> L[纠偏层4: 依赖图分析与调度]

    I --> M[纠偏层5: 自动重试与降级]
    J --> M
    K --> M
    L --> M
    M --> N[纠偏层6: 可观测性与漂移检测]

# OpenAI工具定义格式
tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "Get weather for a location",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {"type": "string"},
                "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
            },
            "required": ["city"]
        },
        "strict": True  # V5新增：Strict Schema模式
    }
}]

# 响应解析
tool_call = response.choices[0].message.tool_calls[0]
# tool_call.function.name, tool_call.function.arguments

# Anthropic工具定义格式
tools = [{
    "name": "get_weather", 
    "description": "Get weather for a location",
    "input_schema": {  # 注意：用input_schema而非parameters
        "type": "object",
        "properties": {
            "city": {"type": "string"},
            "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
        },
        "required": ["city"]
    }
}]

# 2026新增：Programmatic Tool Calling
# Claude可以在extended thinking中编写代码来调用工具

# Gemini工具定义
tools = [{
    "function_declarations": [{
        "name": "get_weather",
        "description": "Get weather for a location",
        "parameters": {
            "type": "OBJECT",
            "properties": {
                "city": {"type": "STRING"},
                "unit": {"type": "STRING", "enum": ["celsius", "fahrenheit"]}
            },
            "required": ["city"]
        }
    }]
}]

# 使用outlines进行约束解码
from outlines import models, generate

model = models.transformers("NousResearch/Hermes-4-14B")
# 定义JSON Schema约束
generator = generate.json(model, weather_schema)
result = generator("What's the weather in Tokyo?")
# 输出保证是合法JSON且Schema合规

from pydantic import BaseModel, ValidationError
from functools import wraps

def validate_tool_call(schema_map):
    """工具调用校验装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(tool_name, arguments):
            if tool_name not in schema_map:
                raise ToolNotFoundError(tool_name)
            schema = schema_map[tool_name]
            try:
                validated = schema(**arguments)
                return func(tool_name, validated.model_dump())
            except ValidationError as e:
                # 将校验错误反馈给模型进行重试
                return retry_with_feedback(tool_name, arguments, str(e))
        return wrapper
    return decorator

# MCPToolRouter的核心策略
class ToolRouter:
    def __init__(self, tool_registry, embedding_model):
        self.tools = tool_registry
        self.embeddings = embedding_model

    def select_tools(self, user_query, max_tools=5):
        """基于语义相似度选择相关工具"""
        query_embedding = self.embeddings.embed(user_query)
        scores = []
        for tool in self.tools:
            desc_embedding = self.embeddings.embed(tool.description)
            similarity = cosine_similarity(query_embedding, desc_embedding)
            scores.append((tool, similarity))

        # 只注入Top-K工具到上下文
        selected = sorted(scores, key=lambda x: -x[1])[:max_tools]
        return [t[0] for t in selected]

class ToolDependencyAnalyzer:
    def build_dependency_graph(self, tool_calls):
        """构建工具调用依赖图"""
        graph = {}
        output_map = {}  # 记录每个工具的输出变量

        for call in tool_calls:
            deps = set()
            for param_value in call.arguments.values():
                if isinstance(param_value, str):
                    # 检测是否引用了前序工具的输出
                    for prev_call, outputs in output_map.items():
                        if any(f"${prev_call}.{out}" in param_value 
                               for out in outputs):
                            deps.add(prev_call)
            graph[call] = deps
            output_map[call.name] = call.output_fields

        return graph

    def schedule(self, graph):
        """拓扑排序生成执行计划"""
        # 可并行的调用放在同一批次
        batches = []
        remaining = set(graph.keys())
        completed = set()

        while remaining:
            batch = {t for t in remaining if graph[t].issubset(completed)}
            batches.append(batch)
            completed.update(batch)
            remaining -= batch

        return batches

class ToolCallRetryManager:
    def __init__(self, max_retries=3, fallback_model=None):
        self.max_retries = max_retries
        self.fallback = fallback_model

    async def execute_with_retry(self, model, messages, tools):
        for attempt in range(self.max_retries):
            response = await model.chat(messages, tools=tools)

            for tool_call in response.tool_calls:
                result = await execute_tool(tool_call)

                if result.success:
                    messages.append(tool_result_msg(tool_call, result))
                else:
                    # 构建纠错反馈
                    correction = format_correction(
                        tool_call, result.error, attempt
                    )
                    messages.append(error_feedback_msg(correction))

                    if attempt == self.max_retries - 1 and self.fallback:
                        # 降级到更强模型做最后一次尝试
                        return await self.fallback.chat(messages, tools=tools)

        return response

class ToolCallMonitor:
    def __init__(self):
        self.metrics = {
            "selection_accuracy": SlidingWindowMetric(window=1000),
            "param_compliance": SlidingWindowMetric(window=1000),
            "retry_rate": SlidingWindowMetric(window=1000),
            "latency_p95": SlidingWindowMetric(window=100),
        }
        self.baselines = {}

    def record(self, tool_call_event):
        self.metrics["selection_accuracy"].record(
            tool_call_event.correct_tool_selected
        )
        # 漂移检测：如果滑动窗口准确率下降超过5%，触发告警
        current = self.metrics["selection_accuracy"].get()
        if self.baselines.get("selection_accuracy"):
            drift = self.baselines["selection_accuracy"] - current
            if drift > 0.05:
                alert_drift("tool_selection", drift, current)

总Token成本 = 基础对话Token + 工具定义Token + 历史工具结果Token + 校验反馈Token

graph LR
    subgraph "生产级工具调用架构"
        U[用户请求] --> R[路由层]
        R --> F[工具过滤器]
        F --> M[LLM推理]
        M --> V1{Schema校验}
        V1 -->|通过| E[工具执行器]
        V1 -->|失败| RT[重试管理器]
        RT -->|≤3次| M
        RT -->|>3次| FB[降级模型]
        FB --> M
        E --> V2{执行结果校验}
        V2 -->|成功| O[结果处理]
        V2 -->|失败| RT
        O --> C[上下文压缩]
        C --> MON[监控面板]
        MON -->|漂移告警| A[运维告警]
    end