想象你刚入职一家新公司，要请三天假。HR 告诉你：得先在 OA 系统里发起一张请假单——这张单子要依次经过直属经理审批、HR 审核、财务登记补贴。每一道关卡都有一个“岗位”处理这张单子：经理签“同意/驳回”，HR 更新休假天数，财务备注补贴处理。这张单子从头到尾始终是同一张，只是内容被一步一步填满；路线也可能因为“同意”或“驳回”而分叉。 这就是 LangGraph 在做的事——它管理的是 AI 应用内部的“审批单流转”。要构建一个有状态的 Agent 工作流，核心就两个构件：节点（Node）负责做事（处理请求、调用模型、更新信息），边（Edge）负责决定下一步去哪儿。而整个过程中被传递、被更新的“请假单”，就是状态（State）。本文会带你从零理解：节点如何定义处理逻辑、边如何控制流转方向、条件分支如何让流程“看情况走”、状态如何在节点间被传递合并，以及如何设计状态的数据结构 Schema。学完你就能用 LangGraph 搭出“客服助手”“多步检索”这样的有状态 Agent 工作流。

岗位关联

对 AI 从业者来说，LangGraph 的节点与边机制是构建生产级 Agent 的骨架：它把业务逻辑、状态管理和流程控制解耦，让 AI 应用的行为可预测、可回放、可维护。

目录
  1. 可视化演示
  2. 节点定义
  3. 边与流转
  4. 条件分支
  5. 状态传递
  6. 状态Schema
  7. 工作流实战
  8. AI 应用场景
  9. Python 演示
  10. 常见误区
  11. 面试题

可视化演示

交互展示LangGraph中节点与边的有状态流转机制：状态传递、条件分支和组合模式。

本节配有交互动效（进入视口后自动加载，约 3KB）。

状态流转：节点间传递 节点接收State并返回更新字典 边控制节点间执行顺序和方向 Node(State) → State Update 状态在节点间有序传递和更新

节点定义

把公司审批流想象成一条流水线：直属经理、HR、财务各有一个工位，每个工位只负责在请假单上完成自己那一步——经理签意见、HR 核天数、财务登记补贴。在 LangGraph 里，这些工位就是节点（Node）：它是图里最小的执行单元，负责接收上一站送来的状态（State），做一段自己的处理，再把更新后的状态交出去。

技术上，节点可以理解为一个“输入一个输出”的函数。在 Python 中，最简单的节点就是一个普通函数：参数 state 是当前这张“请假单”，返回值是你想改的字段增量。LangGraph 会自动把返回值合并回全局状态，不需要你手写 state["approval"] = ...：

节点把一张单子从一个工位交给下一个工位，每个节点只负责自己最熟悉的格子，改完就传下去，这正是 LangGraph 让复杂 Agent 逻辑保持清晰的基本功。

S′=N(S)\displaystyle S' = N(S)
状态更新公式：S 是进到节点前的状态，N(·) 是节点内部的处理函数，S' 是节点处理完后的新状态。整个过程透明可追踪，恰如每个审批工位在工作单上留下新痕迹。
  1. 当前状态 S节点开始处理前，全局“请假单/状态字典”长什么样。
    SS
  2. 节点函数 N(·)针对“这个工位要干什么”而封装的业务逻辑——判断、改写、追加消息、调用模型等。
    N(⋅)N(\cdot)
  3. 更新结果 S'节点处理后返回的字段增量，LangGraph 将其合并进全局状态，形成新状态 S'。
    S′=N(S)S' = N(S)
python2 行
def manager_node(state):    return {"approval": "approved"}

边与流转

在审批流程里，经理签完字，假条会被送到 HR；HR 复核完，又送到财务。这些“下一站送到哪”的指令，就是 LangGraph 里的边（Edge）。边定义了节点之间固定的流转方向，它本身不处理数据，只告诉执行引擎：完成当前节点后，下一个要去哪个节点。

这段代码按顺序连接了入口、经理节点、HR 节点和出口。如果业务流程是“永远固定走同一条路”，那么普通边就足够用了。

如果请假流程永远“经理→HR→财务”固定不变，确实不需要任何判断。但现实中审批往往出现“同意走 A 流程、驳回走 B 流程”的情况——这就到条件分支了。

