Agent 轨迹与协议鲁棒性

Agent 任务和普通问答最大的区别是：最终答案只是结果层，真正决定稳定性的，是中间计划、工具选择、参数拼装、观察解释、重试与回退策略。测试如果只看结果对不对，就会漏掉大量"暂时没炸，但随时会炸"的系统性风险。

教学导读

**定位：**这一章把 Agent 从"会答题"提升到"会执行系统任务"的测试视角，重点在轨迹、协议、权限和恢复能力。 **前置依赖：**建议已学习 Agent、Tool Calling、MCP 和接口契约测试基础。 **适用场景：**客服 Agent、工作流 Agent、工具编排、多工具任务、MCP Server/Client 联调。 **学完产出：**你应该能同时对结果、轨迹、工具契约和权限边界做系统级测试设计。

先说结论。

Agent 最难测的地方在于"答案对了也可能过程错，答案错了也可能过程大体合理"。所以真正成熟的 Agent 测试不会只做结果断言，而会同时关注轨迹合法性、工具调用正确性、权限边界、幂等性、超时和重试行为。这本质上是把 Agent 当成一个分布式系统来测试，而不是把它当成一个问答模型来测试。

01. 轨迹评测：为什么要看 plan / action / observation

一条最常见的 Agent 轨迹可以拆成：

用户目标
→ 任务拆解（plan）
→ 工具选择（action）
→ 工具返回（observation）
→ 解释结果 / 重新规划（replan）
→ 最终回答

这个流程在学术上被称为 ReAct (Reasoning + Acting) 范式。其核心思想是：Agent 在每一步都先"思考"下一步该做什么（Thought），然后执行一个动作（Action），观察结果（Observation），再基于观察进行下一步推理。

阶段	常见错误	测试方式	严重程度
Plan	漏步骤、顺序错、目标理解错	轨迹断言、关键步骤覆盖	高 — 影响整体执行方向
Action	选错工具、参数拼错	schema 与参数校验	高 — 可能产生副作用
Observation	误解工具返回	返回值解释断言	中 — 影响后续决策
Replan	遇错不回退、盲目继续	失败恢复测试	高 — 错误会被放大
Termination	不终止、过早终止	终止条件测试	中 — 影响完整性和成本

1.1 一个"答案对了，但轨迹已经坏了"的例子

用户问题：帮我查订单 123 是否可退款

错误轨迹：
  Thought: 我需要查这个订单的信息
  Action:  search_product(query="订单123")           ← 选错工具
  Obs:     返回了一些商品搜索结果
  
  Thought: 这不对，我应该查订单
  Action:  get_order(order_id="123")                 ← 纠正了
  Obs:     {"order_id":"123", "status":"delivered", "refundable":true}
  
  Thought: 订单状态是已送达，可退款
  Answer:  "该订单可退款"                              ← 答案对了

为什么仍然要判问题：
  1. 轨迹里有多余工具调用，成本和延迟都上升
  2. 如果 search_product 恰好返回冲突信息，后续可能答错
  3. 这说明 plan 层已经不稳定，不是一个可放心放大的系统
  4. 在真实流量下，每一次多余调用都会叠加成本和延迟

这就是 Agent 测试和普通问答测试的分水岭。普通问答里"结果对"常常已经够用；Agent 系统里"结果对但过程错"意味着它只是这次没炸，不代表下次不炸。

01.5 轨迹的形式化定义与度量

要做可量化的轨迹评测，首先需要形式化地定义轨迹：

轨迹的形式化表示

# 轨迹的形式化定义
from dataclasses import dataclass, field
from typing import Optional, Any
from enum import Enum

class StepType(Enum):
    THOUGHT = "thought"
    ACTION = "action"
    OBSERVATION = "observation"
    ANSWER = "answer"

@dataclass
class TrajectoryStep:
    step_id: int
    step_type: StepType
    content: str
    tool_name: Optional[str] = None
    tool_args: Optional[dict] = None
    tool_result: Optional[Any] = None
    timestamp_ms: int = 0
    token_cost: int = 0

@dataclass
class Trajectory:
    task_id: str
    goal: str
    steps: list[TrajectoryStep] = field(default_factory=list)
    final_answer: Optional[str] = None
    total_cost_tokens: int = 0
    total_time_ms: int = 0
    
    @property
    def actions(self) -> list[TrajectoryStep]:
        return [s for s in self.steps if s.step_type == StepType.ACTION]
    
    @property
    def tool_calls(self) -> list[str]:
        return [s.tool_name for s in self.actions if s.tool_name]
    
    @property
    def unique_tools(self) -> set[str]:
        return set(self.tool_calls)
    
    @property
    def num_retries(self) -> int:
        """统计重试次数（同一工具连续调用）"""
        retries = 0
        prev = None
        for call in self.tool_calls:
            if call == prev:
                retries += 1
            prev = call
        return retries

轨迹质量度量指标

定义了形式化表示后，可以设计一组量化指标来衡量轨迹质量：

