数据真实性声明：本文「工具链组合测试结果」来自本包脚本 scripts/run_article17_tool_chains.py 对 CustomAgent 的一次真实调用。Provider：DeepSeek，model=deepseek-chat，temperature=0.3，跑数时间 2026-07-21，产物：results/20260721_081456/article_runs/article17_tool_chains_20260721_081942.json。完整 15 场景用例集为设计目标，本文实测为 3 条代表性工具链。

引子

我们的电商数据分析智能体，日常任务之一是：用 search（经营类输入） 拉取销售摘要，用 calculator 计算环比和同比，用 code_executor 做数据清洗，再用 search（报告类输入，如「生成月度周报」） 拉取 Markdown 模板或交给后续 none 汇总。

流程上常出现的四个工具：search、calculator、code_executor，以及收尾的 search（报告） 或 none（纯 LLM 成文）。

单独用每个工具都没问题。组合在一起，问题就出来了：第一次 search 返回的文本/结构化片段需要 code_executor 清洗字段，calculator 需要上一段数值输入，报告类 search（或 none）需要前面子任务的结果拼接进提示。

工具之间的数据传递、依赖关系、失败传播，是智能体最容易出问题的地方。

这篇文章讲多工具组合场景的集成测试方法。

工具链组合测试

线性组合

工具按顺序调用：A → B → C → D。

工具链	工具	数据流向	测试重点
竞品分析	web_fetch → code_executor → calculator → memory_store	网页 → 清理 → 统计 → 存储	数据格式转换
数据分析	code_executor → calculator → memory_store	读取 → 计算 → 存储	数值精度
报告生成	code_executor → none → memory_store	代码 → 文字 → 存储	内容完整性

分支组合

根据条件选择不同工具：A → (B 或 C) → D。

场景	工具链	分支条件	测试重点
数据源选择	web_fetch 或 code_executor → 后续	数据在网页还是文件	工具选择正确性
计算方式	calculator 或 code_executor → 后续	简单计算 vs 复杂计算	工具选择合理性

并行组合

无依赖的工具并行执行：A 和 B 并行 → C。

场景	并行组	后续	测试重点
多竞品抓取	web_fetch × 3	code_executor 汇总	并行执行、结果合并
多数据源	web_fetch + code_executor	calculator 统计	数据合并、格式统一

依赖失败传播测试

工具 A 失败，依赖 A 的工具 B、C、D 怎么办？

失败工具	依赖工具	期望行为	评分
web_fetch	code_executor → calculator	code_executor 跳过，calculator 跳过	100%
code_executor	calculator → memory_store	calculator 跳过，memory_store 跳过	100%
calculator	memory_store	memory_store 跳过	100%
web_fetch	code_executor	code_executor 重试或换工具	80%

工具间数据传递测试

工具 A 的输出是工具 B 的输入。数据格式是否匹配？数据是否丢失？

数据流向	A 输出格式	B 期望格式	测试重点
web_fetch → code_executor	HTML 文本	纯文本	HTML 清理
code_executor → calculator	数字字符串	数学表达式	格式转换
calculator → memory_store	计算结果	key=value	格式转换
code_executor → code_executor	变量	变量	状态保持

代码：工具链测试与失败传播

#!/usr/bin/env python3
"""
多工具组合场景测试

功能：
1. 工具链组合测试
2. 依赖失败传播测试
3. 工具间数据传递测试
"""

import os
import sys
import time
import json
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass, field


@dataclass
class ToolChainResult:
    """工具链测试结果"""
    chain_id: str
    chain_name: str
    tools: List[str]
    tool_results: Dict[str, Dict]
    data_flow_valid: bool
    chain_success: bool
    elapsed: float
    defects: List[Dict] = field(default_factory=list)


@dataclass
class FailurePropagationResult:
    """失败传播测试结果"""
    failed_tool: str
    dependent_tools: List[str]
    expected_behavior: str
    actual_behavior: str
    passed: bool
    rollback_correct: bool


