LangGraph核心模块拆解：状态机与条件分支构建AI Agent工作流

如果你最近在折腾AI Agent大概率绕不开LangGraph这个名字。我第一次看到它时心里想的是又来了一个花架子框架但把它拆开以后发现它做的事情其实特别朴实——把你脑子里的“流程状态机”变成可以运行的代码。和普通链式调用不一样LangGraph把每一步都当成节点把节点间的跳转规则当成边状态则交给一个全局对象管理。也就是说你可以在任意一步停下来查看状态、改变方向、甚至回溯重跑。这篇文章我打算直接从核心模块入手用一个完整的小项目把代码讲透。适合已经跑过LangChain、对Agent有点概念、但还没系统看过LangGraph的读者。我先保证你看完能上手写一个带条件分支和状态记忆的图然后我们再聊持久化、人工介入这些进阶玩法。看完以后你至少不会再觉得它是个黑盒。1. LangGraph到底在解决什么问题1.1 从普通链式调用到有状态编排先想想以前我们怎么写AI流程。最朴素的做法是链式调用llm.invoke(prompt1) - LLM结果 - 拼到prompt2 - llm.invoke(prompt2) - ...这种写法的问题在于流程是固定的、线性的一旦中间需要根据结果改变路径代码就变得非常难维护。比如“如果用户情绪是愤怒就转人工否则自动回复”你会发现自己写了一堆if else还把每一步的临时变量散落在各处。LangGraph把这个事情抽象成一张图。每个处理步骤是一个节点节点之间的跳转关系是边。所有的中间结果都写进一个全局状态对象里谁都能读谁都可以改。这样一来流程可以用图的方式直接画出来和产品/团队交流成本低。分支、循环、回退都是图的拓扑关系而不是散落的逻辑。运行过程中状态可持久化相当于给工作流加了“存档”。我用一个生活化的类比普通流程就是一条固定传送带零件从A到B到C中间没有岔路LangGraph是一个带分拣口的流水线传送带可以变轨、可以回流每个包裹上的标签就是状态系统靠标签决定下一个该去哪。这个类比我一直跟别人讲基本都能秒懂。1.2 核心设计理念节点、边、状态机LangGraph的底层核心就三个概念State状态一个数据结构保存整个流程运行时需要的信息。Node节点一个Python函数接收State返回对State的增量更新。Edge边从哪个节点走到哪个节点的规则。其中最容易忽略的是State。它不是普通的dict而是带类型的、可叠加更新的对象。LangGraph允许每个节点只返回需要修改的字段剩下的字段自动保留。这个设计让代码非常干净你不需要手动传递一堆上下文。再往下走条件边Conditional Edges让图有“智能”。一个节点返回结果后图会根据状态里的某个字段决定下一步走哪条边。我觉得理解了这三样东西外加一个编译过程langgraph的核心就已经拿下一大半了。2. 核心模块逐个拆解附代码直击2.1 StateGraph整个工作流的骨架StateGraph是LangGraph的主类所有模块都围绕它组织。先看一段最基础的代码from typing import TypedDict from langgraph.graph import StateGraph class MyState(TypedDict): messages: list user_name: str def node_a(state: MyState) - dict: # 读取state返回需要更新的字段 return {user_name: 张三} workflow StateGraph(MyState) workflow.add_node(a, node_a) workflow.set_entry_point(a) workflow.set_finish_point(a) app workflow.compile()这里有几个关键点StateGraph的泛型参数是状态结构不是节点很多新手看示例会误以为StateGraph(MyState)是传入某个内置类型。add_node给节点起名字名字可以随意但建议用有业务含义的词。set_entry_point和set_finish_point是出入口。不能只有start没有end否则编译时会直接报错。compile()返回一个可调用的对象它才是真正用来执行工作流的入口。我实测下来StateGraph本身没什么魔法它做的事情就是把节点和边校验一遍生成一个可执行的编译图。真正决定行为的是State和节点里的逻辑。2.2 State不可变状态与ReducerState定义的是整张图的“公共记忆”。LangGraph推荐用TypedDict定义好处是IDE能自动补全字段结构也非常直观。不过麻烦的地方在于更新规则。默认情况下如果两个节点都返回同一个字段后执行的节点会覆盖先执行的值。这通常在链式流程里没问题但如果你希望“累加”或者“合并”消息列表就要用到Reducer。from typing import Annotated, TypedDict from langgraph.graph import add_messages class ChatState(TypedDict): messages: Annotated[list, add_messages] user_id: strAnnotated的含义是这个字段的更新不走默认覆盖而是走add_messages这个函数。add_messages会把新的消息追加到原有列表末尾而不是整体替换。我开发时遇到过一个问题在一个并行节点里两个分支都返回messages字段如果没有Reducer状态就会变成最后执行分支的消息前一个分支的结果直接丢掉了。后来我统一给列表字段加上Reducer才彻底解决。所以建议所有列表类型的字段都显式配置Reducer这不会带来额外成本但能少踩很多坑。2.3 Node与Edge运算单元和路线规则节点函数是LangGraph里的最小运算单元。它接收当前State返回一个字典字典里的字段会被合并到State中。核心写法如下def process(state: MyState) - dict: new_messages state[messages] [{role: assistant, content: 处理完成}] return {messages: new_messages}注意节点函数不要返回None也不要返回一个包含所有字段的大字典。最佳实践是只返回你真正改动过的字段这样代码更清晰也能减少潜在的冲突。边的类型分两种普通边从一个节点无条件走到另一个节点。条件边根据状态或路由函数的输出动态决定下一步。普通边的代码workflow.add_edge(a, b) workflow.set_entry_point(a) workflow.set_finish_point(b)如果流程是线性的这种写法非常简单。但如果分支多了我建议不要每个分支都手动连边而是用条件边集中管理。2.4 Conditional Edges让流程自己拿主意条件边是LangGraph里最体现“智能”的部分。它的路由函数不是直接返回下一个节点的名字而是返回一个“键值”再把键值映射到节点。这样可以避免路由函数和具体节点名耦合。看一个示例def route_after_check(state: MyState) - str: if state[sentiment] negative: return human return auto_reply workflow.add_conditional_edges( classify, route_after_check, { human: human_handler, auto_reply: auto_reply_node, } )这个写法的好处是route_after_check只返回业务含义的字符串比如“human”“auto_reply”具体哪个节点处理这个字符串由后面的字典统一管理。以后改节点名字只需要改映射不需要动路由函数。我一开始习惯直接在路由函数里返回节点名比如return human_handler。短期没问题但图上节点一多路由函数反复修改很容易不小心把名字写错。后来全部改成语义化返回维护成本低很多。条件边完全可以实现循环效果让一个节点处理完以后路由回更早的节点。这就解决了“如果LLM输出不合法就重试”这类需求。要注意的是循环必须设置一个最大次数否则可能无限跑下去。3. 一个完整的可复现示例智能客服工单处理3.1 需求场景与状态设计我拿一个常见的场景来说智能客服工单分类与回复。用户提交一条问题系统先做意图和情绪识别如果用户情绪负面或者问题属于售后类就转人工否则直接用自动回复模板。这个场景很适合用LangGraph演示因为它同时用到了条件分支和状态持久化。我们定义状态如下from typing import Annotated, TypedDict from langgraph.graph import add_messages class TicketState(TypedDict): messages: Annotated[list, add_messages] category: str sentiment: str need_human: bool ticket_id: str final_answer: strmessages保存对话原文用add_messages做累积。category问题分类比如“技术咨询”“售后”“建议”。sentiment情绪标签。need_human是否转人工由分类节点计算后写入。ticket_id工单号用来模拟业务标识。final_answer最终回复内容。状态里既有简单字段又有列表字段这样能同时展示Reducer和覆盖更新的区别。设计状态的时候我的经验是先写下流程里所有节点需要读写的字段再开始写节点函数不要边写边加。3.2 节点函数与分支逻辑实现先做几个工具函数模拟LLM的输出def classify_and_analyze(state: TicketState) - dict: user_text state[messages][-1][content] # 实际项目里这里会调用LLM或专门的分类模型 if 退款 in user_text or 投诉 in user_text: category 售后 sentiment negative elif 怎么用 in user_text or 安装 in user_text: category 技术咨询 sentiment neutral else: category 其他 sentiment positive need_human category 售后 or sentiment negative return { category: category, sentiment: sentiment, need_human: need_human, }这个节点读取状态里的最后一条消息给出分类、情绪和是否转人工的判断。返回的字段会被写入全局状态供后面的条件边使用。为了让演示简单我没有真正调大模型只用了关键词判断改成LLM调用时逻辑是一样的只是返回之前加一个llm.invoke()。接着写两个下游节点一个是自动回复一个是转人工。def auto_reply(state: TicketState) - dict: reply f【{state[category]}】您好这是一条自动回复。您的工单号是: {state[ticket_id]} return {final_answer: reply, messages: [{role: assistant, content: reply}]} def human_handle(state: TicketState) - dict: reply f【{state[category]}】已将工单 {state[ticket_id]} 转给人工客服请稍候。 return {final_answer: reply, messages: [{role: assistant, content: reply}]}自动回复和人工接待的逻辑都很简单就是读取状态、生成回复、把回复写回状态。注意消息字段用了Reducer所以回传消息时用列表追加语义如果没用Reducer这条回传语句会覆盖之前的messages那就出问题了。条件路由函数def decide_next(state: TicketState) - str: if state[need_human]: return human return auto3.3 编译与调用全流程把节点和图组合起来from langgraph.graph import StateGraph workflow StateGraph(TicketState) workflow.add_node(classify, classify_and_analyze) workflow.add_node(auto_reply, auto_reply) workflow.add_node(human_handle, human_handle) workflow.set_entry_point(classify) workflow.add_conditional_edges( classify, decide_next, { human: human_handle, auto: auto_reply, } ) workflow.add_edge(auto_reply, __end__) workflow.add_edge(human_handle, __end__) app workflow.compile()注意__end__是一个特殊节点名称表示流程结束。LangGraph要求从set_entry_point进入的每个分支最终都要能到达__end__否则编译或运行时可能报错。如果你不想显式连终点边也可以用set_finish_point(auto_reply)指定。调用它initial_state { messages: [{role: user, content: 我要退款你们的服务太差了}], ticket_id: T20250001, } result app.invoke(initial_state) print(result[final_answer])运行后会走到human_handle分支因为消息里包含“退款”和“投诉”sentiment被标记为negativeneed_human为True。参数说明invoke会返回最终完整状态而不是只返回输出。所以你可以直接result[final_answer]也能看到category、sentiment等中间字段。这个特性很实用调试时可以直接把result打出来看整条链路的状态变化省去到处打日志的麻烦。4. 进阶玩法持久化、人工介入与并行4.1 Checkpointer断点续跑和多轮对话前面那个例子图运行完状态就丢了。现实场景里我们往往需要保留状态比如支持多轮对话、允许某次运行中断后再继续。LangGraph用Checkpointer解决这个问题。from langgraph.checkpoint.memory import MemorySaver checkpointer MemorySaver() app workflow.compile(checkpointercheckpointer) config {configurable: {thread_id: thread-1}} result1 app.invoke(initial_state, config) result2 app.invoke({messages: [{role: user, content: 那具体多久能解决}]}, config)关键就两点在compile时传入checkpointer。每次invoke都带上同一个thread_id。带thread_id之后LangGraph会把每一次运行的状态快照保存在检查点里。第二次invoke时它能从同一个thread的旧状态继续而不是从零开始。我在做多轮客服对话时这个功能几乎是必需的否则每轮都要把历史消息手动拼进去。MemorySaver只适合本地调试进程一重启数据就没了。生产环境建议用数据库类的持久化实现LangGraph也提供了基于PostgreSQL等适配器的checkpointer只是我不推荐在初学阶段就把存储搞复杂先用内存版跑通再换存储实现。4.2 人工介入interrupt与resume客服系统里有一种经典需求自动流程走到某一步时停下来等人工确认或填写信息然后继续跑。LangGraph的interrupt机制专门做这件事。from langgraph.types import interrupt def human_approve(state: TicketState) - dict: approval interrupt({ type: 需要人工确认, content: state[messages][-1][content], }) return {need_human: approval.get(approved, False)}当图执行到这个节点时遇到interrupt会抛出一个中断信号把运行状态保存下来然后暂停。之后你通过外部程序把用户的确认结果传回同一个thread图就会从刚才中断的地方继续执行。app.invoke(initial_state, config) # 图执行到interrupt时暂停 resume_command {approved: True} app.invoke(Command(resumeresume_command), config)这个概念理解成“工作流自动暂停等人工回话”就好。它的价值在于把人工确认这个动作变成了图里的一个节点而不是靠外部回调绕过核心流程。我第一次用这个能力时很兴奋因为它直接把我以前的“sleep轮询数据库”方案替换掉了。4.3 并行节点与Map-Reduce模式处理多个工单或者同时跑多个检索任务时线性执行效率太低。LangGraph支持通过Send API动态创建并行分支。from langgraph.constants import Send def continue_to_support_agents(state: TicketState): return [ Send(agent_worker, {ticket_id: t[id], description: t[desc]}) for t in state[sub_tickets] ] workflow.add_conditional_edges( dispatch, continue_to_support_agents, [agent_worker] )Send的含义不是调用一个节点而是把传入的数据作为子状态启动一个新的“agent_worker”节点实例。每个子任务可以独立运行也可以有自己的内部图路径。我做批量工单分派时还会配合Map-Reduce模式先把一批任务分发给多个worker节点各跑各的最后用一个“汇总”节点把结果合并回总状态。这里的汇总节点跟普通节点没什么区别只是需要处理列表类型的返回结果。注意要给列表字段配好Reducer不然多个并行分支返回时会互相覆盖。并行不是越多越好特别是每个worker都要调LLM时并发太高会导致接口限流。我的习惯是先限制并发数比如一次最多处理5个子任务再逐步调大。5. 踩坑实录与排查技巧5.1 高频报错与对策我在本地和项目里跑LangGraph时遇到过不少问题挑几个典型的列出来报错或现象原因解决办法TypeError: NoneType object is not iterable节点函数返回了None或字段被覆盖成了None确保节点函数始终返回dict没有更新就不要返回这个字段并行结果互相覆盖列表字段没有配置Reducer给列表字段加Annotated[list, add_messages]或者自定义merge函数ValueError: Expected node xxx not found条件边映射里写了不存在的节点名检查add_conditional_edges第三个参数的所有目标是否执行过add_node多轮对话历史不增长第二次invoke没带thread_id或没配checkpointer编译时传checkpointerinvoke时带上thread_id图执行卡死条件边形成无终止条件的循环在循环路径上增加次数计数器达到上限强制结束Invalid update返回了State里不存在的字段检查节点函数返回的key是否都定义在State中5.2 几个建议和最佳实践先说说状态设计。State字段越少越好不要把大段临时内容塞进State。有些中间计算结果只在当前节点用那就不要写到State里用节点内部局部变量处理。不然调试时会有大量干扰噪声而且还容易触发Reducer冲突。再说节点函数风格。我建议所有节点函数都写成“纯函数”式不对State做隐式修改只返回更新。这样单元测试也很好写你用假State调一个节点断言返回的dict对不对就行。我一直把节点当独立函数测试最后再组合成图整体bug率明显下降。还有条件边路由函数务必保持轻量。它会跟随每次状态更新被调用如果里面有重IO或慢计算整个流程会被拖慢。需要复杂判断时应提前在一个普通节点里计算好结果把结果放在State里路由函数只做读操作。最后关于调试日志。LangGraph运行时的图信息可以把这个打开from langgraph.pregel import PregelOptions options PregelOptions(debugTrue) result app.invoke(initial_state, configoptions)实测能看到每个节点的输入状态输出状态比自己在节点里print干净得多。不过生产环境不建议一直开着日志量有点大。我在实际项目中有一个小习惯每个节点函数内部都会记录一句日志格式是“当前节点名输入状态关键字段输出状态关键字段”。刚开始觉得麻烦后来排查线上问题全靠这些日志去还原链路。LangGraph的优势就在于状态流清晰日志能把每一步的状态变化串起来定位问题非常快。再提一个容易忽略的细节LangGraph的State字段命名不要用太通用的单词比如result、data因为多个节点都可能返回这些字段一旦字段冲突排查成本很高。我吃过一次亏一个节点的data是列表另一个节点的data是字符串运行时报错后查了半天才明白是字段被覆盖了。现在我都用带有业务含义的字段名比如sub_ticket_list、final_reply_text。如果你只是想在单机脚本里快速验证逻辑用MemorySaver就够了。但如果你准备把这个东西接到一个真实服务里一定要提前想好checkpointer与业务系统的关系。每个thread_id对应一段独立会话所以thread_id最好用业务侧的会话ID或用户ID这样即使重启服务也能通过同样的thread_id恢复上下文。LangGraph不是一个需要背很多API才能上手的框架它的核心就是“状态在哪、节点干什么、边怎么走”。把这三点搞清楚剩下都是技巧问题。希望这篇拆解能让你少走一点弯路至少遇到报错时能第一时间去查State的定义和Reducer配置而不是对着整个框架发呆。