指标	计算方式	含义	参考阈值
步骤效率	最优步骤数 / 实际步骤数	轨迹是否有冗余	> 0.8 为优
工具准确率	正确工具调用 / 总工具调用	工具选择是否正确	> 0.95
参数准确率	参数正确的调用 / 总调用	参数拼装质量	> 0.90
重试率	重试次数 / 总调用次数	一次成功率	< 0.1
回退正确率	正确回退 / 需回退的场景	失败恢复能力	> 0.85
token 效率	最优 token / 实际 token	成本控制	> 0.6

python

class TrajectoryMetrics:
    """轨迹质量度量计算器"""
    
    def compute(self, trajectory: Trajectory, 
                golden_trajectory: Trajectory) -> dict:
        return {
            "step_efficiency": self._step_efficiency(
                trajectory, golden_trajectory),
            "tool_accuracy": self._tool_accuracy(
                trajectory, golden_trajectory),
            "param_accuracy": self._param_accuracy(
                trajectory, golden_trajectory),
            "retry_rate": self._retry_rate(trajectory),
            "token_efficiency": self._token_efficiency(
                trajectory, golden_trajectory),
            "total_cost": trajectory.total_cost_tokens,
            "total_time_ms": trajectory.total_time_ms,
        }
    
    def _step_efficiency(self, actual, golden) -> float:
        if len(actual.actions) == 0:
            return 0.0
        return min(1.0, len(golden.actions) / len(actual.actions))
    
    def _tool_accuracy(self, actual, golden) -> float:
        """检查工具选择序列的匹配度"""
        golden_tools = golden.tool_calls
        actual_tools = actual.tool_calls
        
        if not golden_tools:
            return 1.0 if not actual_tools else 0.0
        
        # 使用最长公共子序列 (LCS) 来衡量
        lcs_len = self._lcs(golden_tools, actual_tools)
        return lcs_len / len(golden_tools)
    
    def _lcs(self, seq1, seq2) -> int:
        """最长公共子序列长度"""
        m, n = len(seq1), len(seq2)
        dp = [[0] * (n + 1) for _ in range(m + 1)]
        for i in range(1, m + 1):
            for j in range(1, n + 1):
                if seq1[i-1] == seq2[j-1]:
                    dp[i][j] = dp[i-1][j-1] + 1
                else:
                    dp[i][j] = max(dp[i-1][j], dp[i][j-1])
        return dp[m][n]
    
    def _retry_rate(self, trajectory) -> float:
        calls = trajectory.tool_calls
        if not calls:
            return 0.0
        return trajectory.num_retries / len(calls)
    
    def _param_accuracy(self, actual, golden) -> float:
        """逐步对比参数正确性"""
        if not golden.actions:
            return 1.0
        
        correct = 0
        total = min(len(actual.actions), len(golden.actions))
        
        for a, g in zip(actual.actions, golden.actions):
            if a.tool_name == g.tool_name and a.tool_args == g.tool_args:
                correct += 1
        
        return correct / total if total > 0 else 0.0
    
    def _token_efficiency(self, actual, golden) -> float:
        if actual.total_cost_tokens == 0:
            return 0.0
        return min(1.0, golden.total_cost_tokens / actual.total_cost_tokens)

01.6 用 LLM 做轨迹 Judge

除了基于规则的指标，还可以用 LLM 做更语义化的轨迹评判：

TRAJECTORY_JUDGE_PROMPT = """你是一个 Agent 轨迹质量评审员。

用户目标: {goal}

Agent 执行轨迹:
{trajectory_text}

最终答案: {final_answer}

请从以下维度评估轨迹质量，每项 1-5 分：

1. 计划合理性：任务拆解是否正确、完整、顺序合理
2. 工具选择：每步是否选择了最合适的工具
3. 参数正确性：工具参数是否语义正确（不只是格式合法）
4. 观察解释：对工具返回的理解是否准确
5. 错误恢复：遇到失败时是否做了合理的重试或回退
6. 效率：是否有冗余步骤、不必要的工具调用
7. 最终答案：结果是否正确、完整

对每项给出分数和简短理由，最后给出总体评价。

输出 JSON 格式：
{{
  "plan_quality": {{"score": 1-5, "reason": "..."}},
  "tool_selection": {{"score": 1-5, "reason": "..."}},
  "param_correctness": {{"score": 1-5, "reason": "..."}},
  "observation_interpretation": {{"score": 1-5, "reason": "..."}},
  "error_recovery": {{"score": 1-5, "reason": "..."}},
  "efficiency": {{"score": 1-5, "reason": "..."}},
  "final_answer": {{"score": 1-5, "reason": "..."}},
  "overall": {{"score": 1-5, "summary": "..."}}
}}"""

