企业官网建设流程全解析

LangGraph 快速上手指南：30 分钟跑通一个有状态 Agent，卡住了怎么排查

【免费下载链接】langgraphBuild resilient agents.项目地址: https://gitcode.com/GitHub_Trending/la/langgraph

LangGraph 是一个用于构建有状态、长任务 Agent 的编排框架：它把"状态流转"拆成节点和边，再叠加持久化、流式输出、人工中断、重试等能力，专门解决"Agent 跑一半崩溃没法续跑、多轮对话没记忆、想在某一步停下来等人确认"这类问题。如果你是刚接触 LangGraph 的开发者，这篇文章会带你跑通主流程，并给出一张参数速查表和一个可直接对照的排障手册。

什么时候该用 LangGraph 而不是别的

一个很典型的场景：你要做一个"查资料 → 写草稿 → 人工审校 → 发布"的流水线。用普通函数链写，它跑挂了只能从头再来；用 LangGraph 写，每一步之间的状态都会被记录下来，任何一步崩了都能从断点恢复，人工审校那一步还能停下来等人在界面上点"通过"。

判断标准很简单：

• 流程有明确的步骤、分支、循环（多智能体协作、ReAct 循环）；
• 单次运行时间长，需要断点续跑或回放；
• 中间需要人来介入（审批、改状态、补充信息）；

满足任意一条，LangGraph 就是合适的工具。反过来，如果只是"调用一次模型、返回一次结果"，直接用模型 SDK 即可，没必要引入图。

跑通主流程要看哪几个文件

LangGraph 的仓库是多包结构，但你第一次跑通主流程只需要理解四个组件，其余都可以先跳过：

1. StateGraph —— 图的声明入口。位于 state.py，你在这里定义状态结构、加节点、连边。add_node()注册一个处理函数，add_edge()声明固定流转，add_conditional_edges()让分支由函数返回值决定，最后compile()得到可执行对象。
2. Pregel 引擎 —— 真正执行图的"总调度"。位于 pregel 包，核心类定义在 main.py。它的工作方式借鉴了 Pregel 消息传递模型：每一"步"先确定哪些节点该跑，并行执行后把各节点的输出写回状态，再进入下一步，直到没有节点需要执行。理解"按步推进、每步同步状态"这一点，后面读源码会轻松很多。
3. Checkpointer —— 状态的"存档系统"。位于 checkpoint 包，最常用的内存版是MemorySaver（memory 子包）。给compile(checkpointer=...)传一个进去，每次invoke时带上thread_id，图的中间状态就会按线程存档，支持暂停、恢复、回放到任意历史节点。
4. interrupt / Command —— 人工介入的开关。定义在 types.py。节点里调用interrupt("问题")会立刻停住整个图并把问题抛给外部；外部用Command(resume=...)带着答案调一次，图就从断点继续。

四个组件串起来，数据流是：输入 → 引擎按步调度节点 → 节点读状态、写状态 → checkpointer 存档 → 流式/一次性返回结果。

下面是最小可运行骨架，你可以直接照着敲一遍：

from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver g = StateGraph(State).add_node("plan", plan).add_node("write", write) g = g.add_edge(START, "plan").add_edge("plan", "write").add_edge("write", END) app = g.compile(checkpointer=MemorySaver()) app.invoke({"task": "写周报"}, {"configurable": {"thread_id": "t1"}})

跑完后，对同一个thread_id再调一次invoke，新输入会叠在旧状态上继续流转——这就是"有状态"的全部秘密。

常用能力与参数怎么配

不用逐个翻文档，先记住下面这张表。它们都出现在compile()和调用配置里，覆盖了 80% 的日常需求：

另外两个进阶入口可以知道但不用深究：channels 包 实现了各种状态通道的合并语义（如LastValue、Topic）；func 包 提供了@entrypoint/@task装饰器，适合"不太想用图 API"的人用纯函数风格搭流程。

三个高频卡点：现象、根因、解法

状态没有持久化，重启后全丢

现象：同一个thread_id调两次invoke，第二次行为跟第一次一样，像是从头开始。

根因：compile()时没传checkpointer，或者传了但调用时config里没有thread_id。两者缺一不可——没有 checkpointer 就没有存档设施，没有thread_id引擎不知道存到哪条线上。

解法：确认compile(checkpointer=MemorySaver())和invoke(inputs, {"configurable": {"thread_id": "..."}})成对出现。生产环境把MemorySaver换成 Postgres/SQLite 版（checkpoint 各后端子包），原理完全一致。

节点改了状态，但读到的是旧值

现象：节点 A 往messages追加了一条，节点 B 里打印state却看不到这条新消息；或者两个节点并发写同一字段，后面的把前面的覆盖了。

根因：状态字段的"合并规则"默认是覆盖（LastValue）。LLM 消息这类天然要累加的字段，需要显式声明 reducer，或者直接使用带消息 reducer 的MessagesState。

解法：在状态定义上给字段挂 reducer（例如Annotated[list, operator.add]或add_messages），让多次写入变成追加；确实需要"整字段替换"的场景则反过来用覆盖语义。改完用debug=True跑一遍，看每一步实际写入的值是否符合预期。

图"卡住"了：不知道停在哪个节点、怎么恢复

现象：开了interrupt_before或调用了interrupt()，图停住后你不确定它停在哪儿、resume 之后走的对不对。

根因：暂停的图其实已经完整存档，只是缺少观察手段；resume 时如果没用同一个thread_id，等于开了个新线程，自然"恢复"不了。

解法：先用app.get_state(config)查看当前断点和各通道值，确认停在哪个节点；恢复时务必复用原thread_id：

from langgraph.types import Command snapshot = app.get_state({"configurable": {"thread_id": "t1"}}) app.invoke(Command(resume="approved"), {"configurable": {"thread_id": "t1"}})

动手清单：接下来三件事

1. 验证持久化：按上面最小骨架跑通后，对同一thread_id连调两次invoke，观察第二次是否继承了第一次的状态；再换成MemorySaver之外任意一个 checkpoint 后端，确认存档机制与后端解耦。
2. 加一个人工中断点：在两个节点之间加interrupt_before，跑一次、get_state看断点、Command(resume=...)恢复，完整走一遍"暂停—观察—继续"循环，这是 LangGraph 相对普通工作流框架的核心差异。
3. 动手改一个示例：仓库自带 examples 目录，里面有 RAG、多智能体、反思式 Agent 等参考实现，挑一个结构最接近你业务的，把模型层换成自己的、状态字段改成自己的，是最快的学习路径。

git clone https://gitcode.com/GitHub_Trending/la/langgraph

把上面三步做完，你就掌握了 LangGraph 的日常使用面：图怎么声明、状态怎么流转、断点怎么恢复。剩下的流式输出、子图、定时任务等能力，都是在这条主链路上长出来的——遇到新需求时，先回来对照"参数速查表"找对应配置，再进源码里看具体实现，比通读整个仓库高效得多。

【免费下载链接】langgraphBuild resilient agents.项目地址: https://gitcode.com/GitHub_Trending/la/langgraph