LangGraph 实战：用条件分支控制循环与结束

痛点一：不清楚如何在 LangGraph 中实现“循环”逻辑。往往只能写成线性的流程，导致业务需求无法满足。不过，

痛点二：状态累积使用不当。容易出现数据错乱或遗漏，调试成本高。话说回来，

痛点三：条件分支写错后图会卡死或无限循环。却难以定位根因，

这里通过一个完整示例。手把手演示如何利用 add_conditional_edges 实现“条件循环”，并结合 reducer 自动累积每一步的执行记录，最终导出流程图 PNG。

. 环境准备

这篇文章使用的运行环境如下：

Python 3.10+
LangGraph
typing-extensions

安装依赖的观点是，

python -m pip install -U langgraph typing-extensions

. 定义可累积状态

我们通过 TypedDict + Annotated 的方式声明一个可以自动合并的列表字段：

import operator
from typing import Annotated。Any
from typing_extensions import TypedDict
class State:
# 使用 list 的加法来累计每个节点返回的记录
aggregate: Annotated,operator.add]

Annotated,operator.add] 表示 aggregate 是列表，并在每次节点返回时执行列表拼接（+)。 再看例如，

 + # →

. 编写条件路由

主要路由函数决定接下来是继续循环还是结束：

from typing import Literal
def router -> Literal:
# 当累计记录不足 6 条时继续循环。否则结束
if len <6:
        return "step2"
    return "end"

• 长度 <6 → 返回 "step2"进入下一轮循环；
• 长度 ≥ 6 → 返回 "end"触发图的终止节点。

随后使用 add_conditional_edges 把路由返回值映射到真实节点或 END：

graph_builder.add_conditional_edges(
"step1",router,{
"step2": "step2",# 正常流向
至于"end"，END,# 特殊终点，需要映射到 LangGraph 提供的 END 常量
}。)

. 建立循环结构

graph_builder.add_edge
graph_builder.add_edge
graph_builder.add_edge
graph_builder.add_edge # 回到路由，实现循环

关键点：每次执行完 step4 后都会回到 step1,触发 router,从而动态决定是否继续。

. 完整代码

import operator
from typing import Annotated,Any,Literal
from langgraph.graph import END,START,StateGraph
from typing_extensions import TypedDict
# ---------- 状态定义 ----------
class State:
aggregate: Annotated。operator.add]
# ---------- 各步骤 ----------
def step1:
print
return {"aggregate": }
def step2:
print
return {"aggregate": }
def step3:
print
return {"aggregate": }
def step4:
print
return {"aggregate": }
# ---------- 路由函数 ----------
def router -> Literal:
if len <6:
        return "step2"
    return "end"
# ---------- 构建图 ----------
graph_builder = StateGraph
# 添加节点
graph_builder.add_node.add_node.add_node.add_node
# 起始边 + 条件边
graph_builder.add_edge
graph_builder.add_conditional_edges(
    "step1",
    router,
    {
        "step2": "step2",
        "end": END,
    },
)
# 循环路径
graph_builder.add_edge
graph_builder.add_edge
graph_builder.add_edge  # 此行仅作占位说明，请忽略
# 编译图并导出 PNG
graph = graph_builder.compile
png_bytes = graph.get_graph.draw_mermaid_png
with open as f:
    f.write
# 执行图
result = graph.invoke
print

. 真实运行输出

python index.py

这是 step1这是 step25步步步步步5步8 步9 步10 步11 步12 步13 步14 步15 步16 执行结果： {' aggregate ': }