class TrajectoryJudge:
    def __init__(self, llm_client):
        self.llm_client = llm_client
    
    def evaluate(self, trajectory: Trajectory) -> dict:
        trajectory_text = self._format_trajectory(trajectory)
        
        prompt = TRAJECTORY_JUDGE_PROMPT.format(
            goal=trajectory.goal,
            trajectory_text=trajectory_text,
            final_answer=trajectory.final_answer
        )
        
        response = self.llm_client.generate(
            prompt, temperature=0.1, response_format="json"
        )
        
        return json.loads(response)
    
    def _format_trajectory(self, trajectory: Trajectory) -> str:
        lines = []
        for step in trajectory.steps:
            if step.step_type == StepType.THOUGHT:
                lines.append(f"Thought: {step.content}")
            elif step.step_type == StepType.ACTION:
                args_str = json.dumps(step.tool_args, ensure_ascii=False)
                lines.append(f"Action: {step.tool_name}({args_str})")
            elif step.step_type == StepType.OBSERVATION:
                lines.append(f"Observation: {step.content[:500]}")
        return "\n".join(lines)

轨迹 Judge 的校准。

和结果 Judge 一样，轨迹 Judge 也需要校准。推荐做法：准备 30-50 条带有人工评分的轨迹作为 calibration set，定期检查 LLM Judge 与人工的相关系数。Spearman 相关系数应 > 0.7 才能投入使用。

02. Tool Calling 失败模式：不只是"有没有调用成功"

工具调用的失败可以分为多个层次，从表面到深层：

L1: 调用级失败

工具不存在（hallucinated tool）
参数缺失或类型错误
超时或网络错误

**检测方式：**JSON Schema 校验即可捕获

L2: 选择级失败

选错工具（该查订单，调了商品搜索）
顺序错误（先支付后验库存）
过度调用（同一问题打 5 个工具）

**检测方式：**轨迹断言 + 工具序列校验

L3: 语义级失败

参数语义错误（合法但含义不对）
返回值误解（空=不存在 vs 空=无权限）
业务前置条件缺失

**检测方式：**语义契约测试

L4: 系统级失败

越权访问
重复执行（幂等失败）
半成功链路（部分步骤成功部分失败）

**检测方式：**端到端系统测试

测试启发。

Tool Calling 的正确性至少有三层：选得对不对、调得对不对、解释得对不对。很多系统只测第一层（调用是否成功），于是"调用成功但逻辑错误"的问题长期漏掉。

02.5 参数合法但语义错误：企业里更阴的坑

工具	语义上为什么错	可能后果
query_order(order_id, include_closed)	如果退款必须看已关闭订单，这个布尔值就会把关键订单过滤掉	用户被告知无法退款，实际可退
search_user(role)	调用成功不代表可以查，语义上已经越过权限边界	泄露管理员信息
query_refund(date_from, date_to)	"创建时间"还是"退款申请时间"没分清	返回错误时间范围的数据
send_notification(user_id, msg)	user_id="all" 可能给全量用户发通知	无法撤回的全量推送

所以企业的 Agent 契约测试通常不只断言 JSON Schema，还会断言参数语义、默认值风险和业务前置条件。

语义参数校验的实现

python

class SemanticParamValidator:
    """超越 JSON Schema 的语义参数校验器"""
    
    def __init__(self):
        self.rules = {}
    
    def register_tool(self, tool_name: str, rules: list[dict]):
        """
        注册工具的语义校验规则
        
        rules 示例:
        [
            {
                "param": "include_closed",
                "condition": "当 action 涉及退款查询时",
                "expected": True,
                "reason": "退款必须包含已关闭订单"
            },
            {
                "param": "user_id",
                "forbidden_values": ["all", "*", "admin"],
                "reason": "禁止全量操作和越权查询"
            },
            {
                "param": "date_from",
                "constraint": "必须与业务语境中的时间字段语义匹配",
                "needs_context_check": True
            }
        ]
        """
        self.rules[tool_name] = rules
    
    def validate(self, tool_name: str, args: dict, 
                 context: dict = None) -> list[str]:
        """返回语义违规列表"""
        violations = []
        rules = self.rules.get(tool_name, [])
        
        for rule in rules:
            param = rule["param"]
            value = args.get(param)
            
            # 检查禁止值
            if "forbidden_values" in rule:
                if value in rule["forbidden_values"]:
                    violations.append(
                        f"参数 {param}={value} 触发禁止规则: "
                        f"{rule['reason']}"
                    )
            
            # 检查期望值（上下文相关）
            if "expected" in rule and "condition" in rule:
                if context and self._check_condition(
                    rule["condition"], context):
                    if value != rule["expected"]:
                        violations.append(
                            f"在{rule['condition']}下，"
                            f"{param} 应为 {rule['expected']}，"
                            f"实际为 {value}: {rule['reason']}"
                        )
        
        return violations

02.6 调用成本与效率分析

在企业环境中，Agent 的每一次工具调用都有成本：API 费用、延迟叠加、下游系统负担。需要系统性地度量和约束：

python

