---
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。

## 扩展阅读

- [[多智能体协作模式]]
- [[上下文工程与记忆管理]]