在 LangGraph 中,Edge 是连接节点的道路,决定了执行流程的路径。箭头方向不仅把节点连起来,还在定义时就固定了「从哪到哪」。

边的三个核心作用:
| 作用 | 说明 |
|---|---|
| 定义执行顺序 | 规定节点之间先执行谁、后执行谁 |
| 控制 State 流转 | 每个节点执行后更新 State,再传给下一节点 |
| 构建完整工作流 | 与 START、END 配合,形成有入口、有出口的完整图 |
LangGraph 中的边分为三类,共同构成图的「道路系统」:
两个普通节点之间的固定连接,表示无条件的线性执行:
graph.add_edge("input", "process")graph.add_edge("process", "output")
特点:固定路径、无条件跳转,适合线性流水线。
每个图都需要一个入口。LangGraph 内置特殊节点 START,通过起始边连接到第一个自定义节点:
from langgraph.graph import STARTgraph.add_edge(START, "input")
START 是框架内置的特殊节点,表示图的执行起点,不可省略。
最后一个节点需要连接到内置特殊节点 END,表示图执行完成:
from langgraph.graph import ENDgraph.add_edge("output", END)
到达 END 后,整个工作流结束。
一个典型的线性图结构如下:
START → Node1 → Node2 → Node3 → END
执行机制:
中间那些箭头,就是 Edge 在代码里的具象化——底层对应的就是「按顺序跑节点、逐步 update State」的执行机制。
所有边都连好之后,图还不能直接跑,必须先 编译,再 调用。
app = graph.compile()
关键点:
app 就是可执行的图应用传入初始 State,同步跑完整个工作流:
result = app.invoke(initial_state)
invoke 的特点:
result 本质是一个字典invoke取值方式两种等价:
result["raw_input"]# 方括号索引result.get("processed", "")# get 方法,可设默认值
完整工作流示例:三个节点分别负责接收输入、处理数据、格式化输出,通过边串成 START → input → process → output → END。这是最基础的线性流程模式;后续还会接触条件路由、循环路由等更复杂结构。
文件: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.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.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()))
可视化的价值:
┌────────┐起始边 ┌────────┐普通边 ┌─────────┐普通边 ┌────────┐结束边 ┌─────┐│ START│ ───────→ │ input│ ───────→ │ process │ ───────→ │ output │ ───────→ │ END │└────────┘└────────┘└─────────┘└────────┘└─────┘ │││ └──── 每步 update State,传给下一节点 ────┘
compile(),才能 invoke() 运行| 要点 | 内容 |
|---|---|
| 本质 | 边 = 连接节点的道路,决定执行路径与 State 流转 |
| 普通边 | 两节点固定连接,线性、无条件 |
| 起始边 | START → 第一个自定义节点,图的入口 |
| 结束边 | 最后一个节点 → END,图的出口 |
| 编译 | graph.compile() 必须执行,编译后结构不可改 |
| 执行 | app.invoke(初始State) 同步跑完,返回完整 State |
| 取值 | result["key"] 或 result.get("key") 均可 |