class AgentCostTracker:
    """Agent 执行成本追踪器"""
    
    def __init__(self):
        self.tool_costs = {}  # 每个工具的平均成本
        self.budgets = {}     # 每个任务类型的成本预算
    
    def analyze_trajectory(self, trajectory: Trajectory) -> CostReport:
        total_token_cost = trajectory.total_cost_tokens
        total_api_calls = len(trajectory.actions)
        total_time = trajectory.total_time_ms
        
        # 冗余调用检测
        redundant = self._detect_redundant_calls(trajectory)
        
        # 与预算对比
        budget = self.budgets.get(trajectory.goal, None)
        over_budget = (budget and total_token_cost > budget)
        
        return CostReport(
            total_tokens=total_token_cost,
            total_api_calls=total_api_calls,
            total_time_ms=total_time,
            redundant_calls=redundant,
            over_budget=over_budget,
            cost_per_call=total_token_cost / max(total_api_calls, 1),
            efficiency_score=self._efficiency_score(trajectory)
        )
    
    def _detect_redundant_calls(self, trajectory) -> list[dict]:
        """检测冗余调用"""
        redundant = []
        seen_calls = set()
        
        for step in trajectory.actions:
            call_sig = f"{step.tool_name}:{json.dumps(step.tool_args, sort_keys=True)}"
            if call_sig in seen_calls:
                redundant.append({
                    "step_id": step.step_id,
                    "tool": step.tool_name,
                    "reason": "完全重复的调用"
                })
            seen_calls.add(call_sig)
        
        return redundant
    
    def _efficiency_score(self, trajectory) -> float:
        """0-1 的效率评分"""
        actions = trajectory.actions
        if not actions:
            return 1.0
        
        unique_tools = len(trajectory.unique_tools)
        total_calls = len(actions)
        retries = trajectory.num_retries
        
        # 惩罚因子
        redundancy_penalty = max(0, 1 - retries / total_calls)
        call_penalty = min(1.0, unique_tools / total_calls)
        
        return (redundancy_penalty + call_penalty) / 2

03. MCP / 协议鲁棒性：为什么 schema 漂移会变成线上事故

MCP（Model Context Protocol）或其他工具协议不是单纯的传输通道，它会把工具描述、参数约束、权限边界都显式暴露给模型。一旦 schema 漂移、字段含义变更、权限声明不清，就可能引发调用错误或越权风险。

协议级问题

字段改名（status → order_status）
必填项变化（可选变必填）
返回结构变化（扁平变嵌套）
超时规则变化
错误码体系变更

业务级问题

工具能调通，但业务含义理解错
同名字段在不同版本含义不同
默认值变化导致行为改变
权限模型升级导致原有调用失败
枚举值增加导致未覆盖的分支

3.1 一个 schema 漂移的课堂例子

版本 v1：
check_refund_policy(status)
  → status: 订单状态, enum["paid", "delivered", "cancelled"]

版本 v2：
check_refund_policy(order_status, payment_status)
  → 拆成了两个独立字段

如果 Agent 还按 v1 的方式调用：

场景 A: 严格校验模式
  → 直接报 400 参数错误
  → 用户看到"系统异常"
  → 影响：可观测，容易发现

场景 B: 兼容填充模式（更危险）
  → SDK 自动把 status 映射到 order_status
  → payment_status 用了默认值 "unknown"
  → 表面调用成功
  → 但判断逻辑失真，返回了错误的退款政策
  → 影响：不可观测，可能长期存在

对测试来说：
"能调通"不是终点，
"是否仍表达同一业务语义"才是关键。

03.5 Schema 漂移自动检测

python

import json
from deepdiff import DeepDiff

class SchemaVersionTracker:
    """工具 Schema 版本跟踪与漂移检测"""
    
    def __init__(self, schema_store):
        self.schema_store = schema_store
    
    def detect_drift(self, tool_name: str, 
                     new_schema: dict) -> DriftReport:
        old_schema = self.schema_store.get_latest(tool_name)
        if not old_schema:
            return DriftReport(has_drift=False, is_new=True)
        
        diff = DeepDiff(old_schema, new_schema, ignore_order=True)
        
        breaking_changes = []
        non_breaking_changes = []
        
        # 分析每种变更的影响
        for change_type, changes in diff.items():
            for path, detail in (changes.items() 
                                 if isinstance(changes, dict) 
                                 else []):
                impact = self._assess_impact(change_type, path, detail)
                if impact == "breaking":
                    breaking_changes.append({
                        "type": change_type,
                        "path": path,
                        "detail": str(detail),
                    })
                else:
                    non_breaking_changes.append({
                        "type": change_type,
                        "path": path,
                        "detail": str(detail),
                    })
        
        return DriftReport(
            has_drift=bool(breaking_changes or non_breaking_changes),
            breaking_changes=breaking_changes,
            non_breaking_changes=non_breaking_changes,
            risk_level="high" if breaking_changes else 
                       "medium" if non_breaking_changes else "low"
        )
    
    def _assess_impact(self, change_type, path, detail) -> str:
        """评估变更是否是破坏性的"""
        # 必填参数被移除或改名 → 破坏性
        if "required" in path and "dictionary_item_removed" in change_type:
            return "breaking"
        
        # 参数类型变更 → 破坏性
        if "type" in path and "values_changed" in change_type:
            return "breaking"
        
        # 新增可选参数 → 非破坏性
        if "dictionary_item_added" in change_type:
            return "non_breaking"
        
        return "breaking"  # 默认保守判断

