Eino 编排详解：Chain、Graph 与 Workflow

核心摘要

Eino compose 提供三种相关但不同的 API：Chain 是简洁的前向管线，Graph 暴露显式拓扑和 Cycle，Workflow 将多个前驱字段映射为类型化输入。三者都编译为支持 Invoke、Stream、Collect 与 Transform 的 Runnable。Compile 负责结构和可表达的类型检查，生产正确性仍取决于运行时限制、授权、取消和结果核验。

核心要点

• 从普通 Go 或 Chain 开始，只有拓扑或字段映射确有价值时才升级为 Graph 或 Workflow。
• Graph Cycle 除 WithMaxRunSteps 外，还必须有应用级终止条件和副作用策略。
• Workflow 分离控制流与数据流，但不支持 Cycle。
• Stream 适配可能缓存或复制数据，必须测试内存与 Cancellation 行为。
• 模型选择的 Branch 只是提议，权限和不可逆操作由确定性代码负责。

三种 API 对应三类任务

官方概述与 Workflow 指南给出了稳定的概念边界：

Chain 是简化的 Graph Builder。Workflow 也不是「功能更多的 Graph」，而是通过控制流与字段映射分离做出不同的所有权选择。

编译期与运行时的边界

Eino 的编排设计原则使用 Go 类型尽可能对齐相邻节点。它能在流量进入 Graph 前发现一类错误，但「编译」有两层：

应把 Compile 当作拓扑构建，不是安全认证。

Runnable 的四种模式

编译后的 Chain、Graph 或 Workflow 提供四种执行模式：

流式编程指南说明 Eino 可以把 Value 封装为 Stream、把 Stream 拼接为 Value、为多消费者复制 Stream，或合并多条 Stream。每次转换都会改变缓冲和失败行为，不能假定四种模式的延迟与内存等价。

Chain：让直线流程保持直线

当每个结果都进入下一步且没有 Branch 时，Chain 最合适。常见结构是：

校验输入 -> 归一化 -> 调用模型 -> 校验输出

不要把全部职责塞进一个 Lambda。Provider Call、Authorization、Persistent Effect 和 Pure Transformation 的重试与观测契约并不相同。

Graph：显式表达 Branch 与 Cycle

当控制流本身很重要时使用 Graph。任何 Cycle 都应定义：

• 不依赖自然语言乐观判断的 Exit Condition；
• WithMaxRunSteps 硬上限；
• Deadline、Cancellation、Token、Cost 与 Fan-out Budget；
• 副作用的 Idempotency 与 Reconciliation；
• 模糊失败后的显式 outcome_unknown 状态。

如果 Payment 或 Message 可能已经提交后发生 Timeout，不能直接回到循环再次执行。应先查询权威下游状态。

Workflow：映射字段，而不是制造胶水对象

Workflow 可以从多个前驱选择字段，组成下游类型化输入，并把控制 Edge 与 Data Mapping 分开。它适合复用输入结构不同的现有业务函数，但仍是无环流程，字段名也属于兼容契约。

映射字段承载敏感或可变数据时，要记录 Source、Revision 与 Authorization Scope。接线正确不代表数值可信。

可运行的 Policy Graph

下面示例以 Eino Core v0.9.21 为目标，构建一个带显式 Branch 和四步运行上限的 Graph。Decision 来自可信应用数据，不来自模型文本。

package main

import (
	"context"
	"fmt"
	"strings"

	"github.com/cloudwego/eino/compose"
)

type Job struct {
	Task     string
	Approved bool
}

type Decision struct {
	Status string
	Task   string
}

