MCP 服务器评估与基准测试：工具选择、参数填充与任务成功率

# 1) 改坏不自知: 优化了工具描述，某些场景反而变差，无人发现
# 2) 无法比较: 两种 schema 设计哪个好？只能拍脑袋
# 3) 无法回归: 模型升级、SDK 升级后，工具可用性可能骤降
# 评估是"让迭代有方向"的基础设施

# 一次评测的原始记录 → 分层指标
def compute_metrics(records: list[dict]) -> dict:
    n = len(records)
    tool_ok = sum(r["selected_tool"] == r["expected_tool"] for r in records)
    param_ok = sum(r["params_match"] for r in records)
    task_ok = sum(r["task_success"] for r in records)
    return {
        "tool_selection_acc": tool_ok / n,
        "param_acc": param_ok / n,
        "task_success_rate": task_ok / n,
        "avg_turns": sum(r["turns"] for r in records) / n,
        "invalid_call_rate": sum(r["invalid_calls"] for r in records)
                             / sum(r["turns"] for r in records),
    }

# 质量之外，还有两个必看维度
# 1) token 成本: 工具定义 + 调用往返 + 结果注入的总 token
# 2) 延迟: P50/P95 端到端耗时（含模型推理与工具执行）
# 一个"准确率 95% 但每次花 3 万 token"的服务器，可能不如"90% 但 8 千 token"
# 质量-成本-延迟三者要一起看，不能只优化一个

{
  "id": "search-001",
  "category": "tool_selection",
  "task": "用户问：帮我找出上周合并的所有 PR",
  "expected_tool": "github__list_prs",
  "expected_params": { "state": "merged", "since": "2026-09-27" },
  "forbidden_tools": ["github__merge_pr"],
  "success_criteria": "调用 list_prs 且 state=merged、时间范围正确",
  "tags": ["read", "github", "time-range"]
}

# 建议构成（一个成熟的 MCP 服务器基准集）
# 1) 核心用例 100-300 条: 覆盖主要工具与典型场景
# 2) 边界用例 30-50 条: 空结果、超长输入、异常参数
# 3) 安全用例 20-40 条: 越权、注入、危险操作
# 4) 负例 20-30 条: 本不该调用任何工具（应拒绝/澄清）
# 5) 多步用例 20-40 条: 需组合多个工具的任务
# 总量 200-400 条是常见起点；太少则噪声大，太多则维护难

async def eval_tool_selection(case: dict, client) -> dict:
    tools = await client.list_tools()          # 服务器暴露的工具集
    # 让模型在"只能选工具、不能执行"的模式下做选择
    choice = await model.select_tool(
        task=case["task"],
        tools=tools,
        # 关键: 考察"选择"而非"执行"
        allow_execute=False,
    )
    selected = choice.tool_name
    return {
        "id": case["id"],
        "expected_tool": case["expected_tool"],
        "selected_tool": selected,
        "correct": selected == case["expected_tool"],
        "forbidden_hit": selected in case.get("forbidden_tools", []),
    }

# 工具选择评估的关键是"干扰项"
# 1) 相似工具: list_prs vs list_issues（名字相近）
# 2) 重叠能力: search_code vs get_file（都能拿代码）
# 3) 误导描述: 描述里说"也可以用来 X"（诱导误选）
# 4) 高频陷阱: 用最熟悉的工具解决所有问题
# 没有干扰项的评估，准确率会虚高（工具少时随便选都对）

from datetime import datetime

def params_match(actual: dict, expected: dict, case: dict) -> dict:
    fields, correct = 0, 0
    details = {}
    for k, exp in expected.items():
        fields += 1
        act = actual.get(k)
        ok = compare_value(act, exp, case.get("fuzzy_fields", {}))
        correct += int(ok)
        details[k] = {"expected": exp, "actual": act, "ok": ok}
    # 额外参数（不该有的）
    extra = set(actual) - set(expected) - set(case.get("allowed_extra", []))
    return {
        "all_match": correct == fields and not extra,
        "field_acc": correct / fields if fields else 1.0,
        "details": details,
        "extra_params": sorted(extra),
    }

# 有些参数不能要求逐字相等
# 1) 时间: "上周" → 允许多种合法区间表达
# 2) 数量: limit=10 vs limit=20 可能都对（在合理范围）
# 3) 路径: ./a/b 与 a/b 等价
# 4) 枚举: "merged" vs "MERGED"（大小写）
# 评估器要支持"语义等价"判定，否则会误判正确为错误

