LangGraph实战教程：控制流详解

1. LangGraph 入门基础全解析
2. LangGraph中的Reducer是什么
3. LangGraph 核心概念详解：从编译到可视化
4. LangGraph实战教程：一文搞懂图的状态（State）管理
5. LangGraph实战教程：状态管理与graph.invoke入参深度解析
6. LangGraph高级教程：Multi Schema多状态管理详解
7. LangGraph实战教程：预定义状态MessagesState与AgentState

说到LangGraph，控制流可以说是整个图编排的骨架——它决定了节点什么时候动、往哪跳。今天这篇，我们就来拆解LangGraph里最基础的两种控制流：顺序结构和分支结构。掌握了这些，构建复杂图应用就有了根基。

一、顺序结构

顺序结构是最直观的控制流形式，节点按定义顺序依次执行，就像流水线上的工序一样。

1.1 add_edge：基础边连接

add_edge用于在两个节点之间添加一条有向边。节点之间的执行顺序、分支跳转、循环控制，最终都靠节点和边一起表达。下面这个例子展示了最基本的链式调用：

from langgraph.graph import StateGraph, START, END
from typing import TypedDict

class OverAllState(TypedDict):
    username: str
    greeting: str
    output: str

def node_a(state: OverAllState) -> OverAllState:
    return {"greeting": "Dear " + state["username"]}

def node_b(state: OverAllState) -> OverAllState:
    return {"output": state["greeting"] + "，你好！"}

builder = StateGraph(state_schema=OverAllState)
builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)
builder.add_edge(START, "node_a")
builder.add_edge("node_a", "node_b")
builder.add_edge("node_b", END)
graph = builder.compile()

res = graph.invoke({"username": "小黄"})
print(res)

输出：

{"username": "小黄", "greeting": "Dear 小黄", "output": "Dear 小黄，你好！"}

执行流程很简单：START -> node_a -> node_b -> END。

1.2 add_sequence：批量添加顺序节点

如果需要构建一组按顺序执行的节点，add_sequence可以省去很多重复代码。它会按照列表顺序依次添加节点，并在相邻节点之间自动添加边。

builder = StateGraph(state_schema=OverAllState)
builder.add_edge(START, "node_a")
builder.add_sequence([node_a, node_b])
builder.add_edge("node_b", END)

这段代码等价于手动调用 add_node 和 add_edge：

builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)
builder.add_edge("node_a", "node_b")

1.3 省略指向END的边

在LangGraph中，END并不是运行阶段真正执行的节点，它更像一个“终止标记”。在一些简单的线性流程里，即使省略指向 END的边，最后一个节点执行完毕后，如果没有后续节点触发，图也能正常结束。

builder = StateGraph(state_schema=OverAllState)
builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)
builder.add_edge(START, "node_a")
builder.add_edge("node_a", "node_b")
# 省略了 builder.add_edge("node_b", END)

不过需要注意：虽然可以省略，但在分支、条件跳转、循环退出等场景中，显式指向 END通常更利于阅读和维护。而 START通常不能省略，因为它告诉LangGraph图运行时应该从哪些节点开始执行。

二、分支结构

分支结构允许图根据运行时状态选择不同的执行路径，这才是图编排真正灵活的地方。

2.1 静态分支（Static Branch）

静态分支的核心特征是：节点的下游候选节点在图编译阶段就完全确定，运行时只是根据条件选择哪条边执行。说白了，编译期就知道下游有哪些节点，但具体走哪条路要运行时决定。

2.1.1 并行节点

并行节点是最简单的静态分支形式。当多个节点都从同一个上游节点触发时，它们会在同一个超步（SuperStep）中被激活。

builder.add_edge(START, "node_a")
builder.add_edge(START, "node_b")

来看一个完整的例子：两个节点并行执行，一个写诗，一个讲笑话。

from langgraph.graph import StateGraph, START, END
from typing import TypedDict
from langchain_deepseek import ChatDeepSeek
from langchain.messages import HumanMessage
from dotenv import load_dotenv

load_dotenv(override=True)
model = ChatDeepSeek(model="deepseek-v4-flash", extra_body={"thinking": {"type": "disabled"}})

class OverAllState(TypedDict):
    topic: str
    poem: str
    joke: str

def node_a(state: OverAllState) -> OverAllState:
    poem = model.invoke([HumanMessage(f"写一首关于 {state['topic']} 的七言绝句")]).content
    return {"poem": poem}

