工具调用测试六栈：从Mock到混沌的工程突围

graph TB
    subgraph "工具调用测试六栈"
        L1["L1: Mock/Replay<br/>evalcraft / pytest-mock"]
        L2["L2: Simulation<br/>LangWatch Scenario / ToolSimulator"]
        L3["L3: Contract<br/>DriftGentic / Schema检测"]
        L4["L4: Golden Dataset<br/>Google ADK Eval / CI门禁"]
        L5["L5: Observability<br/>OTel + Langfuse/LangSmith"]
        L6["L6: Chaos<br/>agent-chaos / 故障注入"]
    end

    L1 -->|"通过签名验证"| L2
    L2 -->|"行为验证通过"| L3
    L3 -->|"契约一致"| L4
    L4 -->|"回归门禁通过"| L5
    L5 -->|"上线后监控"| L6

    L6 -->|"发现脆弱点<br/>反馈修复"| L1

    style L1 fill:#e3f2fd
    style L2 fill:#f3e5f5
    style L3 fill:#fff3e0
    style L4 fill:#e8f5e9
    style L5 fill:#fce4ec
    style L6 fill:#efebe9

# evalcraft磁带回放模式测试
from evalcraft import AgentTest, cassette

class TestWeatherAgent(AgentTest):
    agent = WeatherAgent()

    @cassette("weather_nyc_sunny")
    def test_calls_weather_tool_with_city(self):
        result = self.agent.run("纽约今天天气如何？")
        # 验证调用了正确的工具
        self.assert_tool_called("get_weather")
        # 验证参数正确
        self.assert_tool_args("get_weather", {"city": "New York"})
        # 验证最终回复包含天气信息
        self.assert_response_contains("晴")

    @cassette("multi_tool_forecast")
    def test_parallel_tool_calls(self):
        result = self.agent.run("对比东京和伦敦的天气")
        # 验证并发调用了两次天气工具
        self.assert_tool_call_count("get_weather", 2)
        self.assert_tool_args_at(0, "get_weather", {"city": "Tokyo"})
        self.assert_tool_args_at(1, "get_weather", {"city": "London"})

// LangWatch Scenario: 多轮工具调用模拟测试
import { Scenario } from "langwatch-scenario";

const agentUnderTest = new BookingAgent();

const result = await Scenario.create()
  .agent(agentUnderTest)                    // 被测Agent
  .simulator("openai:gpt-4.1")             // LLM扮演用户
  .judge("anthropic:claude-sonnet-4-20250514")             // LLM评判结果
  .instruction(
    "你是一个挑剔的商务旅客，试图改签航班。" +
    "尝试以下边界情况：改签到已满航班、用过期里程兑换、同时改签两个航段"
  )
  .successCriteria([
    "Agent必须调用search_flights工具查询可用航班",
    "当航班满员时，Agent不应直接调用book_flight",
    "Agent应主动询问替代日期而非直接失败",
    "里程不足时，Agent应提示付费选项"
  ])
  .maxTurns(8)
  .run();

// 输出：通过率、工具调用序列、失败回合详情
console.log(`通过率: ${result.passRate}%`);
console.log(`工具调用序列: ${result.toolCallTrace}`);

# AWS ToolSimulator: 模拟工具返回值
from strands_agents.eval import ToolSimulator

simulator = ToolSimulator()

# 注册工具并配置模拟场景
@simulator.register("database_query")
def simulate_db(tool_name, args, state):
    if state.get("db_down"):
        return {"error": "Connection refused", "status": 500}
    if "SELECT * FROM users" in args["sql"]:
        return {"rows": [{"id": 1, "name": "test_user"}]}
    return {"rows": []}

# 运行模拟评估
result = simulator.evaluate(
    agent=my_agent,
    scenarios=[
        {"input": "查询所有用户", "state": {"db_down": False}},
        {"input": "查询所有用户", "state": {"db_down": True}},  # 故障场景
    ]
)

# DriftGentic: 工具调用契约验证
from driftgentic import ContractGuard, ToolContract

# 定义工具契约
weather_contract = ToolContract(
    name="get_weather",
    input_schema={
        "type": "object",
        "properties": {
            "city": {"type": "string", "minLength": 1},
            "units": {"type": "string", "enum": ["celsius", "fahrenheit"]}
        },
        "required": ["city"]
    },
    output_schema={
        "type": "object",
        "properties": {
            "temperature": {"type": "number"},
            "condition": {"type": "string"}
        },
        "required": ["temperature", "condition"]
    },
    # 后端API的OpenAPI spec URL，用于自动漂移检测
    backend_spec_url="https://api.weather.com/openapi.json",
    drift_detection="auto"  # 自动对比Schema差异
)

# 在Agent中注册契约守卫
guard = ContractGuard(contracts=[weather_contract])

# Agent每次工具调用都会经过验证
agent = MyAgent(middleware=[guard])

# Google ADK: Golden Dataset回归测试
from google.adk.evaluation import AgentEvaluator

