graph TB
    subgraph "工具调用精度七层断层"
        A["第1层：Schema理解<br/>模型能否读懂工具定义"] --> B["第2层：函数选择<br/>从N个工具中选对的"]
        B --> C["第3层：参数填充<br/>类型、枚举、约束全对"]
        C --> D["第4层：单次执行<br/>一次调用即成功"]
        D --> E["第5层：多轮链条<br/>3+步工具链不中断"]
        E --> F["第6层：错误恢复<br/>失败后自动修正"]
        F --> G["第7层：有状态交互<br/>跨调用保持上下文"]
    end

    style A fill:#4CAF50,color:#fff
    style B fill:#4CAF50,color:#fff
    style C fill:#8BC34A,color:#fff
    style D fill:#FFC107,color:#000
    style E fill:#FF9800,color:#fff
    style F fill:#FF5722,color:#fff
    style G fill:#F44336,color:#fff

from datasets import load_dataset
from transformers import AutoModelForCausalLM, AutoTokenizer
from trl import SFTTrainer, SFTConfig

# 加载 xLAM 数据集
dataset = load_dataset("Salesforce/xlam-function-calling-60k")

def format_tool_call(example):
    """将 xLAM 格式转换为对话格式"""
    system_msg = "You are a helpful assistant with access to the following tools:"
    for tool in example["tools"]:
        system_msg += f"\n{tool}"

    return {
        "messages": [
            {"role": "system", "content": system_msg},
            {"role": "user", "content": example["query"]},
            {"role": "assistant", "content": example["answer"]}
        ]
    }

# QLoRA 配置 - 单卡可跑
model = AutoModelForCausalLM.from_pretrained(
    "meta-llama/Meta-Llama-3-8B-Instruct",
    load_in_4bit=True,
    device_map="auto"
)

trainer = SFTTrainer(
    model=model,
    train_dataset=dataset.map(format_tool_call),
    args=SFTConfig(
        max_seq_length=2048,
        per_device_train_batch_size=4,
        gradient_accumulation_steps=8,
        learning_rate=2e-4,
        num_train_epochs=3,
        output_dir="./llama3-tool-calling"
    )
)
trainer.train()

# DeepSeek V4 工具调用示例（简化）
from openai import OpenAI

client = OpenAI(
    base_url="https://api.deepseek.com",
    api_key="your-key"
)

tools = [{
    "type": "function",
    "function": {
        "name": "query_inventory",
        "description": "查询指定SKU的库存状态",
        "parameters": {
            "type": "object",
            "properties": {
                "sku": {"type": "string", "description": "SKU编号"},
                "warehouse": {
                    "type": "string",
                    "enum": ["beijing", "shanghai", "guangzhou"],
                    "description": "仓库位置"
                }
            },
            "required": ["sku"]
        }
    }
}]

response = client.chat.completions.create(
    model="deepseek-v4-pro",
    messages=[{"role": "user", "content": "BJ-001这个SKU在三个仓库的库存情况"}],
    tools=tools,
    tool_choice="auto"
)
# DeepSeek V4 能自动识别需要并行调用三次
# 参数正确拆分到三个仓库

# 差的 Schema
{
    "name": "get_data",
    "description": "获取数据",
    "parameters": {"type": "object", "properties": {"id": {"type": "string"}}}
}

# 好的 Schema  
{
    "name": "get_customer_order_history",
    "description": "根据客户ID查询其所有历史订单，返回订单列表。每个订单包含订单号、下单时间、商品明细、总金额和状态。仅返回最近12个月的订单。",
    "parameters": {
        "type": "object",
        "properties": {
            "customer_id": {
                "type": "string",
                "description": "客户唯一标识符，格式为CUST-XXXX，例如CUST-1234"
            },
            "include_cancelled": {
                "type": "boolean",
                "description": "是否包含已取消的订单，默认false",
                "default": false
            }
        },
        "required": ["customer_id"]
    }
}

