← 工作流编排

LangGraph 的 State、Node、条件边分别解决什么问题?为什么不直接写 while 循环?

高频 中等 LangGraph 状态图 · 第 1 / 2 问 更新于 2026/09/29
LangGraph状态图工作流编排Python
本题落地项目AI Agentic RAG高级企业知识库平台

简化版

LangGraph 把一个多步骤流程拆成三样东西:State 是一个从头传到尾的字典,存着流程跑到现在的全部数据;Node 是一个函数,读 State、返回自己要改的那几个字段,由框架合并回 State;Edge 决定下一步去哪,普通边固定走向,条件边由一个路由函数读 State 返回下一个节点名,回到前面的节点就形成了环。不直接写 while 循环,是因为循环写法里状态散在局部变量、分支判断和业务逻辑缠在一起、加一个步骤要改循环结构;LangGraph 让「流程写在边上、业务写在节点里、状态写在一个地方」,每一步的记录、测试和扩展都变简单了。有环就必须有上限,业务侧用计数字段在路由里拦,框架侧再用 recursion_limit 兜底。

详细版

graph = StateGraph(State)                     # State 是 TypedDict
graph.add_node("retrieve", retrieve_node)     # 节点:输入 State,返回要更新的字段
graph.add_node("grade", grade_node)
graph.add_edge(START, "retrieve")             # 普通边:固定走向
graph.add_edge("retrieve", "grade")
graph.add_conditional_edges(                  # 条件边:路由函数决定下一步
    "grade", route_after_grade,
    {"generate": "generate", "retrieve": "retrieve", "fallback": "fallback"},
)
app = graph.compile()
result = app.invoke(initial_state, {"recursion_limit": 30})
概念解决什么问题要点
State数据在节点之间怎么流动节点只返回改动字段;默认覆盖,需要追加的字段配归并函数
Node每一步做什么纯函数式:读状态、返回更新,不直接调用别的节点
普通边固定的先后顺序add_edge(a, b)
条件边分支和循环路由函数只读状态、返回节点名,第三个参数把返回值映射到目标节点
recursion_limit环停不下来超过后抛 GraphRecursionError,是框架层的兜底

完整版教学

一、while 循环写法的三个问题

先看一个「检索 → 判断资料够不够 → 不够就改写再查 → 够了就生成」的流程,用 while 循环写是这样的:

retry = 0
while retry < 3:
    docs = retrieve(query)
    if grade(docs):          # 判断和控制流写在同一行
        break
    query = rewrite(query)
    retry += 1
answer = generate(docs) if retry < 3 else "未找到"

能跑,但流程一复杂就暴露三个问题。第一,状态散在局部变量里:query、docs、retry 各是一个变量,想把每一步的输入输出记下来,就得在每个分支点手写一遍记录代码,重复且容易漏。第二,分支判断和业务逻辑混在一起:if grade(docs): break 既是业务调用也是控制流,节点多到七八个时,要在脑子里模拟执行才能画出流程图。第三,加一个步骤要改循环结构:想在检索前插一个「查询路由」,就得调整循环嵌套,牵一发动全身。

二、State:一个字典从头传到尾

State 是一个 TypedDict,定义了流程里所有要传递的数据。每个节点收到的是完整的 State,返回的只是自己改了的字段,框架负责合并:

初始      {query: "那个多少钱", retry: 0}
rewrite   返回 {query: "培训服务的收费标准"}       → {query: "培训服务…", retry: 0}
retrieve  返回 {documents: [A, B, C]}             → {query: …, retry: 0, documents: [A, B, C]}
grade     返回 {is_relevant: False, retry: 1}      → {…, is_relevant: False, retry: 1}

合并规则要记清楚:没有配置归并函数的字段,新值直接覆盖旧值;如果希望追加,比如消息列表,要在类型上声明归并函数(例如 Annotated[list, operator.add]),这时节点返回的列表会接在原列表后面。把「覆盖」误当成「追加」,是 State 设计里最常见的 bug。

State 的好处是:每个节点都能拿到之前所有节点的产出;任何时刻 State 的快照就是「流程跑到哪了」,直接序列化就能落库或回放。