async function evalTask(case: TaskCase, agent: Agent): Promise<TaskResult> {
  const session = await agent.start();
  let turns = 0;
  const calls: ToolCall[] = [];
  try {
    for await (const step of agent.run(case.task, { maxTurns: case.maxTurns ?? 12 })) {
      turns++;
      if (step.type === "tool_call") calls.push(step);
    }
    const final = await session.finalAnswer();
    return {
      id: case.id,
      success: await judgeTask(case, final, calls),
      turns,
      calls,
      cost_tokens: session.usage.totalTokens,
    };
  } finally {
    await session.close();
  }
}

# 端到端"成功"的判定要素
# 1) 最终答案正确（对照标准答案/关键点）
# 2) 过程合规（没调用被禁工具、没越权）
# 3) 效率合理（轮次/成本在预算内）
# 4) 无副作用（未产生不该有的写操作）
# 只看"答案对不对"会漏掉"过程是否安全"

JUDGE_PROMPT = """你是 MCP 工具使用评估员。请判断下面这次任务是否成功。

任务: {task}
期望: {expected}
实际工具调用: {calls}
最终答案: {answer}

评分维度（各 0-2 分）:
1. 答案正确性: 是否解决了任务
2. 工具使用: 是否用了正确的工具与参数
3. 过程合规: 是否调用了禁止的工具或越权

只输出 JSON: {{"correctness": n, "tool_use": n, "compliance": n, "reason": "..."}}
"""

async def judge_task(case, answer, calls) -> dict:
    raw = await judge_model.complete(JUDGE_PROMPT.format(
        task=case["task"], expected=case["expected"],
        calls=serialize(calls), answer=answer,
    ))
    return parse_json(raw)

# LLM-as-judge 的三个可靠性问题
# 1) 位置偏差: 倾向选先出现的选项 → 随机化顺序
# 2) 自偏好: 偏好自己模型的输出 → 用不同模型做裁判
# 3) 分数漂移: 同一输入不同次给不同分 → 固定温度、多次取众数
# 关键: 用人工标注的"金标准"校准 judge，报告 judge 与人的一致率
# judge 不是真理，是"可扩展的近似"——必须校准

# 1) 建立基线: 当前版本的指标快照
# 2) 改动: 改工具描述/schema/实现
# 3) 重跑基准集: 用同一模型、同一配置
# 4) 对比: 逐用例 diff，标出变好/变坏
# 5) 决策: 净提升则合入，有回归则分析
# 回归要"逐用例看"，不能只看总分——总分持平可能掩盖此消彼长

# 指标变化是否显著？别被噪声骗了
from statistics import mean, stdev
import math

def is_significant(before: list[float], after: list[float], alpha=0.05) -> bool:
    n1, n2 = len(before), len(after)
    m1, m2 = mean(before), mean(after)
    s1, s2 = stdev(before), stdev(after)
    # 简易双样本 t 检验
    se = math.sqrt(s1**2 / n1 + s2**2 / n2)
    if se == 0:
        return m1 != m2
    t = abs(m1 - m2) / se
    return t > 1.96   # 近似 95% 置信

# 对比两种设计（如两种工具描述）
# 1) 同一基准集，随机分组（每条用例随机分配 A 或 B）
# 2) 固定模型与温度（消除随机性）
# 3) 每条用例跑多次（如 5 次）取平均，降低方差
# 4) 报告均值 ± 置信区间，而非单点
# 5) 同时看质量、成本、延迟三个维度
# "A 比 B 高 2%" 若不显著，就是噪声

# CI 中的 MCP 评估流水线
stages:
  - name: smoke
    cases: benchmark/smoke/*.json      # 20 条，快速反馈（<2min）
  - name: core
    cases: benchmark/core/*.json       # 200 条，合入前跑
    gate: { tool_selection_acc: 0.90, task_success_rate: 0.80 }
  - name: safety
    cases: benchmark/safety/*.json     # 安全用例，必须 100%
    gate: { violation_count: 0 }
  - name: report
    run: mcp-eval report --baseline main --current HEAD

# 评估结果作为合入门禁
def gate(result: dict, thresholds: dict) -> tuple[bool, list[str]]:
    failures = []
    for metric, minimum in thresholds.items():
        if result[metric] < minimum:
            failures.append(f"{metric}={result[metric]:.3f} < {minimum}")
    # 回归门禁: 相比基线不能明显变差
    for metric, delta in result["regressions"].items():
        if delta < -0.03:
            failures.append(f"{metric} regressed by {delta:.3f}")
    return (not failures), failures

# 1) 固定模型版本（不写 "latest"）
# 2) 固定温度与随机种子
# 3) 固定工具集与服务器版本
# 4) 记录完整的评测元数据（模型、版本、时间、配置）
# 5) 基准集版本化（用例变更要可追溯）
# 不可复现的评估等于没有评估