CamTalk/docs/Eino/quick_start/chapter_08_graph_tool.md

tags, create time

tags	create time
Eino Agent GraphTool Compose Workflow Backend	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
• RAG 实现：rag/rag.go

前置条件

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

运行

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

# 设置项目根目录
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（有向无环图），支持汇聚节点、并行分支和非相邻连接：

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 用于并行处理一批独立任务，充分利用计算资源：

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

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 来源：

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 函数实现了五个阶段的流水线：

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：

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 执行流程图

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：

// 在工作流节点中使用 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 系统（下一章）