graph LR
    subgraph "工具路由架构"
        U["用户请求"] --> R["路由层<br/>轻量模型/Qwen3-8B"]
        R -->|Top-5 候选| M["主模型<br/>DeepSeek V4-Pro"]
        M --> T["精确工具调用"]

        DB["工具库<br/>100+ 工具"] --> R
    end

    style R fill:#FF9800,color:#fff
    style M fill:#4CAF50,color:#fff
    style T fill:#2196F3,color:#fff

# 完整 Schema（~150 tokens）
{
    "name": "send_email",
    "description": "发送电子邮件给指定收件人。支持HTML和纯文本格式。可添加附件路径。...",
    "parameters": {
        "type": "object",
        "properties": {
            "to": {"type": "string", "description": "收件人邮箱地址"},
            "subject": {"type": "string", "description": "邮件主题"},
            "body": {"type": "string", "description": "邮件正文内容"},
            "format": {"type": "string", "enum": ["html", "text"]},
            "cc": {"type": "string", "description": "抄送地址"},
            "attachments": {"type": "array", "items": {"type": "string"}}
        },
        "required": ["to", "subject", "body"]
    }
}

# 压缩 Schema（~50 tokens）- 精度损失 < 3%
{
    "name": "send_email",
    "description": "发邮件：to/subject/body必填，cc/attachments可选",
    "parameters": {
        "type": "object",
        "properties": {
            "to": {"type": "string"},
            "subject": {"type": "string"},
            "body": {"type": "string"},
            "cc": {"type": "string"},
            "attachments": {"type": "array", "items": {"type": "string"}}
        },
        "required": ["to", "subject", "body"]
    }
}

# 隐耦合检测器
def detect_coupling(tool_calls: list[dict]) -> list[list[int]]:
    """检测工具调用间的数据依赖，返回依赖组"""
    dependencies = []
    for i, call_a in enumerate(tool_calls):
        for j, call_b in enumerate(tool_calls):
            if i >= j:
                continue
            # 检查 call_b 的参数是否引用 call_a 的返回字段
            params_b = json.dumps(call_b["parameters"])
            return_fields_a = get_expected_return_fields(call_a["name"])
            for field in return_fields_a:
                if field in params_b:
                    dependencies.append([i, j])
    return dependencies

# 使用：有依赖的串行执行，无依赖的并行执行
groups = detect_coupling(tool_calls)
parallel_groups = group_by_independence(tool_calls, groups)
for group in parallel_groups:
    results = await asyncio.gather(*[execute(call) for call in group])

graph TD
    subgraph "混合路由引擎"
        Q["工具调用请求"] --> C{"复杂度分类器"}
        C -->|"简单<br/>单工具+少参数"| S["轻量层<br/>Qwen3-8B微调模型<br/>成本: ¥0.0005/次"]
        C -->|"中等<br/>多工具选择"| M["标准层<br/>DeepSeek V4<br/>成本: ¥0.002/次"]
        C -->|"复杂<br/>多轮链条+推理"| H["旗舰层<br/>GPT-5.4/Claude Opus<br/>成本: ¥0.015/次"]
    end

    S --> R["结果验证器"]
    M --> R
    H --> R
    R -->|"通过"| O["返回结果"]
    R -->|"失败"| U["升级到上层模型"]
    U --> R

    style C fill:#FF9800,color:#fff
    style S fill:#8BC34A,color:#fff
    style M fill:#2196F3,color:#fff
    style H fill:#9C27B0,color:#fff

# 概念示意：基于内部表征的幻觉检测
def detect_hallucination_from_internals(model, tool_call, tools):
    """
    对比模型生成 tool_call 时的内部激活模式
    与"正确调用"的典型激活模式
    """
    # 提取模型生成 tool name 时的 logprobs
    logprobs = model.get_logprobs(tool_call.function.name)

    # 如果模型对工具名的不确定性高（logprob 低），
    # 说明它在"猜"而不是"知道"
    if logprobs.top_logprob < 0.3:
        return {"hallucination_risk": "HIGH", "action": "reject"}

    # 检查参数值是否出现在工具定义的约束中
    for param, value in tool_call.function.arguments.items():
        if param not in tools[tool_call.function.name].parameters:
            return {"hallucination_risk": "HIGH", "action": "reject"}

    return {"hallucination_risk": "LOW", "action": "proceed"}