Graph 图编排
Chain 是直线,Graph 是"节点 + 你手动连的边 + 分支"、且可循环——这是 Agent 的基础。今天讲节点、边、分支,以及循环怎么实现。
Graph 是什么
AddXxxNode 登记节点、AddEdge 手动指定谁连谁、AddBranch 按运行时条件分叉。手动连边比 Chain 麻烦,但换来了任意拓扑——尤其是加一条"回头边"就能形成循环。Chain 底层就是 Graph;Graph 才是真正的引擎,Agent 全靠它。心智模型:一堆节点,你自己用边把它们连起来,可以有分支、有环(循环)。典型就是 Agent 的"模型 ↔ 工具"反复循环:
API 用 NewGraph[I,O]()(generic_graph.go:72)建图,然后手动 AddXxxNode 加节点、AddEdge 连边、AddBranch 加分支,最后 Compile。两个保留节点常量:START="start"、END="end"(graph.go:37)。
• 分支 Branch 就像换乘站:到了西直门,看你要去哪,决定换 2 号线还是 13 号线——运行时根据条件选下一站。
• 循环(回头边)就像 2 号线环线:绕一圈还能回到起点站,可以一直绕,直到你决定出站(走 END)。
Chain 则像一条没有换乘、不成环的单线:从头坐到尾,中途下不了别的线。
加节点:AddXxxNode(真实代码)
Graph 为每种组件提供一个 AddXxxNode(key, 组件, opts...),每个节点一个字符串 key(graph.go):
g := compose.NewGraph[[]*schema.Message, *schema.Message]()
g.AddChatModelNode("chat", chatModel) // graph.go:350
g.AddToolsNode("tools", toolsNode) // graph.go:399
g.AddLambdaNode("parse", myLambda) // graph.go:433 自定义逻辑
g.AddGraphNode("sub", subGraph) // graph.go:441 把子图当一个节点嵌入
AddLambdaNode 让你插入任意自定义函数(Day 06 的 4 种 Lambda 签名);AddGraphNode 能把整张子图当一个节点——支持图的嵌套复用。graph.go:162 的 addNode):图已编译则拒绝、key 不能是 START/END、key 不能重复、用了 state 的节点图必须开了 state。这些检查在"建图时"就报错,而不是等运行时崩——Eino 把错误尽量前移(Day 09 类型检查也是这思路)。组件怎么变成节点(真实代码)
你传给 AddChatModelNode 的是一个 ChatModel 组件,它怎么变成图能跑的节点?看 component_to_graph_node.go:93:
func toChatModelNode(node model.BaseChatModel, ...) (*graphNode, ...) {
return toComponentNode(node, components.ComponentOfChatModel,
node.Generate, // 作为 Invoke
node.Stream, // 作为 Stream
nil, nil, // 无 Collect / Transform → 由 Day 06 的机制自动补齐
opts...)
}
Generate、Stream)当作 Runnable 的 Invoke/Stream 传进去,Collect/Transform 传 nil。于是每个组件都被统一包成一个 Runnable(Day 06 那个"统一插头"),缺的范式自动补齐。toComponentNode 把"五花八门的组件方法"塌缩成"统一的 Runnable"。所以图执行时不用管节点里是模型还是工具还是 Lambda——全是 Runnable,调 4 个方法之一即可。Day 06 的统一抽象,在这里兑现价值。加边:AddEdge
g.AddEdge(from, to)(generic_graph.go:106)——手动指定"from 节点的输出流向 to 节点"。用 key 指定,起点可以是 START、终点可以是 END:
g.AddEdge(compose.START, "chat") // 入口 → 模型
g.AddEdge("tools", "chat") // ★ 工具 → 模型:这条"回头边"形成了循环!
// "chat" → END 由分支决定(见 L06)
AddEdge("A", "B") = "A 算完,把输出交给 B 当输入"。Chain 帮你自动加这些边,Graph 要你自己加——好处是你能加任意边,包括 AddEdge("tools", "chat") 这种"回头边"(工具跑完回到模型),从而形成循环。控制流 vs 数据流(一个高级但重要的区分)
AddEdge 底层其实同时维护两套边(graph.go:232 的 addEdgeWithMappings):
- 控制边 controlEdges(
graph.go:270):决定"谁执行完才轮到我"(执行顺序)。 - 数据边 dataEdges(
graph.go:290):决定"我的数据从哪来"。
普通 AddEdge 两者都建(既传数据又定顺序)。但 Workflow(Day 10)能把两者分开——"我等 A 执行完(控制),但数据从 B 来(数据)"。
分支 Branch:运行时决定走哪
边是"固定"的连接,分支是"运行时根据条件决定往哪走"。NewGraphBranch(branch.go:145):
// 条件函数:看输入,返回"下一个节点的 key"
condition := func(ctx, in T) (endNode string, err error) {
if 有工具调用 { return "tools", nil } // 去工具节点
return compose.END, nil // 否则结束
}
g.AddBranch("chat", compose.NewGraphBranch(condition, map[string]bool{"tools": true, END: true}))
endNodes(那个 map)列出所有可能去的节点,供编译期建"预测边"。{content:"", tool_calls:[{name:"查天气"}]}(想调工具)→ 条件函数看到有 tool_calls,return "tools" → 走到 ToolsNode。下一轮模型输出
{content:"北京今天晴,25℃", tool_calls:[]}(不想调了)→ 条件函数看到没 tool_calls,return END → 结束。同一个 "chat" 节点,靠分支在"去 tools"和"去 END"之间来回切——循环就这么转起来的。
NewGraphBranch(选一个)、NewGraphMultiBranch(branch.go:89,一次选多个并发跑)、NewStreamGraphBranch(branch.go:168,流式分支——条件函数吃的是流,可以只看第一个 chunk 就决定路由,不用等模型全部生成完)。ReAct Agent 的"要不要调工具"判断就用流式分支(Day 12/15 看真实代码)——模型刚开口就能判断有没有 tool_use。分支条件本身也是个 Runnable,享受 Day 06 的 4 范式适配。循环怎么实现:回头边 + 分支
把 L04 的回头边 + L06 的分支拼起来,就是 Agent 的循环:
g.AddEdge(START, "chat") // 入口 → 模型
g.AddBranch("chat", 分支{有工具→"tools", 无→END}) // 模型后判断
g.AddEdge("tools", "chat") // ★ 工具 → 模型(回头边 = 循环)
AnyPredecessor——只要任一前驱到达就触发节点,所以"回头边"能让模型节点被反复触发。而 Workflow 用 DAG 引擎,不允许环。WithMaxRunSteps 设"最多转多少步"(Day 09),超了报错 ErrExceedMaxSteps。Agent 层还有 MaxIterations(Day 02 见过,默认 20)。双保险防死循环。"北京天气?")→ 到 ChatModel 站,模型看了我一眼说"得查工具",给我贴了张 tool_calls 标签 → 到 换乘站(分支),检票员看我有 tool_calls 标签,让我换乘去 ToolsNode 站 → 工具查完把"北京晴 25℃"塞进我兜里 → 回头边(环线)把我送回 ChatModel 站 → 模型这次说"够了",撕掉我的 tool_calls 标签 → 换乘站看我没标签了,放我去 END 站出站,我带着最终答案离开。我在 ChatModel 站停了两次——这就是"循环"。| 轮次(super-step) | 此刻在跑哪个节点 | 关键变量 | 分支去哪 |
|---|---|---|---|
| 1 | ChatModel | tool_calls = [查天气] | → tools |
| 2 | ToolsNode | 工具结果 = "北京晴 25℃" | 回头边 → chat |
| 3 | ChatModel(第二次) | tool_calls = [](空) | → END |
for 循环。其实:Graph 里没有 for——循环是"回头边 + 分支"这个拓扑结构声明出来的,由引擎(Pregel)一轮轮调度自动转,直到分支走 END。你只是"画了张有环的图",没写任何循环代码。今日小结 + 动手
🧠 今天你应该能回答
- Graph 和 Chain 的核心区别?(手动连边、可循环 vs 自动连边、直线)
- 组件怎么变成节点?(toComponentNode 包成 Runnable,接 Day 06)
- 控制流 vs 数据流为什么要分开?
- 三种分支?流式分支为什么对 Agent 重要?
- Agent 的循环怎么用"回头边 + 分支"实现?靠哪个引擎(Pregel)?
✋ 动手:对着真实代码读一遍
# 1. 加节点 + START/END 常量(L02)
sed -n '350,360p' compose/graph.go
sed -n '37,40p' compose/graph.go
# 2. 组件→节点(L03,接 Day 06)
sed -n '93,102p' compose/component_to_graph_node.go
# 3. 加边 + 控制/数据边(L04/L05)
sed -n '232,295p' compose/graph.go | head -40
# 4. 三种分支(L06)
sed -n '89,168p' compose/branch.go | head -40