369 lines
15 KiB
Markdown
369 lines
15 KiB
Markdown
---
|
||
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 系统（下一章）
|