07｜Graph 实战：实现分支、并行与循环

《Eino 实战：用 Go 构建 AI 应用与智能体》· 第 07 篇
关键词：compose.Graph、Node、Edge、Branch、并行汇聚、State、循环
前置条件：已完成 06｜Chain 入门。

Chain 表达的是“一路向前”。但工单处理天然不是一条直线：

• 不同分类走不同处理组——这是分支。
• 一边查知识库、一边查历史工单——这是并行。
• 处理失败时重试，直到成功或超过次数——这是循环。

这三种结构都需要 Graph。它只多加两个概念：节点（Node）和边（Edge）。

一、Graph 的基本模型

g:=compose.NewGraph[string,string]()

两个泛型参数分别是图的输入类型和输出类型，和 Chain 一样。之后：

• AddXxxNode(key, node)添加节点，key是你给节点起的名字，用于连接。
• AddEdge(from, to)连接两个节点。
• compose.START和compose.END是两个特殊端点，表示入口和出口。

_=g.AddLambdaNode("classify",classify)_=g.AddEdge(compose.START,"classify")_=g.AddEdge("classify",compose.END)

这和 Chain 的差别在于：流向是你显式画出来的，因此才有能力表达分支、并行和环。

二、分支：按运行时结果选择路径

分支用NewGraphBranch构建：