# 工具链定义
TOOL_CHAINS = [
    {
        "id": "TC-01",
        "name": "竞品分析（线性）",
        "tools": ["web_fetch", "code_executor", "calculator", "memory_store"],
        "task": "抓取竞品官网信息，提取关键指标，对比分析，存储结果",
        "expected_data_flow": ["网页HTML", "清理文本", "统计结果", "存储"],
    },
    {
        "id": "TC-02",
        "name": "数据分析（线性）",
        "tools": ["code_executor", "calculator", "memory_store"],
        "task": "读取 CSV 数据，计算各品类销售额，存储结果",
        "expected_data_flow": ["CSV数据", "统计结果", "存储"],
    },
    {
        "id": "TC-03",
        "name": "多竞品抓取（并行）",
        "tools": ["web_fetch", "web_fetch", "web_fetch", "code_executor"],
        "task": "同时抓取 3 个竞品官网，汇总分析",
        "expected_data_flow": ["3个网页", "汇总分析"],
    },
]


def test_tool_chain(agent, chain: Dict) -> ToolChainResult:
    """
    测试工具链组合

    Args:
        agent: 智能体实例
        chain: 工具链定义

    Returns:
        ToolChainResult
    """
    start_time = time.time()
    agent.reset()

    result = agent.run(chain["task"])
    elapsed = time.time() - start_time
    meta = result.get("_meta", {})
    subtasks = meta.get("subtasks", [])

    # 构建工具结果映射
    tool_results = {}
    for s in subtasks:
        tool = s.get("tool", "")
        if tool and tool != "none":
            if tool not in tool_results:
                tool_results[tool] = []
            tool_results[tool].append({
                "status": s.get("status", ""),
                "result": s.get("result", "")[:100],
            })

    # 验证数据流
    data_flow_valid = _verify_data_flow(subtasks, chain["expected_data_flow"])

    # 检测缺陷
    defects = []
    for s in subtasks:
        if s.get("status") == "failed":
            defects.append({
                "type": "执行缺陷",
                "severity": "High",
                "detail": f"{s['id']} ({s.get('tool', '')}) 执行失败",
            })

    chain_success = result.get("success", False) and data_flow_valid

    return ToolChainResult(
        chain_id=chain["id"],
        chain_name=chain["name"],
        tools=chain["tools"],
        tool_results=tool_results,
        data_flow_valid=data_flow_valid,
        chain_success=chain_success,
        elapsed=elapsed,
        defects=defects,
    )


def _verify_data_flow(subtasks: List[Dict], expected_flow: List[str]) -> bool:
    """
    验证数据流

    Args:
        subtasks: 子任务列表
        expected_flow: 期望的数据流向

    Returns:
        数据流是否有效
    """
    # 简单验证：检查子任务是否按顺序执行
    if not subtasks:
        return False

    # 检查是否有失败的工具
    failed_tools = [s for s in subtasks if s.get("status") == "failed"]
    if len(failed_tools) > len(subtasks) * 0.5:
        return False

    return True


def test_failure_propagation(agent, chain: Dict,
                             fail_tool: str) -> FailurePropagationResult:
    """
    测试工具失败后的传播

    Args:
        agent: 智能体实例
        chain: 工具链定义
        fail_tool: 要失败的工具

    Returns:
        FailurePropagationResult
    """
    agent.reset()

    # 执行任务
    result = agent.run(chain["task"])
    meta = result.get("_meta", {})
    subtasks = meta.get("subtasks", [])

    # 找到依赖 fail_tool 的子任务
    fail_tool_ids = [s["id"] for s in subtasks if s.get("tool") == fail_tool]
    dependent_ids = []
    for s in subtasks:
        for dep in s.get("depends_on", []):
            if dep in fail_tool_ids:
                dependent_ids.append(s["id"])

    # 检查依赖工具的行为
    expected = "跳过"  # 期望行为：跳过
    actual = "跳过"  # 实际行为（简化验证）
    for s in subtasks:
        if s["id"] in dependent_ids:
            if s.get("status") == "skipped":
                actual = "跳过"
            elif s.get("status") == "failed":
                actual = "失败"
            else:
                actual = "执行"

    passed = actual == expected
    rollback_correct = actual in ("跳过", "失败")  # 回滚正确

    return FailurePropagationResult(
        failed_tool=fail_tool,
        dependent_tools=dependent_ids,
        expected_behavior=expected,
        actual_behavior=actual,
        passed=passed,
        rollback_correct=rollback_correct,
    )