def node_b(state: OverAllState) -> OverAllState:
    joke = model.invoke([HumanMessage(f"写一个关于 {state['topic']} 的笑话")]).content
    return {"joke": joke}

builder = StateGraph(state_schema=OverAllState)
builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)
builder.add_edge(START, "node_a")
builder.add_edge(START, "node_b")
builder.add_edge("node_a", END)
builder.add_edge("node_b", END)
graph = builder.compile()

res = graph.invoke({"topic": "猫咪"})
print(res)

输出：

{"topic": "猫咪", "poem": "《猫咪》\n夜巡檐角步轻悄，昼卧花阴晒暖毛。...", "joke": "有一只猫走进了一家披萨店..."}

这里有两个关键点：

• “并行”主要指调度语义上的并行，两个节点之间没有先后依赖。
• 它们的输出会在当前超步执行完成后统一合并到状态中。
• 如果两个节点写入同一个状态字段，则该字段通常需要配置合适的Reducer，否则可能抛出 InvalidUpdateError。

2.1.2 条件分支

StateGraph提供了 add_conditional_edges方法，用于从某个上游节点出发，根据运行时状态选择下游节点。

方法签名：

def add_conditional_edges(
    self,
    source: str,
    path: Callable[..., Hashable | Sequence[Hashable]],
    path_map: dict[Hashable, str] | list[str] | None = None,
) -> Self:

核心参数：

• source：条件分支的起始节点。
• path：路由规则，是一个可执行对象（通常是函数）。
• path_map：路由规则的返回值到真实节点名之间的映射关系。

不使用path_map

如果不传 path_map，路由函数的返回值应当直接是图中的节点名称。

def router(state: OverAllState) -> Literal["node_a", "node_b"]:
    if "诗" in state["content_type"]:
        return "node_a"
    return "node_b"

builder.add_conditional_edges(START, router)

完整案例：

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langchain.messages import HumanMessage
from langchain_deepseek import ChatDeepSeek
from dotenv import load_dotenv

load_dotenv(override=True)
model = ChatDeepSeek(model="deepseek-v4-flash", extra_body={"thinking": {"type": "disabled"}})

class OverAllState(TypedDict):
    topic: str
    content_type: str
    poem: str
    joke: str

def node_a(state: OverAllState) -> OverAllState:
    poem = model.invoke([HumanMessage(f"写一首关于 {state['topic']} 的七言绝句")]).content
    return {"poem": poem}

def node_b(state: OverAllState) -> OverAllState:
    joke = model.invoke([HumanMessage(f"写一个关于 {state['topic']} 的笑话")]).content
    return {"joke": joke}

def router(state: OverAllState) -> Literal["node_a", "node_b"]:
    if "诗" in state["content_type"]:
        return "node_a"
    return "node_b"

builder = StateGraph(state_schema=OverAllState)
builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)
builder.add_conditional_edges(START, router)
builder.add_edge("node_a", END)
builder.add_edge("node_b", END)
graph = builder.compile()

poem_res = graph.invoke({"topic": "布偶狗", "content_type": "诗"})
joke_res = graph.invoke({"topic": "布偶狗", "content_type": "笑话"})

使用path_map

如果不希望路由函数直接返回节点名，而是返回业务语义更强的标识，可以使用 path_map进行映射。

def router(state: OverAllState) -> Literal["a", "b"]:
    if "诗" in state["content_type"]:
        return "a"
    return "b"

builder.add_conditional_edges(START, router, 
    path_map={"a": "node_a", "b": "node_b"})

这种写法的好处是：

• 路由函数可以返回业务含义更清晰的标签。
• 图节点名称可以保持工程化命名。
• 渲染图结构时，边上可以显示路由标签，使图更容易理解。

同时路由至多个节点

add_conditional_edges也支持一次路由到多个下游节点。

from collections.abc import Sequence

def router(state: OverAllState) -> Sequence[Literal["node_a", "node_b", "node_c"]]:
    if "诗" in state["content_type"]:
        return ["node_a", "node_c"]
    return ["node_b", "node_c"]

builder.add_conditional_edges(START, router, 
    path_map=["node_a", "node_b", "node_c"])

在这个例子中：

• 当 content_type包含"诗"时，同时触发 node_a和 node_c。
• 否则同时触发 node_b和 node_c。

