LangGraph 快速上手指南30 分钟跑通一个有状态 Agent卡住了怎么排查【免费下载链接】langgraphBuild resilient agents.项目地址: https://gitcode.com/GitHub_Trending/la/langgraphLangGraph 是一个用于构建有状态、长任务 Agent 的编排框架它把状态流转拆成节点和边再叠加持久化、流式输出、人工中断、重试等能力专门解决Agent 跑一半崩溃没法续跑、多轮对话没记忆、想在某一步停下来等人确认这类问题。如果你是刚接触 LangGraph 的开发者这篇文章会带你跑通主流程并给出一张参数速查表和一个可直接对照的排障手册。什么时候该用 LangGraph 而不是别的一个很典型的场景你要做一个查资料 → 写草稿 → 人工审校 → 发布的流水线。用普通函数链写它跑挂了只能从头再来用 LangGraph 写每一步之间的状态都会被记录下来任何一步崩了都能从断点恢复人工审校那一步还能停下来等人在界面上点通过。判断标准很简单流程有明确的步骤、分支、循环多智能体协作、ReAct 循环单次运行时间长需要断点续跑或回放中间需要人来介入审批、改状态、补充信息满足任意一条LangGraph 就是合适的工具。反过来如果只是调用一次模型、返回一次结果直接用模型 SDK 即可没必要引入图。跑通主流程要看哪几个文件LangGraph 的仓库是多包结构但你第一次跑通主流程只需要理解四个组件其余都可以先跳过StateGraph —— 图的声明入口。位于 state.py你在这里定义状态结构、加节点、连边。add_node()注册一个处理函数add_edge()声明固定流转add_conditional_edges()让分支由函数返回值决定最后compile()得到可执行对象。Pregel 引擎 —— 真正执行图的总调度。位于 pregel 包核心类定义在 main.py。它的工作方式借鉴了 Pregel 消息传递模型每一步先确定哪些节点该跑并行执行后把各节点的输出写回状态再进入下一步直到没有节点需要执行。理解按步推进、每步同步状态这一点后面读源码会轻松很多。Checkpointer —— 状态的存档系统。位于 checkpoint 包最常用的内存版是MemorySavermemory 子包。给compile(checkpointer...)传一个进去每次invoke时带上thread_id图的中间状态就会按线程存档支持暂停、恢复、回放到任意历史节点。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(checkpointerMemorySaver()) app.invoke({task: 写周报}, {configurable: {thread_id: t1}})跑完后对同一个thread_id再调一次invoke新输入会叠在旧状态上继续流转——这就是有状态的全部秘密。常用能力与参数怎么配不用逐个翻文档先记住下面这张表。它们都出现在compile()和调用配置里覆盖了 80% 的日常需求配置写在哪它解决什么问题checkpointercompile(checkpointer...)状态持久化、断点恢复、多轮对话记忆thread_idinvoke(..., config)区分不同会话/任务的存档线同 ID 累加状态不同 ID 互不干扰interrupt_before/aftercompile(...)在指定节点前后强制暂停等外部审批再继续interrupt()节点函数体内运行时动态决定要不要停配合Command(resume...)恢复storecompile(store...)跨线程的长期记忆用户偏好、知识库与 checkpointer 的线程内短期记忆互补debugTruecompile(debugTrue)打印每一步的节点调度与状态写入定位图为什么这么走retry_policyadd_node(..., retry_policy...)单节点失败自动重试次数、退避、可重试异常类型配置细节见 retry 实现stream_modestream(...)选择输出粒度values看每步完整状态updates只看增量messages直接拿 LLM 流式 token状态 reducer状态 schema 字段上定义同名字段并发写入时如何合并覆盖、追加、自定义是状态更新冲突的根治手段另外两个进阶入口可以知道但不用深究channels 包 实现了各种状态通道的合并语义如LastValue、Topicfunc 包 提供了entrypoint/task装饰器适合不太想用图 API的人用纯函数风格搭流程。三个高频卡点现象、根因、解法状态没有持久化重启后全丢现象同一个thread_id调两次invoke第二次行为跟第一次一样像是从头开始。根因compile()时没传checkpointer或者传了但调用时config里没有thread_id。两者缺一不可——没有 checkpointer 就没有存档设施没有thread_id引擎不知道存到哪条线上。解法确认compile(checkpointerMemorySaver())和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让多次写入变成追加确实需要整字段替换的场景则反过来用覆盖语义。改完用debugTrue跑一遍看每一步实际写入的值是否符合预期。图卡住了不知道停在哪个节点、怎么恢复现象开了interrupt_before或调用了interrupt()图停住后你不确定它停在哪儿、resume 之后走的对不对。根因暂停的图其实已经完整存档只是缺少观察手段resume 时如果没用同一个thread_id等于开了个新线程自然恢复不了。解法先用app.get_state(config)查看当前断点和各通道值确认停在哪个节点恢复时务必复用原thread_idfrom langgraph.types import Command snapshot app.get_state({configurable: {thread_id: t1}}) app.invoke(Command(resumeapproved), {configurable: {thread_id: t1}})动手清单接下来三件事验证持久化按上面最小骨架跑通后对同一thread_id连调两次invoke观察第二次是否继承了第一次的状态再换成MemorySaver之外任意一个 checkpoint 后端确认存档机制与后端解耦。加一个人工中断点在两个节点之间加interrupt_before跑一次、get_state看断点、Command(resume...)恢复完整走一遍暂停—观察—继续循环这是 LangGraph 相对普通工作流框架的核心差异。动手改一个示例仓库自带 examples 目录里面有 RAG、多智能体、反思式 Agent 等参考实现挑一个结构最接近你业务的把模型层换成自己的、状态字段改成自己的是最快的学习路径。git clone https://gitcode.com/GitHub_Trending/la/langgraph把上面三步做完你就掌握了 LangGraph 的日常使用面图怎么声明、状态怎么流转、断点怎么恢复。剩下的流式输出、子图、定时任务等能力都是在这条主链路上长出来的——遇到新需求时先回来对照参数速查表找对应配置再进源码里看具体实现比通读整个仓库高效得多。【免费下载链接】langgraphBuild resilient agents.项目地址: https://gitcode.com/GitHub_Trending/la/langgraph创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考