LangGraph构建带审批流程的智能客服系统实战

1. 项目概述：构建带审批流程的智能客服系统

最近在帮一家金融公司改造他们的智能客服系统时，遇到个有意思的需求：某些敏感业务场景（比如账户余额查询、交易记录调取）需要人工审核才能继续执行，而且当审批通过后，系统要能自动恢复之前的对话流程。这让我开始研究如何用LangGraph搭建支持审批与恢复执行的客服Agent，今天就把实战经验分享给大家。

传统客服Agent遇到审批环节时往往直接中断会话，等人工审核后需要用户重新描述需求。而我们要实现的系统会在审批通过后，自动回到中断点继续执行未完成的操作流程。想象一下这样的场景：用户查询近半年交易记录→触发风控审批→专员审核通过→系统自动返回查询结果。整个过程无需用户重复操作，体验流畅得像没有中断过一样。

2. 技术选型与架构设计

2.1 为什么选择LangGraph？

相比LangChain，LangGraph最大的优势是支持有状态的多步骤工作流。它的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"时：

1. 查询操作触发初级审批（客服组长审核）
2. 导出操作触发高级审批（风控专员审核）
3. 每个审批通过后只执行对应环节

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(f"approval:{session_id}", pickle.dumps(state)) def load_state(session_id): data = r.get(f"approval:{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小时后才得到处理，这时需要特殊处理：

1. 设置TTL自动过期（比如24小时）
2. 恢复时检查数据有效性（如查询条件是否仍然合理）
3. 添加超时提醒模板：

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集成，但审批流程需要额外记录：

1. 为每个审批请求创建独立Trace
2. 记录审批人和决策原因
3. 标注流程恢复点

from langsmith import Client client = Client() def log_approval(session_id, action, metadata): client.create_run( name="approval_decision", inputs={"session_id": session_id}, outputs={"action": action}, metadata=metadata )

5. 典型问题排查指南

5.1 状态恢复失败

症状：审批通过后流程没有从正确节点恢复

• 检查点是否包含足够上下文（至少需要：当前节点、输入参数、历史路径）
• 确认重建工作流时使用了相同的节点注册顺序
• 验证Redis的序列化/反序列化过程没有数据丢失

5.2 审批通知未触发

排查步骤：

1. 检查审批节点是否正确设置了输出边
2. 验证消息服务（邮件/SMS）的API密钥
3. 查看LangGraph的异常日志：

# 查看最近10条错误日志 docker logs --tail 10 langgraph-server | grep "approval"

5.3 性能优化建议

当审批流程较长时（超过5个节点），建议：

1. 对状态对象进行压缩（如使用zlib）
2. 只持久化必要的上下文字段
3. 对大文件附件使用对象存储引用

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%。如果你正在构建需要人工介入的智能客服系统，不妨试试这个方案。