# OpenAI strict mode — 100% schema compliance
response = client.chat.completions.create(
    model="gpt-5",
    messages=[{"role": "user", "content": "查询北京明天的天气"}],
    tools=[{
        "type": "function",
        "function": {
            "name": "get_weather",
            "strict": True,  # 关键：启用CFG约束解码
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string"},
                    "date": {"type": "string", "format": "date"}
                },
                "required": ["city", "date"]
            }
        }
    }],
    tool_choice="auto"
)
# GPT-5 strict模式实测Schema遵从率：99.9%
# 非strict模式：约96-97%

# Anthropic tool_use — implicit schema enforcement
response = client.messages.create(
    model="claude-sonnet-4-6-20260501",
    messages=[{"role": "user", "content": "查询北京明天的天气"}],
    tools=[{
        "name": "get_weather",
        "input_schema": {  # Anthropic使用input_schema而非parameters
            "type": "object",
            "properties": {
                "city": {"type": "string"},
                "date": {"type": "string"}
            },
            "required": ["city", "date"]
        }
    }],
    tool_choice={"type": "tool", "name": "get_weather"}  # 强制调用
)

from pydantic import BaseModel, Field, ValidationError
from datetime import date
from functools import wraps

class WeatherQuery(BaseModel):
    city: str = Field(min_length=1, max_length=50)
    target_date: date = Field(alias="date")  # 别名映射：模型用date，内部用target_date

def validate_tool_call(schema_cls):
    """工具调用输出校验装饰器"""
    def decorator(func):
        @wraps(func)
        async def wrapper(raw_input: dict):
            try:
                validated = schema_cls(**raw_input)
                return await func(validated.model_dump())
            except ValidationError as e:
                # 校验失败：触发重试或降级
                return {"error": "parameter_validation_failed",
                        "details": e.errors(),
                        "retry_hint": "Check date format: YYYY-MM-DD"}
        return wrapper
    return decorator

@validate_tool_call(WeatherQuery)
async def get_weather(params: dict):
    # 此时params已经过Pydantic校验，类型安全
    ...

┌─────────────────────────────────────────┐
│        工具调用精度金字塔                 │
├─────────────────────────────────────────┤
│                                         │
│    第三层：并行耦合消解                   │
│    ┌───────────────────────┐            │
│    │ 依赖图 · 死锁检测      │            │
│    │ 部分失败回滚           │            │
│    └───────────┬───────────┘            │
│                │                        │
│    第二层：参数幻觉治理                   │
│    ┌───────────┴───────────┐            │
│    │ Phantom检测 · 类型强制 │            │
│    │ 枚举白名单 · 漂移监控  │            │
│    └───────────┬───────────┘            │
│                │                        │
│    第一层：Schema强校验                   │
│    ┌───────────┴───────────┐            │
│    │ CFG约束解码 · Pydantic │            │
│    │ 双重校验 · 结构化重试  │            │
│    └───────────────────────┘            │
│                                         │
└─────────────────────────────────────────┘

ALLOWED_TOOLS = {"get_weather", "get_forecast", "search_location"}

def sanitize_tool_call(call: dict) -> dict | None:
    """拦截Phantom Call：工具名不在白名单则拒绝"""
    if call.get("name") not in ALLOWED_TOOLS:
        # 记录幽灵调用用于分析
        log_phantom_call(
            requested=call["name"],
            closest_match=find_closest(call["name"], ALLOWED_TOOLS),
            raw_output=call
        )
        return None  # 拒绝执行
    return call

from difflib import get_close_matches

VALID_CITIES = {"北京", "上海", "广州", "深圳", "杭州"}

def validate_enum_param(value: str, valid_set: set, field_name: str) -> str | None:
    if value in valid_set:
        return value
    # 尝试模糊匹配作为降级
    matches = get_close_matches(value, valid_set, n=1, cutoff=0.6)
    if matches:
        return matches[0]  # 降级到最近似合法值
    return None  # 触发重试

import hashlib
import json

def schema_fingerprint(schema: dict) -> str:
    """计算Schema指纹，检测漂移"""
    return hashlib.sha256(
        json.dumps(schema, sort_keys=True).encode()
    ).hexdigest()[:16]

# 部署时注册预期指纹
EXPECTED_FINGERPRINTS = {
    "get_weather": schema_fingerprint(weather_schema)
}

def check_schema_drift(tool_name: str, actual_output: dict):
    expected = EXPECTED_FINGERPRINTS.get(tool_name)
    actual = schema_fingerprint(infer_schema(actual_output))
    if expected != actual:
        alert_schema_drift(tool_name, expected, actual)

from dataclasses import dataclass
from typing import Set, Dict, List

@dataclass
class ToolCall:
    name: str
    params: dict
    reads_from: Set[str]   # 读取的数据源
    writes_to: Set[str]    # 写入的数据源

