CamTalk/docs/Eino/quick_start/chapter_08_graph_tool.md

---
tags: ["Eino", "Agent", "GraphTool", "Compose", "Workflow", "Backend"]
create time: "2026-04-29 15:30"
---

# 第八章：Graph Tool（复杂工作流）

## 概述

本章引入 Eino 的 **Graph Tool** 能力——将复杂的编排工作流封装为一个可调用的 Tool。通过 `compose.Workflow` 构建包含读取、分块、并行评分、筛选和答案生成的多步骤流水线，让 Agent 能够处理需要多阶段协同的大文件 RAG 场景。

> [!tip] 一句话理解 Graph Tool
>
> **简单 Tool = 单步操作**（如读取文件），**Graph Tool = 完整流水线**（读取 → 分块 → 并行评分 → 筛选 → 生成答案）。它是 compose 编排能力的 Tool 化封装入口。

## 代码位置

- 入口代码：[cmd/ch08/main.go](https://github.com/cloudwego/eino-examples/blob/main/quickstart/chatwitheino/cmd/ch08/main.go)
- RAG 实现：[rag/rag.go](https://github.com/cloudwego/eino-examples/blob/main/quickstart/chatwitheino/rag/rag.go)

## 前置条件

与第一章一致：需要配置一个可用的 ChatModel（OpenAI 或 Ark）。

## 运行

在 `examples/quickstart/chatwitheino` 目录下执行：

```bash
# 设置项目根目录
export PROJECT_ROOT=/path/to/your/project

go run ./cmd/ch08
```

输出示例：

```
you> 请帮我分析 RFC6455 文档中关于 WebSocket 握手的部分
[assistant] 我来帮你分析文档...
[tool call] answer_from_document(file_path: "rfc6455.txt", question: "WebSocket 握手过程")
[tool result] 找到 3 个相关片段，正在生成答案...
[assistant] 根据 RFC6455 文档，WebSocket 握手过程如下...
```

## 从简单 Tool 到 Graph Tool：为什么需要复杂工作流

第四章我们创建了简单的 Tool，每个 Tool 执行单一任务。但实际场景中，很多任务需要多个步骤协同完成。

**简单 Tool 的局限：**

| 局限 | 说明 |
|------|------|
| 单一职责 | 每个 Tool 只能做一件事（读文件、搜索等） |
| 无法并行 | 多个独立子任务不能同时执行 |
| 难以复用 | 复杂逻辑硬编码在调用链中，无法单独测试和复用 |

**重要说明：本章只是展示 compose/graph/workflow 能力的一角。**

从更大的视角看，Eino 的 `compose` 包提供了非常通用、确定性的编排能力：你可以把任何需要"确定性业务流程"的系统，用 `compose` 的 Graph/Chain/Workflow 组织成可执行的流水线，并且它能够**原生编排 Eino 的所有 component**（ChatModel、Prompt、Tools、Retriever、Embedding、Indexer 等），同时具备完整的 **callback** 体系，以及 **interrupt/resume + checkpoint** 支持。

### Graph Tool 的定位

> [!note] Graph Tool vs 简单 Tool
>
> | 对比项 | 简单 Tool | Graph Tool |
> |--------|----------|------------|
> | 本质 | 单步函数 | compose 编排产物的封装 |
> | 编排 | 无 | 由 compose 提供（并行、分支、字段映射） |
> | 状态管理 | 无 | 节点间传递数据 + checkpoint 持久化 |
> | 中断恢复 | 不支持 | 支持（嵌套 interrupt 场景） |

### 核心类比

> [!tip] 厨房做菜类比
>
> - **简单 Tool**：像是一个厨具（菜刀——只负责切东西）
> - **Graph Tool**：像是一条预制菜流水线（备料 → 烹饪 → 摆盘——每一步自动衔接，你只需说"做这道菜"）

## 关键概念

### compose.Workflow

`compose.Workflow` 是 Eino 中构建有状态工作流的核心组件。与线性 Chain 不同，Workflow 允许创建 DAG（有向无环图），支持汇聚节点、并行分支和非相邻连接：

```go
wf := compose.NewWorkflow[Input, Output]()

// 添加节点并建立连接
wf.AddLambdaNode("load", loadFunc).AddInput(compose.START)
wf.AddLambdaNode("chunk", chunkFunc).AddInput("load")
wf.AddLambdaNode("answer", answerFunc).
    AddInput("chunk").
    AddInputWithOptions(compose.START,
        []*compose.FieldMapping{compose.MapFields("Question", "Question")},
        compose.WithNoDirectDependency())
wf.End().AddInput("answer")
```

> [!question] 深入思考
>
> Workflow 为什么需要 `START` 和 `END` 这两个虚拟节点，而不是直接指定输入输出？  
> 提示：想想如果工作流有多个入口点（例如用户可以直接跳转到某个中间节点重试），或者需要在运行时动态插入新节点。START/END 为这些灵活性提供了统一的锚点。

### BatchNode（并行处理）

`BatchNode` 用于并行处理一批独立任务，充分利用计算资源：

```go
scorer := batch.NewBatchNode(&batch.NodeConfig[scoreTask, scoredChunk]{
    Name:           "ChunkScorer",
    InnerTask:      newScoreWorkflow(cm), // 单个 chunk 的评分流程
    MaxConcurrency: 5,                    // 最大并发数
})
```

**工作原理：**

1. 接收任务切片作为输入
2. 按 `MaxConcurrency` 限制并行调度（内部使用 goroutine pool）
3. 所有结果收集后按顺序返回

> [!tip] 选择 MaxConcurrency 的原则
>
> - 过低 → 浪费了并发能力，响应慢
> - 过高 → 资源竞争，LLM API 限流
> - 推荐做法：以 LLM Provider 的 QPS 上限为参考值，一般 3~10 之间调整

### FieldMapping（跨节点数据传递）

FieldMapping 解决非相邻节点间的数据传递问题：当两个节点没有直接的边连接时，你需要显式声明数据的来源和目标字段。

```go
wf.AddLambdaNode("score", scoreFunc).
    // 从 "chunk" 节点取 All 数据，映射到当前节点的 Chunks 字段
    AddInputWithOptions("chunk",
        []*compose.FieldMapping{compose.ToField("Chunks")},
        compose.WithNoDirectDependency()).
    // 从 START 节点取 Question 字段，直接映射到当前节点的 Question 字段
    AddInputWithOptions(compose.START,
        []*compose.FieldMapping{compose.MapFields("Question", "Question")},
        compose.WithNoDirectDependency())
```

**三种 FieldMapping 方式：**

| 方法 | 作用 | 适用场景 |
|------|------|---------|
| `MapFields(src, dst)` | 字段重命名映射 | 两端字段名不一致时 |
| `ToField(dst)` | 整条数据映射到单一字段 | 上游只有一个输出，且需包裹到 struct |
| `All()` | 传入上游全部输出（默认行为） | 相邻节点间的直接传递 |

**为什么非相邻节点需要 `WithNoDirectDependency`？**

Eino 依赖图检测会验证节点的输入是否来自前驱节点。当使用 FieldMapping 跨越层级取值时，必须显式标记 `WithNoDirectDependency()`，否则会被依赖检查拦截。

## Graph Tool 的实现

下面我们以"大文件内容检索并回答"为例，逐步构建一个完整的 Graph Tool。整个流程分为三步：定义 IO 结构 → 构建工作流 → 封装为 Tool。

### 1. 定义输入输出结构

输入和输出定义了 Graph Tool 对外暴露的接口契约，也是 Agent 调用时的参数 schema 来源：

```go
type Input struct {
    FilePath string `json:"file_path" jsonschema:"description=Absolute path to the document"`
    Question string `json:"question"  jsonschema:"description=The question to answer"`
}

type Output struct {
    Answer  string   `json:"answer"`
    Sources []string `json:"sources"`
}
```

> [!note] jsonschema tag 的作用
>
> 这些标签会被自动转换为 JSON Schema，决定了 Agent（LLM）看到的工具参数描述。写得好，模型就能精准理解该传什么值。

### 2. 构建工作流

完整的 `buildWorkflow` 函数实现了五个阶段的流水线：

```go
func buildWorkflow(cm model.BaseChatModel) *compose.Workflow[Input, Output] {
    wf := compose.NewWorkflow[Input, Output]()

    // --- load: 读取文件 ---
    wf.AddLambdaNode("load", compose.InvokableLambda(
        func(ctx context.Context, in Input) ([]*schema.Document, error) {
            data, err := os.ReadFile(in.FilePath)
            if err != nil {
                return nil, err
            }
            return []*schema.Document{{Content: string(data)}}, nil
        },
    )).AddInput(compose.START)

    // --- chunk: 分块 ---
    wf.AddLambdaNode("chunk", compose.InvokableLambda(
        func(ctx context.Context, docs []*schema.Document) ([]*schema.Document, error) {
            var out []*schema.Document
            for _, d := range docs {
                out = append(out, splitIntoChunks(d.Content, 800)...)
            }
            return out, nil
        },
    )).AddInput("load")

    // --- score: 并行评分（核心亮点）---
    scorer := batch.NewBatchNode(&batch.NodeConfig[scoreTask, scoredChunk]{
        Name:           "ChunkScorer",
        InnerTask:      newScoreWorkflow(cm),
        MaxConcurrency: 5,
    })

    wf.AddLambdaNode("score", compose.InvokableLambda(
        func(ctx context.Context, in scoreIn) ([]scoredChunk, error) {
            tasks := make([]scoreTask, len(in.Chunks))
            for i, c := range in.Chunks {
                tasks[i] = scoreTask{Text: c.Content, Question: in.Question}
            }
            return scorer.Invoke(ctx, tasks)
        },
    )).
        AddInputWithOptions("chunk", []*compose.FieldMapping{compose.ToField("Chunks")}, compose.WithNoDirectDependency()).
        AddInputWithOptions(compose.START, []*compose.FieldMapping{compose.MapFields("Question", "Question")}, compose.WithNoDirectDependency())

    // --- filter: 筛选 top-k ---
    wf.AddLambdaNode("filter", compose.InvokableLambda(
        func(ctx context.Context, scored []scoredChunk) ([]scoredChunk, error) {
            sort.Slice(scored, func(i, j int) bool {
                return scored[i].Score > scored[j].Score
            })
            if len(scored) > 3 {
                scored = scored[:3]
            }
            return scored, nil
        },
    )).AddInput("score")

    // --- answer: 生成最终答案 ---
    wf.AddInputWithOptions("filter", []*compose.FieldMapping{compose.ToField("TopK")}, compose.WithNoDirectDependency()).
        AddInputWithOptions(compose.START, []*compose.FieldMapping{compose.MapFields("Question", "Question")}, compose.WithNoDirectDependency())

    wf.End().AddInput("answer")

    return wf
}
```

> [!note] 代码解读：为什么 score 和 answer 都有两处 AddInput？
>
> **score 节点**需要两个数据来源：
> - `chunk` 的输出（待评分的文本块）
> - `START` 的 `Question`（用户的问题，用来给每个 block 打分）
>
> **answer 节点**同理也需要：
> - `filter` 的输出（top-k 的相关片段）
> - `START` 的 `Question`（拼接到 prompt 中）
>
> 这就是为什么需要 `WithNoDirectDependency()`——它们跳过了中间节点，直接向源头要数据。

### 3. 封装为 Tool

最后一步是将编译后的工作流包装成 Agent 可调用的标准 Tool：

```go
func BuildTool(ctx context.Context, cm model.BaseChatModel) (tool.BaseTool, error) {
    wf := buildWorkflow(cm)
    return graphtool.NewInvokableGraphTool[Input, Output](
        wf,
        "answer_from_document",          // Tool 名称（Agent 看到的名字）
        "Search a large document for relevant content and synthesize an answer.", // Tool 描述
    )
}
```

> [!warning] 编译时机
>
> `graphtool.NewInvokableGraphTool` 内部会对 Workflow 执行编译检查，验证节点连通性、类型兼容性。如果在运行时才发现错误，排查会比较困难——建议在单元测试中对 buildWorkflow 的返回值做一次 compile-time check。

## Graph Tool 执行流程图

```mermaid
flowchart TD
    A["输入: file_path, question"] --> B["load\n读取文件\n→ []*Document"]
    B --> C["chunk\n分块 (800 tokens)\n→ []*Document"]

    C --> D["score\n并行评分\n(MaxConcurrency=5)\n→ []scoredChunk"]

    D --> E["filter\n排序并取 top-k\n→ []scoredChunk"]

    E --> F["answer\n结合问题和\nTop-K 片段生成答案\n→ Output"]

    F --> G["返回: {answer, sources}"]

    style A fill:#e3f2fd
    style G fill:#e8f5e9
    style D fill:#fff3e0
```

**流程中的关键设计决策：**

| 阶段 | 决策点 | 原因 |
|------|--------|------|
| chunk | 固定 800 token 分块 | 平衡上下文窗口与检索精度 |
| score | 并行评分（MaxConcurrency=5） | 避免串行等待，利用 LLM API 并发能力 |
| filter | 保留 top-3 | 控制后续 token 消耗，避免信息过载 |

## 可中断恢复

Graph Tool 天然继承 Eino 的中断恢复机制。当工作流内部的某个节点触发 `interrupt` 时，Runner 会暂停整个工作流，等待用户输入后 resume：

```go
// 在工作流节点中使用 interrupt
func myNode(ctx context.Context, input MyInput) (MyOutput, error) {
    wasInterrupted, _, stored := tool.GetInterruptState[string](ctx)
    if !wasInterrupted {
        return MyOutput{}, tool.StatefulInterrupt(ctx, &commontool.ApprovalInfo{
            ToolName: "my_workflow_step",
            ArgumentsInJSON: stored,
        }, stored)
    }
    // Resume 后继续执行...
    return process(stored), nil
}
```

> [!tip] Graph Tool 的中断优势
>
> 由于每个节点都是独立的 lambda 函数，可以在任意节点插入 interrupt 逻辑，而无需修改其他节点。这种细粒度的可控性是简单 Tool 无法做到的。

## 本章小结

| 概念 | 一句话理解 |
|------|-----------|
| **Graph Tool** | 将 compose 编排产物封装为 Agent 可调用的 Tool 入口 |
| **compose.Workflow** | 支持 DAG 结构的有状态工作流，可表达复杂业务逻辑 |
| **BatchNode** | 并行处理批量任务的内置组件，受 MaxConcurrency 限制 |
| **FieldMapping** | 跨节点传递数据的机制，解决非相邻节点间的通信 |
| **可中断恢复** | Graph Tool 完整继承 interrupt/resume + checkpoint 能力 |

## 扩展思考

### Graph Tool 的典型应用场景

| 场景 | 说明 | 收益 |
|------|------|------|
| **多文档 RAG** | 并行检索多个文档源并综合回答 | 减少 Token 往返次数，一次 Tool Call 覆盖全部 |
| **多模型协作** | 不同模型处理不同阶段（摘要 → 翻译 → 总结） | 各取所长，降低单次请求成本 |
| **审批流水线** | 工作流中包含需要人工确认的步骤 | 兼顾自动化与安全合规 |
| **数据管道** | ETL（抽取、转换、加载）流程的 Agent 化 | 用自然语言驱动数据处理 |

### 性能优化建议

1. **调整 MaxConcurrency**：根据 LLM API 的速率限制调参，一般 3~10 为宜
2. **缓存层**：对相同 input + question 组合的结果做缓存，避免重复计算
3. **自适应 chunk 大小**：根据文档类型（代码、散文、日志）动态调整分块策略
4. **Early Exit**：当 top-1 分数远高于第二名时，跳过 filter 直接回答

## 关联笔记

- [[Eino/quick_start/_index]]
- [[Eino/quick_start/chapter_04_tool_and_filesystem]] — 简单 Tool 的创建与文件系统访问（第二章的工具章节）
- [[Eino/quick_start/chapter_07_interrupt_resume]] — Interrupt/Resume 机制（上一章）
- [[Eino/quick_start/chapter_09_skill_console]] — Skill 系统（下一章）