LangGraph 边(Edge)的概念与核心作用详解
在 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 START
graph.add_edge(START, "input")
注意,START 是框架内置的节点,代表图的执行起点,这一步骤不可省略。
3. 结束边(End Edge)
最后一个节点需要连接到内置的 END 节点,表示图执行完成:
from langgraph.graph import END
graph.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 TypedDict
class DataState(TypedDict):
raw_input: str
processed: str
output: 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 TypedDict
from langgraph.graph import StateGraph, START, END
class DataState(TypedDict):
raw_input: str
processed: str
output: 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}
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)
此时图结构已经完整:
START → input → process → output → END
但还没有 compile() 和 invoke(),所以直接运行不会有输出——下一个脚本将解决这个问题。
脚本 3:compile 与 invoke 实战
文件:3_compile和invoke.py
在脚本2的基础上,补上编译、执行和结果读取:
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
class DataState(TypedDict):
raw_input: str
processed: str
output: 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}
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)
app = graph.compile()
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, display
display(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") 两种方式均可 |