def build_dependency_graph(calls: List[ToolCall]) -> Dict[str, Set[str]]:
    """构建工具调用依赖图，检测隐式耦合"""
    deps = {}
    # 先建立写入映射
    writers = {}  # data_source -> [tool_name]
    for call in calls:
        for ds in call.writes_to:
            writers.setdefault(ds, []).append(call.name)

    # 再检查读取依赖
    for call in calls:
        deps[call.name] = set()
        for ds in call.reads_from:
            if ds in writers:
                deps[call.name].update(writers[ds])
    return deps

def detect_coupling(calls: List[ToolCall]) -> List[List[str]]:
    """检测并行分组中的耦合冲突"""
    deps = build_dependency_graph(calls)
    conflicts = []
    for tool, dep_set in deps.items():
        if dep_set:  # 存在依赖
            conflicts.append([tool] + list(dep_set))
    return conflicts

def parallel_with_deps(calls: List[ToolCall]) -> List[dict]:
    """依赖图感知的并行执行"""
    deps = build_dependency_graph(calls)
    results = {}
    executed = set()

    while len(executed) < len(calls):
        # 找出所有依赖已满足的调用
        ready = [
            c for c in calls 
            if c.name not in executed 
            and all(d in executed for d in deps.get(c.name, set()))
        ]
        if not ready:
            raise DeadlockError("工具调用存在循环依赖")

        # 并行执行就绪组
        group_results = await asyncio.gather(*[
            execute_tool(c) for c in ready
        ])
        for call, result in zip(ready, group_results):
            results[call.name] = result
            executed.add(call.name)

    return results

# 一个"中等复杂度"工具的Token开销示例
tool_definition = {
    "name": "search_flights",
    "description": "搜索航班信息，支持按出发地、目的地、日期、舱位等条件筛选",
    "parameters": {
        "type": "object",
        "properties": {
            "origin": {"type": "string", "description": "出发城市IATA代码"},
            "destination": {"type": "string", "description": "到达城市IATA代码"},
            "date": {"type": "string", "format": "date"},
            "cabin_class": {"type": "string", "enum": ["economy", "business", "first"]},
            "passengers": {"type": "integer", "minimum": 1, "maximum": 9},
            "direct_only": {"type": "boolean"},
            "max_stops": {"type": "integer", "minimum": 0, "maximum": 3},
            "price_max": {"type": "number"}
        },
        "required": ["origin", "destination", "date"]
    }
}
# 上方定义 ≈ 420 tokens（Claude tokenizer）

async def semantic_tool_select(user_query: str, tool_catalog: list) -> list:
    """用轻量模型做工具预选，只加载相关工具到上下文"""
    # 步骤1：用embedding检索Top-K相关工具
    query_emb = await embed(user_query)
    scores = [cosine_sim(query_emb, t["embedding"]) for t in tool_catalog]
    top_k = sorted(range(len(scores)), key=lambda i: scores[i], reverse=True)[:5]

    # 步骤2：只加载Top-5工具的完整Schema
    selected_tools = [tool_catalog[i]["full_schema"] for i in top_k]

    # 步骤3：附加"其他可用工具"的简要列表
    other_tools = [
        {"name": tool_catalog[i]["name"], "one_liner": tool_catalog[i]["summary"]}
        for i in range(len(tool_catalog)) if i not in top_k
    ]
    return selected_tools, other_tools

sequenceDiagram
    participant User
    participant Agent
    participant MCP_Registry
    participant Tool_Server

    User->>Agent: "帮我订明天去上海的机票"
    Agent->>MCP_Registry: semantic_search("flight booking travel")
    MCP_Registry-->>Agent: [book_flight, search_flights, get_airport_info]
    Agent->>Tool_Server: tools/list (filtered: 3 tools)
    Tool_Server-->>Agent: 完整Schema (3 tools ≈ 1200 tokens)
    Agent->>Agent: 选择并调用 search_flights
    Agent->>Agent: 选择并调用 book_flight
    Agent-->>User: "已为您预订CA1234航班..."

class ProductionToolMiddleware:
    """生产级工具调用中间件：集成三层精度工程"""

    def __init__(self, config):
        self.schema_validator = PydanticValidator(config.schemas)
        self.tool_whitelist = set(config.allowed_tools)
        self.dep_analyzer = DependencyAnalyzer()
        self.token_budget = TokenBudget(max_per_turn=config.max_tokens)
        self.drift_monitor = SchemaDriftMonitor(config.fingerprints)

    async def execute(self, model_output: dict) -> ToolResult:
        # 第一层：Schema强校验
        validated = self.schema_validator.validate(model_output)
        if not validated.ok:
            return self._retry_with_feedback(validated.errors)

        # 第二层：Phantom Call + 参数幻觉
        if model_output["name"] not in self.tool_whitelist:
            return self._reject_phantom(model_output)

        # 第三层：并行耦合检测
        conflicts = self.dep_analyzer.check(model_output)
        if conflicts:
            model_output = self._serialize_conflicting(model_output, conflicts)

        # Token预算检查
        if not self.token_budget.check(model_output):
            return self._trim_tool_descriptions(model_output)

        # 执行 + Schema漂移监控
        result = await self._execute_with_monitoring(model_output)
        self.drift_monitor.check(model_output["name"], result)

        return result