传统模式：  User → LLM → [从全部工具中选择] → 执行
图编排模式：User → 图引擎 → [缩小候选集] → LLM → [从候选集中选择] → 执行

# 第三代TDG构建器核心逻辑
class TDGBuilder:
    def __init__(self, tools: list[Tool]):
        self.graph = nx.DiGraph()
        self.tools = {t.name: t for t in tools}

    def build_from_traces(self, traces: list[ExecutionTrace]):
        """从执行trace中挖掘依赖关系"""
        for trace in traces:
            for i, call in enumerate(trace.calls):
                self.graph.add_node(call.tool_name)
                # 参数依赖：如果call_i的参数值出现在call_j的返回值中
                for j in range(i):
                    prev = trace.calls[j]
                    if self._value_flows(prev.result, call.params):
                        self.graph.add_edge(
                            prev.tool_name, call.tool_name,
                            dep_type="data_flow", confidence=0.8
                        )
        # 置信度聚合：多次trace中出现同一依赖则提升置信度
        self._aggregate_confidence()
        # 环检测
        cycles = list(nx.simple_cycles(self.graph))
        if cycles:
            self._resolve_cycles(cycles)

┌─────────────────────────────────┐
│         Upper: Navigator         │
│  输入: task + TDG               │
│  输出: tool_path = [A→C→F]      │
│  方法: GNN + LLM混合推理         │
└──────────────┬──────────────────┘
               │ path
┌──────────────▼──────────────────┐
│         Lower: Executor          │
│  输入: tool_path + 原始任务      │
│  输出: 具体tool_call JSON        │
│  方法: 轻量LLM (如8B模型即可)    │
└─────────────────────────────────┘

# NaviAgent简化实现
class NaviAgent:
    def __init__(self, tdg: ToolDependencyGraph, 
                 navigator_model, executor_model):
        self.tdg = tdg
        self.navigator = navigator_model  # 可以是GNN+大模型
        self.executor = executor_model    # 可以是8B小模型

    def plan_and_execute(self, task: str, max_retries=3):
        for attempt in range(max_retries):
            # Upper: 在TDG上规划路径
            path = self.navigator.plan(
                task=task,
                graph=self.tdg.to_graph_description(),
                available_tools=self.tdg.get_candidate_tools(task)
            )
            # Lower: 逐步执行
            context = {"task": task}
            for step_idx, tool_name in enumerate(path):
                call_result = self.executor.execute_step(
                    tool_name=tool_name,
                    context=context,
                    step_index=step_idx,
                    total_steps=len(path)
                )
                if call_result.success:
                    context[tool_name] = call_result.output
                else:
                    # 闭环：反馈Navigator
                    path = self.navigator.replan(
                        task=task, failed_step=step_idx,
                        error=call_result.error, context=context
                    )
                    break
            else:
                return AgentResult(success=True, output=context)
        return AgentResult(success=False, error="max retries exceeded")

flowchart TD
    A[收集执行Traces] --> B[模式挖掘<br/>发现频繁工具子序列]
    B --> C{支持度≥阈值?}
    C -->|Yes| D[生成Meta-Tool<br/>封装子序列为单步调用]
    C -->|No| E[保留原子工具]
    D --> F[注入工具列表<br/>替换原始子序列]
    F --> G[Agent使用Meta-Tool<br/>减少推理步数]

    style D fill:#4CAF50,color:white
    style G fill:#2196F3,color:white

工具集合 → 依赖图生成 → 路径采样 → 轮次级对话生成 → 验证 → 训练数据

# ToolWeave风格合成数据生成核心逻辑
class ToolWeaveSynthesizer:
    def __init__(self, tools: list[Tool], tdg: ToolDependencyGraph):
        self.tools = tools
        self.tdg = tdg

    def synthesize_dialogue(self, num_turns=5):
        # 1. 从TDG中采样一条可行路径
        path = self.tgd.sample_valid_path(length=num_turns)

        # 2. 生成任务描述（反向：从路径推导任务）
        task = self._reverse_engineer_task(path)

        # 3. 轮次级生成（每一步考虑前面步骤的结果）
        dialogue = []
        context = {}
        for step_idx, tool_name in enumerate(path):
            tool = self.tools[tool_name]
            # 生成该轮的用户意图
            user_intent = self._generate_intent(tool, context, step_idx)
            # 生成工具调用（基于约束，确保参数合法）
            tool_call = self._generate_constrained_call(tool, context)
            # 模拟工具返回（基于工具Schema的输出类型）
            tool_result = self._simulate_result(tool, tool_call)

            dialogue.append({
                "turn": step_idx,
                "intent": user_intent,
                "tool_call": tool_call,
                "result": tool_result
            })
            context[tool_name] = tool_result

        # 4. 验证：检查依赖关系是否被正确满足
        if self._validate_dialogue(dialogue, path):
            return dialogue
        return self.synthesize_dialogue(num_turns)  # 重试

// MCP Server Composition配置示例
{
  "composition": {
    "mode": "orchestrator",
    "servers": [
      {"id": "db", "transport": "stdio", "tool_prefix": "db."},
      {"id": "web", "transport": "sse", "tool_prefix": "web."},
      {"id": "fs", "transport": "stdio", "tool_prefix": "fs."}
    ],
    "dependency_graph": {
      "db.query": {"requires": ["db.connect"], "provides": "query_result"},
      "web.scrape": {"requires": [], "provides": "raw_html"},
      "fs.write": {"requires": ["format.data"], "provides": "file_path"}
    },
    "conflict_resolution": "namespace_prefix"
  }
}