START→e1N1→e2N2→e3END\displaystyle \text{START} \xrightarrow{e_1} N_1 \xrightarrow{e_2} N_2 \xrightarrow{e_3} \text{END}
边的链路公式：流程从 START 出发，依次经过 N1、N2，最终到达 END；每条边 e_i 代表一次无条件的流转。
  1. 起点 START图执行的唯一入口，所有状态都从这里进入。
    START\text{START}
  2. 普通边 e_i从一个节点指向下一个节点的有向线，代表“无条件、固定走这里”。
    →ei\xrightarrow{e_i}
  3. 终点 END所有流程的出口，执行到这里即完成一次完整调用。
    END\text{END}
python7 行
from langgraph.graph import StateGraph, START, END g = StateGraph(State)g.add_node("manager", manager_node)g.add_edge(START, "manager")g.add_edge("manager", "hr")g.add_edge("hr", END)

条件分支

职场里，经理在假条上签“同意”，单子去 HR；签“驳回”，单子退回员工修改。决定下一步走哪条路的不是固定边，而是条件分支（Conditional Edge）：一个带判断的函数，看完当前状态后返回一个“路线 key”，LangGraph 根据 key 走对应分支。

使用 add_conditional_edges 时，条件函数返回的是字符串 key，映射表再把 key 翻译成具体节点：

条件分支是 Agent 工作流的灵魂：让流程“带着脑子”走，而不是机械地顺序执行。那枚小小的章，是审批单能自己拐弯的原因。