03.6 协议 Fuzzing：主动发现边界问题

除了被动检测 schema 变更，还可以主动对工具协议做 fuzzing 测试，发现隐藏的边界问题：

python

import random
import string

class ToolFuzzer:
    """对工具调用进行模糊测试"""
    
    FUZZ_STRATEGIES = [
        "empty_string",      # 空字符串
        "null_value",        # null
        "extreme_long",      # 超长字符串
        "special_chars",     # 特殊字符
        "type_mismatch",     # 类型不匹配
        "boundary_values",   # 边界值
        "sql_injection",     # SQL 注入尝试
        "missing_required",  # 缺少必填字段
        "extra_fields",      # 多余字段
        "unicode_edge",      # Unicode 边缘 case
    ]
    
    def generate_fuzz_cases(self, tool_schema: dict) -> list[dict]:
        """基于 schema 自动生成 fuzz 用例"""
        cases = []
        params = tool_schema.get("parameters", {}).get("properties", {})
        required = tool_schema.get("parameters", {}).get("required", [])
        
        for strategy in self.FUZZ_STRATEGIES:
            case = self._apply_strategy(strategy, params, required)
            cases.append({
                "strategy": strategy,
                "args": case,
                "expected": "error_handled_gracefully"
            })
        
        return cases
    
    def _apply_strategy(self, strategy, params, required) -> dict:
        base_args = self._generate_valid_args(params)
        
        if strategy == "empty_string":
            first_str = next(
                (k for k, v in params.items() 
                 if v.get("type") == "string"), None)
            if first_str:
                base_args[first_str] = ""
        
        elif strategy == "extreme_long":
            first_str = next(
                (k for k, v in params.items() 
                 if v.get("type") == "string"), None)
            if first_str:
                base_args[first_str] = "A" * 100000
        
        elif strategy == "type_mismatch":
            first_int = next(
                (k for k, v in params.items() 
                 if v.get("type") == "integer"), None)
            if first_int:
                base_args[first_int] = "not_a_number"
        
        elif strategy == "missing_required":
            if required:
                del base_args[required[0]]
        
        elif strategy == "sql_injection":
            first_str = next(
                (k for k, v in params.items() 
                 if v.get("type") == "string"), None)
            if first_str:
                base_args[first_str] = "'; DROP TABLE users; --"
        
        return base_args

Fuzzing 的目标不是破坏系统。

而是验证系统在异常输入下是否能优雅降级。一个好的 Agent 系统在收到 fuzz 输入时，应该返回清晰的错误信息而不是 500 崩溃、静默失败或泄露堆栈。

04. 权限、隔离与安全边界

维度	为什么重要	测试点	典型案例
权限隔离	避免越权查询和写操作	不同角色、不同租户、不同数据域	普通用户不应能调用 admin 工具
幂等性	重试不能造成重复下单、重复扣费	失败重试、重复提交	支付超时后重试导致双扣
补偿机制	中间步骤失败后系统能否回滚	半成功链路、超时链路	锁座成功但支付失败
超时与熔断	避免单工具拖垮整条 Agent	超时、断路、降级策略	外部 API 挂了导致 Agent 卡死
数据隔离	租户 A 的数据不能被租户 B 访问	跨租户请求、参数篡改	通过修改 tenant_id 查看他人数据

4.1 一个企业最怕的链路

场景：Agent 帮用户自动改签机票

步骤 1：查询可用航班 → 成功
步骤 2：锁座 → 成功 (座位被锁定 15 分钟)
步骤 3：计算差价 → 成功 (补 ¥320)
步骤 4：支付补差价 → 超时 (网络波动)
步骤 5：Agent 自动重试支付
步骤 6：第一次支付实际上已经成功了
步骤 7：重试支付也成功了
结果：用户被扣了 ¥640

事故分析：
1. 支付接口没有幂等键 → 重试导致重复扣费
2. Agent 没有在重试前查询支付状态 → 盲目重试
3. 没有补偿机制 → 无法自动退还多扣款项
4. 没有人工审核门槛 → 金额直接扣除

这类事故里，最终答案写得再体面也没有意义。

一个非常现实的判断。

企业往往不是因为 Agent "答错一句话"而暂停上线，而是因为它可能越权、重复执行、超时卡死或者在失败时没有补偿。这些都是协议和系统层鲁棒性问题，需要用系统测试的方法来覆盖。

04.5 幂等性测试与补偿机制

幂等性是指：同一个操作执行一次和执行多次的效果完全一致。在 Agent 系统中，由于网络不稳定和重试机制，幂等性至关重要。

python

