graph TB
    subgraph "第一世代：Schema合规"
        A1["BFCL V1-V2<br/>2023-2024"]
        A2["核心问题：模型能否生成<br/>正确格式的函数调用？"]
        A3["测试方式：静态JSON Schema<br/>→ 单次调用准确率"]
        A1 --> A2 --> A3
    end

    subgraph "第二世代：工作流编排"
        B1["Tau-bench / ToolBench<br/>2024-2025"]
        B2["核心问题：Agent能否在<br/>多轮对话中编排工具链？"]
        B3["测试方式：模拟用户交互<br/>→ 多步骤任务完成率"]
        B1 --> B2 --> B3
    end

    subgraph "第三世代：协议生态"
        C1["MCP-Bench / MCP Atlas<br/>2025-2026"]
        C2["核心问题：Agent能否在<br/>真实协议环境中发现并使用工具？"]
        C3["测试方式：28个MCP服务器<br/>→ 工具发现+编排全链路"]
        C1 --> C2 --> C3
    end

    A3 -->|"准确率天花板~90%"| B1
    B3 -->|"完成率天花板~67%"| C1
    C3 -->|"现实失败率12-18%"| C1

    style A1 fill:#4a90d9,color:#fff
    style B1 fill:#e67e22,color:#fff
    style C1 fill:#27ae60,color:#fff

# BFCL V4评估核心流程（简化）
from bfcl import BFCLPipeline

pipeline = BFCLPipeline(version="v4")

# 定义测试用例：给定工具schema和用户查询
test_case = {
    "tools": [
        {
            "name": "get_weather",
            "parameters": {
                "location": {"type": "string"},
                "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
            }
        },
        {
            "name": "search_flights",
            "parameters": {
                "origin": {"type": "string"},
                "destination": {"type": "string"},
                "date": {"type": "string", "format": "date"}
            }
        }
    ],
    "query": "What's the weather in Tokyo and find flights from NYC to Tokyo on March 15?",
    "expected": [
        {"name": "get_weather", "parameters": {"location": "Tokyo"}},
        {"name": "search_flights", "parameters": {
            "origin": "NYC", "destination": "Tokyo", "date": "2026-03-15"
        }}
    ]
}

# 评估：不仅检查格式，还执行验证
result = pipeline.evaluate(
    model="claude-opus-4-1",
    test_case=test_case,
    mode="executable"  # V4新增：实际执行验证
)
# Output: {accuracy: 0.7036, ...}

# Tau-bench核心评估循环（简化）
class TauBenchEvaluator:
    def __init__(self, domain="airline"):
        self.tool_schemas = load_domain_tools(domain)
        self.policies = load_policies(domain)
        self.db = init_database(domain)  # 带状态的真实数据库

    def evaluate_agent(self, agent_fn, test_cases):
        results = []
        for case in test_cases:
            db_state = self.db.snapshot()
            user_sim = UserSimulator(
                intent=case.user_intent,
                persona=case.user_persona,
                patience=case.max_turns
            )

            conversation = []
            for turn in range(case.max_turns):
                user_msg = user_sim.generate(conversation)
                agent_response = agent_fn(
                    user_msg, 
                    tools=self.tool_schemas,
                    policies=self.policies,
                    db_state=db_state
                )

                # 执行Agent的工具调用并更新数据库
                if agent_response.tool_calls:
                    for call in agent_response.tool_calls:
                        db_state = self.execute_tool(call, db_state)

                conversation.append((user_msg, agent_response))

                if user_sim.is_satisfied(conversation):
                    results.append({"success": True, "turns": turn})
                    break
            else:
                results.append({"success": False, "turns": turn})

        return sum(r["success"] for r in results) / len(results)

# MCP-Bench核心评估架构（简化）
from mcp_bench import MCPBenchRunner, MCPServerPool

# 初始化28个真实MCP服务器
servers = MCPServerPool([
    "mcp://finance-server:8080",    # 金融数据查询
    "mcp://medical-db:8081",         # 医疗记录检索
    "mcp://devops-tools:8082",       # CI/CD操作
    "mcp://data-analytics:8083",     # 数据分析工具
    # ... 24个更多服务器
])

runner = MCPBenchRunner(servers)