def print_tool_chain_report(results: List[ToolChainResult]):
    """打印工具链测试报告"""
    print(f"\n{'='*70}")
    print(f"工具链组合测试报告")
    print(f"{'='*70}")

    total = len(results)
    passed = sum(1 for r in results if r.chain_success)

    print(f"\n汇总:")
    print(f"  总工具链: {total}")
    print(f"  通过: {passed} ({passed/total:.0%})")
    print(f"  失败: {total - passed} ({(total-passed)/total:.0%})")

    print(f"\n详细结果:")
    print(f"{'ID':8s} | {'名称':12s} | {'工具数':6s} | {'数据流':6s} | {'耗时':6s} | {'缺陷数':6s}")
    print("-" * 70)
    for r in results:
        icon = "" if r.chain_success else ""
        data_flow = "" if r.data_flow_valid else ""
        print(f"{r.chain_id:8s} | {r.chain_name:12s} | {len(r.tools):6d} | {data_flow:6s} | {r.elapsed:5.1f}s | {len(r.defects):6d} {icon}")

    # 缺陷详情
    all_defects = []
    for r in results:
        all_defects.extend(r.defects)
    if all_defects:
        print(f"\n缺陷详情:")
        for i, d in enumerate(all_defects[:10], 1):
            print(f"  {i}. [{d['severity']}] {d['type']}: {d['detail']}")

    print(f"{'='*70}\n")


def print_failure_propagation_report(results: List[FailurePropagationResult]):
    """打印失败传播测试报告"""
    print(f"\n{'='*70}")
    print(f"依赖失败传播测试报告")
    print(f"{'='*70}")

    print(f"\n{'失败工具':12s} | {'依赖工具':12s} | {'期望':6s} | {'实际':6s} | {'回滚':6s}")
    print("-" * 70)
    for r in results:
        passed_icon = "" if r.passed else ""
        rollback_icon = "" if r.rollback_correct else ""
        print(f"{r.failed_tool:12s} | {', '.join(r.dependent_tools):12s} | {r.expected_behavior:6s} | {r.actual_behavior:6s} | {rollback_icon}")

    print(f"{'='*70}\n")


def run_demo():
    """演示"""
    print("=" * 70)
    print("多工具组合场景测试演示")
    print("=" * 70)

    sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
    from agents.custom_agent.agent import CustomAgent

    agent = CustomAgent(temperature=0.3)

    # 工具链测试
    print("\n--- 工具链组合测试 ---")
    chain_results = []
    for chain in TOOL_CHAINS[:2]:  # 演示用 2 个
        result = test_tool_chain(agent, chain)
        chain_results.append(result)
    print_tool_chain_report(chain_results)

    # 失败传播测试
    print("\n--- 依赖失败传播测试 ---")
    prop_results = []
    for chain in TOOL_CHAINS[:1]:
        for tool in ["web_fetch", "code_executor"]:
            result = test_failure_propagation(agent, chain, tool)
            prop_results.append(result)
    print_failure_propagation_report(prop_results)

    print("=" * 70)


if __name__ == "__main__":
    run_demo()

实测说明：对 CustomAgent（DeepSeek / deepseek-chat / temperature=0.3）运行 3 个多工具组合场景。2026-07-21 实测，产物见 article17_tool_chains_20260721_081942.json。

数据：工具链组合测试结果

工具链	工具调用数	成功率	耗时	输出长度	说明
竞品分析（线性）	3	100%	5.1s	430 字符	web_fetch → code_executor → calculator → memory_store（代码 TC-01 定义）
数据分析（线性）	10	100%	19.2s	1310 字符	code_executor / calculator / memory_store 串联（规划偏碎；Agent 将 3 种工具拆成 10 个子任务）
多竞品抓取（并行示意）	5	100%	5.8s	332 字符	web_fetch×3 → code_executor（代码 TC-03 定义）

测试环境：CustomAgent，DeepSeek deepseek-chat，temperature=0.3，2026-07-21。

关键发现：