class IdempotencyTester:
    """幂等性测试框架"""
    
    def test_idempotency(self, tool_name: str, args: dict, 
                         n_retries: int = 3) -> IdempotencyResult:
        """
        对同一个工具调用执行多次，
        验证系统状态是否一致
        """
        results = []
        
        # 记录初始状态
        initial_state = self._capture_system_state()
        
        # 执行第一次
        first_result = self._call_tool(tool_name, args)
        first_state = self._capture_system_state()
        results.append(first_result)
        
        # 执行 N 次重试
        for i in range(n_retries):
            retry_result = self._call_tool(tool_name, args)
            retry_state = self._capture_system_state()
            results.append(retry_result)
            
            # 检查状态是否与第一次执行后一致
            state_diff = self._compare_states(first_state, retry_state)
            if state_diff:
                return IdempotencyResult(
                    is_idempotent=False,
                    failure_reason=f"第 {i+2} 次执行后状态不一致",
                    state_diff=state_diff,
                    results=results
                )
        
        return IdempotencyResult(
            is_idempotent=True,
            results=results
        )

class CompensationTester:
    """补偿机制测试框架"""
    
    def test_partial_failure(self, workflow_steps: list[dict],
                             fail_at_step: int) -> CompensationResult:
        """
        模拟在指定步骤失败，验证补偿是否正确执行
        """
        executed = []
        
        for i, step in enumerate(workflow_steps):
            if i == fail_at_step:
                # 模拟失败
                break
            result = self._execute_step(step)
            executed.append(result)
        
        # 等待补偿机制触发
        import time
        time.sleep(5)
        
        # 检查补偿结果
        final_state = self._capture_system_state()
        
        return CompensationResult(
            steps_executed=len(executed),
            failed_at=fail_at_step,
            compensation_triggered=self._check_compensation_log(),
            final_state_clean=self._verify_clean_state(final_state),
        )

权限隔离测试矩阵

角色	查询订单	修改订单	查询用户	管理员操作	跨租户访问
普通用户	仅自己	仅自己	禁止	禁止	禁止
客服	所属租户	有限修改	所属租户	禁止	禁止
管理员	所有	所有	所有	允许	有限

# 权限隔离测试用例生成
def generate_permission_test_cases(roles, tools, tenants):
    """
    自动生成权限测试矩阵
    """
    test_cases = []
    
    for role in roles:
        for tool in tools:
            for tenant in tenants:
                # 正常场景：在权限范围内
                if tool.allowed_for(role, tenant):
                    test_cases.append({
                        "role": role,
                        "tool": tool.name,
                        "tenant": tenant,
                        "expected": "success",
                        "description": f"{role} 应能访问 {tool.name}"
                    })
                
                # 越权场景：超出权限范围
                else:
                    test_cases.append({
                        "role": role,
                        "tool": tool.name,
                        "tenant": tenant,
                        "expected": "permission_denied",
                        "description": f"{role} 不应能访问 {tool.name}"
                    })
    
    return test_cases

04.6 契约测试与 Mock Harness

python

class MockToolRegistry:
    """可控的工具 Mock 环境"""
    
    def __init__(self):
        self.tools = {}
        self.call_log = []
    
    def register(self, name: str, handler, schema: dict):
        self.tools[name] = {
            "handler": handler,
            "schema": schema,
            "call_count": 0
        }
    
    def call(self, name: str, args: dict) -> dict:
        if name not in self.tools:
            raise ToolNotFoundError(f"工具 {name} 不存在")
        
        tool = self.tools[name]
        
        # Schema 校验
        self._validate_schema(args, tool["schema"])
        
        # 记录调用
        tool["call_count"] += 1
        self.call_log.append({
            "tool": name,
            "args": args,
            "timestamp": time.time()
        })
        
        # 执行 mock handler
        return tool["handler"](args)
    
    def inject_failure(self, tool_name: str, 
                       failure_type: str = "timeout",
                       after_n_calls: int = 0):
        """注入故障用于测试恢复能力"""
        original = self.tools[tool_name]["handler"]
        call_count = [0]
        
        def failing_handler(args):
            call_count[0] += 1
            if call_count[0] > after_n_calls:
                if failure_type == "timeout":
                    time.sleep(30)
                    raise TimeoutError("工具超时")
                elif failure_type == "error":
                    return {"error": "internal_error", "code": 500}
                elif failure_type == "empty":
                    return {}
            return original(args)
        
        self.tools[tool_name]["handler"] = failing_handler

# 使用示例
registry = MockToolRegistry()

registry.register("get_order", 
    handler=lambda args: {
        "order_id": args["order_id"],
        "status": "delivered",
        "refundable": True
    },
    schema={
        "parameters": {
            "required": ["order_id"],
            "properties": {
                "order_id": {"type": "string"}
            }
        }
    }
)

# 测试正常流程
result = registry.call("get_order", {"order_id": "123"})

# 注入故障，测试 Agent 的恢复能力
registry.inject_failure("get_order", "timeout", after_n_calls=1)

python

def assert_semantic_contract(trace):
    """语义契约断言"""
    first = trace["actions"][0]
    second = trace["actions"][1]
    
    assert first["tool"] == "get_order", \
        "第一步应该查订单"
    assert second["tool"] == "check_refund_policy", \
        "第二步应该检查退款政策"
    assert second["args"]["status"] in {"paid", "delivered", "cancelled"}, \
        "状态必须是合法的业务枚举值"
    assert trace["final_answer"] in {
        "该订单可退款",
        "该订单不可退款",
        "需要人工审核"
    }, "最终答案必须是预定义的业务结论之一"

