边的概念与核心作用

在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

具体的执行机制可以拆解为：

1. 从START节点启动
2. 按照边的连接顺序，逐个执行节点
3. 每个节点执行后，更新全局State
4. 下一个节点会接收到更新后的State
5. 到达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") 均可