branch:=compose.NewGraphBranch(func(_context.Context,categorystring)(string,error){returncategory,nil// 返回值即目标节点名},map[string]bool{"hardware":true,"network":true,"other":true},// 哪些是终点分支)_=g.AddBranch("classify",branch)

两个参数值得说明：

• 第一个是分支函数，它接收上游输出，返回一个节点名。返回什么，流程就走哪条路。
• 第二个标记哪些目标是“终点分支”。写在里面的节点不需要再连出边；否则编译时可能报图未连通。

注意分支函数在运行时执行，所以同一条输入在不同条件下可能走不同路径。这正好呼应第一篇的说法：预定义拓扑不等于输出确定。

三、并行与汇聚

并行不需要特殊 API，一个节点的输出连到多个下游即可：

_=g.AddEdge("split","slow")_=g.AddEdge("split","fast")

难点在汇聚：两个上游同时指向一个下游时，类型必须能对齐。如果两个节点都输出string，而汇聚节点期待map[string]any，编译会直接报错。

解决办法是WithOutputKey，把节点输出包装成带键的 map：

_=g.AddLambdaNode("slow",slowCheck,compose.WithOutputKey("knowledge"))_=g.AddLambdaNode("fast",fastCheck,compose.WithOutputKey("history"))

此时两个节点的输出类型都变成map[string]any，汇聚节点接收到的就是按 key 合并的结果。这是 Graph 里最实用、也最容易卡住的技巧之一。

四、完整示例：分支

这段程序不访问网络，可以直接运行：

packagemainimport("context""fmt""log""strings""github.com/cloudwego/eino/compose")funcmain(){iferr:=run();err!=nil{log.Fatal(err)}}funcrun()error{ctx:=context.Background()g:=compose.NewGraph[string,string]()// 节点 1：根据关键词做简单分类（真实项目里常由模型完成）。classify:=compose.InvokableLambda(func(_context.Context,instring)(string,error){switch{casestrings.Contains(in,"打印")||strings.Contains(in,"离线"):return"hardware",nilcasestrings.Contains(in,"网")||strings.Contains(in,"连接"):return"network",nildefault:return"other",nil}})// 分支节点：根据分类结果选择下游节点。branch:=compose.NewGraphBranch(func(_context.Context,categorystring)(string,error){returncategory,nil// 返回值即目标节点名},map[string]bool{"hardware":true,"network":true,"other":true},// 标记哪些是终点分支)handleHardware:=compose.InvokableLambda(func(_context.Context,instring)(string,error){return"转硬件支持组处理："+in,nil})handleNetwork:=compose.InvokableLambda(func(_context.Context,instring)(string,error){return"转网络支持组处理："+in,nil})handleOther:=compose.InvokableLambda(func(_context.Context,instring)(string,error){return"转综合支持组处理："+in,nil})_=g.AddLambdaNode("classify",classify)_=g.AddLambdaNode("hardware",handleHardware)_=g.AddLambdaNode("network",handleNetwork)_=g.AddLambdaNode("other",handleOther)_=g.AddBranch("classify",branch)_=g.AddEdge(compose.START,"classify")_=g.AddEdge("hardware",compose.END)_=g.AddEdge("network",compose.END)_=g.AddEdge("other",compose.END)runnable,err:=g.Compile(ctx)iferr!=nil{returnfmt.Errorf("编译 Graph 失败: %w",err)}for_,ticket:=range[]string{"三楼打印机离线，无法出纸","会议室网络连接不稳定","希望调整账号权限",}{out,err:=runnable.Invoke(ctx,ticket)iferr!=nil{returnerr}fmt.Printf("%s -> %s\n",ticket,out)}returnnil}

实机输出：

三楼打印机离线，无法出纸 -> 转硬件支持组处理：hardware 会议室网络连接不稳定 -> 转网络支持组处理：network 希望调整账号权限 -> 转综合支持组处理：other

同一张图，三条不同路径。分支函数决定了实际走向。

五、完整示例：并行与循环

这段程序演示两个进阶结构，同样无需密钥：

packagemainimport("context""fmt""log""time""github.com/cloudwego/eino/compose")// state 用于在多次节点执行之间保存数据。typestatestruct{AttemptsintDonebool}funcmain(){iferr:=run();err!=nil{log.Fatal(err)}}funcrun()error{ctx:=context.Background()iferr:=runParallel(ctx);err!=nil{returnerr}returnrunLoop(ctx)}// runParallel 演示并行分支与汇聚。funcrunParallel(ctx context.Context)error{g:=compose.NewGraph[string,map[string]any]()// 扇出：一个节点的输出同时送往两个并行节点。split:=compose.InvokableLambda(func(_context.Context,instring)(string,error){returnin,nil})slowCheck:=compose.InvokableLambda(func(_context.Context,instring)(string,error){time.Sleep(20*time.Millisecond)return"知识库检索结果("+in+")",nil})fastCheck:=compose.InvokableLambda(func(_context.Context,instring)(string,error){return"工单历史查询结果("+in+")",nil})// 扇入：接收来自多个上游的结果，输入类型是 map[string]any。merge:=compose.InvokableLambda(func(_context.Context,inmap[string]any)(map[string]any,error){returnin,nil})_=g.AddLambdaNode("split",split)// WithOutputKey 把节点输出变成 map[string]any{key: value}，// 使多个上游结果在下游按 key 汇聚。_=g.AddLambdaNode("slow",slowCheck,compose.WithOutputKey("knowledge"))_=g.AddLambdaNode("fast",fastCheck,compose.WithOutputKey("history"))_=g.AddLambdaNode("merge",merge)_=g.AddEdge(compose.START,"split")_=g.AddEdge("split","slow")_=g.AddEdge("split","fast")_=g.AddEdge("slow","merge")_=g.AddEdge("fast","merge")_=g.AddEdge("merge",compose.END)runnable,err:=g.Compile(ctx)iferr!=nil{returnfmt.Errorf("编译并行 Graph 失败: %w",err)}start:=time.Now()out,err:=runnable.Invoke(ctx,"打印机离线")iferr!=nil{returnfmt.Errorf("执行并行 Graph 失败: %w",err)}fmt.Printf("[并行] 耗时约 %v\n",time.Since(start).Round(time.Millisecond))fork,v:=rangeout{fmt.Printf("[并行] %s = %v\n",k,v)}returnnil}// runLoop 演示带条件回退的循环，直到满足退出条件。funcrunLoop(ctx context.Context)error{g:=compose.NewGraph[string,string](compose.WithGenLocalState(func(context.Context)*state{return&state{}}))// 每个节点执行前，把当前状态读出来做判断。attempt:=compose.InvokableLambda(func(ctx context.Context,instring)(string,error){iferr:=compose.ProcessState(ctx,func(_context.Context,s*state)error{s.Attempts++returnnil});err!=nil{return"",err}returnin,nil})check:=compose.InvokableLambda(func(ctx context.Context,instring)(string,error){varnintiferr:=compose.ProcessState(ctx,func(_context.Context,s*state)error{n=s.Attemptsreturnnil});err!=nil{return"",err}returnfmt.Sprintf("第 %d 次尝试",n),nil})// 分支：尝试次数不足则回到 attempt 节点，否则结束。branch:=compose.NewGraphBranch(func(ctx context.Context,_string)(string,error){vardonebooliferr:=compose.ProcessState(ctx,func(_context.Context,s*state)error{ifs.Attempts>=3{s.Done=true}done=s.Donereturnnil});err!=nil{return"",err}ifdone{returncompose.END,nil}return"attempt",nil},map[string]bool{compose.END:true,"attempt":true},)_=g.AddLambdaNode("attempt",attempt)_=g.AddLambdaNode("check",check)_=g.AddEdge(compose.START,"attempt")_=g.AddEdge("attempt","check")_=g.AddBranch("check",branch)runnable,err:=g.Compile(ctx)iferr!=nil{returnfmt.Errorf("编译循环 Graph 失败: %w",err)}out,err:=runnable.Invoke(ctx,"开始")iferr!=nil{returnfmt.Errorf("执行循环 Graph 失败: %w",err)}fmt.Println("[循环] 最终输出:",out)returnnil}

实机输出（map 遍历顺序不固定，两次运行的键顺序可能不同；耗时也会波动）：

[并行] 耗时约 20ms [并行] knowledge = 知识库检索结果(打印机离线) [并行] history = 工单历史查询结果(打印机离线) [循环] 最终输出: 第 3 次尝试

几点解读：

• 并行是真的并行。其中一个节点故意Sleep(20ms)，总耗时仍在 21ms 左右，而不是 20ms 加上另一节点的时间。
• 循环靠分支回退实现。分支函数返回"attempt"就是回到上一节点，返回compose.END就退出。没有单独的“loop”原语。
• 状态是循环的支点。计数存在 state 里，分支函数据此判断是否退出。

循环必须有退出条件

如果分支函数总是返回"attempt"，流程会无限循环下去。务必设置可终止的条件，例如最大次数、超时或明确的成功判定：

ifs.Attempts>=3{returncompose.END,nil}

上面的示例用的是次数上限，这是最简单也最稳妥的兜底。

六、状态：一次执行内的共享数据

循环示例里用到了状态，这里单独说明。

创建时声明：

g:=compose.NewGraph[string,string](compose.WithGenLocalState(func(context.Context)*state{return&state{}}))

在节点内读写：

err:=compose.ProcessState(ctx,func(_context.Context,s*state)error{s.Attempts++returnnil})

三点必须记住：

1. 状态按“一次执行”隔离。每次Invoke都会调用WithGenLocalState里的函数生成新状态，两次执行之间不共享。它不是跨请求的会话存储。
2. 状态类型要能对上。ProcessState[T]中的类型必须和声明时的类型一致，否则运行时报找不到状态。
3. 框架对状态读写加了锁，但你的业务逻辑仍需自己保证一致性。涉及多步依赖时，要考虑并发下的正确性。

另外，读写状态的提示函数也要做错误检查。上面示例中忽略了ProcessState的部分返回值，仅为了聚焦流程结构；真实代码里应逐个处理。

七、Chain 还是 Graph？

经验做法：先用 Chain 把流程跑通，出现环或复杂汇聚时再换成 Graph。两者不是替代关系，Chain 本身就是 Graph 的封装。

八、常见问题

编译报图未连通
有节点没有连到START或END。分支目标如果不会连出边，要写进NewGraphBranch的第二个参数。

汇聚节点编译报类型不匹配
多个上游的输出类型无法统一。给上游加WithOutputKey，让它们都输出map[string]any。

运行时报找不到 state
ProcessState[T]的泛型类型与WithGenLocalState声明的类型不一致，或图创建时没有开启状态。

流程无限循环
分支缺少退出条件。加次数上限、超时或明确的成功判定。

回调里读不到状态
回调必须返回传入的context。返回一个全新的context.Background()会丢掉框架注入的状态——这一点在第 10 篇会实测演示。

并行看起来没并行
检查是否存在隐式的数据依赖，或上游节点串行执行。另外，节点数量少时并行收益有限，不必强求。

九、小结

• Graph 用节点和边表达流程，流向显式可见。
• 分支用NewGraphBranch，由运行时函数决定走哪条路。
• 并行靠“一个节点连多个下游”实现，汇聚靠WithOutputKey统一类型。
• 循环通过分支回退实现，必须有可终止的退出条件。
• 状态按一次执行隔离，是循环和跨节点共享数据的支点。
• 预定义拓扑不等于输出确定——分支和外部调用都会让实际路径变化。

下一篇《Workflow 实战：用字段映射表达业务流程》，看另一种思路：用字段级依赖描述流程。

参考资料与版本说明

资料核对日期为2026 年 10 月 9 日：

1. Chain/Graph 编排介绍
2. 编排的设计理念
3. Eino 主仓库

本文示例按Eino v0.9.21编写，两个示例均实机运行，正文输出为真实结果。