LangGraph实战：搭建一个带人工介入的智能客服系统

序言：当你发现AI客服遇到复杂问题时只会"车轱辘话"来回转，而用户已经气得要投诉了，你就该考虑让真人介入了。

一、那个差点搞砸的上线日

三个月前，我负责的智能客服系统正式上线。团队熬了两个月，做了意图识别、知识库检索、多轮对话，自以为万无一失。

上线第一小时，数据还行。80%的问题AI都能自己解决，用户满意度看起来也凑合。

但第二小时，问题开始冒头。

一个用户在咨询"退款流程"，AI识别了意图，也调出了知识库的答案。但用户的情况比较特殊：他买的商品已经拆封了，而且超过了7天无理由退货期。AI按照标准流程回复"可以申请退款"，用户填了退款申请，结果被后台驳回——拆封商品不支持退。

用户炸了。他在对话框里连发了五条消息，语气越来越冲。AI还在机械地重复"抱歉给您带来不好的体验，退款流程是……"

那一刻，运营主管在群里@我：“这种明显搞不定的，为什么不转人工？”

我无言以对。因为当时的系统架构是：用户问→AI答，答完结束。没有"判断AI搞不定"的逻辑，没有"转人工"的通道，更没有"AI和人工之间交接状态"的机制。

那天晚上，我重新梳理了需求。一个靠谱的客服系统，至少需要三种能力：

1. 自知之明：AI知道自己什么时候搞不定，主动求助；
2. 循环追问：信息没收集全时，能反复问用户，而不是瞎猜；
3. 断点续传：用户聊到一半退出了，下次进来还能接着聊。

这三件事，用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-1）kb_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_input=state["user_input"]messages=state.get("messages",[])llm=ChatOpenAI(model="gpt-4o",temperature=0)prompt=f"""你是一位意图识别专家。根据用户输入判断意图类别和置信度。 可选意图：退款咨询、物流查询、产品咨询、投诉建议、其他。 用户输入：{user_input}历史对话：{messages[-3:]iflen(messages)>3elsemessages}请严格按以下格式输出： 意图：<类别> 置信度：<0-1之间的小数> 缺失信息：<如果需要额外信息才能回答，列出缺失字段，否则填"无">"""response=llm.invoke(prompt)content=response.content# 解析输出intent="其他"confidence=0.5missing="无"forlineincontent.split("\n"):ifline.startswith("意图："):intent=line.replace("意图：","").strip()elifline.startswith("置信度："):try:confidence=float(line.replace("置信度：","").strip())except:passelifline.startswith("缺失信息："):missing=line.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:"""根据缺失信息，追问用户"""collected=state.get("collected_info",{})intent=state.get("intent","")# 根据意图定义必填字段required_fields={"退款咨询":["订单号","退款原因"],"物流查询":["订单号"],"产品咨询":["产品型号"],"投诉建议":["联系方式"]}missing=[]forfieldinrequired_fields.get(intent,[]):iffieldnotincollectedornotcollected[field]:missing.append(field)ifmissing:question=f"为了帮您处理{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:"""从知识库检索答案"""intent=state.get("intent","")collected=state.get("collected_info",{})# 实际项目中这里调用RAG检索# 为演示，用模拟数据kb_data={"退款咨询":"退款流程：1.进入我的订单 2.点击申请退款 3.选择退款原因 4.提交审核（1-3个工作日）","物流查询":"物流查询：进入我的订单→查看物流→复制快递单号到快递公司官网查询","产品咨询":"产品参数请查看商品详情页，或咨询专属客服获取技术白皮书","投诉建议":"您的反馈已记录，客服专员将在24小时内致电回访"}answer=kb_data.get(intent,"抱歉，暂时无法回答您的问题，为您转接人工客服。")# 如果涉及敏感词，标记需要人工介入sensitive_words=["投诉","举报","工商局","媒体","曝光","律师"]needs_human=any(wordinstate["user_input"]forwordinsensitive_words)return{"kb_answer":answer,"needs_human":needs_human,"messages":[{"role":"assistant","content":answer}]}

这个节点有两个职责：一是检索知识库，二是敏感词检测。如果用户提到了"工商局""媒体曝光"等关键词，直接标记needs_human=True，后续路由会强制转人工。

节点四：人工介入（interrupt机制）

这是整个系统最关键、也最优雅的节点。