Next(S)={Nhr,S.approval=approvedNrewrite,otherwise\displaystyle \text{Next}(S) = \begin{cases} N_{hr}, & S.\text{approval} = \text{approved} \\ N_{rewrite}, & \text{otherwise} \end{cases}
条件路由公式：根据状态里的审批意见，动态决定下一个节点是 HR 节点还是退回修改节点。
  1. 条件函数 Next(S)以当前状态 S 为输入、输出“下一站 key”的纯函数，类似审批人盖章。
    Next(S)\text{Next}(S)
  2. 分支决策条件state["approval"] == "approved" 这类业务判断，决定取哪个分支。
    S.approval=approvedS.\text{approval} = \text{approved}
  3. 路由映射表分支 key 与目标节点的字典，LangGraph 通过它找到实际要跳转的节点。
    key↦Ntarget\text{key} \mapsto N_{target}
python4 行
def route_after_manager(state):    if state["approval"] == "approved":        return "hr"    return "rewrite"
python2 行
g.add_conditional_edges("manager", route_after_manager,                        {"hr": "hr", "rewrite": "rewrite"})

状态传递

请假单始终是同一张，但从经理到 HR 再到财务，单子上不断多出新格子：经理签了意见，HR 补了天数，财务备注了补贴。这个现象在 LangGraph 里叫状态传递与更新（State Passing）。同一次运行中，状态是全局共享的一份数据对象（如一个 dict），节点把它放进函数参数，返回的东西则通过合并策略累加回去。

LangGraph 默认采用增量更新：每个节点不必返回完整状态，只返回你要变更的字段。引擎会把返回值与旧状态合并，生成新状态后再传给下一个节点。

这也正是“同一张请假单被不断填充”的严谨表述：没有人重新打印单子，只是在原单子上继续书写；如果两个节点都改了同一个字段，就看那个字段的“合并规则”来决定谁覆盖谁。

St+1=St⊕ΔSt\displaystyle S_{t+1} = S_t \oplus \Delta S_t
状态合并公式：当前状态 + 节点产生的增量 = 下一状态。
  1. 旧状态 S_t进入节点前已有的全局状态，可能包含历史消息、步骤计数、用户输入等。
    StS_t
  2. 增量 ΔS_t节点返回值里需要新增或修改的字段集合，只描述变化本身。
    ΔSt\Delta S_t
  3. 合并操作符 ⊕LangGraph 根据字段的 reducer 决定如何合并——标量覆盖、列表追加，还是自定义策略。
    St⊕ΔStS_t \oplus \Delta S_t

状态Schema

如果公司不规定请假单格式，有的员工写“三天”，有的写“3 days“，HR 根本没法统计。状态 Schema（结构定义） 要做的就是给”这张单子长什么样”立规矩——哪些字段、什么类型、如何合并。

在 LangGraph 中，最常用 TypedDict 定义状态 schema：

没有 schema，LangGraph 无法知道字段怎么合并；有了 schema，节点可以放心只返回一个小增量——正如每张请假单上每个格子都有明确的主人。

S={ messages:list,  step:int,  user_id:str }\displaystyle S = \{\, \text{messages}: \text{list},\; \text{step}: \text{int},\; \text{user\_id}: \text{str} \,\}
状态结构公式：状态是所有字段的集合，每个字段都有名字、类型和合并策略。
  1. 字段集合状态由若干字段组成，例如 messages 是 list，step 是 int。
    messages:list\text{messages}: \text{list}
  2. Annotated 元数据Annotated[T, reducer] 附加合并信息，告诉引擎“新值到来时如何合并”。
    Annotated[T,reducer]\text{Annotated}[T, \text{reducer}]
  3. Reducer 合并策略决定“两个值相遇时谁说了算”，如 add_messages 会把新消息追加到旧列表后。
    list→list+[new]\text{list} \to \text{list} + [\text{new}]
python7 行
from typing import Annotated, TypedDictfrom langgraph.graph.message import add_messages class State(TypedDict):    messages: Annotated[list, add_messages]   # 消息自动追加    step: int                                  # 流程计数，普通覆盖    user_id: str                               # 用户标识

工作流实战

现在把岗位（节点）、路线（边）、盖章决策（条件分支）、单子（状态 Schema）全部安排好，只需要把它们拼装成一个完整的“公司流程系统”——这就是构建与编译图（Build & Compile）。

所有单子（状态）会在一次调用中自动走完全程——这就是 LangGraph 构建有状态 Agent 工作流的基本套路：先定义状态，再加节点，再连边，最后编译运行。

G=compile(StateGraph(S, {Ni}, {Ej}))\displaystyle G = \text{compile}\big(\text{StateGraph}(S,\, \{N_i\},\, \{E_j\})\big)
工作流构建公式：用状态 Schema 构造 StateGraph，注册节点集合与边集合，编译后即可调用。
  1. 构造 StateGraph(S)传入状态 Schema，初始化一个图对象。
    StateGraph(S)\text{StateGraph}(S)
  2. 节点集合 {N_i}通过 add_node 将函数打包成图里的节点。
    {Ni}\{N_i\}
  3. 边集合 {E_j}通过 add_edge 与 add_conditional_edges 定义边和条件路由。
    {Ej}\{E_j\}
  4. 编译执行compile() 返回可直接调用的应用对象，invoke() 正式运行整个流程。
    compile(⋯ )→invokeresult\text{compile}(\cdots) \xrightarrow{\text{invoke}} \text{result}
python14 行
from langgraph.graph import StateGraph, START, END g = StateGraph(State)g.add_node("manager", manager_node)g.add_node("hr", hr_node)g.add_node("finance", finance_node)g.add_edge(START, "manager")g.add_conditional_edges("manager", route_after_manager,                        {"hr": "hr", "rewrite": "rewrite"})g.add_edge("hr", "finance")g.add_edge("finance", END) app = g.compile()result = app.invoke({"messages": [], "step": 0, "user_id": "007"})

AI 应用场景

智能客服分流

阶段：Agent 编排与条件路由

解决的问题：传统客服对话靠 if-else 硬编码分支，新增一个类目就要改代码；用 LangGraph 的条件分支和状态传递，路由逻辑变成一张可维护的映射表，用户消息和决策记录都沉淀在状态里，可回溯。

具体案例：用户说“我要退款”，客服 Agent 先经过“意图识别节点”，条件边根据识别结果把请求路由到“退款政策节点”或“人工客服节点”；每一步都把模型输出追加到 messages 列表留痕，最终由“回复生成节点”输出答案。

多步 RAG 检索

阶段：检索增强生成（有状态工作流）

解决的问题：单次检索往往质量不稳定，需要“检索→评分→重写→再检索”的循环；用条件分支控制是否重写，用状态里的 retry_count 防止死循环。

具体案例：用户提问进入“文档检索节点”，拿到候选文档后经“相关性评分节点”打分；若分数低于阈值，条件边把流程送回“问题重写节点”再进检索，直到分数达标或重试次数耗尽，最后交给“生成回答节点”。

代码生成修复 Agent

阶段：Agent 工具调用循环

解决的问题：大模型一次生成的代码常有语法错误，人工回填太慢；用 LangGraph 编排“生成→编译→修复”循环，让修复节点自动返回并重试。

具体案例：“代码生成节点”产出源码写入 state.code，“编译检查节点”执行检查；条件边根据编译结果决定走“输出节点”还是“修复节点”，max_retries 字段控制终止条件，避免无限循环。

Python 演示

python37 行
from typing import Annotated, TypedDictfrom langgraph.graph import StateGraph, START, ENDfrom langgraph.graph.message import add_messages  class State(TypedDict):    messages: Annotated[list, add_messages]    step: int  def inquiry_node(state: State) -> dict:    return {"messages": [{"role": "user", "content": "我想查快递"}],            "step": state["step"] + 1}  def solution_node(state: State) -> dict:    return {"messages": [{"role": "assistant", "content": "请提供运单号"}],            "step": state["step"] + 1}  def route_by_step(state: State) -> str:    return "solution" if state["step"] >= 1 else "end"  g = StateGraph(State)g.add_node("inquiry", inquiry_node)g.add_node("solution", solution_node)g.add_edge(START, "inquiry")g.add_conditional_edges("inquiry", route_by_step,                        {"solution": "solution", "end": END})g.add_edge("solution", END) app = g.compile() result = app.invoke({"messages": [], "step": 0})print(result["step"])print([m.content for m in result["messages"]])
目的：构建一个“查询→条件分流→回答”的有状态 Agent 工作流，演示节点、边、条件边、状态合并的完整组合。
  1. class State(TypedDict):定义状态 Schema，声明 messages 与 step 两个字段
  2. messages: Annotated[list, add_messages]给消息字段配置追加型 reducer，新消息自动拼接
  3. def inquiry_node(state: State) -> dict:节点函数，接收当前状态并返回增量字段
  4. g.add_node("inquiry", inquiry_node)把函数注册成图里的节点
  5. g.add_conditional_edges("inquiry", route_by_step, ...)给节点添加条件边，按路由函数动态选择下一站
  6. g.compile()编译状态图，返回可执行的应用对象
  7. app.invoke({"messages": [], "step": 0})传入初始状态运行工作流

常见误区

误区我以为每个节点拿到的是一份独立拷贝，改完不影响其他节点，所以节点里随便改 `state` 都没关系。

正解LangGraph 中单次 invoke 流程里，所有节点共享同一份全局状态，节点返回的增量会被合并回去，下一个节点能看到上一个节点的改动。它不是“每站复印一份”，而是“同一张假条轮流填写”。

为何常见：受普通函数“参数按值传递”的直觉影响，很容易把 Python 的默认行为套到图执行引擎上。

误区我以为条件边只能返回“是/否”两个方向，顶多加一个默认值，分支再多就做不了。

正解条件函数可以返回任意多个 key，路由映射表是普通字典，可以配置 3 个、5 个甚至更多分支；同一 key 也可以映射到不同节点，分支本质是“从当前节点到下一节点的动态编号”。

为何常见：生活里“同意/驳回”的二值直觉太强，容易忽略路由表其实是个通用映射。

误区我以为必须把 `state` 里的每个字段都重新写一遍再返回，否则其他字段就会被清空。

正解LangGraph 的合并机制决定，节点只需返回要修改的字段增量，其余字段自动沿用旧值；如果返回了 Schema 之外的字段，可能被忽略或校验失败。

为何常见：写普通 Python 函数时习惯把所有变量一起 return，天然担心“不发完整数据就会被覆盖”。

误区我以为 `messages` 这类列表字段每次直接赋值就好，新节点写入时不会影响历史数据。

正解列表字段如果不配 add_messages 之类的 reducer，新节点返回的列表会整组覆盖旧列表，导致历史消息丢失；配了 reducer 后才会自动追加。

为何常见：把“谁后写谁占格”的直觉带到图引擎，忽略了在 LangGraph 里 reducer 才是合并规则的最终裁判。

面试题

Q1. 在 LangGraph 中，一个节点（Node）最常见的定义方式是？

单选题 · easy

  • A.异步生成器对象
  • B.接收状态返回更新字典← 正确答案
  • C.一种只读配置文件
  • D.数据库的连接语句

答案：B。节点本质是一个普通函数，它接收当前状态（State），返回要更新的字段增量（dict），LangGraph 会自动把返回值合并进全局状态。生成器、配置文件、数据库语句都不是节点的标准形态。

Q2. 边（Edge）在 LangGraph 中的核心作用是？

单选题 · easy

  • A.缓存中间计算结果
  • B.决定节点之间的流转顺序← 正确答案
  • C.存储所有用户聊天记录
  • D.对模型进行参数微调

答案：B。边定义的是“完成当前节点后下一步去哪”，它决定流程的流转顺序，本身不处理业务数据。缓存、存储、微调是其他组件或外部系统的职责。

Q3. 关于 START 与 END 节点，下列说法正确的是？

单选题 · easy

  • A.必须显式添加多个 START
  • B.前者是入口，后者是出口← 正确答案
  • C.二者都不能连接任何边
  • D.它们是两个特殊的数据节点

答案：B。START 是图执行的唯一入口，END 是流程出口；两者都必须存在，但各自只能有一个。它们不是普通的数据处理节点，而是图结构的边界标记。

Q4. 实现条件分支，需要用到的方法是？

单选题 · mid

  • A.add_conditional_edges← 正确答案
  • B.add_message_history
  • C.update_state_schema
  • D.compile_and_execute

答案：A。add_conditional_edges 专门用于给节点添加条件边，它接收一个路由函数和分支映射表。其余方法分别与消息历史、Schema 更新、编译执行相关，都不负责条件路由。

Q5. 状态在节点间传递时，LangGraph 默认采用的更新方式是？

单选题 · mid

  • A.全量复制一份新状态
  • B.节点返回增量，自动合并← 正确答案
  • C.每次清空状态重新输入
  • D.由用户手动锁字段

答案：B。LangGraph 默认增量更新：节点只需返回要修改的字段，引擎将其与旧状态合并成新状态。不需要全量复制，也不会清空状态，更不需要用户手动加锁。

Q6. 以下哪些属于 LangGraph 中定义状态 Schema 的常用方式？

多选题 · mid

  • A.使用 TypedDict 声明字段类型← 正确答案
  • B.使用 Annotated 指定合并策略← 正确答案
  • C.使用 JSX 描述界面结构
  • D.使用 CSV 文件保存记录

答案：AB。TypedDict 用来声明状态字段及类型，Annotated 用来给字段附加 reducer 合并策略。JSX 是前端语法，CSV 是数据存储格式，都与状态 Schema 无关。

Q7. 关于条件边，正确的说法有哪些？

多选题 · mid

  • A.路由函数返回字符串 key← 正确答案
  • B.key 映射到具体节点← 正确答案
  • C.分支数量最多不能超过三个
  • D.每个节点必须有条件边

答案：AB。条件函数输出字符串 key，路由映射表把 key 翻译成目标节点，因此分支数量没有硬性上限。普通节点完全可以只用普通边，不强制使用条件边。

Q8. 若节点的返回 dict 包含状态 Schema 中未定义的字段，最可能出现什么结果？

单选题 · hard

  • A.自动创建一个新分支
  • B.字段被 Schema 校验拦截← 正确答案
  • C.直接修改全局配置文件
  • D.静默把字段转成字符串

答案：B。LangGraph 依赖 Schema 进行状态合并，未定义的字段不会被自动加入状态，通常会被忽略或触发类型校验错误。它不会创建分支、修改配置文件，也不会被静默改成字符串。

Q9. 同一状态里若两个节点先后给 list 字段赋值且未配 reducer，最终结果是？

单选题 · hard

  • A.后一个节点整组覆盖前一个← 正确答案
  • B.两个列表自动拼接
  • C.LangGraph 随机选取一个
  • D.触发异常并停止运行

答案：A。没有 reducer 的列表字段采用“覆盖”语义，后一个节点返回的列表会整体替换旧值。两个列表自动拼接需要 add_messages 之类的 reducer 才能实现。

Q10. 关于 add_messages reducer，以下说法最准确的是？

单选题 · extreme

  • A.它会删除重复的历史消息
  • B.它把新消息追加到现有列表← 正确答案
  • C.它只允许保存一条最新消息
  • D.它只能处理纯文本消息

答案：B。add_messages 的核心行为是把新消息追加到现有消息列表末尾，并完成消息格式标准化。它不会去重，不限制列表只留一条，也不只接受纯文本——结构化消息同样支持。