1. 3 个多工具组合场景全部执行成功（100%）
2. 数据分析场景耗时最长（19.2s），工具调用也最多（10 次），与「复杂串联更碎」一致
3. 竞品分析与多竞品场景约 5–6s 完成；本轮未观察到并行合并失败，但受限于样本量（仅 3 条链路），尚无法给出并行稳定性结论，需在更大规模测试中验证
4. 用例设计集仍为 15 个场景；本轮只覆盖 3 条代表性链路，失败传播/回滚仍以文中脚本为方法说明

交付物

1. 工具链组合用例集（15 个场景）

ID	名称	工具链	类型	难度
TC-01	竞品分析	web_fetch→code_executor→calculator→memory_store	线性	中等
TC-02	数据分析	code_executor→calculator→memory_store	线性	简单
TC-03	多竞品抓取	web_fetch×3→code_executor	并行	困难
TC-04	报告生成	code_executor→none→memory_store	线性	中等
TC-05	安全扫描	safety_checker→web_fetch→code_executor	线性	中等
TC-06	数据清洗	code_executor→code_executor→calculator	线性	中等
TC-07	多数据源	web_fetch+code_executor→calculator	并行	困难
TC-08	条件分支	(web_fetch或code_executor)→calculator	分支	中等
TC-09	循环处理	code_executor×N→memory_store	循环	困难
TC-10	错误恢复	calculator(失败)→code_executor(替代)	分支	中等

2. 依赖失败传播测试脚本

见上方代码 test_failure_propagation() 函数。

3. 回滚验证脚本

def verify_rollback(result: Dict) -> bool:
    """验证回滚是否正确"""
    meta = result.get("_meta", {})
    subtasks = meta.get("subtasks", [])

    # 检查失败工具后的依赖工具是否被跳过
    failed_ids = [s["id"] for s in subtasks if s["status"] == "failed"]
    for s in subtasks:
        for dep in s.get("depends_on", []):
            if dep in failed_ids and s["status"] not in ("skipped", "failed"):
                return False
    return True

4. 集成测试报告模板

部分	内容
汇总	总工具链、通过率、平均耗时、平均 Token
工具链详情	每个工具链的工具数、数据流、成功率、耗时
失败传播	失败工具、依赖工具、期望/实际行为、回滚验证
缺陷分布	按类型统计缺陷数量
结论	是否通过、改进建议

总结

多工具组合场景是智能体的"压力测试"。本轮 DeepSeek 实测 3 条代表性链路均成功；设计集中的线性/并行/分支场景仍需扩大样本后再给通过率区间。

测试三个维度：工具链组合（线性/分支/并行）、依赖失败传播（失败后跳过）、工具间数据传递（格式转换）。

并行结果合并与失败传播是后续重点——本轮 3 条链路均未观测到并行合并异常，但样本过小，不能据此认为并行不是问题。

本文重点展示了工具链组合测试的方法与代码框架。失败传播测试需要运行时注入失败（如 mock web_fetch 返回错误），涉及模拟环境和断言验证，将在下一篇《缺陷注入与鲁棒性测试》中单独展开实测。

多工具组合的测试，本质上是在测 Agent 的"系统工程能力"——而不仅仅是模型能力。下一篇，我们将进一步拆解：当工具链出错时，如何做缺陷分类与根因分析。

---

面试题模块

Q1：多工具组合场景测试的核心难点是什么？

A：工具之间的数据传递。A 工具的输出是 B 工具的输入，如果 A 返回的数据格式 B 不能解析，整个任务就会中断。测试需要覆盖"工具链"的完整路径，而不是单个工具的"一锤子买卖"。

Q2：多工具场景下，工具调用的顺序依赖怎么测试？

A：两种方法：1) 固定的拓扑序——测试工具 A→B→C 的调用顺序是否正确，检查是否试图先调 C 再调 A；2) 动态重排——测试 Agent 是否能根据返回结果动态调整工具调用顺序（如 A 失败后是否尝试 B 再尝试 A）。第二种更能体现智能体"非确定性"的特点。

Q3：多工具场景中，最常出现的 Bug 类型是什么？

A：传递错误的数据。Agent 把工具 A 的原始响应直接传递给工具 B，而没有做格式转换或数据提取。比如工具 A 返回 {"data": [{"sales": 100}]}，传给工具 B 时应该传 100，但 Agent 传了整串 JSON。这属于"工具接口适配失败"。