# schema 只是"长得像"；
# semantic contract 才是"意思没跑偏"。

05. 多 Agent 协作的测试挑战

当系统中不止一个 Agent，而是多个 Agent 协作完成任务时，测试复杂度呈指数级上升：

消息传递正确性

Agent A 传给 Agent B 的上下文是否完整、准确？信息是否在传递过程中丢失或变形？

任务委托边界

哪些任务应该委托、哪些不应该？委托后的结果是否被正确采纳？

冲突解决

当两个 Agent 给出矛盾结论时，系统如何仲裁？

循环引用

Agent A 请求 B、B 又请求 A，是否有死循环保护？

python

class MultiAgentTestHarness:
    """多 Agent 协作测试框架"""
    
    def __init__(self, agents: dict):
        self.agents = agents
        self.message_log = []
    
    def test_collaboration(self, task: str, 
                          expected_flow: list[str]) -> TestResult:
        """
        测试多 Agent 协作流程
        expected_flow: 预期的消息传递顺序
        """
        actual_flow = []
        
        # 监听所有 Agent 间的消息
        for name, agent in self.agents.items():
            agent.on_message = lambda msg, sender=name: (
                actual_flow.append(f"{sender}->{msg['to']}"),
                self.message_log.append(msg)
            )
        
        # 执行任务
        result = self.agents["orchestrator"].execute(task)
        
        # 验证流程
        flow_match = self._match_flow(expected_flow, actual_flow)
        
        # 检查循环引用
        has_cycle = self._detect_cycle(actual_flow)
        
        # 检查信息完整性
        info_loss = self._detect_info_loss(self.message_log)
        
        return TestResult(
            flow_correct=flow_match,
            has_cycle=has_cycle,
            info_loss=info_loss,
            total_messages=len(self.message_log),
            result=result
        )
    
    def _detect_cycle(self, flow: list[str]) -> bool:
        """检测是否存在消息循环"""
        seen_patterns = set()
        window_size = 3
        
        for i in range(len(flow) - window_size + 1):
            pattern = tuple(flow[i:i+window_size])
            if pattern in seen_patterns:
                return True
            seen_patterns.add(pattern)
        
        return False

06. 企业实战案例：电商客服 Agent 的系统级测试

6.1 场景描述

某电商平台的 AI 客服 Agent 需要处理：订单查询、退款申请、物流追踪、商品咨询、投诉升级。Agent 可以调用 8 个工具，需要根据用户意图选择正确的工具组合。

6.2 测试体系设计

测试层	覆盖内容	用例数	运行频率
结果正确性	最终答案是否正确	200	每次提交
轨迹合理性	plan/action/observation 质量	100	每日
工具契约	schema 校验 + 语义校验	80	工具变更时
权限隔离	跨用户/跨角色访问控制	50	每周
幂等/补偿	重试不重复、失败可回滚	30	发版前
异常恢复	超时/空返回/错误码处理	40	每日
Fuzzing	异常输入下的降级能力	100+	每周

6.3 发现的典型问题

问题 1: 退款流程中的幂等缺陷
  发现方式: 幂等性测试
  影响: 重试导致 2% 的退款金额翻倍
  修复: 在支付接口增加 idempotency_key
  回归: 加入 L1 Golden Set

问题 2: 物流工具 schema 升级导致静默失败
  发现方式: Schema 漂移检测
  影响: 物流状态字段从 string 变 enum，
       Agent 传了旧格式但 SDK 做了兼容，
       返回的状态映射错误
  修复: 升级 Agent 的工具描述
  回归: 加入契约测试集

问题 3: Agent 面对工具超时时无限重试
  发现方式: 异常注入测试
  影响: 单个请求的 token 成本达到正常的 8 倍
  修复: 增加最大重试次数 (3次) 和退避策略
  回归: 加入异常恢复测试集

问题 4: 跨租户数据泄露
  发现方式: 权限隔离测试
  影响: 客服角色通过修改 order_id 可查询
       其他租户的订单详情
  修复: 在工具层增加 tenant_id 校验
  回归: 加入安全测试集 (L2)

07. 测试设计清单

结果断言之外，增加轨迹断言和关键步骤断言。
建立轨迹质量度量体系：步骤效率、工具准确率、参数准确率、重试率。
对工具 schema 变更建立漂移检测和契约测试。
实施语义级参数校验，不只是 JSON Schema 校验。
对超时、空返回、部分失败做异常注入和恢复测试。
对高风险工具做权限隔离和幂等性回归。
对多工具场景记录调用次数、顺序和总成本。
部署协议 Fuzzing 发现隐藏的边界问题。
多 Agent 系统增加消息传递、循环检测和冲突仲裁测试。
用 LLM Judge 做轨迹语义评估，并定期校准。

7.1 课堂练习

