上周有个读者朋友私信问我:“LangGraph入门篇看完了,图模型确实有意思,但身边没人真正放在业务里用,这到底能解决什么实际问题?”我想了想,没有直接跟他掰扯概念,而是把一个刚做完的智能客服Demo代码丢了过去。这个Demo规模不大,但覆盖了订单查询、售后流转、商品咨询、转人工这四类最典型的客服意图,而且完全用LangGraph做流程编排。他看完之后回了一句:“原来这玩意儿是干这个用的。”
这篇就把这个案例完整拆一遍。我会从“为什么这套逻辑必须用图来编排”讲起,然后给出一版可以直接跑起来的代码,再把我实际开发中踩过的坑和排查手段整理成清单。适合已经了解LangGraph基础概念、想看看它在真实业务场景里怎么落地的开发者,也适合被“客服机器人”需求困扰、想找一个可维护架构的同学。
1. 为什么智能客服场景天然适合用LangGraph编排
1.1 链式调用与单纯Agent循环的局限
如果你只做过“输入一段话,返回一段回答”的问答机器人,用LangChain的链式调用确实足够。prompt加载进来,模型跑一遍,输出拿出去,完事。但智能客服不是单轮问答:用户可能会追问“那大概几天能到”,这句话单拎出来没有任何人能回答,必须结合上文订单号;用户也可能先说“我要退货”,等你回复后又补一句“订单号是A1001”,这又是一种跨轮次的上下文补全。
更麻烦的是分支。同一个入口进来,订单查询、售后、商品咨询、投诉转人工,各自的处理逻辑完全不同。用链式调用写分支,本质上就是在外层套一堆if-else,哪个意图走哪条链。写两三个分支还好,凑到十个八个的时候,代码已经不是“能维护”的状态了,更像一盘不断往下缠的意大利面。
有人会说,那直接用单个Agent不就行了?模型自己决定调什么工具、什么时候结束。但真实客服场景恰恰最怕“模型自由发挥”。比如售后规则要求必须先确认订单状态才能发起退款,如果让Agent自己决定,它可能跳过这一步,也可能在没有核实订单号的情况下就承诺用户“可以退款”。业务上这叫流程失控。我们要的是一种既能用模型理解自然语言,又能把关键业务步骤牢牢锁死的架构。
这就是LangGraph这类图编排框架出现的原因:把业务流程图直接变成代码结构。节点是“做什么”,边是“做完之后下一步去哪”,条件边是“根据当前状态决定去哪”。模型负责节点内部的智能部分,流程控制则交还给开发者。
1.2 状态、路由、中断:图模型给客服场景的三个关键能力
LangGraph对客服系统来说,最值钱的是三件事。
第一是统一状态管理。整个会话的所有信息都放在一个State对象里流转,节点读它、改它、再传下去。聊天的历史messages列表通过add_messages这个reducer自动追加,不会出现新消息把旧消息覆盖掉的问题。你在任何节点都能看到整个会话的完整脉络。
第二是条件路由。从一个节点出发,可以连一个普通Python函数,函数内部读一下当前State里的字段,返回下一个要进入的节点名。业务规则在路由函数里一目了然:用户表达的是售后意图,就进售后节点;售后节点判断当前订单不满足自助处理条件,就把need_human置为True,路由函数看到这个标记就会转去人工节点。这种“规则显式写在代码里”的方式,比纯模型决策可靠得多。
第三是中断机制。LangGraph可以在执行到某个节点之前暂停整个图,等外部系统(比如一个真人客服)介入后,再恢复执行。这正好对智能客服最头疼的转人工场景:机器处理不了的case,不能让它硬编一段“请拨打热线”的假话就结束,而是真正把会话挂起、等人工接管。
这三个能力组合在一起,客服系统的核心闭环就成立了:多轮上下文有状态管理兜底,流程分支有条件路由控制,机器搞不定的有中断机制兜底。
1.3 和手写状态机、低代码平台横向对比
在决定用LangGraph之前,我也对比过另外两条路。手写状态机是最直接的想法:定义状态枚举、写状态转移表、在转移函数里处理业务逻辑。简单场景下它完全可行,但把LLM调用塞进状态机之后,问题就来了:状态枚举会随业务膨胀,每个状态又要维护“进入时干什么”“离开时干什么”“异常怎么办”,代码量翻倍得很快。而且状态机本身不关心对话上下文,多轮消息的存取还是要自己另写一套。
低代码平台的思路则相反,平台把节点、工具、数据库都封装成了可视化组件,拖拽就能搭一个客服流。快速验证原型确实爽,但真上生产就会撞到墙:复杂的条件分支在可视化界面里很难维护,内部系统对接要通过平台提供的脚本语言,私有化部署更是绕不开的成本。最重要的是,平台把执行细节藏得太深,出了问题不好定位。
LangGraph等于把“状态机+代码可控+LLM能力”揉在了一起。它没有把流程抽象成黑盒,图本身就是一个数据结构,可以打印、可以单节点调试、可以接入自己的监控体系。付出的代价是需要熟悉一套新API,但换来的是对系统每一步的完全掌控。
我用一个表格把这几种方案的差异整理了出来:
| 方案 | 流程可控性 | 多轮对话支持 | 业务扩展成本 | 典型适用阶段 |
|---|---|---|---|---|
| 手写状态机 | 高,但全手写 | 需自行实现 | 状态枚举膨胀后较高 | 固定、简单流程 |
| 链式调用 | 低,靠外部if-else | 弱,需自行拼装 | 分支多了迅速恶化 | 单意图问答 |
| 单Agent循环 | 中,模型自由决策 | 中 | 中,但会流程失控 | 工具调用密集、无强规则场景 |
| 低代码平台 | 中,受平台限制 | 平台内置 | 中,深度定制困难 | 原型验证 |
| LangGraph | 高,代码即流程 | 框架原生支持 | 中,学习曲线换可控性 | 生产级复杂业务流程 |
2. 案例需求与图结构设计
2.1 业务场景与意图清单
这个案例的背景我设定为某在线商城的智能客服Demo,用户从对话框进来,机器人需要处理四类请求:订单查询、售后申请、商品咨询、转人工。这四类基本覆盖了客服系统80%以上的日常请求,用来演示图编排足够典型。
对应到意图标签,我定义如下:
| 意图 | 触发示例 | 处理方式 |
|---|---|---|
| order_query | “查一下A1001的物流” | 调用订单查询工具,返回状态和物流单号 |
| aftersale | “我要退货”“申请退款” | 核验订单信息,简单情况自动登记,复杂情况转人工 |
| product_query | “这个有现货吗”“能开发票吗” | 直接基于商品知识库问答 |
| human | “转人工”“投诉” | 进入人工接管的暂停节点 |
真实业务的意图肯定不止四个,可能还要拆出优惠券、发票、地址修改等等。但设计思路完全相同:先在图上把意图和节点规划好,再逐类填实现。
2.2 节点边界划分:多细才算合适
图设计最容易犯的毛病是节点粒度失控。我见过有人把整个客服对话写成一个超级节点,内部塞入意图识别、工具调用、多轮总结,结果图是画出来了,本质还是一个大函数,没享受到图编排的任何好处。也见过另一个极端,把“检查用户是否输入了订单号”“检查订单状态”“判断是否发货”这种函数级操作各拆成一个节点,路由表膨胀到看不清主流程。
我的经验是,节点粒度对标“一张业务工单”。一个节点应该完成一件业务上完整、可独立验收的事。比如“查订单并生成回复”是一个节点,“判断能否自动退款”是另一个节点;“历史订单是否超过退货期限”这种细节,放在“判断能否自动退款”节点内部作为逻辑分支就好,没必要单独开节点。
节点拆得合适,整张图读起来就像一份业务流程图。每个节点可以单独打日志、单独测试、单独替换实现。以后要改动售后规则,只需要动一个节点和一条条件边,不用牵一发动全身。
2.3 状态字段设计:把流程变量显式化
LangGraph里,State是贯穿整个图的共享数据结构。设计State字段时要问自己一个问题:有哪些信息,是后面某个节点必须依赖、但当前节点不产生、必须从上游传下来的?
我这个案例里定义了三类字段:messages、intent、need_human。
messages是会话历史,用Annotated[list, add_messages]声明,框架会自动把新消息追加到历史里,而不是覆盖。intent是意图识别的结果,后面的条件路由函数靠它决定去哪个节点。need_human是售后节点写给条件路由函数看的标志位:售后节点判断当前请求需要人工介入时,把它置为True,路由函数读到就转去人工节点。
真实项目里,你大概率还要加用户ID、订单号、会话来源渠道这些字段。这些字段的取值都来自某个上游节点或外部系统,通过返回值写进State,下游节点从State里取出即可。
用生活化的方式理解:State就像流水线上挂在工件上的工单卡片。每个工位读完卡片上的信息,决定做什么操作,再把自己的处理结果写回卡片,传给下一个工位。全程只有一个卡片在流转,所以不会出现信息丢在半路的情况。
3. 实战落地:核心代码一步步写
3.1 环境准备与版本选择
先装依赖。我的示例用到了langgraph和langchain-core,聊天模型接的是你项目里现成的大模型API,适配包按你自己的平台装就行。
pip install langgraph langchain-coreLangGraph的API演进很快,不同版本之间有一些命名差异。我这篇文章的代码基于当前主流版本:StateGraph、START、END都从langgraph.graph导入,add_messages从langgraph.graph.message导入。如果你用的版本较老,可能还需要微调导入路径,认准这三个核心导入位置就不会差太远。
安装好之后,先别急着写客服逻辑。我建议你先写一个只有两个节点的最小图,从START到第一个节点再到END,跑通invoke,确认环境没问题。把环境变量和版本问题先排掉,后面调试业务代码时才不会混在一起。
3.2 定义State与模拟工具
先定义状态类型。这里用TypedDict声明字段,messages字段必须用Annotated包裹并指定add_messages作为reducer。
from typing import Annotated, TypedDict from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langchain_core.messages import HumanMessage, AIMessage class AgentState(TypedDict): messages: Annotated[list, add_messages] intent: str need_human: bool再写一个模拟的订单查询工具。真实场景里这个函数大概率是查内部订单系统的API,但在这里我用一个字典模拟数据库,返回字符串格式的查询结果。
ORDER_DB = { "A1001": "已发货,物流单号:SF123456", "A1002": "待发货,预计2天后出库", } def query_order_status(order_no: str) -> str: """根据订单号查询订单状态。""" return ORDER_DB.get(order_no, "未找到该订单,请核对订单号")工具函数要注意一点:返回内容必须是模型或下游节点能直接理解的信息。比如查询结果的字符串里既包含状态也包含物流单号,这样后续节点可以直接拿它组装回复,不需要再猜字段含义。
3.3 意图识别节点:先用规则跑通图
开发的第一步,我建议意图识别用最简单、最死板的规则实现,不要一上来就接大模型。原因很简单:先把图的骨架跑通,确认节点之间的流转正常,然后再把“识别”这种内部实现替换成模型,出现问题时排查范围小得多。
def extract_order_no(text: str) -> str: for token in text.split(): if token.upper().startswith("A") and token[1:].isdigit(): return token.upper() return "" def parse_intent(state: AgentState) -> dict: last_text = state["messages"][-1].content if any(k in last_text for k in ("订单", "物流", "快递", "发货")): intent = "order_query" elif any(k in last_text for k in ("退货", "退款", "换货", "售后")): intent = "aftersale" elif any(k in last_text for k in ("转人工", "投诉", "人工")): intent = "human" else: intent = "product_query" return {"intent": intent}节点函数的返回很关键:它只返回需要写入State的字段增量。也就是说,你不需要把整个State返回去,返回一个dict,框架会把里面的键合并进State。parse_intent只改了intent字段,messages没有动,那它就是只返回intent就好。
真实项目里,这个节点会替换成一个结构化输出的LLM调用,让模型输出意图标签,再写进State。但图的节点边界完全不用改,这正好体现了图和内部实现的解耦。
3.4 业务节点、条件路由与图编译
接下来把业务节点实现出来。order_lookup节点负责查订单并生成回复;aftersale节点负责售后规则判断;product_query节点做商品咨询;human_handoff节点输出转人工提示。
def order_lookup(state: AgentState) -> dict: last_text = state["messages"][-1].content order_no = extract_order_no(last_text) if not order_no: # 也可以先查最近对话记录里的订单号,这里简化处理 return {"messages": [AIMessage(content="请提供订单号,格式如:A1001。")]} status = query_order_status(order_no) return {"messages": [AIMessage(content=f"您的订单{order_no}当前状态:{status}。")]} def aftersale(state: AgentState) -> dict: last_text = state["messages"][-1].content order_no = extract_order_no(last_text) if not order_no: return { "need_human": True, "messages": [AIMessage(content="为了确认您的订单信息,我正在为您转接人工客服。")], } status = query_order_status(order_no) if "已发货" in status: return { "need_human": True, "messages": [AIMessage(content="您的订单已发货,退货需要人工审核,正在为您转接人工客服。")], } return { "need_human": False, "messages": [AIMessage(content=f"您的订单{order_no}符合售后条件,已经为您登记退货申请。")], } def product_query(state: AgentState) -> dict: return {"messages": [AIMessage(content="这款商品目前有现货,可以正常下单。发票会在发货后3个工作日内开具。")]} def human_handoff(state: AgentState) -> dict: return {"messages": [AIMessage(content="已为您转接人工客服,请稍候。工单编号:H-2024-001。")]}注意aftersale节点的设计:它根据订单状态决定是否把need_human置为True,然后返回给框架。真正决定“下一步去哪”的不是它自己,而是后面的条件路由函数。这样写的好处是,售后流程的规则都集中在aftersale节点内部,路由只负责看标志位,职责非常干净。
现在把节点挂到图上,配置边。先从入口连到意图识别节点,然后按意图做条件分发。
builder = StateGraph(AgentState) builder.add_node("parse_intent", parse_intent) builder.add_node("order_lookup", order_lookup) builder.add_node("aftersale", aftersale) builder.add_node("product_query", product_query) builder.add_node("human_handoff", human_handoff) builder.add_edge(START, "parse_intent") builder.add_conditional_edges( "parse_intent", lambda state: state["intent"], { "order_query": "order_lookup", "aftersale": "aftersale", "product_query": "product_query", "human": "human_handoff", }, ) builder.add_conditional_edges( "aftersale", lambda state: "human_handoff" if state["need_human"] else END, {"human_handoff": "human_handoff", END: END}, ) builder.add_edge("order_lookup", END) builder.add_edge("product_query", END) builder.add_edge("human_handoff", END)add_conditional_edges是LangGraph里最核心的API:第一个参数是起始节点,第二个参数是一个路由函数,第三个参数是路由值的映射表。路由函数返回的字符串,必须能在映射表里找到对应目标节点,否则会报错。这里aftersale的路由函数返回了END常量,而END在映射表里也作为键存在,注意这个技巧:LangGraph允许把END作为条件边的一个目标。
最后编译并跑一次。
app = builder.compile() result = app.invoke({ "messages": [HumanMessage(content="查一下A1001的物流")], }) print(result["messages"][-1].content)输出结果就是order_lookup节点生成的回复。到这里,一个最小可用的客服流程已经跑通了。
3.5 用Checkpointer开启多轮会话
上面的图能跑,但每次invoke都是独立的,没有“会话记忆”。用户上一轮刚报过订单号A1001,下一轮追问“那大概几天能到”,图面对这一轮输入时State里只有当前这一条消息,前面的订单号信息已经丢了。这显然不符合客服场景。
LangGraph的解决方案是Checkpointer。编译时传入一个MemorySaver,invoke时带上thread_id,框架就会把这个会话的所有状态存下来,同thread_id的下一次调用会自动恢复历史。
from langgraph.checkpoint.memory import MemorySaver app = builder.compile(checkpointer=MemorySaver()) config = {"configurable": {"thread_id": "session-001"}} app.invoke( {"messages": [HumanMessage(content="查一下A1001的物流")]}, config, ) app.invoke( {"messages": [HumanMessage(content="那这个订单大概几天能到")]}, config, )第二次调用时,State里的messages已经包含了前面所有轮次的对话。你可以在节点函数里通过state["messages"]拿到完整上下文,然后组装回答。这就是add_messages的作用:框架自动把新消息追加到messages列表末尾,而不是覆盖。
这里有一个非常容易踩的坑:如果你invoke时忘记传config,或者thread_id每次都不一样,那么即使编译时加了checkpointer,框架也会把每次调用当成新会话。我排查多轮不记忆的问题时,第一件事永远是确认thread_id是否一致。
3.6 转人工的进阶做法:中断机制
前面的human_handoff节点只是“输出一段转人工的提示”,实际上并没有真正把会话交接给人工。真实场景里,转人工应该做到:机器人的执行流停在某个节点前,等人工客服介入处理,人工处理完后再恢复机器人的后续逻辑。
LangGraph提供了interrupt机制。编译时指定interrupt_before,框架会在进入目标节点前暂停执行:
app = builder.compile( checkpointer=MemorySaver(), interrupt_before=["human_handoff"], )当用户触发了转人工意图,图执行到human_handoff节点之前会停住。此时调用get_state查看当前状态,next字段会显示接下来要执行的节点是human_handoff,说明会话正在等人。
snapshot = app.get_state(config) print("当前状态:", snapshot.next)人工客服处理完、在系统里点击“已接管”之后,调用app.invoke(None, config),框架会从刚才暂停的位置继续往下执行。这个机制比在节点内部硬编码if-else要优雅得多,因为暂停、恢复、状态保存全部由框架接管,你不需要自己维护“人工处理到一半系统重启了怎么办”这种问题。
4. 常见问题与排查技巧实录
4.1 高频报错速查表
开发这个案例的过程中,我把几个反复出现的问题整理成了下面的速查表。这些问题不是玄学,根因都很明确,掌握了之后排查速度会快很多。
| 现象 | 根因 | 处理方法 |
|---|---|---|
| 节点返回dict后,报未定义的键 | 返回的键没有在State类型里声明 | 回到TypedDict补字段,或删除多余返回 |
| 多轮对话历史被覆盖 | messages字段没有加add_messages | 类型改成Annotated[list, add_messages] |
| 条件路由报KeyError | 路由函数返回值不在映射表里 | 检查路由函数返回的字符串,和映射表键名逐一对应 |
| 多轮不记忆、上下文丢失 | 没传checkpointer,或thread_id不一致 | 编译时加checkpointer,invoke时传同一个config |
| 工具调用结果没有生效 | 工具函数返回格式不规范,下游解析失败 | 确认工具返回内容包含所有必要信息 |
| 图执行没有走到预期节点 | 条件边的目标被END短路 | 检查路由函数对每个分支是否都返回了正确值 |
拿“条件路由报KeyError”来说,这种问题多半发生在意图标签写法和映射表键名不一致的时候。比如parse_intent里写了intent = "order_query",但映射表键名写成了"order"或者"query_order"。这类错误不会在编译时暴露,要等图执行到条件边才炸。排查时最直接的办法是在路由函数里加print,把return值打出来看,或者用get_state查看当前state快照。
4.2 调试Graph的三板斧
LangGraph开发调试,我基本只用三个手段。
第一是可视化。编译后的图对象自带get_graph()方法,可以打印ASCII流程图。我把这个命令放在调试的第一位,因为它能直观地告诉你节点和边连得对不对。
app.get_graph().print_ascii()输出会展示每个节点和执行路径,连接关系一目了然。如果图上出现了你预期的边,那问题大概率在路由函数返回逻辑;如果边本身就缺了,回到add_conditional_edges的映射表检查。
第二是查状态快照。图执行到任何一步,都可以通过get_state查当前State的值和下一个要执行的节点。这个能力在处理中断、调试转人工时尤其有用。你可以直接看到need_human字段被设置成了什么,next指向哪里,不需要靠猜。
snapshot = app.get_state(config) print(snapshot.values) print(snapshot.next)第三是单节点回归。LangGraph的节点函数本质上是普通Python函数,接收state返回新字段。这意味着你可以不启动整个图,单独构造一个模拟state,调用节点函数,断言返回值是否符合预期。我用这个方式在本地把每个节点的逻辑都测了一遍,确认无误后再组合到图里,定位问题快很多。
4.3 几个实战设计原则
踩过几次坑之后,我总结了几条做LangGraph客服系统的设计原则,现在基本当成固定习惯。
先规则跑通,再换模型。意图识别、回复生成都可以先用最简单的硬编码实现,让整张图能完整跑通。骨架稳定之后再逐个节点替换成LLM实现,每一步都可验证,不会出现“整个系统跑不起来不知道是图配置错还是模型返回错”的窘境。
节点粒度取“业务动作”而不是“函数功能”。判断是否发货不应单独开节点,但“判断能否自助售后”应该单独开节点。划分的依据是:这个动作是否会被其他流程复用?是否需要在日志里单独追踪?两者都是yes才值得开节点。
转人工放在图边界,用中断机制而不是硬编码。直接在售后节点里生成“转人工”回复也能实现业务,但你会失去对会话的控制权:无法暂停、无法人工介入、无法恢复后续逻辑。把转人工设计成一条通往图边界节点的条件边,配合interrupt_before,才是生产级做法。
条件路由的返回值字符串和映射表键名,要集中管理。我习惯在文件顶部定义一组常量,把节点名、意图名都存进去,避免在多处手写魔法字符串。手写顺手,但一旦改了一个名字,排查成本够你喝一壶。
我现在做这类Agent项目,习惯先把图在纸上画出来,哪怕是随手画的草稿。画不出来说明业务还没想清楚,画出来之后再写代码,基本就是按图施工。跑起来之后,print_ascii看一眼节点连接,get_state看一眼State快照,绝大部分问题都能定位。这个习惯本身就是LangGraph这类框架给开发者的最大价值——它逼着你先把流程想清楚,而不是让模型天马行空地自由发挥。