序言当你发现AI客服遇到复杂问题时只会车轱辘话来回转而用户已经气得要投诉了你就该考虑让真人介入了。一、那个差点搞砸的上线日三个月前我负责的智能客服系统正式上线。团队熬了两个月做了意图识别、知识库检索、多轮对话自以为万无一失。上线第一小时数据还行。80%的问题AI都能自己解决用户满意度看起来也凑合。但第二小时问题开始冒头。一个用户在咨询退款流程AI识别了意图也调出了知识库的答案。但用户的情况比较特殊他买的商品已经拆封了而且超过了7天无理由退货期。AI按照标准流程回复可以申请退款用户填了退款申请结果被后台驳回——拆封商品不支持退。用户炸了。他在对话框里连发了五条消息语气越来越冲。AI还在机械地重复抱歉给您带来不好的体验退款流程是……那一刻运营主管在群里我“这种明显搞不定的为什么不转人工”我无言以对。因为当时的系统架构是用户问→AI答答完结束。没有判断AI搞不定的逻辑没有转人工的通道更没有AI和人工之间交接状态的机制。那天晚上我重新梳理了需求。一个靠谱的客服系统至少需要三种能力自知之明AI知道自己什么时候搞不定主动求助循环追问信息没收集全时能反复问用户而不是瞎猜断点续传用户聊到一半退出了下次进来还能接着聊。这三件事用LangChain的Chain几乎无法实现。Chain是线性的走完A→B→C就结束不支持循环不支持中途暂停等人也不支持状态持久化。LangGraph就是来解决这些问题的。二、用StateGraph定义对话状态给货车装货LangGraph的核心思想是状态驱动。整个对话过程就是一辆货车在各个站点之间跑每个站点加工货物最终把完整的订单送到终点。我们先把这辆货车能拉什么货定义清楚。fromtypingimportTypedDict,Annotated,Listfromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.checkpoint.memoryimportMemorySaverfromlanggraph.typesimportinterrupt,Commandfromlangchain_openaiimportChatOpenAIimportoperator# 定义对话状态classCustomerServiceState(TypedDict):user_input:str# 用户当前输入intent:str# 识别出的意图collected_info:Annotated[dict,operator.or_]# 已收集的信息如订单号、手机号confidence:float# AI置信度0-1kb_answer:str# 知识库检索结果needs_human:bool# 是否需要转人工human_feedback:str# 人工反馈内容messages:Annotated[List[dict],operator.add]# 对话历史turn_count:int# 对话轮次防死循环这个TypedDict就是货车的货舱清单。Annotated配合operator.add或operator.or_告诉LangGraph当新的数据进来时是追加到列表里还是合并到字典里。collected_info用operator.or_意味着每次节点返回新的字典会自动和已有的字典合并而不是覆盖。这样信息收集节点可以分多次补充字段比如第一次拿到订单号第二次拿到手机号两者都会保留。messages用operator.add意味着每次返回的新消息会自动追加到历史列表末尾实现对话历史的累积。三、设计节点五个站点构成完整服务链节点是图中的执行单元每个节点只做一件事。我们的客服系统设计了五个核心节点。节点一意图识别defintent_recognition_node(state:CustomerServiceState)-dict:识别用户意图并评估置信度user_inputstate[user_input]messagesstate.get(messages,[])llmChatOpenAI(modelgpt-4o,temperature0)promptf你是一位意图识别专家。根据用户输入判断意图类别和置信度。 可选意图退款咨询、物流查询、产品咨询、投诉建议、其他。 用户输入{user_input}历史对话{messages[-3:]iflen(messages)3elsemessages}请严格按以下格式输出 意图类别 置信度0-1之间的小数 缺失信息如果需要额外信息才能回答列出缺失字段否则填无responsellm.invoke(prompt)contentresponse.content# 解析输出intent其他confidence0.5missing无forlineincontent.split(\n):ifline.startswith(意图):intentline.replace(意图,).strip()elifline.startswith(置信度):try:confidencefloat(line.replace(置信度,).strip())except:passelifline.startswith(缺失信息):missingline.replace(缺失信息,).strip()return{intent:intent,confidence:confidence,messages:[{role:user,content:user_input}],turn_count:state.get(turn_count,0)1}这个节点不仅识别意图还评估置信度。如果用户输入很模糊比如那个东西怎么回事置信度会偏低后续路由会把它导向信息收集节点而不是直接瞎答。节点二信息收集definfo_collection_node(state:CustomerServiceState)-dict:根据缺失信息追问用户collectedstate.get(collected_info,{})intentstate.get(intent,)# 根据意图定义必填字段required_fields{退款咨询:[订单号,退款原因],物流查询:[订单号],产品咨询:[产品型号],投诉建议:[联系方式]}missing[]forfieldinrequired_fields.get(intent,[]):iffieldnotincollectedornotcollected[field]:missing.append(field)ifmissing:questionf为了帮您处理{intent}还需要您提供以下信息{, .join(missing)}。请问您方便提供吗return{messages:[{role:assistant,content:question}],collected_info:{}# 本轮无新增但operator.or_会保留已有的}# 信息已收集全标记为可进入知识库检索return{collected_info:{_complete:True}}这个节点实现了循环追问。如果用户没给订单号就问订单号没给退款原因就问退款原因。直到collected_info里集齐了所有必填字段才放行到下一个节点。节点三知识库检索defkb_retrieval_node(state:CustomerServiceState)-dict:从知识库检索答案intentstate.get(intent,)collectedstate.get(collected_info,{})# 实际项目中这里调用RAG检索# 为演示用模拟数据kb_data{退款咨询:退款流程1.进入我的订单 2.点击申请退款 3.选择退款原因 4.提交审核1-3个工作日,物流查询:物流查询进入我的订单→查看物流→复制快递单号到快递公司官网查询,产品咨询:产品参数请查看商品详情页或咨询专属客服获取技术白皮书,投诉建议:您的反馈已记录客服专员将在24小时内致电回访}answerkb_data.get(intent,抱歉暂时无法回答您的问题为您转接人工客服。)# 如果涉及敏感词标记需要人工介入sensitive_words[投诉,举报,工商局,媒体,曝光,律师]needs_humanany(wordinstate[user_input]forwordinsensitive_words)return{kb_answer:answer,needs_human:needs_human,messages:[{role:assistant,content:answer}]}这个节点有两个职责一是检索知识库二是敏感词检测。如果用户提到了工商局媒体曝光等关键词直接标记needs_humanTrue后续路由会强制转人工。节点四人工介入interrupt机制这是整个系统最关键、也最优雅的节点。defhuman_handoff_node(state:CustomerServiceState)-dict:触发人工介入暂停工作流等待人工反馈# 组装当前会话摘要供人工客服参考summaryf【会话摘要】 用户意图{state.get(intent,未知)}已收集信息{state.get(collected_info, {})} AI回答{state.get(kb_answer,无)}转人工原因{敏感词触发ifany(winstate[user_input]forwin[投诉,举报,工商局,媒体,曝光,律师])else置信度低或用户要求}# interrupt会暂停图的执行把控制权交还给外部系统# 外部系统如客服后台收到中断信息后人工客服介入human_responseinterrupt({type:human_handoff,summary:summary,prompt:人工客服请处理后输入反馈或输入pass让AI继续处理})# 当人工客服在后台提交反馈后工作流从interrupt处恢复return{human_feedback:human_response.get(feedback,),needs_human:False,# 人工已处理重置标记messages:[{role:assistant,content:f【人工客服回复】{human_response.get(feedback,)}}]}interrupt是LangGraph 0.2.31引入的机制。它的工作方式是图执行到human_handoff_node时遇到interrupt()调用图立即暂停当前状态被自动保存到checkpointerinvoke()返回外部系统如Web后台收到包含__interrupt__字段的结果人工客服在后台看到会话摘要处理用户问题人工处理完后外部系统调用graph.invoke(Command(resume{feedback: ...}), config)LangGraph从checkpointer读取断点状态interrupt()返回人工输入节点继续执行图恢复流转。这意味着工作流暂停期间不占用任何运行时资源但状态完整保留。人工客服可以过十分钟再处理甚至换一台机器处理流程都能无缝恢复。节点五结束defend_node(state:CustomerServiceState)-dict:结束节点生成最终回复kb_answerstate.get(kb_answer,)human_fbstate.get(human_feedback,)ifhuman_fb:finalhuman_fbelse:finalkb_answer\n\n如果还有其他问题随时问我。return{messages:[{role:assistant,content:final}]}四、配置条件边三个红绿灯控制交通节点定义好了但怎么决定用户走到哪个节点靠条件边Conditional Edge。defroute_after_intent(state:CustomerServiceState)-str:意图识别后的路由逻辑confidencestate.get(confidence,0)intentstate.get(intent,)turn_countstate.get(turn_count,0)# 规则1轮次超限强制结束防止死循环ifturn_count10:returnend# 规则2置信度低于0.6且不是信息收集阶段转人工ifconfidence0.6andstate.get(collected_info,{}).get(_complete):returnhuman# 规则3信息未收集全进入信息收集节点collectedstate.get(collected_info,{})required{退款咨询:[订单号,退款原因],物流查询:[订单号],产品咨询:[产品型号],投诉建议:[联系方式]}missingany(fnotincollectedornotcollected[f]forfinrequired.get(intent,[]))ifmissing:returncollect# 规则4信息已全进入知识库检索returnkbdefroute_after_kb(state:CustomerServiceState)-str:知识库检索后的路由逻辑ifstate.get(needs_human,False):returnhumanreturnenddefroute_after_human(state:CustomerServiceState)-str:人工介入后的路由逻辑# 人工反馈后如果用户还有追问可以回到意图识别继续# 这里简化为直接结束returnend三条条件边分别对应三个红绿灯置信度低时转人工confidence 0.6直接进human_handoff_node不让AI在不确定的情况下瞎答。缺必填信息时循环追问missing为True时进info_collection_node追问到信息补齐为止。这是一个循环信息收集→回到意图识别→再判断→如果还缺→再收集。直到_completeTrue才放行。敏感词触发升级needs_humanTrue时强制进人工节点无论AI多自信。五、组装完整图把零件拼成机器# 创建StateGraphworkflowStateGraph(CustomerServiceState)# 注册节点workflow.add_node(intent,intent_recognition_node)workflow.add_node(collect,info_collection_node)workflow.add_node(kb,kb_retrieval_node)workflow.add_node(human,human_handoff_node)workflow.add_node(end,end_node)# 设置入口workflow.add_edge(START,intent)# 条件边1意图识别后分流workflow.add_conditional_edges(intent,route_after_intent,{collect:collect,# 缺信息去收集kb:kb,# 信息全了去检索human:human,# 置信度低转人工end:end# 超限结束})# 信息收集后回到意图识别重新判断循环workflow.add_edge(collect,intent)# 条件边2知识库检索后分流workflow.add_conditional_edges(kb,route_after_kb,{human:human,# 敏感词触发转人工end:end# 正常结束})# 人工介入后结束workflow.add_edge(human,end)# 编译图加上checkpointer实现持久化memoryMemorySaver()# 开发用内存生产环境换SqliteSaver或PostgresSaverappworkflow.compile(checkpointermemory)这段代码的精髓在于图的拓扑结构一眼就能看懂。从START进入意图识别然后有三个分支缺信息→收集→回到意图识别循环信息全了→检索→结束或转人工置信度低→直接转人工。六、Checkpointer让对话有记忆让中断能续传前面的代码里有一行memory MemorySaver()这行代码背后是整个LangGraph最值钱的能力之一状态持久化。为什么需要持久化想象一个场景用户和AI客服聊了5轮收集了订单号和退款原因AI正在检索知识库。这时候服务器突然重启了。如果没有持久化用户刷新页面后重新进入对话AI会问“请问您要咨询什么问题”——用户崩溃了“我刚才不是已经说了订单号了吗”LangGraph的checkpointer在每个super-step节点执行完自动保存状态快照。这些快照按thread_id归档同一个thread_id的对话无论调用多少次、无论中间中断多久LangGraph都会从最新的checkpoint恢复状态继续执行。三种持久化方案开发环境InMemorySaverfromlanggraph.checkpoint.memoryimportInMemorySaver memoryInMemorySaver()appworkflow.compile(checkpointermemory)状态存在内存里进程退出就丢。适合调试。测试/小生产环境SqliteSaverimportsqlite3fromlanggraph.checkpoint.sqliteimportSqliteSaver connsqlite3.connect(customer_service.db,check_same_threadFalse)memorySqliteSaver(connconn)appworkflow.compile(checkpointermemory)状态存在本地SQLite文件里程序重启后还能恢复。适合中小规模部署。生产环境PostgresSaverfromlanggraph.checkpoint.postgresimportPostgresSaverimportpsycopg connpsycopg.connect(postgresql://user:passlocalhost:5432/cs_db,autocommitTrue,row_factorypsycopg.rows.dict_row)checkpointerPostgresSaver(conn)checkpointer.setup()# 首次运行创建表结构appworkflow.compile(checkpointercheckpointer)PostgreSQL支持高并发、事务安全、集群部署是生产环境的首选。断点续传的实战用法# 第一次调用用户问退款config{configurable:{thread_id:user_9527}}resultapp.invoke({user_input:我要退款,collected_info:{},turn_count:0},config)# 假设执行到info_collection_nodeAI追问请提供订单号# 用户关闭页面去翻找订单号# 十分钟后用户带着订单号回来了# 再次调用同一个thread_idresultapp.invoke({user_input:订单号是TB20240615001},config# 同一个thread_id)# LangGraph会自动加载user_9527的历史状态# 发现上一轮在collect节点等待订单号# 新的输入被合并到collected_info里# 然后自动流转到下一个节点这就是断点续传的魔力用户不需要重复已经说过的话系统记得你们聊到哪了。查看历史状态# 查看当前会话的最新状态snapshotapp.get_state(config)print(f当前状态{snapshot.values})print(f下一个待执行节点{snapshot.next})# 查看完整历史historylist(app.get_state_history(config))fori,snapinenumerate(history):print(f第{i}步{snap.values.get(intent,N/A)}→ 下一步{snap.next})get_state_history返回所有checkpoint你可以像看录像一样回放整个对话流程排查哪一步出了问题。七、完整运行示例把上面的代码串起来我们模拟一次完整的对话# 场景用户咨询退款但没给订单号config{configurable:{thread_id:demo_001}}# 第1轮用户输入stateapp.invoke({user_input:我要退款,collected_info:{},turn_count:0},config)# 输出AI追问请提供订单号和退款原因print(state[messages][-1][content])# 第2轮用户提供订单号模拟断点续传stateapp.invoke({user_input:订单号是TB20240615001},config)# AI继续追问退款原因print(state[messages][-1][content])# 第3轮用户说明原因stateapp.invoke({user_input:商品质量有问题屏幕有坏点},config)# 信息收集全了进入知识库检索返回退款流程print(state[messages][-1][content])# 第4轮用户不满意提到我要投诉到工商局stateapp.invoke({user_input:你们处理太慢了我要投诉到工商局},config)# 敏感词触发进入human_handoff_node# interrupt暂停等待人工介入if__interrupt__instr(state):print(【系统】已转人工客服请等待...)# 模拟人工客服处理完后恢复fromlanggraph.typesimportCommand finalapp.invoke(Command(resume{feedback:您好我是客服主管已为您加急处理预计2小时内退款到账。}),config)print(final[messages][-1][content])这个示例展示了系统的全部能力循环追问用户没给全信息时AI反复问断点续传用户中途离开再回来对话无缝衔接敏感词升级提到工商局自动转人工interrupt协作人工处理完后系统恢复执行把人工反馈带给用户。八、写在最后从问答机器人到服务系统写完这个项目我对LangGraph最大的感受是它让AI客服从问答机器人进化成了服务系统。问答机器人的逻辑是问→答结束。服务系统的逻辑是识别→收集→判断→检索→人工→结束中间有循环、有分支、有暂停、有恢复。LangGraph的StateGraph让我们能精确控制这个流程的每一个分支。条件边不是黑盒是写死在代码里的业务规则。interrupt不是魔法是显式的暂停点。checkpointer不是抽象概念是实实在在的状态快照。如果你还在用Chain硬凑多轮对话或者用Agent黑盒赌运气不妨试试LangGraph。当你第一次画出一张完整的客服流程图看到用户在不同节点之间流转看到断点续传无缝恢复看到人工客服优雅地介入——你会明白这不是又一个框架这是AI服务工程的成人礼。