defhuman_handoff_node(state:CustomerServiceState)->dict:"""触发人工介入，暂停工作流等待人工反馈"""# 组装当前会话摘要，供人工客服参考summary=f"""【会话摘要】 用户意图：{state.get('intent','未知')}已收集信息：{state.get('collected_info', {})} AI回答：{state.get('kb_answer','无')}转人工原因：{'敏感词触发'ifany(winstate['user_input']forwin['投诉','举报','工商局','媒体','曝光','律师'])else'置信度低或用户要求'}"""# interrupt会暂停图的执行，把控制权交还给外部系统# 外部系统（如客服后台）收到中断信息后，人工客服介入human_response=interrupt({"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+引入的机制。它的工作方式是：

1. 图执行到human_handoff_node时，遇到interrupt()调用；
2. 图立即暂停，当前状态被自动保存到checkpointer；
3. invoke()返回，外部系统（如Web后台）收到包含__interrupt__字段的结果；
4. 人工客服在后台看到会话摘要，处理用户问题；
5. 人工处理完后，外部系统调用graph.invoke(Command(resume={"feedback": "..."}), config)；
6. LangGraph从checkpointer读取断点状态，interrupt()返回人工输入，节点继续执行，图恢复流转。

这意味着：工作流暂停期间不占用任何运行时资源，但状态完整保留。人工客服可以过十分钟再处理，甚至换一台机器处理，流程都能无缝恢复。

节点五：结束

defend_node(state:CustomerServiceState)->dict:"""结束节点，生成最终回复"""kb_answer=state.get("kb_answer","")human_fb=state.get("human_feedback","")ifhuman_fb:final=human_fbelse:final=kb_answer+"\n\n如果还有其他问题，随时问我。"return{"messages":[{"role":"assistant","content":final}]}

四、配置条件边：三个红绿灯控制交通

节点定义好了，但怎么决定用户走到哪个节点？靠条件边（Conditional Edge）。

defroute_after_intent(state:CustomerServiceState)->str:"""意图识别后的路由逻辑"""confidence=state.get("confidence",0)intent=state.get("intent","")turn_count=state.get("turn_count",0)# 规则1：轮次超限，强制结束，防止死循环ifturn_count>=10:return"end"# 规则2：置信度低于0.6，且不是信息收集阶段，转人工ifconfidence<0.6andstate.get("collected_info",{}).get("_complete"):return"human"# 规则3：信息未收集全，进入信息收集节点collected=state.get("collected_info",{})required={"退款咨询":["订单号","退款原因"],"物流查询":["订单号"],"产品咨询":["产品型号"],"投诉建议":["联系方式"]}missing=any(fnotincollectedornotcollected[f]forfinrequired.get(intent,[]))ifmissing:return"collect"# 规则4：信息已全，进入知识库检索return"kb"defroute_after_kb(state:CustomerServiceState)->str:"""知识库检索后的路由逻辑"""ifstate.get("needs_human",False):return"human"return"end"defroute_after_human(state:CustomerServiceState)->str:"""人工介入后的路由逻辑"""# 人工反馈后，如果用户还有追问，可以回到意图识别继续# 这里简化为直接结束return"end"

三条条件边分别对应三个"红绿灯"：

• 置信度低时转人工：confidence < 0.6直接进human_handoff_node，不让AI在不确定的情况下瞎答。
• 缺必填信息时循环追问：missing为True时进info_collection_node，追问到信息补齐为止。这是一个循环：信息收集→回到意图识别→再判断→如果还缺→再收集。直到_complete=True才放行。
• 敏感词触发升级：needs_human=True时强制进人工节点，无论AI多自信。

五、组装完整图：把零件拼成机器

# 创建StateGraphworkflow=StateGraph(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实现持久化memory=MemorySaver()# 开发用内存，生产环境换SqliteSaver或PostgresSaverapp=workflow.compile(checkpointer=memory)

这段代码的精髓在于：图的拓扑结构一眼就能看懂。从START进入意图识别，然后有三个分支：缺信息→收集→回到意图识别（循环）；信息全了→检索→结束或转人工；置信度低→直接转人工。

六、Checkpointer：让对话有"记忆"，让中断能"续传"

前面的代码里有一行memory = MemorySaver()，这行代码背后是整个LangGraph最值钱的能力之一：状态持久化。

为什么需要持久化？

想象一个场景：用户和AI客服聊了5轮，收集了订单号和退款原因，AI正在检索知识库。这时候，服务器突然重启了。

如果没有持久化，用户刷新页面后重新进入对话，AI会问：“请问您要咨询什么问题？”——用户崩溃了：“我刚才不是已经说了订单号了吗？”

LangGraph的checkpointer在每个super-step（节点执行完）自动保存状态快照。这些快照按thread_id归档，同一个thread_id的对话，无论调用多少次、无论中间中断多久，LangGraph都会从最新的checkpoint恢复状态，继续执行。

