最近做 Agent 应用的人越来越多但真正一上手就卡住的往往不是模型怎么选而是流程怎么控制。用 LangChain 写线性链路很顺手可一旦出现“需要多轮思考”“要根据结果调用不同工具”“多个分支要并行推进”代码就开始变得混乱状态散落在各种变量里分支判断堆在 if/else 里加一个新功能就要动一大片旧逻辑。LangGraph 的热度越来越高本质上就是因为它在解决这个问题。它不是用来替代 LangChain 的而是把流程重新定义成一张图每个节点只做一件事节点之间用边连接条件分支通过路由函数控制状态由框架统一管理。这种设计让复杂 Agent 的编排变得可控、可追溯、可扩展。这篇文章的目标很明确带你把 LangGraph 的核心用法完整跑通包括最常用的 conditional_edge 条件路由、子图、并行分支以及状态如何修改、如何做持久化。读完你不仅能看懂官方文档还能自己动手实现一个带分支和循环的 Agent 流程同时知道哪些坑是新手一定会踩的。1. 为什么 LangGraph 值得学先下一个判断LangGraph 解决的不是“能不能调大模型”的问题而是“复杂流程如何可控地跑起来”的问题。如果你只是在单次问答里调用一次模型完全用不上它但如果你的业务需要多个步骤配合甚至需要 Agent 自己决定下一步干什么LangGraph 就是当前最值得学的编排框架之一。传统开发里多步骤 Agent 通常靠手写状态机或者层层回调来实现。这样做有几个很明显的痛点状态管理靠全局变量或类成员流程一长就难以追踪。分支逻辑写在业务代码里新增一条路径需要改多个地方。循环控制很别扭尤其是“一直思考直到满足条件才退出”这类需求。链路不可视出了问题只能靠日志反推。LangGraph 把这些问题收敛到了几个核心抽象上State 负责状态Node 负责业务动作Edge 负责流转路径Conditional Edge 负责动态路由。你只需要把做的事情拆成节点再把节点之间的依赖关系画出来LangGraph 就能替你执行这张图。从学习曲线看LangGraph 的 API 本身并不复杂真正难的是“先想清楚流程结构再写代码”。很多人一上来就写节点写到一半发现分支条件设计不合理只能推翻重来。所以这篇文章后面会反复强调一件事画图先行。2. LangGraph 核心概念一张图编排 Agent 流程LangGraph 的核心思想可以用一个类比来理解它像一张地铁线路图。节点是每个站点站点之间用边连接列车根据线路图从一个站开到另一个站。如果前方有岔路就需要一个判断逻辑决定走哪条线。2.1 State整个流程的“车厢”State 是一个 TypedDict 类型定义描述整个流程图运行过程中需要维护的数据。它可以是一个字符串列表、一个计数器、一个配置文件也可以是消息对象列表。from typing import TypedDict class State(TypedDict): messages: list[str] step: int这里messages可以存对话历史或中间日志step可以当作计数器使用。LangGraph 的模型是每个节点接收当前 State执行自己的逻辑后返回一个“部分更新”框架再把这个更新合并到全局 State 中。这里有一个新手最容易误解的地方节点内部不应该直接修改传入的 state 对象而是返回一个 dict告诉 LangGraph 哪些字段需要更新。State 在 LangGraph 里是不可变视角的修改必须通过返回值完成。2.2 Node每个站点只干一件事Node 就是一个普通的 Python 函数输入是当前 State输出是一个 dict表示要对 State 做的更新。节点函数的设计应该保持单一职责一个节点只做一件事情这样后续维护和测试都会容易很多。def process_node(state: State): return {step: state[step] 1, messages: state[messages] [processed]}这个函数把step加一并在messages后面追加一条记录。LangGraph 会把这个返回值合并到全局状态。2.3 Edge 与 Conditional Edge决定流程走向Edge 是普通的静态连线表示固定流程路径。Conditional Edge 则是动态路由根据当前 State 的内容决定下一步进入哪个节点。条件路由是 LangGraph 最灵活的部分也是 Agent 能自主决策的关键。2.4 概念对比表格概念作用类比易错点State定义流程中共享的数据结构地铁的车厢字段过多、职责不清晰Node执行一个具体动作地铁站点在一个节点里堆太多逻辑Edge连接两个节点的静态路径普通轨道忘记连接 END 或 STARTConditional Edge根据状态动态决定路径岔路道岔路由函数返回值与映射键不一致3. LangGraph 和 LangChain 的区别LangGraph 常常被拿来和 LangChain 比较但它俩并不是同一个层面的东西。LangChain 的抽象偏向上层应用比如 Prompt 管理、模型调用、工具定义LangGraph 则更偏底层流程编排它定义了“状态如何流转”这件事。用一句话总结LangChain 帮你“调用大模型”LangGraph 帮你“编排整个调用流程”。两者可以配合使用在 LangGraph 的节点内部调用 LangChain 的模型封装或工具封装完全没有问题。维度LangChainLangGraph抽象层应用层提供链式封装调度层提供图编排流程模型Chain偏线性StateGraph支持分支和循环状态管理弱靠外部传递强内置 State 机制适合场景简单链式调用复杂 Agent、多步骤动态流程如果你在 LangChain 里用 Chain 写过比较复杂的链路最后一定会遇到“链条之间怎么跳转”的问题。LangGraph 正是为这类问题设计的用图代替链用节点代替每一步操作用条件路由代替写死的顺序。4. 环境准备与安装LangGraph 官方要求 Python 3.9 及以上版本。建议使用虚拟环境隔离依赖避免和系统 Python 环境相互污染。安装 LangGraph 本体pip install langgraph如果项目中还需要调用 OpenAI 等模型通常还会安装对应的 LangChain 集成包pip install langchain-openai安装完成后可以验证版本python -c import langgraph; print(langgraph.__version__)这里说明一点LangGraph 的版本迭代节奏比较快API 在不同大版本之间有过变化。比如早期版本中的START_NODE后来统一为START条件路由的写法也有调整。本文示例基于当前主流写法如果运行时报错提示某个符号找不到优先检查官方文档中对应版本的使用方式。5. 第一个 LangGraph 应用从零跑通 StateGraph接下来我们实现一个最基础的流程图开始节点、处理节点、结束节点依次执行。为了让示例不依赖任何外部服务和 API Key节点内部只做状态操作不实际调用模型。创建文件langgraph_demo.pyfrom typing import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): messages: list[str] step: int def start_node(state: State): return { messages: state[messages] [start], step: state[step] 1, } def process_node(state: State): return { messages: state[messages] [process], step: state[step] 1, } def end_node(state: State): return { messages: state[messages] [end], step: state[step] 1, } graph StateGraph(State) graph.add_node(start, start_node) graph.add_node(process, process_node) graph.add_node(end, end_node) graph.add_edge(START, start) graph.add_edge(start, process) graph.add_edge(process, end) graph.add_edge(end, END) app graph.compile() result app.invoke({messages: [], step: 0}) print(result)运行方式python langgraph_demo.py预期输出{messages: [start, process, end], step: 3}看到这个输出说明你已经跑通了一个最简单的 StateGraph 流程。下面拆解几个关键点StateGraph(State)表示创建一个基于State类型定义的图。add_node注册节点第一个参数是节点名第二个参数是节点函数。add_edge(START, start)表示整个流程从start节点进入。add_edge(end, END)表示流程执行到end节点后终止。compile()把图编译成可调用的应用对象。invoke()传入初始状态触发整个图执行返回最终的 State。这个示例体现了一个重要机制每个节点函数接收的是当前 State返回的是需要更新的字段。LangGraph 自动把这些更新合并到 State 中并作为下一个节点的输入。节点函数返回的键不需要包含 State 的全部字段只需要包含要新增或修改的部分。6. 条件路由与分支控制conditional_edge 深度解析很多人学 LangGraph 卡住的第一个地方就是条件路由。它本质上回答一个问题执行完当前节点后下一步去哪里这个决定不是写死的而是根据当前 State 的内容动态计算出来的。6.1 条件路由的三段式结构LangGraph 的条件路由由三个部分组成起始节点执行完哪个节点之后做判断。路由函数接收当前 State返回一个字符串键。映射表把字符串键映射到下一个目标节点。三段式结构保证了逻辑清晰路由函数只负责给出决策结果不直接涉及流程控制具体走向由映射表声明。6.2 设计一个带循环的 Agent 流程假设我们要实现一个 Agent 思考循环Agent 每次执行都会让轮次加一直到达到最大轮数才结束。这是很多 Agent 应用的真实需求也是 LangGraph 相比普通链式流程最突出的优势之一。创建文件conditional_demo.pyfrom typing import TypedDict from langgraph.graph import StateGraph, START, END class AgentState(TypedDict): messages: list[str] rounds: int def think_node(state: AgentState): return { messages: state[messages] [fthinking round {state[rounds] 1}], rounds: state[rounds] 1, } def finish_node(state: AgentState): return {messages: state[messages] [finished]} def route_after_think(state: AgentState): if state[rounds] 3: return finish return continue graph StateGraph(AgentState) graph.add_node(think, think_node) graph.add_node(finish, finish_node) graph.add_edge(START, think) graph.add_conditional_edges( think, route_after_think, { continue: think, finish: finish, }, ) graph.add_edge(finish, END) app graph.compile() result app.invoke({messages: [], rounds: 0}) print(result[rounds]) print(result[messages])运行这个脚本输出如下3 [thinking round 1, thinking round 2, thinking round 3, finished]这里的add_conditional_edges是 LangGraph 条件路由的标准写法。第一个参数声明在think节点执行之后做判断第二个参数是路由函数第三个参数是映射表路由函数的返回值必须能在映射表里找到对应的 key。6.3 循环检测与递归上限这个示例里出现了自环边think节点可以连接到它自己。这在传统流程框架里是不允许的但 LangGraph 明确支持循环因为 Agent 思考、工具调用这类场景本质上就需要循环。不过循环结构也意味着图可能在极端情况下进入死循环。LangGraph 提供了recursion_limit配置作为保护机制默认限制递归调用次数。如果超过限制会抛出异常。建议在调用时主动配置一个合理的上限result app.invoke( {messages: [], rounds: 0}, config{recursion_limit: 20}, )这里真正容易踩坑的地方是路由函数返回值忘记和映射表 key 对齐。比如路由函数返回ready但映射表里写的是{continue: think}运行时就会报错。排查这类问题第一步永远是检查路由函数的返回值。7. 子图与并行分支当业务流程比较大时把所有节点堆在一张图里会变得很难维护。LangGraph 支持把一部分节点封装成子图再作为父图中的一个节点使用。这种组合方式非常适合团队协作不同人负责不同子图最终在父图里拼接。7.1 子图示例from typing import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): messages: list[str] step: int def sub_node(state: State): return { messages: state[messages] [sub_node], step: state[step] 1, } # 构建子图 sub_graph StateGraph(State) sub_graph.add_node(sub_node, sub_node) sub_graph.add_edge(START, sub_node) sub_graph.add_edge(sub_node, END) sub_app sub_graph.compile() # 构建父图 def parent_start(state: State): return {messages: state[messages] [parent_start]} def parent_end(state: State): return {messages: state[messages] [parent_end], step: state[step] 1} parent_graph StateGraph(State) parent_graph.add_node(start, parent_start) parent_graph.add_node(sub, sub_app) parent_graph.add_node(end, parent_end) parent_graph.add_edge(START, start) parent_graph.add_edge(start, sub) parent_graph.add_edge(sub, end) parent_graph.add_edge(end, END) parent_app parent_graph.compile() result parent_app.invoke({messages: [], step: 0}) print(result)预期输出{messages: [parent_start, sub_node, parent_end], step: 2}子图编译后的对象可以作为普通节点加入父图这是 LangGraph 的设计核心之一。子图内部可以使用自己独立的命名空间父图通过“sub”这个名字引用它。这种嵌套结构非常适合复杂业务比如每个子图负责一个领域任务父图负责调度。7.2 并行分支LangGraph 还支持从同一个节点分出多条边让多个节点并行执行最后汇聚到同一个节点。这在需要同时调用多个工具、做多路检索时非常有用。graph StateGraph(State) graph.add_node(start, start_node) graph.add_node(branch_a, branch_a_node) graph.add_node(branch_b, branch_b_node) graph.add_node(join, join_node) graph.add_edge(START, start) graph.add_edge(start, branch_a) graph.add_edge(start, branch_b) graph.add_edge(branch_a, join) graph.add_edge(branch_b, join) graph.add_edge(join, END)执行顺序上branch_a和branch_b会以并行方式执行两个节点都结束之后才会进入join节点。这里的“并行”主要体现在调度层LangGraph 会调度多个任务并发执行。并行分支需要特别注意状态合并问题。如果两个分支节点都返回同一个 State 字段默认行为是后执行完的那个覆盖前一个这会导致数据丢失。正确的做法是使用 Reducer 定义合并规则让多个分支的更新能正确拼接。8. 状态修改与长期记忆正确理解 state 更新机制LangGraph 的热搜词里经常出现“如何在节点函数改变 state 状态值”这个问题的答案其实在前面已经提到过不要在节点内部修改 state 对象而是返回一个 dict 作为部分更新。这是 LangGraph 最核心的状态更新方式。但在并行分支、多节点协同的场景下默认的“覆盖”策略不够用。LangGraph 提供了字段级的 Reducer 机制通过Annotated类型声明合并规则。8.1 使用 Reducer 合并列表字段from typing import TypedDict, Annotated import operator from langgraph.graph import StateGraph, START, END class State(TypedDict): messages: Annotated[list[str], operator.add] def node_a(state: State): return {messages: [message_from_a]} def node_b(state: State): return {messages: [message_from_b]} def join_node(state: State): return {messages: [join]} graph StateGraph(State) graph.add_node(a, node_a) graph.add_node(b, node_b) graph.add_node(join, join_node) graph.add_edge(START, a) graph.add_edge(START, b) graph.add_edge(a, join) graph.add_edge(b, join) graph.add_edge(join, END) app graph.compile() result app.invoke({messages: []}) print(result)输出可能是以下任意一种顺序{messages: [message_from_a, message_from_b, join]}由于两个分支节点并行执行最终列表的拼接顺序不一定固定。这里Annotated[list[str], operator.add]的含义是当出现多个节点更新messages字段时使用operator.add把新的列表追加到已有列表后面而不是覆盖。8.2 自定义 Reducer如果默认的operator.add不满足业务需求还可以自定义合并函数。合并函数接收两个参数当前 State 中已有的值和节点返回的新值返回合并后的结果。例如保留去重逻辑from typing import TypedDict, Annotated def merge_unique(existing: list[str], new: list[str]) - list[str]: for item in new: if item not in existing: existing existing [item] return existing class State(TypedDict): tags: Annotated[list[str], merge_unique]这里要注意合并函数应该尽量保持纯粹不要依赖外部可变状态否则在并行执行时会产生难以预料的副作用。8.3 Checkpointer 与长期记忆LangGraph 中的“记忆”不是一个简单的缓存而是基于状态检查点机制。框架会在每个节点执行前后保存 Checkpoint记录当前 State 的快照。使用 Checkpointer 之后即使流程中断也能从最近的检查点恢复执行。从 langgraph 的 checkpoint 模块导入内存版实现from langgraph.checkpoint.memory import MemorySaver memory MemorySaver() app graph.compile(checkpointermemory) config {configurable: {thread_id: thread-001}} app.invoke({messages: [你好]}, configconfig) app.invoke({messages: [你好]}, configconfig)这里的thread_id相当于一次会话的标识。同一个thread_id下的多次调用共享历史状态不同thread_id之间状态隔离。这个机制是构建聊天机器人、多轮 Agent 应用的基础。需要注意的是MemorySaver只是把状态保存在内存里进程重启后数据就会丢失。生产环境通常需要接入持久化存储LangGraph 官方有对应的 checkpoint 持久化方案。对于长期记忆场景最新版本中还引入了跨线程的 Store 机制适合保存用户偏好、实体关系等跨会话信息。这部分 API 仍在迭代中建议直接查阅最新的官方文档避免被版本差异误导。9. LangGraph 常见问题与排查思路在学习和实践过程中你会遇到一些重复率很高的问题。我整理了一张排查表覆盖了最常见的几类现象。问题现象可能原因排查方式解决方案运行时报找不到 START语言环境版本过旧检查 langgraph 版本升级到新版本改用新版 API条件路由函数返回值不在映射表中返回值与 key 不匹配在路由函数里打点输出返回值修正返回值或补充映射表 key节点函数修改 state 后没有生效直接修改了传入对象检查函数是否返回更新 dict改为返回 partial update并行分支某个字段数据丢失默认覆盖策略导致检查是否有多个分支写同字段为字段配置 Reducer循环流程报递归超限未设置 recursion_limit查看异常栈中循环链路配置合理的 recursion_limit多次 invoke 状态串了未使用 thread_id 隔离检查 config 是否传入为不同会话设置不同 thread_id子图内部状态不更新子图与父图 State 类型不一致检查子图和父图的 TypedDict 定义保持状态结构一致或做映射下面挑两个最常见的展开说一下。9.1 节点修改状态不生效很多初学者会写出这样的代码def bad_node(state: State): state[step] 1 # 没有 return这样写不会报错但 LangGraph 不会感知到任何状态变化因为节点函数没有返回更新。LangGraph 的约定是“通过返回值修改状态”而不是通过引用传参。正确写法是def good_node(state: State): return {step: state[step] 1}如果确实需要依赖上一次的状态做复杂计算也应该先读取传入 state再把计算结果通过返回值传给框架。9.2 条件路由异常条件路由报错时最常见的原因是路由函数返回的值在映射表里找不到。可以先用一段打印日志定位def route_after_think(state: AgentState): result finish if state[rounds] 3 else continue print(route result:, result) return result定位到返回值之后再去检查add_conditional_edges的映射表。这两个位置必须对应上少一个 key 都会出问题。10. 最佳实践与工程建议10.1 先画图再写代码这是最重要的一条建议。LangGraph 的代码结构本质上就是流程图的翻译。开始写代码之前先在纸上或白板上画出节点、边和条件路由确认每一步的状态流转。状态设计如果一开始就是乱的后面写再多节点也是在错误的地基上盖楼。10.2 节点函数保持单一职责每个节点只完成一个业务动作不要在一个节点里既调模型又做判断又改状态。判断逻辑放到路由函数里模型调用放到专门的节点里状态更新通过返回值声明。这样后续测试、日志追踪、替换实现都会容易很多。10.3 State 字段要克制State 是整个图的“共享内存”字段越多节点之间的隐式耦合越强。建议只放真正需要跨节点流转的数据把临时变量留在节点内部。字段命名要体现业务含义比如messages、rounds、retry_count。10.4 善用日志与追踪在本地调试时可以在每个节点入口打印一段日志记录当前 State 的关键字段。LangGraph 本身有对应的追踪能力但在入门阶段最简单的 print 也能帮你快速定位流程走到哪一步、状态变成什么样。10.5 提前设计 Checkpointer 方案即使当前项目不涉及多轮对话也建议在 compile 阶段就接入 checkpointer。后续要加记忆能力时只需要改配置不用重构节点逻辑。生产环境不要使用MemorySaver要选择持久化方案并设计好thread_id的生成规则。10.6 安全与权限提醒如果 LangGraph 流程中的某个节点会执行外部工具或访问数据库务必遵循最小权限原则。工具的调用权限、数据库账号权限都要严格控制避免 Agent 被提示注入攻击后执行危险操作。任何涉及删除、变更生产数据的节点在接入前都要做好备份、回滚和操作审批流程。10.7 测试策略对每个节点写独立的单元测试输入一个构造好的 State断言返回的更新是否符合预期。对整张图写集成测试验证主要条件路由能否走到正确分支。这样在调整路由逻辑时可以快速发现问题。11. 总结与后续学习方向把 LangGraph 学明白真正用好的关键点可以概括为三句话用图的方式思考流程用返回 dict 的方式修改状态用条件路由和子图控制复杂逻辑。入门阶段不需要一次性掌握所有 API先跑通线性流程再分别吃透条件路由和子图最后再研究持久化与并行执行这个路径是最平滑的。下一步你可以尝试一个综合练习构建一个“工具调用型 Agent”包含模型节点、工具节点、条件路由和循环控制。先从模拟函数开始不接真实模型跑通之后再替换成真实 LLM接入你实际业务里的工具。你会发现之前纠结的“多步骤流程怎么写才不烂”这个问题在 LangGraph 里会变得非常清晰。建议把本文收藏备用尤其当你准备把 Agent 从 Demo 推进到生产环境时第 9 节的问题排查表和最后几节最佳实践会经常用到。