最近一段时间不少做AI应用的朋友跑来问我LangGraph到底是不是那个“必学框架”跟以前写的Agent脚本到底差在哪。说实话光靠概念解释很容易把人绕晕真正能讲清楚的还是拿一个场景从头到尾做一遍。这篇文章我会从最简单的LangGraph状态图开始用AI Agent应用开发的思路一步步把它扩展成一个能应对真实业务的智能客服系统再往上一层做成多Agent协作架构。适合已经写过一点Agent脚本、但还没有系统理解状态流转、分支路由和持久化设计的开发者。读完你至少能手写一个带工具调用的客服Agent独立设计一个多Agent协作的Supervisor结构知道生产环境里怎么加人工介入、怎么做会话持久化也大概清楚哪些坑能绕开。1. 为什么是LangGraphAgent开发的核心痛点与方案选择1.1 从脚本到Agent状态管理为什么是硬骨头很多人最开始写Agent其实就是一段while循环模型在循环里决定要不要调用工具调完把结果塞回去直到模型给出最终回答。这个写法二三十行就能跑通Demo阶段非常舒服但问题也在它身上。一旦场景切到真实业务循环脚本就开始变形了。拿一个客服机器人来说用户问了一句“订单还没发货”Agent要先查订单系统的状态再判断该走“催发货”还是“退款流程”中途可能还需要跟用户确认收货地址确认完之后又得回到原来的对话上下文继续处理。这时候你会发现自己得手动维护大量的临时状态历史消息列表、工具调用中间结果、当前分支处于哪一步、用户上一轮填的订单号……这些状态散落在循环各处的局部变量里一旦流程分叉多了几乎没法复盘。更麻烦的是这种状态很难序列化保存。客服系统需要支持用户隔了几个小时回来继续聊需要能够接回上一次的会话现场。你当然可以自己写一个全局缓存把所有变量手动塞进去再取出来但每加一个新分支就得改一遍存取逻辑维护成本会越来越高。LangGraph的解决思路是把这类问题重新定义成“图遍历”问题整个Agent流程是一张有向图节点就是处理函数边代表跳转规则状态则是贯穿整个图执行的共享对象。节点函数只负责“接收当前状态、返回状态增量”图框架负责把状态按顺序、按分支送进不同的节点并在每次节点执行之后把返回值合并进全局状态里。这个设计的价值在于完整执行轨迹、状态快照、每一步的分支决策全部变成可记录、可恢复的数据。对需要“翻旧账”的线上系统来说这一点比把代码写得再漂亮都更重要。1.2 LangGraph的关键设计节点、边、State与CheckpointerLangGraph最核心的四个概念我得用自己的理解给它们拆开说。节点Node就是一个普通函数输入是整个状态对象输出是一个字典表示要更新哪些字段。比如一个“查询订单”节点接收状态里的订单号返回一个“订单状态”字段。因为节点之间不共享任何局部变量所以每个节点都可以单独测试这是它比脚本循环优雅的根本原因。边Edge分成两种普通边固定把流程从一个节点带到下一个节点条件边则根据状态里某个字段的值动态决定下一个节点往哪走。比如根据意图字段走“售前咨询”还是“售后投诉”。图上还有START和END两个特殊节点标注流程从哪开始、在哪结束。状态State是用TypedDict定义的字段集合所有节点共享。字段可以挂Reducer最常用的Reducer是operator.add它的作用是告诉框架当两个节点同时返回同一个字段时是直接覆盖还是追加合并。这个设计特别适合消息列表多个节点产生的消息会自动追加不会互相覆盖。Checkpointer是持久化组件它会在每次节点执行结束后把状态快照保存下来并绑定一个thread_id。有了它用户关掉页面再回来系统可以拉取同一个thread_id的历史状态把对话“续起来”。这也是LangGraph做多轮记忆的官方姿势。如果拿流程图来类比StateGraph是一张流程图节点是流程框边是带箭头的连线状态是流动的工单Checkpointer则是给每个工单拍照片存档的相机。模型每一步的决策体现在条件边上业务动作体现在节点上现场复原靠照片。1.3 什么时候才需要用LangGraph这一点我必须站在实际工程角度多说几句不是所有Agent都值得上LangGraph。如果只是一个模型加两三个工具的循环输出直接返回给用户没有多轮上下文也不涉及复杂分支那用普通脚本完全够引入图结构反而会增加心智负担。但当你出现下面这些信号时就可以认真考虑迁移了一是多轮记忆开始被迫自己写缓存而且缓存逻辑已经被改到第三版。二是分支判断越来越多if-else嵌套到你自己都不想维护。三是有流程需要人工审批、重试、断点续跑。四是多个Agent要协作并且需要共享同一份会话上下文。五是线上出了问题你希望有一份完整的状态历史可以直接回放而不是让用户重新复现一遍。我自己判断的标准很简单当Agent开始变成一个“需要被运营和排障的系统”而不是“一个跑完就结束的脚本”时就该上图形化管理了。2. 环境准备与第一个可运行的Agent2.1 起步依赖先把环境收拾干净开始动手前先准备一个干净的Python环境。推荐用虚拟环境避免把LangGraph的依赖装到你机器上的其他项目里。Python版本建议3.10以上3.11、3.12的兼容性表现都比较稳。python -m venv .venv source .venv/bin/activate pip install --upgrade pip pip install langgraph langchain-core解释一下这两个包的关系langgraph是图编排框架本体langchain-core提供的是消息类型、工具定义这类基础类型两者配合使用。模型调用部分LangGraph本身不绑定任何具体服务商你需要按照自己实际使用的模型SDK去安装对应的包然后封装出一个“输入消息列表、返回AI消息”的函数接入图里。下面所有示例我都会用占位函数get_chat_model()来表示这一步你在本地替换成自己能跑通的模型接口就行。2.2 最小Agent带工具调用的对话循环我们先写一个最简单但五脏俱全的Agent它能理解用户问题并且能按需调用一个查询工具。这个例子跑通了后面客服系统的骨架就都有了。先定义状态from typing import Annotated, TypedDict from langgraph.graph import StateGraph, START, END from langchain_core.messages import HumanMessage, AIMessage, ToolMessage import operator class AgentState(TypedDict): messages: Annotated[list, operator.add] # 用一个模型封装函数替代具体SDK接入 def get_chat_model(): # 请替换为实际使用的模型SDK例如通过环境变量读取密钥后创建模型实例 raise NotImplementedError(替换为你的模型接入代码)状态里只放了一个messages字段并用operator.add作为Reducer这样任何节点返回的新消息都会追加而不是覆盖。接下来定义一个查询工具和一个真实节点def query_order_status(order_id: str) - str: 模拟根据订单号查询发货状态 return f订单{order_id}当前状态已发货预计明天送达 def agent_node(state: AgentState): tools {query_order_status: query_order_status} model_with_tools get_chat_model().bind_tools(tools) response model_with_tools.invoke(state[messages]) return {messages: [response]}现在把Graph搭出来from langgraph.prebuilt import ToolNode def route_after_agent(state: AgentState): last_message state[messages][-1] if getattr(last_message, tool_calls, None): return tools return end builder StateGraph(AgentState) builder.add_node(agent, agent_node) builder.add_node(tools, ToolNode([query_order_status])) builder.add_edge(START, agent) builder.add_conditional_edge(agent, route_after_agent, {tools: tools, end: END}) builder.add_edge(tools, agent) graph builder.compile()这个图的实际运行逻辑是用户消息进入agent节点模型判断要不要调工具如果要就返回一个tool_calls指令条件边把它导入tools节点执行真实函数执行结果通过ToolMessage回填到消息列表再回到agent节点让模型继续推理直到模型决定不需要工具了给用户一个最终回复。跑起来试试result graph.invoke({messages: [HumanMessage(content帮我查一下订单20240711的发货状态)]}) print(result[messages][-1].content)这里有个容易忽略的细节节点返回的内容必须是一个包含新消息的字典框架会按Reducer规则合并进状态而你的节点函数内部不要去修改原来的state[messages]直接返回新列表就行。这也是LangGraph和普通脚本的一个思维切换点节点不该有“自己维护的变量”它的一切输出都要体现为状态增量。2.3 编译、运行与流式输出的基本姿势上面调用的compile()很重要它把图定义编译成一个Runnable对象之后才能invoke、stream。编译过程会做很多校验比如条件边返回的节点名是否存在、节点是否有重复注册这些错误在运行前就能暴露大部分。流式输出是实际开发中非常高频的需求。前端对话框希望模拟打字机效果而不是等模型全部生成完再一次性吐出。LangGraph里可以用stream方法for chunk in graph.stream( {messages: [HumanMessage(content帮我查一下订单)]}, stream_modeupdates ): for node_name, update in chunk.items(): print(f节点 {node_name} 的增量{update})stream_modeupdates会按节点粒度输出每一步的状态增量如果希望拿到模型生成token级别的增量还需要结合模型自身的流式参数来做更细的封装。实践里我一般前端用流式接口把节点级更新和token级内容都推给前端用户体感会自然很多。3. 实战智能客服记忆、路由与人工介入3.1 用Checkpointer实现多轮记忆上面那个最小Agent跑完一轮状态就丢掉了。做智能客服的第一件事就是让系统记住用户说了什么。LangGraph的实现方式不是把历史消息塞到一个外部变量里而是用Checkpointer配合thread_id。from langgraph.checkpoint.memory import MemorySaver checkpointer MemorySaver() graph builder.compile(checkpointercheckpointer) config {configurable: {thread_id: customer-001}} graph.invoke( {messages: [HumanMessage(content我刚咨询过订单20240711现在想退款)]}, configconfig )核心在于第二次invoke时同一个thread_id会让框架自动把历史状态并入新一轮输入。新一轮消息会拼接在旧消息后面一起送给模型模型自然知道“刚才咨询过订单”这个上下文。这里有个常见的坑如果你忘了传config或者每次thread_id不同那等于每次都开了一个全新会话前面聊的内容模型一概不知。排查多轮记忆问题时第一件事永远是看thread_id有没有正确从请求里带出来。MemorySaver适合开发和测试生产环境不建议用因为它是纯内存实现进程重启数据就没了。生产至少要用能落盘的持久化实现我在第5章会单独说。3.2 意图识别与条件路由客服系统里不可能所有问题都用同一个Agent硬啃。售前咨询、订单查询、退款投诉、闲聊它们的工具调用、话术模板、处理流程完全不同硬塞进一个Agent会让prompt越来越臃肿出问题也很难定位。我倾向的做法是先加一个“意图识别节点”该节点只干一件事把用户最新消息归类到预设意图。然后通过条件边根据意图字段分发到不同处理子节点。def intent_router(state: AgentState): # 实际开发中让模型用结构化输出返回意图这里简化成模拟逻辑 last_text state[messages][-1].content if 退款 in last_text or 退货 in last_text: return after_sales if 发货 in last_text or 物流 in last_text: return order_query return pre_sales然后在图里注册三条分支。这样的好处是逻辑结构非常清晰想让某个业务线换prompt只需要替换对应的节点完全不影响其他分支。同时意图分发结果作为一个独立字段保存运营同学可以拿这些数据做统计看看用户高频诉求集中在哪一类这对产品迭代有直接价值。有一点需要注意不要让意图识别节点顺带做其他事。它只负责分类分类完马上把控制权交给下游节点这样才能保证“路由决策”和“业务处理”解耦。3.3 客服场景的人工介入与边界控制客服场景特别容易出现模型胡说的风险比如用户说“我要全额赔偿”Agent直接答应又比如用户要求提供身份证号Agent顺手就给。要控制这类问题除了在prompt里加限制更可靠的做法是引入人工审批节点。LangGraph里的实现方式有两种一种是在编译时指定interrupt_before让流程执行到某个节点前停下来等待人工介入另一种是更灵活的interrupt()函数在节点内部主动中断等外部恢复后继续执行。from langgraph.types import interrupt def refund_approval(state: AgentState): # 向外部系统发起审批请求并要求返回审批意见 action interrupt({ type: refund_request, order_id: state.get(order_id), amount: state.get(refund_amount), }) if action.get(approved): return {approval_status: approved} return {approval_status: rejected}外部系统接管这个中断后调用graph.invoke传入恢复数据即可让流程继续。生产环境里这个中断点通常就是接一个审批工单系统APP推送审批消息给人工客服人工点击通过再把结果返回来。除了人工介入边界控制还要做几件事给Agent的工具列表做白名单不要在工具列表里暴露非授权的内部系统接口在状态里维护一个“对话轮数”字段超过最大轮数就强制转人工敏感信息检测节点要放在所有输出之前一旦识别出手机号、地址等隐私信息就做脱敏再展示。4. 多Agent协作系统从单兵作战到团队分工4.1 Supervisor模式让一个Agent做调度客服业务分支多了以后图会变得很宽每条分支里的逻辑也各自复杂。这时一个自然的想法是让每个业务分支都成为一个独立的专业Agent然后由一个“总控Agent”来调度它们谁该干活就派谁上场这就是Supervisor模式。Supervisor节点本身也是一个模型调用节点但它调用的是“调度工具”。具体实现上会给Supervisor绑定一个工具列表每个工具对应一个下属Agent的处理能力比如“查订单工具”、“处理退款工具”、“售前咨询工具”。Supervisor根据用户输入决定要不要调用某个工具如果调用了真实调用对应的Agent并把执行结果返回给SupervisorSupervisor再决定下一步是继续找别的Agent还是给出最终答复。def supervisor_node(state: AgentState): tools [ order_query_agent_tool, after_sales_agent_tool, pre_sales_agent_tool, ] supervisor get_chat_model().bind_tools(tools) response supervisor.invoke(state[messages]) return {messages: [response]}其实这个模式和前面单个Agent调用工具的模式非常像区别只是“工具”从查一个普通接口变成了“再跑一个Agent图”。这正是LangGraph有意思的地方多Agent协作不是另起炉灶而是把Agent当作更高级的工具来编排。这种设计的可扩展性很强每新增一个业务方向只需要注册一个新的Agent工具描述Supervisor的prompt里增加一句说明不影响其他Agent。不过代价是Supervisor的复杂度会随Agent数量上升如果下属超过五六个我建议改成层级式管理不要让一个总控面对所有细节。4.2 层次化子图把客服拆成多个专业AgentSupervisor模式解决了“谁上”的问题但每个专业Agent内部其实仍然是一张有状态的图。售前Agent要回答商品问题、要查询库存售后Agent要查订单、要判断退款条件、要走审批。把每个专业能力都独立成子图再注册到父图中是LangGraph处理复杂业务的主要玩法。实现上每个子图就是一个用StateGraph实例预留的状态图能单独编译、单独测试。然后把子图实例挂到父图的某个节点上。调用子图用的是invoke但要注意父子图的状态兼容问题子图有自己的State定义父图节点调用子图时需要把父状态里的相关字段映射成子图的初始输入。我一般会在子图节点里做显式映射def run_after_sales(state: AgentState): sub_config {configurable: {thread_id: state.get(thread_id) -aftersales}} result after_sales_graph.invoke( {messages: state[messages], order_id: state.get(order_id)}, configsub_config ) return {messages: result[messages], after_sales_result: result.get(final_result)}这样做的好处非常实际每个子图可以独立维护自己的内部逻辑并且可以分别上线。比如售后Agent想升级一个更聪明的模型只需要改动子图内部实现父图完全不需要知道。层级化会带来一个新的管理成本子图的thread_id最好基于父会话的thread_id派生保证同一个用户会话下子图状态不会串到其他用户那里。这个细节我踩过坑曾经因为漏传thread_id导致两个不同用户的售后流程共用了同一个状态排查了半个下午。4.3 并行执行与结果聚合多Agent场景里还有一类常见需求一个任务需要同时咨询多个数据源再把结果拼起来。比如用户问“这个商品有货吗什么时候能送到”客服Agent需要同时查库存系统和物流预估这两个查询互相独立串行执行会让用户等待时间翻倍。StateGraph的fan-out机制可以支持并行从一个节点出发的多条边同时执行每条边指向一个独立节点等所有并行节点执行完后再汇入聚合节点。builder.add_edge(intent_router, query_stock) builder.add_edge(intent_router, query_logistics) builder.add_edge(query_stock, aggregate) builder.add_edge(query_logistics, aggregate)并行节点返回的状态合并规则很关键。如果两个节点都往同一个列表字段追加内容配置了operator.add就没有问题如果是两个节点要写同一个标量字段后执行的就会覆盖先执行的结果不稳定。我的习惯是让每个并行节点把结果写入独立的字段比如stock_result、logistics_result最后由聚合节点统一处理这些字段拼成回复。这样既避免竞态也让状态内容更清晰。并行执行还有一个隐藏好处如果一个节点崩了聚合节点可以知道哪个查询失败了并针对性地做降级处理而不是整个流程全挂。这在真实客服系统里比想象中重要得多第三方物流接口偶尔不可用是常态。5. 生产级工程化持久化、调试和常见坑5.1 状态持久化的选择前面提到的MemorySaver只能做本地调试进程一重启所有对话历史消失。生产环境必须用能落盘的Checkpointer实现。单机场景可以用SQLite多实例部署建议用Postgres。选型的逻辑很简单你的Agent服务是单点还是有多个副本有多个副本就必须用共享存储的Checkpointer否则用户请求落到不同实例上就会丢失会话状态。Checkpointer保存的不只是历史消息还包括状态里所有字段比如意图、订单号、审批状态、临时变量。这既是优势也是负担。生产用久了以后checkpoint表会膨胀得很厉害需要设计状态清理策略。我的做法是消息列表只保留最近N轮完整消息更早的消息由专门的节点做摘要压缩把压缩后结果存进一个history_summary字段既保留长期记忆的语义又控制存储膨胀。需要特别提醒一个容易忽略的地方不要往状态里塞二进制大对象或者超长的工具返回原文。工具返回结果如果很长先做截断或摘要再放进状态。否则每次checkpoint都保存一份超大状态持久化存储和网络传输都会被拖垮。5.2 可观测性与线上调试很多人把LangGraph跑通之后就急着上线结果线上出了问题才发现根本没法排查。原因在于图编排把流程切碎成很多节点普通日志里可能只看到一次请求的入口和出口中间发生了什么完全没有记录。好在LangGraph的状态模型天然适合观测。核心思路是每次invoke都带上thread_id同时把完整状态历史保存下来。排查问题时可以使用get_state_history()拉取所有历史快照像翻聊天记录一样把每一步节点的输入输出翻出来看。history graph.get_state_history(config) for snapshot in history: print(snapshot.values.get(latest_intent), snapshot.next)在接入生产监控时至少要把下面几个指标暴露出来每个节点的执行耗时、模型调用的token消耗、条件边的决策结果、中断等待时长。我习惯在整个图外面包装一层日志中间件以thread_id和request_id为维度打点所有日志写入统一日志平台。线上用户反馈“机器人答错了”时能根据thread_id快速拉出完整链路这一步能不能做好往往决定了一个AI项目在团队里是受信任还是被嫌弃。5.3 常见问题速查表下面这些坑是我在实际项目里遇到过的整理成速查表按频率排列。现象常见原因解决办法多轮对话总是不记得前文invoke时没传thread_id或checkpointer没配检查配置里是否有checkpointer并且每次invoke都传同一个thread_id条件边报错节点不存在条件边返回的字符串和add_node注册名不一致条件边函数里动态返回节点名一定是注册过的名字建议用常量字典节点返回后状态没变化返回的字段没有在State定义里声明State的TypedDict必须声明该字段Removed字段不会自动新增并行节点相互覆盖两个节点写同一个标量状态字段并行结果写入不同字段聚合节点再做合并流式输出没有token级效果节点内模型没开启流式或stream_mode没设对先确认模型调用开启stream参数再按节点事件流拼接输出工具调用后模型不继续输出工具结果没转成ToolMessage塞回消息列表检查工具节点返回类型LangGraph prebuilt的ToolNode会自动处理手写节点要确保返回ToolMessageAgent陷入死循环工具结果又触发同一工具或条件边逻辑有回环加最大节点执行次数限制设置recursion_limit条件边里增加退出判断审批节点恢复后流程没继续中断恢复时传入的数据格式不对恢复调用时传入的数据要和interrupt()等待的格式一致最后一个提醒新写一个复杂图时不要一上来就把所有节点都接满。先注册两三个关键节点用print方式打印每个节点的状态变更跑通一条主链路后再逐步加分支。这样出问题时你知道一定是最近加的那部分引起的排查范围会小很多。6. 一些踩坑之后的个人体会我自己最早接触LangGraph时也被“图”这个抽象概念吓退过。后来动手跑通第一个案例才发现它的核心其实只有三件事状态怎么设计、边怎么跳、状态怎么保存。把这三件事想明白剩下的事情都是往框架里填代码而已。实际排查问题的时候最大的感触是状态设计永远是第一位的。状态字段的粒度、Reducer的选择、哪些数据该放状态哪些不该放直接决定了后面所有节点好不好写、问题好不好排查。我甚至有几次重构改的不是节点逻辑而是把State里几个字段拆细、加上合适的Reducer整个图的稳定性和可读性立刻就不一样了。多Agent架构也不是越复杂越好。能用一条条件边解决的问题不要硬生生拆成Supervisor能用一个节点处理完的逻辑不要为了“架构感”搞子图嵌套。我见过不少项目其实业务复杂度根本不需要多Agent但为了套用最佳实践把系统搞到难以维护。多Agent应该是对业务复杂度的回应而不是对技术时髦的追求。如果这篇文章让你少踩几个坑我觉得就值了。你从最简单的最小Agent开始对着第2章的代码跑一遍再按第3章的思路加路由按第4章的思路拆子图很快就能上手LangGraph。等你真正用LangGraph做完一两个线上项目再回头看最初觉得困难的“图”概念可能只剩下一个感觉原来它就是把复杂系统做得可控的那张底图。