三、总结

四、相关面试题

面试题1：add_edge(START, node_a)和 add_conditional_edges(START, router)有什么区别？

参考答案：

• add_edge(START, node_a)表示从START无条件地连接到 node_a，node_a一定会被执行。
• add_conditional_edges(START, router)表示从START出发，根据 router函数的返回值动态决定执行哪个节点。

前者是固定路径，后者是条件路径。在渲染图结构时，条件边通常会显示为虚线，并且边上会标注路由标签。

面试题2：并行执行时，如果两个节点都更新同一个状态字段会发生什么？

参考答案：

如果两个节点都更新同一个状态字段，行为取决于该字段是否配置了Reducer：

1. 有Reducer的字段（如 Annotated[list[str], add]）：两个节点的更新会按照Reducer规则合并。例如 addReducer会将两个列表拼接在一起。
2. 没有Reducer的字段：后执行的节点会覆盖先执行节点的更新，但由于并行节点在同一超步中执行，最终的覆盖顺序是不确定的，可能导致竞态条件。LangGraph会抛出 InvalidUpdateError异常。

最佳实践：对于可能被多个节点同时更新的字段，一定要配置合适的Reducer。

面试题3：add_sequence和手动调用 add_node+ add_edge相比有什么优势？

参考答案：

add_sequence的优势在于：

1. 代码更简洁：一行代码代替多行 add_node和 add_edge调用。
2. 减少出错：自动在相邻节点之间添加边，避免了手动添加时可能遗漏边的问题。
3. 语义清晰：一眼就能看出这是一组顺序执行的节点。

缺点是灵活性较低，不支持在节点之间插入条件分支或其他逻辑。

面试题4：path_map参数的作用是什么？什么时候需要使用它？

参考答案：

path_map的作用是将路由函数的返回值映射到实际的节点名称。

使用场景：

1. 业务语义映射：路由函数返回业务标签（如 "poem"、"joke"），通过 path_map映射到节点名（如 "node_a"、"node_b"）。
2. 图渲染优化：使用 path_map后，渲染图结构时边上会显示路由标签，使图更容易理解。
3. 多目标路由：当路由函数返回序列时，path_map可以帮助渲染器正确展示条件边。

面试题5：省略指向 END的边有什么风险？

参考答案：

虽然在简单线性流程中省略指向 END的边也能正常运行，但在以下场景中存在风险：

1. 分支结构：如果某个分支没有显式指向 END，可能会意外地继续执行到其他节点。
2. 循环结构：在循环退出时，如果没有显式指向 END，循环可能无法正确终止。
3. 可读性：省略 END会使图结构不完整，其他人阅读代码时难以判断流程何时结束。

建议：在所有路径的终点都显式添加指向 END的边，使图结构完整、语义清晰。

面试题6：条件分支的路由函数可以访问哪些状态字段？

参考答案：

路由函数的入参类型决定了它可以访问哪些字段。如果没有显式声明入参类型，路由函数可以访问所有已记录的状态字段。

# 可以访问所有状态字段
def router(state: OverAllState) -> str:
    return "node_a" if state["content_type"] == "诗" else "node_b"

# 也可以使用特定的输入状态类型
def router(state: InputState) -> str:
    return "node_a" if state["content_type"] == "诗" else "node_b"

注意：路由函数通常只应该读取状态字段来做决策，不应该修改状态。如果需要在路由的同时修改状态，应使用 Command机制。

面试题7：如何实现一个"先并行执行，再汇总结果"的Fan-out/Fan-in模式？

参考答案：

可以通过组合并行节点和汇聚节点来实现：

# Fan-out: 从START并行分发到多个节点
builder.add_edge(START, "node_a")
builder.add_edge(START, "node_b")
builder.add_edge(START, "node_c")

# 各节点独立执行...

# Fan-in: 多个节点汇聚到同一个汇总节点
builder.add_edge("node_a", "summary_node")
builder.add_edge("node_b", "summary_node")
builder.add_edge("node_c", "summary_node")

# 汇总节点处理所有结果
builder.add_edge("summary_node", END)

在 summary_node中，可以从状态中读取所有并行节点的输出，进行汇总处理。注意需要为可能被多个节点写入的字段配置合适的Reducer。

希望这篇教程能帮助你掌握LangGraph的控制流机制，从简单的顺序执行到复杂的分支路由，构建出灵活高效的图应用！