三、Node:只管业务,不管下一步

节点是普通函数(同步或 async),输入 State、输出更新。它不调用别的节点,也不决定下一步去哪,这是和 while 写法最大的区别。

这个约束带来两个直接的好处:节点可以单独测试,构造一个 State 字典喂进去、检查返回的字段即可;同一个节点可以被走多次,行为由 State 决定。比如生成节点第一次进来时直接写答案;被校验打回、再次进来时,State 里多了上一轮的问题清单,它就照着清单改。

易错点:原问题和「当前拿去检索的问题」要分成两个字段。改写节点只改后者;判断资料够不够、生成答案时用的都应该是用户的原问题,否则改写一偏,后面全跟着偏。

四、Edge 与条件边:流程写在边上

普通边表达固定顺序,条件边表达分支。条件边由两部分组成:一个路由函数,读 State 返回一个字符串;一个映射表,把字符串对到目标节点:

def route_after_grade(state) -> str:
    if state.get("is_relevant"):
        return "generate"
    if state.get("retry", 0) >= MAX_RETRY:     # 业务上限:超了走兜底
        return "fallback"
    return "rewrite"                            # 回到前面的节点,形成环

路由函数只读状态、只返回节点名,不做任何业务。想改流程就改路由函数和边,不用碰节点里的代码;想看流程,读一遍建图函数里的 add_edge 和 add_conditional_edges,就是一张完整的流程图。环也是这样来的:条件边指回前面的节点,图里就出现了回路,这正是 while 循环要用 continue 和 break 硬凑的东西。

五、有环就要有两层上限

环意味着可能停不下来,比如资料一直被判为不相关,就会一直改写重查,每一轮都在消耗 token。上限要设两层:

层次做法作用
业务上限State 里放计数字段,路由函数判断超限就走兜底节点正常情况下由它收尾,用户看到友好的兜底结果
框架上限调用时传 recursion_limit路由逻辑写错导致死循环时,框架抛 GraphRecursionError 兜住

recursion_limit 按「执行了多少步」计数(顺序执行的图里,每走一个节点算一步)。用一个三节点的循环算一下:审查、重写各一个节点,最后一个汇总节点,最多 5 轮时,一直不通过的完整路径是 5 次审查 + 5 次重写 + 1 次汇总 = 11 步。上限设成 5 × 2 + 2 = 12,正好留一步余量;上限设得远大于最坏路径,路由写错时就要多空转很多步才会被拦下。检索场景下多轮循环什么时候该停,见「Agentic RAG 与普通 RAG 有什么区别?如何决定继续检索还是停止?」。

最坏路径步数 = 轮数 × 每轮节点数 + 收尾节点数 = 5 × 2 + 1 = 11
recursion_limit = 11 + 1(余量) = 12

六、为什么说三者分离的收益是实打实的

把状态、节点、边拆开以后,前面 while 写法的三个问题都有了对应的解法:

while 写法的问题LangGraph 的解法
每个分支都要手写记录代码在每个节点里调用同一个记录函数(或在外面注入回调),记录代码只写一遍
要模拟执行才能看懂流程建图函数就是流程图,路由函数只有几行
加步骤要改循环结构加一个 add_node 和两条边,不动其他节点

代价也要说清:多了一层框架概念,简单的线性流程用它反而啰嗦;State 字段一多,谁在改哪个字段需要约定清楚。所以只有两三步、没有分支和回路的流程,普通函数顺序调用就够了;出现条件分支和回路、需要逐步记录时,状态图的收益才明显。

