很多人学 LangGraph 卡住的点其实不在 API 写法而在于思维方式没转过来。我也一样刚开始对着文档看 State、Node、Edge、Conditional Edge 这些概念第一反应是“这不就是个有向图框架吗”写个 hello world 跑通之后依然没有体感。直到后来拿一个真实的内部业务流程去重构才发现 LangGraph 真正逼你做的不是学会怎么调函数而是把一个模糊的流程变成一张精确的、可执行的、能随时停下来检查的图。这篇笔记我不打算复述官方文档而是围绕“建立工作流思维”这条主线展开为什么用图来组织 Agent 逻辑、核心概念怎么映射到业务、一个从零搭起来的工作流长什么样、以及我在实操中踩过哪些坑。适合两类人看一类是已经会调大模型接口、想往 Agent 方向走的开发者另一类是团队里负责把业务需求落地成技术方案的工程师你会发现它跟画流程图其实是同一件事只不过这张图真的能跑起来。1. 先理解LangGraph 真正让你改变的是什么1.1 从“单次问答”到“流程编排”的思维转变如果用最直白的话描述大模型应用开发大多数人的第一版代码长这样response llm.invoke(user_input) print(response)这没什么不对很多工具型产品一个 Prompt 就能解决。但真实业务不是这样的。拿我最近在做的技术支持工单处理来说完整的流程至少包含判断工单类型、检索内部知识库、生成回复草稿、交给人工审核、审核通过后归档不通过的还得退回修改。每一步之间不是简单的顺序关系而是有分支、有循环、有“人等机器”的节点。如果在普通 Python 代码里表达这个流程通常会是几种写法要么一个大函数里面套三层 if/else 和 for 循环要么单独抽象一个 Orchestrator 类用while加switch硬转。这两种方式在小规模时都能跑但一旦分支超过十几个你基本就失去了对流程的掌控——出问题时不知道卡在哪一步日志只能看到“报错了”状态散落在各个局部变量里测试更是无从下手。LangGraph 的解法是把整个流程显式建模成一张图每个处理环节是一个节点节点与节点之间用边表达走向所有环节共享一份全局状态。你的工作重心从“写一段顺序执行的函数”变成了“定义一张图 设计状态结构 写每个节点的逻辑”这就是工作流思维的起点。提示学 LangGraph 之前值得先问自己一句——我的业务到底有没有“流程”如果只是单轮问答、一次性生成确实没必要上框架。反过来只要是多步、有分支、需要中间判断的场景就值得用图来表达。1.2 为什么图结构是表达工作流更好的方式有一回我在给别人讲 LangGraph 的时候用了餐厅后厨的类比对方一下就通了单人大排档是一位师傅从头做到尾洗菜切菜炒菜全是自己适合菜品少、客流小的场景但是要开正经餐厅就必须把后厨拆成不同岗位配菜员只管切配炒锅只管烹饪传菜员把成品端出去收银员记每一单的账。这个类比能对应上 LangGraph 的所有核心概念。岗位就是节点每道工序有明确的输入和输出传菜路线就是边菜从哪个岗位送到哪个岗位是规定好的收银台账就是全局 State所有岗位都能往这张单子上记信息如果菜品卖完了服务员得临时决定推荐别的菜这就是条件边——根据现场状态决定下一步走哪条分支。这种表达方式最大的好处是拓扑即结构结构即文档。你把流程图画出来业务方哪怕不会代码也能对着图点头说对你把图画出来了开发也只是照着图把每个节点填上逻辑。相比之下几百行 if/else 嵌套的逻辑外人根本看不出整体结构只有你本人在维护期内还勉强记得住。LangGraph 把这种“先画图、再填逻辑”的开发方式落到了代码层面而且解决了一个关键痛点在执行过程中图引擎会维护一份全局状态任意节点只要声明了对某个字段的读写就能拿到其他节点产生的结果。这比在函数之间手动传参要省心得多尤其当流程深度超过三四层的时候参数列表会膨胀到根本不想维护。2. 工作流思维的最小知识体系五个核心概念2.1 State全局共享的便签本在 LangGraph 里State 是整个图的“共享内存”。它本质上是 Python 的类型结构官方推荐用TypedDict定义字段就是你要在节点之间传递的数据。from typing import TypedDict class ReviewState(TypedDict): proposal: str # 原始方案文本 has_sections: bool # 是否包含必要章节 feedback: str # 审核意见 decision: str # 最终结论每个节点函数接收当前 State 作为参数返回一个字典返回的键值对会更新到 State 上。节点不需要把整个 State 返回只需要返回自己产生的增量信息。这里有一个很重要的概念叫 Reducer。默认情况下多个节点往同一个字段写值后写的会覆盖先写的。这对大多数业务字段没问题但有一种场景不行如果需要累积、追加而不是覆盖比如对话消息列表、日志数组。LangGraph 允许用Annotated给字段挂一个合并函数from typing import Annotated from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages]add_messages会把新消息追加进原有列表而不是整个替换。这个细节是新手最容易踩的坑我在后面第 5 章专门展开讲。提示State 设计是整个图的灵魂。我在实操中会把“每个节点到底读写哪些字段”先写在纸上再动笔写代码。字段设计得合理节点之间的依赖关系就清晰设计得不好后面所有节点都会被迫塞进一堆无关字段。2.2 节点每一个环节只做一件事节点就是图中真正干活的函数。形式上它接收 State返回一个包含更新字段的字典。但思想上它应该被当成一个“岗位”来设计这个环节只负责一件事输入输出都明确定义。我见过不少人包括我自己初期把节点写得特别大一个节点里既查数据库、又调模型、又做业务判断结果一张图名义上有五六个节点实际逻辑跟一个大函数没区别条件边根本找不到合适的判断依据。后来我把设计原则改成节点粒度要小到“能被条件边引用”即每个节点最好产出 1-2 个关键字段并且这些字段是后续路由判断可以直接使用的。比如方案评审的例子check_sections这个节点就只做一件事——检查文本包含哪些章节输出一个布尔结果。有了这个布尔结果后面的条件边才能做出清晰路由。2.3 边与条件边控制走向的分拣员边就是图的路线LangGraph 里分为两种普通边表示无条件转移A 做完之后必须去 B条件边则是在节点执行完之后根据当前 State 的某些字段动态决定走哪个方向。条件边函数的写法很直接def route_after_check(state: ReviewState) - str: if state[has_sections]: return pass_node return revise_node builder.add_conditional_edges(check_node, route_after_check)只要函数返回的字符串恰好是图中的某个节点名LangGraph 就会把执行权交给那个节点。这种设计等于把“路由决策”从业务代码里抽出来单独变成一个纯函数特别好测试——给定 State断言返回值就行完全不需要跑整张图。有一条硬性经验条件边的返回值一定要能命中节点名或者显式传入一个{key: node_name}映射表。否则执行引擎会报“未知节点”或者直接挂起调试起来比较挠头。2.4 循环与终止让 Agent 学会自己决定下一步图之所以比线性链式结构强大核心在于循环是“一等公民”。从 2.2 节的例子继续如果判断“打回”之后用户修改了方案再回来提交流程就要回到检查节点而不是直接结束——这就需要一条从“打回”节点指回“检查”节点的边。LangGraph 里循环没有任何特殊语法你只要把边指回去就行。但要注意循环必须保证有明确的终止条件否则图会无限跑下去。我带的两个判断维度是状态字段是否达到目标值比如“人工确认通过”、更新次数是否达到上限比如最多打回两次就强制转人工。后面 4.1 节的 ReAct 模式就是循环的典型应用Agent 节点决定“要不要调用工具”条件边根据最后一条消息的内容判断如果要调用就进入工具节点工具执行完再回到 Agent 节点直到模型判断不再需要工具为止。2.5 持久化与 Checkpoint一切可恢复工作流思维里很重要的一环是“随时能停下随时能恢复”。LangGraph 的持久化机制用 Checkpointer 实现每次节点执行完整个图的快照都会保存下来。from langgraph.checkpoint.memory import MemorySaver checkpointer MemorySaver() graph builder.compile(checkpointercheckpointer)加了这个参数之后你可以用thread_id管理一次完整执行中途挂了可以恢复需要人工介入的时候可以等待、断点后继续跑。生产环境可以换用基于数据库的 Checkpointer支持跨进程恢复。我的体会是不要等到上生产才考虑这件事最初设计状态结构的第一天就把thread_id作为运行维度考虑进去。后面接人工审核、做异常恢复、回看历史记录全都依赖这个机制省了非常多事。3. 动手实操从 0 搭一个“方案评审”工作流3.1 业务场景拆解先把图画出来我选一个很常见的团队内部场景技术方案评审。日常操作是一个成员写方案文档评审人看完后给出结论“通过”或“打回”。现在要做一个小工具来自动预审先检查方案是否包含必要章节背景、方案、风险章节齐全的送人工终审不齐全的直接打回并给出修改建议。动手之前我习惯在纸上画流程提交方案 - 检查章节完整性 - 不齐全打回附修改建议- 结束 - 齐全通过预审 - 结束这张图很朴素只有一条条件边和两个终止分支但已经能说明问题。后面第 4 章会在这个基础上加循环、加人工介入、加并行它会自然长成一棵更完整的流程树而图的表达力始终跟得上。3.2 状态 Schema 设计明确流动的数据基于流程图我需要哪些字段就一目了然了proposal最原始的方案文本检查节点要读has_sections检查节点的输出条件边要读feedback打回时给用户的修改建议decision最终结论外部调用方关心的结果from typing import TypedDict class ReviewState(TypedDict): proposal: str has_sections: bool feedback: str decision: str这里我刻意不把中间过程塞进状态里状态里的字段都是“最终会被某个节点使用”的。有些中间计算结果如果只是临时用一下就不占状态字段直接在节点函数内部消化。3.3 节点实现写出每个环节的逻辑三个节点的实现都不复杂。检查节点做关键词判断为了让示例跑起来不需要 API Key我暂时用文本包含检测代替大模型语义判断等第 4 章再升级成真正的 LLM 节点。def check_sections(state: ReviewState) - dict: proposal state[proposal] required [背景, 方案, 风险] has_sections all(section in proposal for section in required) return {has_sections: has_sections} def pass_review(state: ReviewState) - dict: return { decision: 通过, feedback: 章节完整预审通过等待人工终审。 } def request_revision(state: ReviewState) - dict: return { decision: 打回, feedback: 缺少必要章节请补充背景说明、方案描述或风险评估后重新提交。 }注意看每个节点返回的字典只包含与当前职责相关的字段没有多余内容。这保证了后续条件边的判断来源单一、清晰。3.4 连接成图编译与运行图构建的核心代码就是定义节点、定义边、定义条件路由from langgraph.graph import StateGraph, START, END def route_after_check(state: ReviewState) - str: if state[has_sections]: return pass_node return revise_node builder StateGraph(ReviewState) builder.add_node(check_node, check_sections) builder.add_node(pass_node, pass_review) builder.add_node(revise_node, request_revision) builder.add_edge(START, check_node) builder.add_conditional_edges(check_node, route_after_check) builder.add_edge(pass_node, END) builder.add_edge(revise_node, END) graph builder.compile()这里用的 API 是 LangGraph 0.2 的风格习惯用START常量代替旧版的set_entry_point()。add_conditional_edges只传两个参数时LangGraph 默认把路由函数返回的字符串当作目标节点名所以route_after_check的返回值必须是pass_node或revise_node这种精确的节点标识。运行整张图的方法很直观和调用函数几乎一样result graph.invoke({ proposal: 本方案包含背景说明、技术方案、风险评估。, }) print(result[decision]) # 输出通过如果把proposal里的“风险评估”删掉第二次运行就会得到“打回”。同一个图结构完全由状态驱动不同的走向这就是工作流思维的基本形态。3.5 把图画出来可视化与调参的几个心得LangGraph 内置了导出流程图的接口开发调试的时候把图导出来看比对着代码脑补要高效得多。# 打印 ASCII 形式的图终端里直接看 graph.get_graph().print() # 或者导出图片文件 graph.get_graph().draw_mermaid_png(output_file_pathreview_graph.png)不需要额外装什么可视化库一个文件路径就够。我每次调图结构都会先导出一张图肉眼检查拓扑结构是不是符合预期——尤其注意是不是所有节点都能从 START 到达、所有路径最终都通向 END。一张图里如果存在永远走不到的分支或者某一环忘记画回边看代码很难发现看图一眼就能看出来。关于节点命名我也有一条经验节点名不要用node1、node2这种无意义编号直接用业务动作命名比如check_node、pass_node、revise_node。这直接影响两个体验——条件边路由函数的可读性以及报错日志的可读性。等到图里面有十几个节点的时候VSCode 断点调试看调用栈见名知意的节点名能帮你少掉一半头发。4. 进阶把工作流思维用在更复杂的真实场景4.1 ReAct 循环手把手搭一个会用工具的 Agent从上面这个线性图再往前一步就是循环和 Agent。先想一个问题为什么while True if也能写 ReAct但写出来的东西很难维护因为循环的三要素——继续条件、退出条件、状态累积——都散落在代码各个角落。而 LangGraph 把循环的各个要素显式拆给了不同的角色。class AgentState(TypedDict): messages: Annotated[list, add_messages] def agent_node(state: AgentState) - dict: response model.invoke(state[messages]) return {messages: [response]} def tools_node(state: AgentState) - dict: tool_calls state[messages][-1].tool_calls results [execute_tool(tool) for tool in tool_calls] return {messages: results} def should_continue(state: AgentState) - str: last_message state[messages][-1] if last_message.tool_calls: return tools_node return END graph StateGraph(AgentState) graph.add_node(agent_node, agent_node) graph.add_node(tools_node, tools_node) graph.add_edge(START, agent_node) graph.add_conditional_edges(agent_node, should_continue) graph.add_edge(tools_node, agent_node)这个模式就是很多 Agent 框架的底层架构区别在于 LangGraph 让你完全控制循环的每一步。should_continue这个条件边就是循环的终止判断模型说还要用工具就继续跑工具模型说不用了就直接到 END。每一次工具返回的新消息通过add_messages追加进消息列表作为模型下一轮的上下文。我强烈建议把should_continue单独提炼成纯函数并伴以单元测试因为它决定了 Agent 会不会死循环是整个系统里最值得保护的一行逻辑。4.2 人工介入关键节点停下来等人拍板全自动的流程永远需要留一个人肉卡口尤其是涉及对外发送消息、财务操作、内容发布这类高风险动作。LangGraph 提供了interrupt机制能在节点内部暂停整个图把控制权还给外部程序。from langgraph.types import interrupt def human_review(state: ReviewState) - dict: decision interrupt({ question: 是否批准该方案, proposal_snippet: state[proposal][:200], }) # 外部程序通过 Command(resume...) 传回人工决策 return {decision: decision}图执行到这个节点时会停下来保存所有状态。外部服务通过 API 把人工意见传回来图再从断点继续跑。我不喜欢把人工动作放在图外“假想存在”因为只要系统里有一步是要等真人的你就必须把它当做一个正式节点建模状态里记录“等待中”条件边判断“已处理”。这种显式建模有一个额外好处所有 Agent 的运行日志都完整保留审计时能回答“为什么这件事走到了这一步”。对 B 端产品来说这个能力几乎是必须的。4.3 并行执行让多个节点同时干活有些流程天然是扇出扇入的一个方案提交后需要同时给架构评审、安全评审、成本评审三个不同的节点一起看。LangGraph 的SendAPI 支持动态创建多个并行分支。from langgraph.types import Send def fan_out(state: ReviewState): reviewers [architecture, security, cost] return [Send(review_node, {reviewer: r, proposal: state[proposal]}) for r in reviewers] def review_node(state): reviewer state[reviewer] result llm.invoke(f作为{reviewer}请用一条意见评审{state[proposal]}) return {comments: {reviewer: result}} def fan_in(state): comments state[comments] # 汇总所有意见 return {summary: summarize(comments)}LangGraph 会为每个Send创建独立的分支任务都完成后统一汇入下一个节点。这种模式对应大数据里的 Map-Reduce 思想把一个大任务拆成多个同构小任务并行执行再把结果合并。用图来表达并行和聚合变成了结构的一部分而不是藏在并发代码里的隐式逻辑。4.4 子图把复用的能力打包成独立模块流程一旦多起来就会发现几个业务共用同一个子流程比如“预审打回”这套逻辑既出现在方案评审里也出现在文档发布审核里。LangGraph 的compile()产物可以当作一个普通节点嵌到更大的图里这就是子图。review_subgraph builder.compile() def ask_plan(state): plan llm.invoke(...) return {proposal: plan} def dispatch_review(state): return {review_result: review_subgraph.invoke(state)} big_graph StateGraph(OverallState) big_graph.add_node(plan_node, ask_plan) big_graph.add_node(review_node, dispatch_review) big_graph.add_edge(START, plan_node) big_graph.add_edge(plan_node, review_node) big_graph.add_edge(review_node, END)我建议子图的边界画在“对外输入输出稳定”的地方。子图内部怎么改只要输入输出结构不变外面的图就不用动。这跟代码函数抽象的收益一致但层级更清晰大图讲业务流程子图讲操作细节互不干扰。5. 从踩坑中学到的常见问题速查与调试技巧5.1 最容易让人卡住的 5 类问题实录下面这张表我是翻着自己过去的排错记录整理的基本覆盖了新手会遇到的八成情况现象根因解决思路节点函数返回了状态里没有的字段TypedDict 限制了合法字段检查键名拼写或先定义字段再使用条件边报错“Unknown node”路由函数返回值不是节点名打印返回值确认和add_node名称一致图一直跑停不下来循环缺少终止条件在 State 里加计数或强化条件边判断消息列表越跑越短messages字段没有挂 reducer改用Annotated[list, add_messages]同一次执行两次结果串了没传thread_id每次执行指定唯一的thread_id第五个问题在生产环境特别隐蔽多个请求共用同一个执行上下文状态互相覆盖。这是典型的“本地跑没问题、上线一并发就错”的怪现象。尤其要注意 Checkpointer 启用之后thread_id就是隔离维度的唯一依据。5.2 条件边写错导致无限循环这是我在本地跑 ReAct 模式时最常踩的坑。should_continue一开始写成def should_continue(state): if state[messages][-1].tool_calls: return tools_node return agent_node # 这里写错了应该是 END结果模型一次工具调用之后不管有没有新的工具请求循环都回到 agent_node 重新跑。表现就是终端里日志刷屏token 消耗飞起。排查方法是用graph.stream()逐步观察节点访问顺序很快就看到 agent_node 被反复执行。经验是循环的退出判断永远放在条件边函数里并且以“明确的最终节点”作为返回值。每次改完条件边我都顺手写一条测试用例覆盖“还需继续”和“到此为止”两条路径比任何运行时检查都靠得住。5.3 状态里的上下文越滚越大Token 爆炸用add_messages配合 ReAct 跑几十轮之后消息列表会非常长一方面 token 成本上涨另一方面模型响应质量会下降。这个问题的本质是“状态设计”里没有做上下文管理而不是框架的问题。我的做法是在循环里加一个“压缩节点”每隔几轮把旧消息摘要拼接到最新一轮对话前。这个节点属于“基础设施”放在工具节点和 agent 节点之间用条件边判断本轮是否触发压缩。实际操作中不需要追求高深的压缩策略一个“请总结之前对话的关键事实”的 Prompt 就够日常使用。提示把上下文管理做成图里的一个节点而不是在模型调用处手动截断好处是可以复用、可测试、可观测。每次压缩后消息列表的变化都会体现在状态快照里查问题的时候能看得很清楚。5.4 调试手段与本地可视化我发现很多人不知道graph.stream()这个调试神器它比invoke()好用太多。它逐节点输出执行结果能清楚看到每个节点产生了什么、状态变成了什么。for step in graph.stream(initial_state, stream_modeupdates): print(step)每一轮输出是一个字典键是节点名值是节点返回的增量字段。对着输出看一遍流程走向比打断点还要直观。另一个手段是graph.get_graph().print()在控制台打印 ASCII 流程图用来核对拓扑结构。这两种方式组合起来基本能覆盖 95% 的调试场景。5.5 一个容易被忽略的测试策略先单测节点再测整图工作流思维最容易忽视的一环是测试。图结构本身就不适合只在“集成层”测试因为分支组合数量爆炸。我的习惯分两层做节点级测试每个节点是纯函数给定输入 State 对象断言输出字典的字段和值。特别是条件边函数单独测试每一种路由返回。不需要跑整张图速度极快。图级集成测试用几个代表性的 State 跑graph.invoke()覆盖每条主要分支路径。比如方案评审图至少要覆盖“通过”“打回”“完整跑一遍带循环”三种情况。这样做下来改动节点逻辑时回归成本极低而且通过看哪一层测试挂了能非常快地定位问题。LangGraph 的节点函数设计天然适合这种测试方式——它不依赖外部单例输入输出都显式可构造写测试几乎没有阻碍。最后分享一点个人体会。我最近用 LangGraph 重构内部工具的时候最明显的变化不是性能变快了而是终于知道程序在执行过程中的任何一刻走到哪了。以前看一堆日志猜流程现在打开图就知道每个节点在干什么。如果你正准备学 LangGraph我建议第一步不是装包而是拿一张纸把你要实现的业务流程老老实实画出来。这个习惯一旦建立后面的框架学习就会顺畅很多。