# 测试用例：模糊指令 → 工具发现 → 编排执行
task = {
    "instruction": "帮我把上周的销售数据做个趋势分析，然后发给团队",
    # Agent需要: 1) 发现数据分析MCP服务器
    #           2) 找到合适的查询和分析工具
    #           3) 确定时间参数"上周"
    #           4) 编排查询→分析→发送的工具链
    "evaluation": {
        "tool_discovery": "是否找到了正确的MCP服务器和工具",
        "parameter_inference": "是否正确推断'上周'的日期范围",
        "orchestration": "是否按正确顺序编排工具链",
        "error_handling": "工具执行失败时是否有恢复策略"
    }
}

result = runner.evaluate(model="claude-opus-4-1", tasks=[task])
# MCP-Bench多维度评分，非单一准确率

graph LR
    subgraph "BFCL优势型模型"
        A["格式精准<br/>参数提取强<br/>JSON Schema理解深"]
        A --> A1["适用：API网关<br/>结构化数据提取"]
    end

    subgraph "Tau-bench优势型模型"
        B["策略遵循<br/>多轮对话<br/>状态管理"]
        B --> B1["适用：客服自动化<br/>业务流程编排"]
    end

    subgraph "MCP-Bench优势型模型"
        C["工具发现<br/>模糊理解<br/>协议适配"]
        C --> C1["适用：通用Agent平台<br/>工具市场"]
    end

    A1 -->|"互补"| B1
    B1 -->|"互补"| C1
    C1 -->|"互补"| A1

    style A fill:#3498db,color:#fff
    style B fill:#e74c3c,color:#fff
    style C fill:#2ecc71,color:#fff

# 生产级工具调用容错架构（简化）
class ResilientToolCaller:
    def __init__(self, model_client, max_retries=3):
        self.model = model_client
        self.max_retries = max_retries
        self.circuit_breaker = CircuitBreaker(
            failure_threshold=5,
            recovery_timeout=60
        )
        self.schema_cache = SchemaCache(ttl=3600)  # 缓存工具schema

    async def call_tool(self, user_query, available_tools):
        # 1. 获取最新schema（处理schema漂移）
        fresh_schemas = await self.schema_cache.get_or_refresh(
            available_tools
        )

        for attempt in range(self.max_retries):
            try:
                # 2. 模型推理（带超时）
                tool_call = await asyncio.wait_for(
                    self.model.generate_tool_call(
                        query=user_query,
                        tools=fresh_schemas
                    ),
                    timeout=10.0
                )

                # 3. 参数验证（防止幻觉调用）
                if tool_call.name not in fresh_schemas:
                    raise ToolNotFoundError(tool_call.name)
                validate_params(tool_call, fresh_schemas[tool_call.name])

                # 4. 执行工具（带熔断器）
                result = await self.circuit_breaker.call(
                    execute_tool, tool_call
                )
                return result

            except ToolNotFoundError:
                # 幻觉调用：重新提示模型
                continue
            except ParamValidationError:
                # 参数错误：带错误信息重试
                continue
            except CircuitOpenError:
                # 熔断：使用降级策略
                return self.fallback_response(user_query)
            except asyncio.TimeoutError:
                # 超时：记录并重试
                continue

        raise ToolCallExhaustedError(attempt=self.max_retries)

flowchart TD
    A[需要评估Agent工具使用能力] --> B{生产场景是什么？}

    B -->|结构化API调用| C[运行BFCL V4评测]
    B -->|多轮业务流程| D[运行Tau-bench评测]
    B -->|MCP生态集成| E[运行MCP-Bench评测]
    B -->|复杂推理链| F[运行ToolHop评测]
    B -->|大规模工具库| G[运行ToolACE评测]

    C --> H{BFCL V4 > 65%？}
    D --> I{Tau-bench > 50%？}
    E --> J{MCP-Bench工具发现 > 60%？}
    F --> K{ToolHop多跳 > 40%？}
    G --> L{ToolACE@200工具 > 50%？}

    H -->|是| M[进入生产验证阶段]
    H -->|否| N[优化Prompt或换模型]
    I -->|是| M
    I -->|否| N
    J -->|是| M
    J -->|否| N
    K -->|是| M
    K -->|否| N
    L -->|是| M
    L -->|否| N

    M --> O[部署12%容错层<br/>重试+熔断+降级]
    N --> P[针对性优化<br/>Schema设计/Prompt调优/模型切换]

    O --> Q[持续监控<br/>生产失败率 < 8%]
    P --> B

    style A fill:#2c3e50,color:#fff
    style M fill:#27ae60,color:#fff
    style Q fill:#27ae60,color:#fff
    style N fill:#e74c3c,color:#fff