最新下载
热门教程
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
5.6 LangGraph-Edges解读-Agent图的道路系统
时间:2026-07-02 11:55:00 编辑:袖梨 来源:一聚教程网
边的概念与核心作用
在 LangGraph 中,Edge 是连接节点的道路,决定了执行流程的路径。箭头方向不仅把节点连起来,还在定义时就固定了「从哪到哪」。

边的三个核心作用:
| 作用 | 说明 |
|---|---|
| 定义执行顺序 | 规定节点之间先执行谁、后执行谁 |
| 控制 State 流转 | 每个节点执行后更新 State,再传给下一节点 |
| 构建完整工作流 | 与 START、END 配合,形成有入口、有出口的完整图 |
边的三种类型
LangGraph 中的边分为三类,共同构成图的「道路系统」:
1. 普通边(Normal Edge)
两个普通节点之间的固定连接,表示无条件的线性执行:
graph.add_edge("input", "process")graph.add_edge("process", "output")
特点:固定路径、无条件跳转,适合线性流水线。
2. 起始边(Start Edge)
每个图都需要一个入口。LangGraph 内置特殊节点 START,通过起始边连接到第一个自定义节点:
from langgraph.graph import STARTgraph.add_edge(START, "input")
START 是框架内置的特殊节点,表示图的执行起点,不可省略。
3. 结束边(End Edge)
最后一个节点需要连接到内置特殊节点 END,表示图执行完成:
from langgraph.graph import ENDgraph.add_edge("output", END)
到达 END 后,整个工作流结束。
图的执行流程
一个典型的线性图结构如下:
START → Node1 → Node2 → Node3 → END
执行机制:
- 从 START 节点开始
- 按照边的连接顺序依次执行节点
- 每个节点执行后更新全局 State
- 下一个节点接收更新后的 State
- 到达 END 节点后结束
中间那些箭头,就是 Edge 在代码里的具象化——底层对应的就是「按顺序跑节点、逐步 update State」的执行机制。
编译与执行
所有边都连好之后,图还不能直接跑,必须先 编译,再 调用。
编译图:compile()
app = graph.compile()
关键点:
- 编译是必须步骤:不编译,图无法执行
- 编译后结构不可修改:如需改节点或边,须重新构建并再次编译
- 返回可调用对象:
app就是可执行的图应用
执行图:invoke()
传入初始 State,同步跑完整个工作流:
result = app.invoke(initial_state)
invoke 的特点:
- 同步执行:所有中间节点全部跑完才返回
- 返回最终完整 State:
result本质是一个字典 - 主测试场景最常用:需要等待完整结果时,优先用
invoke
取值方式两种等价:
result["raw_input"]# 方括号索引result.get("processed", "")# get 方法,可设默认值
完整工作流示例:三个节点分别负责接收输入、处理数据、格式化输出,通过边串成 START → input → process → output → END。这是最基础的线性流程模式;后续还会接触条件路由、循环路由等更复杂结构。
脚本 1:普通边连接节点
文件:1_普通边连接节点.py
在引入 add_edge 之前,先手动按顺序调用节点,理解边最终要表达的就是这个执行顺序:
from typing import TypedDictclass DataState(TypedDict):raw_input: strprocessed: stroutput: str# 第一个节点:清理输入(去掉首尾空格)def input_node(state: DataState):cleaned = state["raw_input"].strip()return {"raw_input": cleaned}# 第二个节点:处理数据(转大写)def process_node(state: DataState):processed = state["raw_input"].upper()return {"processed": processed}# 第三个节点:整理输出def output_node(state: DataState):output = f"Result: {state['processed']}"return {"output": output}# 先手动按顺序调用节点,理解边最终要表达的就是这个执行顺序state = {"raw_input": "hello world", "processed": "", "output": ""}print("初始状态:", state)state.update(input_node(state))print("清理输入后:", state)state.update(process_node(state))print("转大写后:", state)state.update(output_node(state))print("最终状态:", state)
运行结果:
初始状态: {'raw_input': 'hello world', 'processed': '', 'output': ''}清理输入后: {'raw_input': 'hello world', 'processed': '', 'output': ''}转大写后: {'raw_input': 'hello world', 'processed': 'HELLO WORLD', 'output': ''}最终状态: {'raw_input': 'hello world', 'processed': 'HELLO WORLD', 'output': 'Result: HELLO WORLD'}
每一步只有目标字段变化:第一个节点只更新了 raw_input,第二个更新了 processed,第三个增加了格式化的 output。连接好边之后,框架做的就是这件事——依次运行节点,对每个节点返回的字典做 state.update()。
脚本 2:START 与 END
文件:2_START和END.py
理解手动流转后,用 StateGraph 和 add_edge 把节点与 START、END 连起来:
from typing import TypedDictfrom langgraph.graph import StateGraph, START, ENDclass DataState(TypedDict):raw_input: strprocessed: stroutput: strdef input_node(state: DataState):cleaned = state["raw_input"].strip()return {"raw_input": cleaned}def process_node(state: DataState):processed = state["raw_input"].upper()return {"processed": processed}def output_node(state: DataState):output = f"Result: {state['processed']}"return {"output": output}# 创建图并添加节点graph = StateGraph(DataState)graph.add_node("input", input_node)graph.add_node("process", process_node)graph.add_node("output", output_node)# 连接边:START → input → process → output → ENDgraph.add_edge(START, "input")graph.add_edge("input", "process")graph.add_edge("process", "output")graph.add_edge("output", END)
此时图结构已经完整:
START → input → process → output → END
但还没有 compile() 和 invoke(),所以直接运行不会有输出——下一个脚本解决这个问题。
脚本 3:compile 与 invoke
文件:3_compile和invoke.py
在脚本 2 的基础上,补上编译、执行和结果读取:
from typing import TypedDictfrom langgraph.graph import StateGraph, START, ENDclass DataState(TypedDict):raw_input: strprocessed: stroutput: strdef input_node(state: DataState):cleaned = state["raw_input"].strip()return {"raw_input": cleaned}def process_node(state: DataState):processed = state["raw_input"].upper()return {"processed": processed}def output_node(state: DataState):output = f"Result: {state['processed']}"return {"output": output}graph = StateGraph(DataState)graph.add_node("input", input_node)graph.add_node("process", process_node)graph.add_node("output", output_node)graph.add_edge(START, "input")graph.add_edge("input", "process")graph.add_edge("process", "output")graph.add_edge("output", END)# compile 把图配置转成真正可执行的应用app = graph.compile()# invoke 从初始 State 出发,跑完整个图initial_state = {"raw_input": "hello world","processed": "","output": "",}result = app.invoke(initial_state)print("清理后的输入:", result["raw_input"])print("处理后的结果:", result["processed"])print("最终输出:", result["output"])print("完整结果:", result)
运行结果:
清理后的输入: hello world处理后的结果: HELLO WORLD最终输出: Result: HELLO WORLD完整结果: {'raw_input': 'hello world', 'processed': 'HELLO WORLD', 'output': 'Result: HELLO WORLD'}
result 就是一个字典(State),除了 result["key"],也可以用 result.get("key") 取值,效果一致。
图可视化
LangGraph 内置可视化能力,编译后可以直观查看图结构。在 Jupyter Notebook 中:
from IPython.display import Image, displaydisplay(Image(app.get_graph().draw_mermaid_png()))
可视化的价值:
- 帮助理解图的执行流程
- 快速发现结构问题
- 便于团队沟通和文档化
边与 State、Node 的关系(一图看懂)
┌────────┐起始边 ┌────────┐普通边 ┌─────────┐普通边 ┌────────┐结束边 ┌─────┐│ START│ ───────→ │ input│ ───────→ │ process │ ───────→ │ output │ ───────→ │ END │└────────┘└────────┘└─────────┘└────────┘└─────┘ │││ └──── 每步 update State,传给下一节点 ────┘
- 边负责把节点执行过程用箭头串起来
- 添加边之后还需
compile(),才能invoke()运行 - 传入初始 State,框架自动沿图跑完所有节点和边,返回最终 Result
小结
| 要点 | 内容 |
|---|---|
| 本质 | 边 = 连接节点的道路,决定执行路径与 State 流转 |
| 普通边 | 两节点固定连接,线性、无条件 |
| 起始边 | START → 第一个自定义节点,图的入口 |
| 结束边 | 最后一个节点 → END,图的出口 |
| 编译 | graph.compile() 必须执行,编译后结构不可改 |
| 执行 | app.invoke(初始State) 同步跑完,返回完整 State |
| 取值 | result["key"] 或 result.get("key") 均可 |
相关文章
- 修图软件免费版下载合集 手机修图软件推荐 07-02
- 央视体育版本检测方式 07-02
- DNF手游弹弹乐玩法在哪 07-02
- DNF手游:绚彩果糖弹弹乐玩法 07-02
- DNF手游弹弹乐玩法详解 07-02
- 绚彩夏日冥想音乐播放指引 07-02