1.1 LangGraph 是什么：面向有状态、多步骤 LLM 应用的编排框架
我第一次接触 LangGraph 的时候，脑子里冒出来的第一个问题就是：LangChain 不是已经能搭 LLM 应用了吗，怎么又来了个新东西？后来翻了几遍文档，自己动手跑了两三个项目，才慢慢品出味道来。LangGraph 是一个用来编排有状态、多步骤 LLM 应用的框架。它的核心想法特别朴素——把工作流画成一张图，节点代表计算步骤，边代表步骤之间的流转关系。
我刚开始用 LangChain 那会儿，写个简单的问答链挺顺手，链式调用一串到底，代码读起来也清楚。问题是，真实场景里的 LLM 应用根本不是一条直线。用户问一句话，我可能需要先判断意图，再决定调哪个工具，工具返回结果之后也许要回到上一步重新判断，中间还可能穿插人工审核。这种带循环、带分支的流程，用传统的链式结构去写，代码会变得又长又绕，调试起来跟走迷宫似的。
LangGraph 就是冲着这个痛点来的。它把整个执行过程建模成一张有向图，状态在节点之间传递，每个节点都能读取和修改这份共享状态。我可以在图里画循环，可以画条件分支，也可以让多个智能体并行跑再汇总结果。框架本身不替我写业务逻辑，它提供的是结构、是骨架、是让复杂流程变得可控的那套抽象。用一句话概括：LangGraph 让 LLM 应用的编排从“写链”变成了“画图”。
1.2 为什么需要 LangGraph：复杂工作流、循环、分支与多智能体协作需求
我拿自己做过的一个客服助手项目举例吧。最初的版本用 LangChain 的 AgentExecutor 加上几个工具，跑起来能回答问题，用户一多，问题就露馅了。有些请求需要先查订单，再根据订单状态决定要不要退款，退款金额还得算一遍，算完之后如果超过某个阈值就得转人工。这套逻辑写成提示词塞给 Agent，它能跑，但非常不稳定，偶尔会跳步骤，偶尔会陷入死循环。
换成 LangGraph 之后，我把每个步骤拆成独立的节点，用条件边控制流转。查订单是一个节点，判断是否退款是一个节点，计算金额是一个节点，转人工又是一个节点。图的结构清清楚楚，哪一步出了错，日志里一看就知道。循环也不再是隐患，我可以用状态里的计数器来限制迭代次数，超过上限就强制走兜底路径。
多智能体协作是另一个让我觉得 LangGraph 不可或缺的场景。我试过让一个主管智能体去调度三个专家智能体，分别负责技术问题、账单问题和投诉处理。主管根据用户输入决定把任务派给谁，专家处理完再把结果交回来，主管决定是继续追问还是直接回复用户。这种模式如果用纯代码加提示词去拼，光是管理对话历史和角色切换就能把人逼疯。LangGraph 把智能体之间的消息传递、状态共享、路由决策都变成了图上的边和节点，协作关系一下子就有了形状。
1.3 LangGraph 核心能力全景：状态图、检查点、人机协同、流式与可观测性
我每次跟别人介绍 LangGraph，都会先画一张图：圆圈是节点，箭头是边，中间飘着一个叫 State 的盒子。状态图是整个框架的地基。我定义一份状态结构，可以是字典、可以是 Pydantic 模型，每个节点收到的都是这份状态，处理完再吐回去。节点之间靠边连接，普通边是无条件跳转，条件边则根据状态里的某个值决定下一步走哪条路。这套东西听起来简单，真正用起来会发现灵活度非常高。
检查点是我最喜欢的功能之一。LangGraph 会在每一步执行后把状态存下来，存到内存、SQLite、Postgres 都行。这意味着什么？意味着我可以在某个节点中断执行，等人类审批完了再恢复，而且是从断点处接着跑，不用从头再来。我做过一个人工审核的流程，智能体生成回复草稿之后暂停，运营人员在界面上改几个字，点确认，图从那个检查点继续往下走，把修改后的内容发出去。整个过程丝滑得不像话。
流式输出和可观测性算是日常开发里的刚需。LangGraph 支持按 token 流式返回，也支持按节点流式返回。用户能看到文字一个一个蹦出来，体验上就舒服很多。调试的时候我会接上 LangSmith，每个节点的输入输出、耗时、token 消耗都记录得明明白白。哪一步慢了，哪一步 prompt 写歪了，翻一眼追踪记录就能定位。这些东西单独拿出来可能不算惊艳，凑在一起就构成了一套完整的生产级编排能力。
1.4 学习路径与前置知识：Python、LangChain 基础与 LangGraph 入门教程导览
如果你问我学 LangGraph 之前需要准备什么，我会说三样东西：Python 基础、对 LangChain 的基本了解、还有一颗愿意画图的心。Python 不用多说了，装饰器、类型注解、异步编程这些概念最好熟悉一下，因为 LangGraph 的 API 里到处都是它们的身影。LangChain 的基础部分——模型调用、提示词模板、工具定义、输出解析器——也建议先摸一遍，LangGraph 并不是要取代 LangChain，它更像是站在 LangChain 肩膀上的一层编排层。
我自己的学习路径大概是这样的：先花一个下午把官方文档里的“核心概念”页面读一遍，然后照着 Quickstart 敲一个最小的状态图示例。那个示例特别简单，就两个节点加一条边，跑通之后我对 StateGraph 的运作方式就有了感觉。接着我会去改那个示例，加一个条件边，再加一个循环，看看状态怎么流动、怎么变化。动手改比光看文档有用得多，很多细节都是在改代码的过程中才注意到的。
官方教程之外，我还会去翻 GitHub 上的示例仓库，看看别人怎么组织节点、怎么设计状态结构。LangGraph 的社区还挺活跃的，能找到不少实战案例。我会建议你别一上来就啃多智能体那种复杂场景，先从单智能体的 ReAct 循环开始，把基础打牢了再往上叠加。学习这件事急不得，画图也是一样，先把简单的图跑顺了，复杂的图自然就有思路了。
2.1 状态（State）：共享数据结构、Reducer 与消息通道
我一开始画图的时候，把状态想得太简单了。以为它就是个字典，节点往里塞点东西，下一个节点拿出来用。后来在真实项目里踩了坑才明白，状态是整张图的血液。每个节点都围绕这份共享数据做读写，节点之间不直接说话，全靠状态传递信息。设计状态结构的时候，我会问自己：整个流程里哪些信息需要跨节点流动？用户原始输入、中间推理结果、工具返回内容、迭代次数、错误标记，这些东西都得放进状态里。它像一份契约，约束着每个节点能拿到什么、能改什么。
Reducer 是状态里最容易忽略又最要命的部分。默认情况下，节点返回的字典会直接覆盖状态里同名的键。我做过一个对话历史节点，每次调用模型后把新消息塞进 state["messages"]，结果下一轮发现历史全没了，只剩下最新一条。查了半天文档才搞懂，需要用 Annotated 给字段挂一个 reducer。比如 Annotated[list, add_messages]，它告诉 LangGraph 这个字段不是覆盖，而是把新消息追加进去。消息通道就是这么来的，多个节点往同一个列表里写消息，reducer 负责合并。我现在的习惯是，凡是需要累积的字段——对话历史、工具调用记录、中间步骤日志——都挂上合适的 reducer。不挂 reducer 的字段就当普通变量用，谁写谁覆盖。
状态设计的好坏直接影响后面的调试难度。我见过有人把所有东西塞进一个大字典，节点返回整个状态，改一个字段要复制一大坨。我更喜欢用 TypedDict 或 Pydantic 定义明确的字段，每个节点只返回自己修改的那部分。这样图的意图更清晰，出问题也容易定位。状态不是垃圾桶，它是节点之间对话的语言。
2.2 节点（Node）：函数、工具、模型调用与子图节点
节点就是函数。我刚开始写节点的时候，总想在里面干太多事——调模型、解析输出、判断意图、调工具，全塞一个函数里。后来把节点拆细了，每个节点只做一件事，图反而更好维护。节点接收状态，返回一个字典，字典里是这次执行后需要更新的字段。同步函数可以，异步函数也可以。我用得最多的是调用模型的节点，里面写提示词、发请求、把回复塞回消息列表。模型调用节点通常比较重，耗时也长，我会把它和纯逻辑节点分开，方便后面做重试和监控。
工具节点和模型调用节点不太一样。工具节点负责执行外部操作——查数据库、调 API、读文件。我通常会把工具包装成节点，输入从状态里取，输出再写回状态。LangGraph 不限制节点里写什么代码，只要能接收状态、返回更新就行。这种自由度让我可以把现有的 Python 函数直接搬进来，不用为了框架改代码。子图节点是另一个有意思的东西。我可以把一整张编译好的图当成一个节点塞进更大的图里。做多智能体项目的时候，我把每个专家智能体都做成子图，主管智能体调度它们的时候，调用子图节点就像调用普通函数一样。子图内部有自己的状态和节点，对外只暴露一个接口，封装得干干净净。
节点命名也很讲究。我习惯用动词加名词，比如 “call_model”、“execute_tool”、“route_intent”。名字起得清楚，图一画出来就能看懂流程。节点返回值要尽量小，只返回变化的字段。我见过有人返回整个状态，虽然也能跑，但状态大了之后性能会受影响。节点是图里的工人，每个工人干好自己的活，图才能顺畅运转。
2.3 边（Edge）：普通边、条件边与动态路由
边决定节点之间怎么走。普通边最简单，从 A 节点直接连到 B 节点，没有条件，没有犹豫。我在画线性流程的时候用普通边，比如“接收用户输入 -> 调用模型 -> 返回结果”。这种边写起来就是 graph.add_edge("a", "b")。普通边的好处是结构清晰，执行路径唯一，调试的时候不用猜。坏处也明显，真实业务很少一条直线走到底。
条件边是 LangGraph 的灵魂。它背后是一个函数，接收当前状态，返回下一个节点的名字。我做过一个意图路由节点，根据用户问题判断是查订单、退款还是投诉，条件边把流程引向三个不同的处理分支。写条件边的时候，返回值必须是图里已经存在的节点名，写错了编译就会报错。这个报错机制帮我省了不少时间。动态路由更灵活一些，可以用 Command 对象在节点内部直接决定跳转目标。我还在摸索这个功能的阶段，目前用得最多的是条件边。
循环也是靠条件边实现的。我在一个迭代优化的图里，让“生成答案”节点连到“评估答案”节点，评估节点通过条件边决定是回到生成节点继续改，还是走向结束节点。为了防止死循环，我在状态里放了一个计数器，每次循环加一，条件边里判断超过上限就强制结束。边是图的骨架，条件边是图的大脑。没有条件边，LangGraph 就和一条链没什么区别。
2.4 图编译与执行：StateGraph、START/END、invoke/stream/ainvoke
StateGraph 是我每天都要打交道的类。我定义状态结构，创建 StateGraph 实例，把节点和边一个个加进去，再调用 compile()。编译这一步会做校验，检查有没有孤立的节点、条件边的返回值是否合法、START 和 END 有没有正确连接。我刚开始学的时候，经常忘了把某个节点连到图上，compile 直接报错，提示节点不可达。这个校验让我少写了很多低级 bug。START 和 END 是两个特殊节点，START 代表图的入口，END 代表结束。每个图都必须有从 START 出发的路径，也得有走向 END 的路径，不然执行起来会卡住。
编译完的图可以调用。invoke 是最直接的方式，传一个初始状态进去，它执行完整张图，返回执行完的状态。我调试简单流程的时候用 invoke，结果一把出来，方便看。stream 则是一边执行一边吐出中间结果。做聊天应用的时候，我用 stream 来展示打字机效果，用户能实时看到模型输出的 token。stream 还能按节点流式返回，每个节点执行完就吐一个事件，调试复杂流程特别有用。ainvoke 是异步版本，我在 FastAPI 服务里用它，避免阻塞主线程。同步和异步的选择取决于运行环境，逻辑本身没有区别。
执行的时候，状态在节点之间按边流动。每个节点执行完，LangGraph 会合并节点返回的更新到状态里，再根据边决定下一步。我见过有人把 invoke 和 stream 混用，结果状态对不上。我的经验是，一个流程要么全程用 invoke，要么全程用 stream，别来回切。编译和执行分开的设计，让图的构建和执行解耦，我可以把图定义放在一个模块里，在多个地方调用。
2.5 持久化与检查点：MemorySaver、线程记忆与中断恢复
检查点是我觉得 LangGraph 最实用的功能。每次节点执行完，框架可以把当前状态存下来。MemorySaver 把检查点保存在内存里，适合开发和测试。我写 demo 的时候用 MemorySaver，跑一遍流程，看看每步状态对不对。内存保存有个问题，进程一重启数据就没了。生产环境我换成了 SqliteSaver 或 PostgresSaver，状态持久化到数据库，服务重启也不怕。检查点让长流程变得可恢复，哪一步失败了，可以从最近的检查点重跑，不用从头开始。
线程记忆是检查点的组织方式。每个检查点都会关联一个 thread_id。我用 thread_id 来区分不同用户的对话。同一个 thread_id 下，图会记住之前的状态，用户再次提问时，对话历史还在。做多轮对话应用的时候，这个机制特别省事。我只需要在 invoke 或 stream 的时候传入 config={"configurable": {"thread_id": user_id}}，剩下的交给框架。不同 thread 之间状态隔离，互不干扰。
中断恢复和人机协同靠的就是检查点。我可以在某个节点前或节点后设置中断，让图暂停执行。运营人员在界面上审核内容、修改状态，再调用恢复接口，图从断点继续往下走。我做过一个内容审核流程，智能体生成草稿后中断，人工编辑完再恢复，发布出去。整个过程中，检查点保存了每一步的状态，修改后的状态会被写回检查点，后续节点拿到的就是最新版本。MemorySaver 适合快速验证中断逻辑，生产环境一定要换持久化后端。检查点不仅是保存进度，它让人类和智能体可以在同一个状态上协作。 from typing import TypedDict from langgraph.graph import StateGraph, START, END
class State(TypedDict):
text: str
result: str
def upper_node(state: State):
return {"result": state["text"].upper()}
def exclaim_node(state: State):
return {"result": state["result"] + "!!!"}
builder = StateGraph(State) builder.add_node("upper", upper_node) builder.add_node("exclaim", exclaim_node) builder.add_edge(START, "upper") builder.add_edge("upper", "exclaim") builder.add_edge("exclaim", END)
graph = builder.compile() output = graph.invoke({"text": "hello langgraph", "result": ""}) print(output)
from typing import TypedDict, Literal from langgraph.graph import StateGraph, START, END, MessagesState
class SupervisorState(MessagesState):
next: str
task: str
def supervisor_node(state: SupervisorState):
decision = supervisor_llm.invoke(state["messages"])
return {"next": decision["next"], "task": decision["task"]}
def research_node(state: SupervisorState):
result = research_llm.invoke(state["task"])
return {"messages": [("assistant", result)]}
def coder_node(state: SupervisorState):
result = coder_llm.invoke(state["task"])
return {"messages": [("assistant", result)]}
def route_supervisor(state: SupervisorState):
return state["next"]
builder = StateGraph(SupervisorState) builder.add_node("supervisor", supervisor_node) builder.add_node("researcher", research_node) builder.add_node("coder", coder_node) builder.add_edge(START, "supervisor") builder.add_conditional_edges("supervisor", route_supervisor, {
"researcher": "researcher",
"coder": "coder",
"FINISH": END
}) builder.add_edge("researcher", "supervisor") builder.add_edge("coder", "supervisor") graph = builder.compile()
from langgraph.graph import StateGraph, START, END from langgraph.types import RetryPolicy from langgraph.checkpoint.postgres import PostgresSaver
def call_model(state):
...
def call_tool(state):
...
builder = StateGraph(dict) builder.add_node("model", call_model, retry=RetryPolicy(max_attempts=3)) builder.add_node("tool", call_tool, retry=RetryPolicy(max_attempts=2, backoff_factor=2)) builder.add_edge(START, "model") builder.add_edge("model", "tool") builder.add_edge("tool", END)
checkpointer = PostgresSaver.from_conn_string("postgresql://...") graph = builder.compile(checkpointer=checkpointer)