七、常见误区与追问

  • 误区:节点返回完整的 State。 节点只返回自己改的字段,框架负责合并,返回整个字典容易把别人的字段意外覆盖成旧值。
  • 误区:列表字段返回新列表就会自动追加。 没有声明归并函数时是直接覆盖,要追加必须在类型上声明归并函数。
  • 误区:路由函数里顺手做点业务。 路由函数只读状态、返回节点名,业务放进节点,否则流程和业务又缠在一起。
  • 误区:有 recursion_limit 就不用业务上限。 框架上限触发时是抛异常,用户看到的是报错;业务上限才能走兜底节点给出友好结果。
  • 误区:所有多步骤流程都该用状态图。 没有分支和回路的短流程,顺序调用函数更直接。
  • 追问:recursion_limit 设多少合适? 按最坏路径算出步数再留一点余量,既不误伤正常路径,也能在路由写错时尽快拦下。
  • 追问:节点能不能直接调用另一个节点? 不应该。节点之间只通过 State 传数据,走向由边决定,否则就退化回了函数嵌套调用。

八、加强记忆

LangGraph 记「三样东西两层上限」:State 是从头传到尾的字典,节点只返回改动字段,默认覆盖、要追加得声明归并函数;Node 只管业务,不调用别人、不决定下一步,所以能单测、能被重复进入;Edge 管走向,条件边由只读状态的路由函数加映射表组成,指回前面的节点就形成环。它解决 while 写法的三个问题:状态散落、判断和业务缠绕、加步骤要改结构。有环就要两层上限:业务上限在路由里走兜底,框架的 recursion_limit 按最坏路径步数加余量设,路由写错时兜底报错。

项目实战落地

项目里怎么做的

《AI Agentic RAG高级企业知识库平台》的 Agentic 问答用 LangGraph 搭了一张六个节点的图(改写、检索、相关性评分、生成、幻觉校验、兜底),三样东西各有讲究:

  • State:AgentState(TypedDict, total=False),字段有原问题 original_query、当前检索用的 query、召回片段、答案、改写次数、重新生成次数、实际检索次数、两个判断结论和结束原因;改写节点只改 query,相关性评分和生成都用 original_query;
  • 条件边:两个路由函数 route_after_grade、route_after_check 只读状态、只返回节点名,相关就去生成、不相关且没到上限回到改写、超上限去兜底;建图函数 build_graph 就是流程图的代码形式;
  • 两层上限:业务上限写死在图模块顶部,由路由函数判断后走兜底节点;调用时再传 LangGraph 的 recursion_limit,取业务步数上限的 3 倍,路由逻辑写错导致死循环时由框架兜住。

《AI Agent智能会议纪要辅助系统》的纪要自检是一张更小的图:审查、重写、汇总三个节点,route_after_review、route_after_refine 两条条件边;recursion_limit 按 max_rounds * 2 + 2 计算,max_rounds 为 5 时上限 12 步,一直不通过的完整路径是 11 步。

为什么这样取舍

  • 不用 while 循环:要把每一步的入参和返回记进步骤表,循环写法得在每个分支点手写落库;状态图里每个节点调用同一个记录函数,代码写一遍覆盖所有节点。
  • 原问题和当前查询分开:判断「这批资料能不能回答问题」,问的是用户真正想问的那个问题,不是改写出来的那个。
  • 上限写死、框架再兜一层:业务上限靠路由函数生效,万一路由写错,框架这一层还能拦住。

面试官还会追问

  • 相关性评分节点一条资料都没召回时,还会调用模型吗?
  • 判断结论为什么用包含关系解析,而且要先判断「不相关」再判断「相关」?
  • 改写节点第一轮为什么不做多查询扩展,而是直接用原问题?

学完《AI Agentic RAG高级企业知识库平台》,上面这些追问你都会迎刃而解。

本题落地项目地狱锤炼AI Agentic RAG高级企业知识库平台基于企业知识库、PGVector 向量检索、Agentic RAG、FastAPI + LangGraph、Function Calling,实现向量 + BM25 混合检索与 RRF 融合、Rerank 重排、HyDE 与多查询改写、父子分块召回、LangGraph 状态图自适应检索、Agent 执行时间线和 LLM-as-judge 四指标评测,覆盖从基础 RAG 到高级 RAG 调优的完整闭环。FastAPILangChainLangGraphRAGPGVector源码+SQL喂饭学习教程配套面试文档环境安装文档项目运行文档 学习这个项目 也可以学AI Agent智能会议纪要辅助系统地狱锤炼 查看项目