写一个"答案对但轨迹错"的 Agent bad case，说明为什么不能放过，并给出你会添加的轨迹断言。
给某个 Tool Calling 场景设计一组"参数合法但语义错误"的测试样本，覆盖至少 3 种错误模式。
为一个高风险写操作 Agent 列出你会强制加入的幂等、补偿和权限断言，并说明每个断言的设计理由。
设计一个 Schema 漂移检测的自动化方案，说明如何集成到 CI/CD 流水线中。
为一个多 Agent 协作场景设计测试用例，覆盖正常流程、循环引用和冲突解决。

7.2 参考答案要点

"答案对但轨迹错"通常意味着计划冗余、成本异常、依赖偶然正确，应该作为系统脆弱性的信号。轨迹断言应覆盖：必经步骤、工具选择序列、最大步骤数、token 预算。
语义错误样本要优先覆盖默认值、时间范围、枚举含义和权限字段，因为这几类最容易"schema 合法但业务跑偏"。每种错误模式都应有明确的预期结果（拒绝调用 or 返回正确默认值）。
高风险写操作至少要断言：角色权限（谁能执行）、重复请求幂等键（执行 N 次效果等于 1 次）、失败后的补偿路径（已执行的步骤能否回滚）、以及超时后的降级策略（告知用户还是静默失败）。
Schema 漂移检测应在工具 SDK 更新时自动触发，对比新旧 schema 的 diff，标记破坏性变更并阻断部署，非破坏性变更发出告警。
多 Agent 测试用例应包含：正常委托和结果采纳、A→B→A 的循环保护、两个 Agent 结论矛盾时的仲裁策略和最终输出的一致性。

7.3 自测标准

学完这一页后，你应该能：

把 Agent 测试拆成 plan / action / observation / replan 四层，并为每层设计测试。
识别 tool calling 从调用级到系统级的各层失败模式。
区分协议级错误和业务级错误，知道各自的检测方法。
把权限、幂等、补偿纳入 Agent 测试范围，理解其在企业中的重要性。
设计和实现轨迹度量、语义契约、Schema 漂移检测等自动化工具。
理解多 Agent 协作场景下的额外测试挑战。

Agent 轨迹与协议鲁棒性 ​

01. 轨迹评测：为什么要看 plan / action / observation ​

1.1 一个"答案对了，但轨迹已经坏了"的例子 ​

01.5 轨迹的形式化定义与度量 ​

轨迹的形式化表示 ​

轨迹质量度量指标 ​

01.6 用 LLM 做轨迹 Judge ​

02. Tool Calling 失败模式：不只是"有没有调用成功" ​

L1: 调用级失败 ​

L2: 选择级失败 ​

L3: 语义级失败 ​

L4: 系统级失败 ​

02.5 参数合法但语义错误：企业里更阴的坑 ​

语义参数校验的实现 ​

02.6 调用成本与效率分析 ​

03. MCP / 协议鲁棒性：为什么 schema 漂移会变成线上事故 ​

协议级问题 ​

业务级问题 ​

3.1 一个 schema 漂移的课堂例子 ​

03.5 Schema 漂移自动检测 ​

03.6 协议 Fuzzing：主动发现边界问题 ​

04. 权限、隔离与安全边界 ​

4.1 一个企业最怕的链路 ​

04.5 幂等性测试与补偿机制 ​

权限隔离测试矩阵 ​

04.6 契约测试与 Mock Harness ​

05. 多 Agent 协作的测试挑战 ​

消息传递正确性 ​

任务委托边界 ​

冲突解决 ​

循环引用 ​

06. 企业实战案例：电商客服 Agent 的系统级测试 ​

6.1 场景描述 ​

6.2 测试体系设计 ​

6.3 发现的典型问题 ​

07. 测试设计清单 ​

7.1 课堂练习 ​

7.2 参考答案要点 ​

7.3 自测标准 ​

Agent 轨迹与协议鲁棒性

01. 轨迹评测：为什么要看 plan / action / observation

1.1 一个"答案对了，但轨迹已经坏了"的例子

01.5 轨迹的形式化定义与度量

轨迹的形式化表示

轨迹质量度量指标

01.6 用 LLM 做轨迹 Judge

02. Tool Calling 失败模式：不只是"有没有调用成功"

L1: 调用级失败

L2: 选择级失败

L3: 语义级失败

L4: 系统级失败

02.5 参数合法但语义错误：企业里更阴的坑

语义参数校验的实现

02.6 调用成本与效率分析

03. MCP / 协议鲁棒性：为什么 schema 漂移会变成线上事故

协议级问题

业务级问题

3.1 一个 schema 漂移的课堂例子

03.5 Schema 漂移自动检测

03.6 协议 Fuzzing：主动发现边界问题

04. 权限、隔离与安全边界

4.1 一个企业最怕的链路

04.5 幂等性测试与补偿机制

权限隔离测试矩阵

04.6 契约测试与 Mock Harness

05. 多 Agent 协作的测试挑战

消息传递正确性

任务委托边界

冲突解决

循环引用

06. 企业实战案例：电商客服 Agent 的系统级测试

6.1 场景描述

6.2 测试体系设计

6.3 发现的典型问题

07. 测试设计清单

7.1 课堂练习

7.2 参考答案要点

7.3 自测标准