func main() {
	ctx := context.Background()
	graph := compose.NewGraph[Job, Decision]()

	validate := compose.InvokableLambda(func(ctx context.Context, input Job) (Job, error) {
		if err := ctx.Err(); err != nil {
			return Job{}, err
		}
		input.Task = strings.TrimSpace(input.Task)
		if input.Task == "" {
			return Job{}, fmt.Errorf("task is required")
		}
		return input, nil
	})
	execute := compose.InvokableLambda(func(_ context.Context, input Job) (Decision, error) {
		return Decision{Status: "executed", Task: input.Task}, nil
	})
	deny := compose.InvokableLambda(func(_ context.Context, input Job) (Decision, error) {
		return Decision{Status: "denied", Task: input.Task}, nil
	})

	if err := graph.AddLambdaNode("validate", validate); err != nil {
		panic(err)
	}
	if err := graph.AddLambdaNode("execute", execute); err != nil {
		panic(err)
	}
	if err := graph.AddLambdaNode("deny", deny); err != nil {
		panic(err)
	}
	if err := graph.AddEdge(compose.START, "validate"); err != nil {
		panic(err)
	}
	branch := compose.NewGraphBranch(
		func(_ context.Context, input Job) (string, error) {
			if input.Approved {
				return "execute", nil
			}
			return "deny", nil
		},
		map[string]bool{"execute": true, "deny": true},
	)
	if err := graph.AddBranch("validate", branch); err != nil {
		panic(err)
	}
	if err := graph.AddEdge("execute", compose.END); err != nil {
		panic(err)
	}
	if err := graph.AddEdge("deny", compose.END); err != nil {
		panic(err)
	}

	runnable, err := graph.Compile(ctx, compose.WithMaxRunSteps(4))
	if err != nil {
		panic(err)
	}
	result, err := runnable.Invoke(ctx, Job{Task: "send invoice", Approved: false})
	if err != nil {
		panic(err)
	}
	fmt.Printf("%s: %s\n", result.Status, result.Task)
}

预期输出：

denied: send invoice

示例刻意把 Approval 留在模型调用之外。真实服务应从经过认证的持久审批记录取得它，并绑定准确 Action 和 Input Revision。

失败与恢复契约

每个 Node 都应区分失败类别：

Graph State 不是通用事务。Checkpoint 可以恢复框架执行，外部系统却可能已经发生变化。

评测清单

• 在启动时 Compile Graph，定义错误应阻止部署。
• 覆盖每个 Branch、Join、Cycle Exit、最大步数与 Cancellation Point。
• 分别运行 Stream 和 Invoke，记录 Buffer 增长与首尾 Chunk 行为。
• 在副作用前后注入 Provider 与 Tool Failure。
• 与普通 Go 基线比较，证明编排复杂度有真实价值。
• 版本化 Graph Topology、Node Implementation、Option、Prompt 与 Policy。

常见问题

Chain 一定比 Graph 简单吗？

只在真正顺序执行时成立。如果 Branch 被隐藏在一个超大 Lambda 内，小型显式 Graph 反而更容易测试和运维。

Graph 能表达 ReAct Loop 吗？

可以，但 Eino ADK 已为常见 Model-Tool Loop 提供 ChatModelAgent。只有自定义循环的行为与证据值得额外维护时，才应自行构建。

Workflow 支持 Branch 吗？

当前 Workflow 文档包含 Branch 能力，但 Workflow 仍保持无环，并以字段映射为核心。设计时应核对准确的 v0.9.21 API。

WithMaxRunSteps 足以控制成本吗？

不足。它限制 Graph Step，不限制 Provider Token、Tool Fan-out、Byte、Wall Time 或外部费用，这些预算都要独立执行。

Callback 能让 Graph 完整可观测吗？

它提供生命周期 Hook，但 Coverage、Payload Redaction、Sampling、Exporter Failure 与 Stream Error 仍需专项测试。

一手资料

• Eino v0.9.21 Chain 源码
• Eino v0.9.21 Graph 源码
• Eino 编排设计原则
• Eino Chain 与 Graph 入门
• Eino Workflow 指南
• Go context Package

相关阅读

• Eino 框架解析
• Eino 核心组件
• Eino 流式处理与 Callback
• Agentic Workflow 术语
• Graph Orchestration 术语