style E fill:#4CAF50,color:white
style F fill:#FF9800,color:white

实测场景：在一个数据分析Agent中，`sql_query`工具因数据库超时失败。自愈路由在TDG上找到替代路径：`sql_query` → `cache_lookup` → `api_fallback`，用缓存数据+API降级完成了任务。

**性能数据**：

| 指标 | 无自愈 | 图驱动自愈 |
|------|--------|-----------|
| 单工具故障任务完成率 | 62% | 84% |
| 双工具故障任务完成率 | 38% | 67% |
| 平均恢复延迟 | N/A | 230ms |
| 额外Token开销 | 0 | +12% |

关键权衡：自愈路由的12%额外Token开销来自替代路径的验证——系统需要用LLM确认替代路径是否适合当前上下文。在生产环境中，这12%的开销通常被"减少了人工介入"的价值所覆盖。
## 三、六栈横向对比

| 维度 | TDG构建器 | NaviAgent | AWO Meta-Tool | ToolWeave | MCP Composition | 自愈路由 |
|------|----------|-----------|---------------|-----------|-----------------|---------|
| **核心价值** | 结构化依赖 | 规模化选择 | 模式固化 | 数据合成 | 多源整合 | 容错恢复 |
| **适用规模** | 20+工具 | 100+工具 | 有重复模式 | 训练阶段 | 多Server | 生产环境 |
| **实现复杂度** | 中 | 高 | 中 | 高 | 中 | 中 |
| **额外延迟** | <5ms | 依赖模型 | 0（减少） | N/A | 15-20ms | 230ms |
| **LLM依赖** | 构建阶段 | 推理阶段 | 离线 | 离线 | 路由阶段 | 失败时 |
| **开源可用性** | 多个库 | 论文代码 | 微软开源 | 部分开源 | MCP协议 | 论文代码 |
| **变现潜力** | ★★★★ | ★★★ | ★★★★ | ★★★★★ | ★★★ | ★★★★ |
## 四、推荐方案：按场景选型

### 场景A：初创团队（工具数<20）

不需要图编排。直接用LLM做工具选择，但做好两件事：
1. 工具Schema要严格定义参数类型和返回类型
2. 用ToolWeave生成测试数据，验证边界case

**投资回报比最高的行动**：花2天时间用ToolWeave跑一遍合成测试，大概率会发现3-5个你没想到的边界case。

### 场景B：成长期产品（工具数20-100）

引入TDG + AWO组合：
1. 先手动构建TDG（第一代方法），不需要运行时学习
2. 部署AWO，从2周的执行trace中挖掘Meta-Tool
3. Meta-Tool能将平均推理步数从4.2步降至2.8步

**预估收益**：Token成本降低30-35%，可靠性提升约2倍。

### 场景C：企业级平台（工具数100+）

全栈部署，但有优先级：
1. **P0**：MCP Server Composition（解决多源整合）
2. **P1**：NaviAgent双层规划（解决规模选择）
3. **P2**：自愈路由（解决生产可靠性）
4. **P3**：AWO Meta-Tool（持续优化）

**关键指标**：在200工具场景下，全栈方案将任务完成率从63%（纯LLM）提升到87%（全栈），但工程投入约需要3-4人月。

### 场景D：工具编排训练营/课程设计

最有变现潜力的知识组合：
1. **Day 1**：TDG基础 + 手动构建练习
2. **Day 2**：NaviAgent原理 + 50工具vs200工具对比实验
3. **Day 3**：AWO Meta-Tool + ToolWeave实战
4. **Day 4-5**：综合项目——构建一个50工具的企业Agent，从TDG到自愈路由

目标学员：已有Agent开发经验、面临工具规模瓶颈的工程师。定价区间：¥2999-5999/人。
## 五、架构总览：图编排Agent系统

```mermaid
flowchart TB
    subgraph Offline["离线阶段"]
        T1[工具Schema收集] --> TDG[Tool Dependency Graph]
        TR[执行Traces] --> AWO[AWO Meta-Tool发现]
        TDG --> TW[ToolWeave合成数据]
        TW --> TRAIN[编排模型训练/微调]
    end

    subgraph Online["在线阶段"]
        USER[用户任务] --> NAV[NaviAgent Navigator]
        TDG -.-> NAV
        NAV --> |工具路径| EXE[NaviAgent Executor]
        EXE --> |调用| TOOLS[工具执行层]
        TOOLS --> |失败| HEAL[自愈路由]
        HEAL --> |替代路径| EXE
        AWO -.-> |Meta-Tool列表| NAV
        MCP[MCP Composition] -.-> TOOLS
    end

    subgraph Feedback["反馈闭环"]
        TOOLS --> LOG[执行日志]
        LOG --> TR
        LOG --> TDG_UPDATE[TDG增量更新]
    end

    style Offline fill:#E3F2FD,stroke:#1565C0
    style Online fill:#FFF3E0,stroke:#E65100
    style Feedback fill:#E8F5E9,stroke:#2E7D32