evaluator = AgentEvaluator(
    agent=my_agent,
    test_files=["tests/golden/booking.test.json"],
    metrics=[
        "tool_selection_accuracy",     # 工具选择正确率
        "tool_args_match_rate",         # 参数匹配率
        "trajectory_similarity",        # 轨迹相似度（非精确匹配）
        "final_response_quality"        # 最终回复质量
    ],
    # 设置通过阈值
    thresholds={
        "tool_selection_accuracy": 0.95,
        "tool_args_match_rate": 0.90,
        "trajectory_similarity": 0.85
    }
)

# 运行评估——可作为pytest用例
results = evaluator.evaluate()
# 如果任何指标低于阈值，返回非零退出码，CI失败
assert results.all_passed, f"回归测试失败: {results.failures}"

# promptfoo: 工具调用回归测试配置
tests:
  - description: "天气查询应该调用get_weather工具"
    vars:
      query: "北京明天天气"
    assert:
      - type: tool-call
        tool: get_weather
        args:
          city: "北京"  # 支持模糊匹配
      - type: contains
        value: "温度"

  - description: "航班预订应先搜索再预订"
    vars:
      query: "帮我订明天北京到上海的机票"
    assert:
      - type: tool-call-sequence
        sequence: [search_flights, book_flight]
      - type: latency
        threshold: 5000  # 响应时间不超过5秒

graph LR
    A["Agent Run Span<br/>trace_id: abc123"] --> B["LLM Call Span<br/>model: gpt-4.1"]
    A --> C["Tool Call Span<br/>tool: get_weather"]
    C --> D["HTTP Request Span<br/>api.weather.com"]
    C --> E["Tool Response Span<br/>status: 200"]
    A --> F["LLM Call Span<br/>model: gpt-4.1"]
    F --> G["Tool Call Span<br/>tool: book_flight"]

    style C fill:#fff3e0
    style G fill:#fff3e0
    style D fill:#e8f5e9
    style E fill:#e8f5e9

# agent-chaos: 工具调用故障注入
from agent_chaos import ChaosMiddleware, FaultConfig

chaos = ChaosMiddleware(
    faults=[
        # 10%概率让天气API超时
        FaultConfig(
            tool_name="get_weather",
            fault_type="timeout",
            fault_rate=0.10,
            delay_ms=30000  # 30秒超时
        ),
        # 5%概率让支付API返回500错误
        FaultConfig(
            tool_name="process_payment",
            fault_type="error",
            fault_rate=0.05,
            error_code=500,
            error_message="Internal Server Error"
        ),
        # 3%概率让搜索API返回空结果
        FaultConfig(
            tool_name="search_flights",
            fault_type="empty_response",
            fault_rate=0.03
        ),
    ],
    # 安全限制
    blast_radius="single_tool",  # 只影响单个工具
    auto_rollback=True,          # 异常时自动回滚
    max_concurrent_faults=1      # 最多同时注入1个故障
)

# 包装Agent
agent = BookingAgent(middleware=[chaos])
result = agent.run("帮我订一张北京到上海的机票")

# 检查Agent是否正确处理了故障
print(chaos.report())
# 输出：故障注入次数、Agent重试次数、降级行为统计

graph TD
    subgraph "混沌工程四阶段"
        A["1. 基线测量<br/>正常状态下工具调用成功率"] --> B["2. 故障注入<br/>超时/错误/空结果"]
        B --> C["3. 行为观测<br/>Agent是否重试/降级/报错"]
        C --> D["4. 韧性评分<br/>Recovery Rate / Cascade / Graceful"]
    end

    D -->|"不达标"| E["修复Agent重试/降级逻辑"]
    E --> A

    D -->|"达标"| F["进入下一轮更强故障"]
    F --> B

    style A fill:#e3f2fd
    style B fill:#fff3e0
    style C fill:#f3e5f5
    style D fill:#e8f5e9

# GitHub Actions: Agent工具调用测试Pipeline
name: Agent Tool Call Tests

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      # L1: Mock/Replay测试（最快，必须通过）
      - name: Unit Mock Tests
        run: pytest tests/unit/ --with-evalcraft

      # L3: Contract验证
      - name: Schema Drift Detection
        run: driftgentic check --contracts ./contracts/

      # L4: Golden Dataset回归
      - name: Golden Dataset Regression
        run: |
          adk evaluate --test-dir tests/golden/ \
            --threshold tool_selection_accuracy=0.95 \
            --threshold trajectory_similarity=0.85

      # L2: Simulation测试（可选，PR时触发）
      - name: Simulation Tests
        if: github.event_name == 'pull_request'
        run: scenario run --simulator gpt-4.1 --judge claude-sonnet-4-20250514

      # L6: Chaos测试（仅main分支，夜间运行）
      - name: Chaos Engineering
        if: github.ref == 'refs/heads/main'
        run: agent-chaos run --config chaos-config.yaml --report chaos-report.json