三种持久化方案

开发环境：InMemorySaver

fromlanggraph.checkpoint.memoryimportInMemorySaver memory=InMemorySaver()app=workflow.compile(checkpointer=memory)

状态存在内存里，进程退出就丢。适合调试。

测试/小生产环境：SqliteSaver

importsqlite3fromlanggraph.checkpoint.sqliteimportSqliteSaver conn=sqlite3.connect("customer_service.db",check_same_thread=False)memory=SqliteSaver(conn=conn)app=workflow.compile(checkpointer=memory)

状态存在本地SQLite文件里，程序重启后还能恢复。适合中小规模部署。

生产环境：PostgresSaver

fromlanggraph.checkpoint.postgresimportPostgresSaverimportpsycopg conn=psycopg.connect("postgresql://user:pass@localhost:5432/cs_db",autocommit=True,row_factory=psycopg.rows.dict_row)checkpointer=PostgresSaver(conn)checkpointer.setup()# 首次运行创建表结构app=workflow.compile(checkpointer=checkpointer)

PostgreSQL支持高并发、事务安全、集群部署，是生产环境的首选。

断点续传的实战用法

# 第一次调用：用户问退款config={"configurable":{"thread_id":"user_9527"}}result=app.invoke({"user_input":"我要退款","collected_info":{},"turn_count":0},config)# 假设执行到info_collection_node，AI追问"请提供订单号"# 用户关闭页面，去翻找订单号# 十分钟后，用户带着订单号回来了# 再次调用，同一个thread_idresult=app.invoke({"user_input":"订单号是TB20240615001"},config# 同一个thread_id)# LangGraph会自动加载user_9527的历史状态# 发现上一轮在collect节点等待订单号# 新的输入被合并到collected_info里# 然后自动流转到下一个节点

这就是断点续传的魔力：用户不需要重复已经说过的话，系统记得你们聊到哪了。

查看历史状态

# 查看当前会话的最新状态snapshot=app.get_state(config)print(f"当前状态：{snapshot.values}")print(f"下一个待执行节点：{snapshot.next}")# 查看完整历史history=list(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轮：用户输入state=app.invoke({"user_input":"我要退款","collected_info":{},"turn_count":0},config)# 输出：AI追问"请提供订单号和退款原因"print(state["messages"][-1]["content"])# 第2轮：用户提供订单号（模拟断点续传）state=app.invoke({"user_input":"订单号是TB20240615001"},config)# AI继续追问退款原因print(state["messages"][-1]["content"])# 第3轮：用户说明原因state=app.invoke({"user_input":"商品质量有问题，屏幕有坏点"},config)# 信息收集全了，进入知识库检索，返回退款流程print(state["messages"][-1]["content"])# 第4轮：用户不满意，提到"我要投诉到工商局"state=app.invoke({"user_input":"你们处理太慢了，我要投诉到工商局"},config)# 敏感词触发，进入human_handoff_node# interrupt暂停，等待人工介入if"__interrupt__"instr(state):print("【系统】已转人工客服，请等待...")# 模拟人工客服处理完后恢复fromlanggraph.typesimportCommand final=app.invoke(Command(resume={"feedback":"您好，我是客服主管，已为您加急处理，预计2小时内退款到账。"}),config)print(final["messages"][-1]["content"])

这个示例展示了系统的全部能力：

1. 循环追问：用户没给全信息时，AI反复问；
2. 断点续传：用户中途离开再回来，对话无缝衔接；
3. 敏感词升级：提到"工商局"自动转人工；
4. interrupt协作：人工处理完后，系统恢复执行，把人工反馈带给用户。

八、写在最后：从"问答机器人"到"服务系统"

写完这个项目，我对LangGraph最大的感受是：它让AI客服从"问答机器人"进化成了"服务系统"。

问答机器人的逻辑是：问→答，结束。服务系统的逻辑是：识别→收集→判断→检索→人工→结束，中间有循环、有分支、有暂停、有恢复。

LangGraph的StateGraph让我们能精确控制这个流程的每一个分支。条件边不是黑盒，是写死在代码里的业务规则。interrupt不是魔法，是显式的暂停点。checkpointer不是抽象概念，是实实在在的状态快照。

如果你还在用Chain硬凑多轮对话，或者用Agent黑盒赌运气，不妨试试LangGraph。当你第一次画出一张完整的客服流程图，看到用户在不同节点之间流转，看到断点续传无缝恢复，看到人工客服优雅地介入——你会明白：这不是又一个框架，这是AI服务工程的成人礼。