当前位置：首页 > news >正文

【LangGraph从入门到精通】010、实战项目：从零构建一个企业级智能客服工单系统

news 2026/6/18 1:04:17

010、实战项目：从零构建一个企业级智能客服工单系统

一、那个让我加班的工单路由问题

上周三晚上十点，我还在调试一个诡异的工单路由逻辑——客户明明选择了“退款问题”，工单却跑到了“技术故障”的客服组。查了半天，发现是关键词匹配时把“退不出款”里的“退款”和“出款”拆成了两个意图。这种问题在传统规则系统里太常见了：规则越写越多，漏洞越补越大，最后变成一坨没人敢动的祖传代码。

这就是为什么我要用LangGraph重构这套系统。传统工单系统像是个死板的流水线，而我们需要的是个能思考、会判断、还能自我调整的智能体工作流。

二、架构设计：别把智能体当函数调用

先看我们最终要的架构长什么样：

# 工单处理的核心工作流定义# 注意：这不是完整代码，先看结构classTicketWorkflow:def__init__(self):# 关键在这里——用StateGraph而不是一堆if-elseself.workflow=StateGraph(TicketState)# 定义节点self.workflow.add_node("classify_intent",self.classify_intent)self.workflow.add_node("extract_entities",self.extract_entities)self.workflow.add_node("route_ticket",self.route_ticket)self.workflow.add_node("human_escalation",self.human_escalation)# 设置条件边——这是智能决策的核心self.workflow.add_conditional_edges("classify_intent",self.should_extract_entities,# 这个函数返回下一个节点名{"extract":"extract_entities","direct_route":"route_ticket","human":"human_escalation"})

很多新手容易犯的错是把智能体节点写成普通的函数调用。记住：每个节点都应该是个独立的“智能体单元”，它有自己的记忆、工具和决策能力。

三、实现细节：这些坑我都踩过

3.1 工单状态设计

fromtypingimportTypedDict,List,Optional,Annotatedfromlanggraph.graph.messageimportadd_messagesclassTicketState(TypedDict):"""工单状态——这是整个工作流的共享内存"""customer_id:stroriginal_text:str# 用户原始描述intent:Optional[str]# 识别出的意图entities:Annotated[List[dict],add_messages]# 提取的实体urgency:int# 紧急程度 1-5department:Optional[str]# 分配部门requires_human:bool# 是否需要人工介入history:List[str]# 处理历史记录# 踩坑提醒：别在这里放LLM实例或数据库连接# 状态应该是可序列化的，否则持久化会报错

状态设计要遵循“最小必要信息”原则。早期版本我把整个用户会话历史都塞进去，结果状态对象膨胀到几十KB，内存直接炸了。

3.2 意图分类节点实现

asyncdefclassify_intent(state:TicketState)->dict:"""智能意图分类——这里用了本地小模型+规则兜底"""# 先尝试用本地模型快速分类（省钱啊兄弟们）local_result=awaitfast_local_model.predict(state["original_text"])iflocal_result["confidence"]>0.8:# 置信度高就直接用intent=local_result["intent"]else:# 置信度低就调用GPT-4（这钱不能省）# 但要注意：别在提示词里直接写“请分类”——要给出明确格式prompt=f""" 用户问题：{state['original_text']}请从以下选项中选择最匹配的意图： - 退款问题 - 技术故障 - 账户问题 - 产品咨询 - 投诉建议 只返回意图名称，不要解释。 """# 这里有个细节：设置temperature=0.1，让输出更稳定response=awaitopenai_chat(prompt,temperature=0.1)intent=response.strip()# 记录处理历史（调试用）state["history"].append(f"意图分类结果：{intent}")return{"intent":intent,"requires_human":False}

注意那个temperature参数——生产环境一定要调低。之前设成0.7，同样的工单今天分到A部门明天分到B部门，客服组长差点找我打架。

3.3 条件路由的逻辑

defshould_extract_entities(state:TicketState)->str:"""决定下一步怎么走——这是工作流的大脑"""# 情况1：用户明确要求转人工if"转人工"instate["original_text"]or"找真人"instate["original_text"]:return"human"# 情况2：简单咨询直接路由simple_intents=["产品咨询","订单状态"]ifstate["intent"]insimple_intents:return"direct_route"# 情况3：复杂问题需要实体提取complex_intents=["退款问题","技术故障","投诉建议"]ifstate["intent"]incomplex_intents:return"extract"# 默认兜底：转人工# 重要原则：不确定时就转人工，别让AI瞎猜return"human"

这个函数我重构了三次。第一次写得太复杂，各种嵌套判断；第二次又太简单，漏了很多边界情况。现在的版本遵循“明确规则优先，AI判断兜底”的原则。

四、持久化与监控：别等上线了才想起来

# 工作流持久化配置defsetup_persistence(graph):"""配置检查点和持久化——没这个功能就别上线"""# 使用内存存储（开发环境）# memory = MemorySaver()# 生产环境一定要用数据库# 我用的PostgreSQL，别用SQLite，并发撑不住fromlanggraph.checkpoint.postgresimportPostgresSaver checkpoint=PostgresSaver.from_conn_string(conn_string=os.getenv("DB_URL"),table_name="ticket_checkpoints"# 表名要有业务含义)# 编译图时绑定持久化app=graph.compile(checkpointer=checkpoint)# 设置检查点策略：每个节点执行后都保存app.checkpoint_config=CheckpointConfig(at=CheckpointAt.END_OF_STEP,# 每个步骤结束都保存root_type="return")returnapp# 监控埋点asyncdeftrack_node_performance(node_name:str,start_time:float):"""记录每个节点的执行时间和成功率"""duration=time.time()-start_time# 推送到监控系统metrics_client.timing(f"ticket.node.{node_name}.duration",duration)# 超过3秒要报警ifduration>3.0:alert(f"节点{node_name}执行超时:{duration:.2f}秒")