161 lines
5.2 KiB
Markdown
161 lines
5.2 KiB
Markdown
|
---
|
||
|
|
tags: [ai/agent, eino-dag, workflow, parallel-step, conditional-branch]
|
||
|
|
create time: 2026-08-08 18:43
|
||
|
|
update time: 2026-08-08 18:43
|
||
|
|
---
|
||
|
|
|
||
|
|
# Eino DAG 工作流设计
|
||
|
|
|
||
|
|
## 概述
|
||
|
|
|
||
|
|
Eino 是 IBM 开源的 AI 编排框架，核心理念是将多步 AI 任务建模为有向无环图（DAG），通过节点表示原子操作、边定义数据流向。与线性 Chain 不同，DAG 天然支持并行执行、条件分支和错误恢复，是构建复杂 Agent 工作流的骨架。
|
||
|
|
|
||
|
|
## 核心概念
|
||
|
|
|
||
|
|
### Node — 节点
|
||
|
|
|
||
|
|
节点是 DAG 的最小执行单元，每个节点封装一个独立能力：
|
||
|
|
|
||
|
|
- **LLM Node**：调用大模型生成文本
|
||
|
|
- **Tool Node**：执行预注册的外部工具（搜索、数据库查询等）
|
||
|
|
- **Prompt Node**：模板渲染，组装上下文
|
||
|
|
- **State Node**：读写共享状态
|
||
|
|
- **Lambda Node**：自定义 Go 函数闭包
|
||
|
|
|
||
|
|
每个节点拥有明确的输入类型和输出类型，节点间通过 State Schema 进行类型安全的传递。
|
||
|
|
|
||
|
|
### Edge — 边
|
||
|
|
|
||
|
|
边定义节点的执行顺序和数据依赖：
|
||
|
|
|
||
|
|
- **普通边**：上游节点完成后触发下游
|
||
|
|
- **条件边**：基于中间结果决定走向
|
||
|
|
- **广播边**：一对多 fan-out，并行触发多个子任务
|
||
|
|
|
||
|
|
### Workflow — 有向无环图构建
|
||
|
|
|
||
|
|
Workflow 在编译期完成拓扑排序，确保图中无环。运行时按拓扑序执行所有没有未满足依赖的节点，已就绪节点可并发执行。
|
||
|
|
|
||
|
|
```mermaid
|
||
|
|
graph TD
|
||
|
|
A["用户请求"] --> B["Prompt 节点\n路由意图"]
|
||
|
|
B --> C{"Choice\n条件判断"}
|
||
|
|
C -->|"问答"| D["RAG 检索节点"]
|
||
|
|
C -->|"创作"| E["知识库加载节点"]
|
||
|
|
D --> F["LLM 节点\n生成回答"]
|
||
|
|
E --> F
|
||
|
|
F --> G["质量评估节点"]
|
||
|
|
G --> H{"再检查"}
|
||
|
|
H -->|"不达标"| D
|
||
|
|
H -->|"达标"| I["结果输出"]
|
||
|
|
```
|
||
|
|
|
||
|
|
> [!NOTE]
|
||
|
|
> DAG 的核心优势在于并行性——如果两个节点没有数据依赖，它们可以在不同 Goroutine 中同时运行，显著缩短端到端延迟。
|
||
|
|
|
||
|
|
## 详解
|
||
|
|
|
||
|
|
### 状态传递机制
|
||
|
|
|
||
|
|
Eino 使用 `State` 对象作为节点间通信的管道。State 是一个强类型字典，节点声明自己的读集合（reads）和写集合（writes）。
|
||
|
|
|
||
|
|
```go
|
||
|
|
// 定义 State Schema
|
||
|
|
state := eino.StateSchema{
|
||
|
|
Fields: []eino.StateField{
|
||
|
|
{Name: "query", Type: eino.TypeString},
|
||
|
|
{Name: "context", Type: eino.TypeSlice},
|
||
|
|
{Name: "answer", Type: eino.TypeString},
|
||
|
|
},
|
||
|
|
}
|
||
|
|
|
||
|
|
// 节点声明权限
|
||
|
|
llmNode := &eino.LLMNode{
|
||
|
|
Reads: []string{"query", "context"},
|
||
|
|
Writes: []string{"answer"},
|
||
|
|
}
|
||
|
|
```
|
||
|
|
|
||
|
|
> [!TIP]
|
||
|
|
> 面试常考点：State 的读写范围决定了节点能否被安全并行——两个节点如果没有重叠的 writes，就一定能并发执行。
|
||
|
|
|
||
|
|
### Choice / Branch — 条件分支
|
||
|
|
|
||
|
|
Choice 节点根据上一个节点的输出选择后续路径，底层基于回调函数返回目标节点 ID：
|
||
|
|
|
||
|
|
```go
|
||
|
|
workflow.AddChoice(
|
||
|
|
retrieverNode, // 上游节点
|
||
|
|
func(state *eino.State) (string, error) { // 路由函数
|
||
|
|
hits := state.MustGet("hits").([]*Doc)
|
||
|
|
if len(hits) == 0 {
|
||
|
|
return "fallback", nil // 走备用路径
|
||
|
|
}
|
||
|
|
return "primary", nil // 走主路径
|
||
|
|
},
|
||
|
|
map[string]string{ // 分支 → 目标节点映射
|
||
|
|
"primary": answerNode,
|
||
|
|
"fallback": fallbackNode,
|
||
|
|
},
|
||
|
|
)
|
||
|
|
```
|
||
|
|
|
||
|
|
### ParallelStep — 并行执行
|
||
|
|
|
||
|
|
对同一上游 fan-out 到多个下游，用 `ParallelStep` 包装一组子图：
|
||
|
|
|
||
|
|
```go
|
||
|
|
parallelStep := &eino.ParallelStep{
|
||
|
|
Inputs: queryNode, // 输入分发策略
|
||
|
|
Steps: []*eino.Node{ // 并行执行的节点列表
|
||
|
|
searchAgent,
|
||
|
|
docAnalyzer,
|
||
|
|
sentimentChecker,
|
||
|
|
},
|
||
|
|
Merge: mergeResults, // 结果合并函数
|
||
|
|
}
|
||
|
|
```
|
||
|
|
|
||
|
|
三个 Agent 并行启动，各自处理 query 的一个维度，最后在 `mergeResults` 处汇聚。Merge 函数负责将结构化数据聚合为最终输出。
|
||
|
|
|
||
|
|
### 错误处理与重试策略
|
||
|
|
|
||
|
|
DAG 层面的容错机制包括：
|
||
|
|
|
||
|
|
| 策略 | 说明 | 适用场景 |
|
||
|
|
|------|------|---------|
|
||
|
|
| 节点级超时 | 单个节点执行超过阈值自动中断 | LLM API 慢响应 |
|
||
|
|
| 失败重试 | 指数退避重试失败的节点 | 网络抖动、限流 |
|
||
|
|
| 降级路径 | Choice 切换至替代节点 | RAG 检索为空时回退 |
|
||
|
|
| 熔断器 | 连续失败 N 次后跳过整条链路 | 上游服务不可用 |
|
||
|
|
|
||
|
|
```go
|
||
|
|
node.WithRetry(&eino.RetryConfig{
|
||
|
|
MaxRetries: 3,
|
||
|
|
BackoffBase: 1 * time.Second,
|
||
|
|
BackoffMul: 2,
|
||
|
|
RetryOn: func(err error) bool {
|
||
|
|
return errors.Is(err, context.DeadlineExceeded)
|
||
|
|
},
|
||
|
|
})
|
||
|
|
```
|
||
|
|
|
||
|
|
## 实践场景
|
||
|
|
|
||
|
|
**面试高频题**：如何用 DAG 实现一个多轮对话系统？
|
||
|
|
|
||
|
|
答案要点：
|
||
|
|
1. 每轮对话作为一个 DAG 实例，State 携带历史消息
|
||
|
|
2. IntentNode 判断用户意图类型
|
||
|
|
3. Choice 分发给不同的专业 Agent
|
||
|
|
4. ParallelStep 并行调用工具和模型
|
||
|
|
5. 质量评估节点做 self-check，不达标则重新检索
|
||
|
|
|
||
|
|
> [!WARNING]
|
||
|
|
> DAG 虽然灵活，但复杂度随节点数呈组合增长。实际项目中建议控制在 15 个节点以内，超出则拆分为嵌套子 DAG。
|
||
|
|
|
||
|
|
## 扩展阅读
|
||
|
|
|
||
|
|
- [[多智能体协作模式]]
|
||
|
|
- [[上下文工程与记忆管理]]
|