1. 项目概述构建带审批流程的智能客服系统最近在帮一家金融公司改造他们的智能客服系统时遇到个有意思的需求某些敏感业务场景比如账户余额查询、交易记录调取需要人工审核才能继续执行而且当审批通过后系统要能自动恢复之前的对话流程。这让我开始研究如何用LangGraph搭建支持审批与恢复执行的客服Agent今天就把实战经验分享给大家。传统客服Agent遇到审批环节时往往直接中断会话等人工审核后需要用户重新描述需求。而我们要实现的系统会在审批通过后自动回到中断点继续执行未完成的操作流程。想象一下这样的场景用户查询近半年交易记录→触发风控审批→专员审核通过→系统自动返回查询结果。整个过程无需用户重复操作体验流畅得像没有中断过一样。2. 技术选型与架构设计2.1 为什么选择LangGraph相比LangChainLangGraph最大的优势是支持有状态的多步骤工作流。它的Graph特性让我们可以用节点(Node)表示每个处理步骤如意图识别、信息提取用边(Edge)控制流程跳转逻辑通过持久化检查点(Checkpoint)实现流程中断与恢复from langgraph.graph import Graph workflow Graph() # 定义节点 workflow.add_node(intent_recognizer, recognize_intent) workflow.add_node(approval_checker, check_approval_required) workflow.add_node(human_approval, send_for_approval) # 定义边 workflow.add_edge(intent_recognizer, approval_checker) workflow.add_conditional_edge( approval_checker, lambda x: requires_approval if x[needs_approval] else auto_pass, {requires_approval: human_approval, auto_pass: end} )2.2 审批流程的三种实现模式根据业务需求不同我们设计了三种审批集成方案方案类型触发条件审批方式恢复机制硬拦截式敏感关键词匹配强制人工审批审批通过后重放完整对话软拦截式风险评分阈值人工审批或自动放行仅执行未完成步骤混合式规则模型判断分级审批初级/高级选择性步骤回滚金融场景推荐使用混合式方案。比如当用户要求查询最近三个月交易记录并导出Excel时查询操作触发初级审批客服组长审核导出操作触发高级审批风控专员审核每个审批通过后只执行对应环节3. 核心实现细节3.1 状态持久化设计实现中断恢复的关键是完整保存对话上下文。我们采用三级存储策略class ApprovalState: def __init__(self): self.current_node # 当前执行节点 self.history [] # 已完成的步骤日志 self.pending_data {} # 待审批数据 self.context {} # 完整对话上下文 # 使用Redis持久化状态 import pickle import redis r redis.Redis() def save_state(session_id, state): r.set(fapproval:{session_id}, pickle.dumps(state)) def load_state(session_id): data r.get(fapproval:{session_id}) return pickle.loads(data) if data else None重要提示状态对象必须包含完整的可序列化数据避免存储LangChain的Memory对象这类不可序列化的内容。3.2 审批接口实现审批系统通常需要提供HTTP接口供人工处理。这里给出FastAPI的实现示例from fastapi import FastAPI, HTTPException from pydantic import BaseModel app FastAPI() class ApprovalRequest(BaseModel): session_id: str approve: bool comment: str None app.post(/approve) async def handle_approval(req: ApprovalRequest): state load_state(req.session_id) if not state: raise HTTPException(404, Session expired) if req.approve: # 恢复执行 workflow rebuild_workflow(state.current_node) result await workflow.arun(state.context) return {status: resumed, result: result} else: # 终止流程 delete_state(req.session_id) return {status: rejected}4. 实战中的经验技巧4.1 审批超时处理我们发现30%的审批请求会在2小时后才得到处理这时需要特殊处理设置TTL自动过期比如24小时恢复时检查数据有效性如查询条件是否仍然合理添加超时提醒模板def check_timeout(state): elapsed time.time() - state.last_update if elapsed 3600 * 2: # 2小时 send_notification( f审批任务#{state.session_id}已超时, f请及时处理{get_approval_url(state.session_id)} )4.2 可视化调试技巧LangGraph自带LangSmith集成但审批流程需要额外记录为每个审批请求创建独立Trace记录审批人和决策原因标注流程恢复点from langsmith import Client client Client() def log_approval(session_id, action, metadata): client.create_run( nameapproval_decision, inputs{session_id: session_id}, outputs{action: action}, metadatametadata )5. 典型问题排查指南5.1 状态恢复失败症状审批通过后流程没有从正确节点恢复检查点是否包含足够上下文至少需要当前节点、输入参数、历史路径确认重建工作流时使用了相同的节点注册顺序验证Redis的序列化/反序列化过程没有数据丢失5.2 审批通知未触发排查步骤检查审批节点是否正确设置了输出边验证消息服务邮件/SMS的API密钥查看LangGraph的异常日志# 查看最近10条错误日志 docker logs --tail 10 langgraph-server | grep approval5.3 性能优化建议当审批流程较长时超过5个节点建议对状态对象进行压缩如使用zlib只持久化必要的上下文字段对大文件附件使用对象存储引用import zlib def compress_state(state): return zlib.compress(pickle.dumps(state)) def decompress_state(data): return pickle.loads(zlib.decompress(data))这个方案已经在生产环境处理了超过12万次带审批的客服请求平均审批响应时间从原来的47分钟降低到19分钟。最让我惊喜的是通过合理设置检查点流程恢复成功率达到了99.3%。如果你正在构建需要人工介入的智能客服系统不妨试试这个方案。