最近在做一个智能客服项目老系统在高并发下经常卡顿扩展起来也特别麻烦。调研了一圈最终决定用 LangGraph 来重构。经过一番折腾新系统不仅吞吐量上去了响应也快了不少。今天就来分享一下我的实战经验希望能帮到有类似需求的同学。1. 为什么选择 LangGraph先聊聊传统方案的痛点在动手之前我们得先搞清楚老系统为什么不行。我们之前的客服系统是基于一个流行的对话框架这里就不点名了搭建的主要遇到了下面几个问题状态管理混乱对话状态比如用户问到了哪一步、填了什么信息散落在各个地方有的在内存里有的在数据库里维护起来像在玩“大家来找茬”。一旦需要增加新的业务分支改代码就特别痛苦。扩展性差系统是单体架构所有逻辑耦合在一起。当用户量上来想通过加机器来分担压力发现很多状态是单点存储的没法简单地水平扩展。响应慢并发低处理用户请求的流程是同步的一个请求没处理完后面的就得等着。遇到需要调用外部接口比如查订单、查库存的时候等待时间就更长了直接影响了用户体验。基于这些痛点我开始寻找新的解决方案。对比了 Rasa、Dialogflow 和 LangGraphRasa开源定制能力强但它的对话管理Tracker Store在复杂、高并发的场景下状态同步和持久化容易成为瓶颈自己优化的工作量不小。Dialogflow谷歌家的开箱即用对于标准问答很友好。但它的定制化程度有限想把我们的复杂业务逻辑比如结合内部CRM系统做精准推荐嵌进去就像戴着镣铐跳舞不太灵活。LangGraph它本质上是一个用于构建有状态、多步骤应用的工作流框架。它的核心优势在于“图”的思维。你可以把客服对话的每一步接收问题、理解意图、查询知识库、生成回复等定义成图上的节点用边来控制流程走向。这种设计天然就是模块化、可编排的状态流转清晰而且它和 LangChain 生态结合紧密处理大语言模型LLM的调用非常顺手。对于我们这种需要高度定制业务流并且对性能有要求的场景LangGraph 成了更合适的选择。2. 核心实现用 LangGraph 搭建模块化客服引擎确定了技术选型接下来就是动手搭建。我们的目标是构建一个清晰、可维护、高性能的对话引擎。2.1 模块化架构设计我们把一个完整的客服对话流程抽象成一张有向图。每个节点负责一个单一的职责节点之间通过边连接。这样做的好处是高内聚低耦合每个节点的功能独立修改一个节点不会影响其他。易于测试可以单独对每个节点进行单元测试。动态编排可以根据不同的用户意图动态选择不同的节点路径实现复杂的业务流程。我们的核心图主要包含以下几类节点路由节点 (Router)分析用户输入判断意图是咨询产品、售后问题还是闲聊。知识库查询节点 (Knowledge Search)针对产品咨询类问题去向量数据库里检索最相关的答案。业务API调用节点 (API Caller)针对需要实时数据的请求比如“我的订单到哪了”调用后端的订单服务API。LLM生成节点 (Response Generator)将检索到的信息或API返回的数据组织成自然、友好的回复。状态管理节点 (State Manager)负责对话状态的更新、保存和加载。2.2 异步消息处理流程代码示例LangGraph 支持异步操作这对于提升并发能力至关重要。下面是一个简化版的核心图构建代码import asyncio from typing import Annotated, TypedDict from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langgraph.checkpoint.aiosqlite import AsyncSqliteSaver # 1. 定义对话状态结构 class ChatState(TypedDict): messages: Annotated[list, add_messages] # 消息历史 user_intent: str # 识别出的用户意图 retrieved_info: list # 从知识库检索到的信息 api_result: dict # 业务API调用结果 needs_human: bool # 是否需要转人工 # 2. 定义各个节点函数异步 async def intent_router_node(state: ChatState) - ChatState: 路由节点分析用户意图 latest_message state[“messages”][-1].content # 这里可以接入一个轻量级的意图分类模型例如用 fasttext 或 small LLM # 为了示例我们简单判断 if “订单” in latest_message: state[“user_intent”] “query_order” elif “怎么用” in latest_message or “如何” in latest_message: state[“user_intent”] “product_qa” else: state[“user_intent”] “general_chat” return state async def knowledge_search_node(state: ChatState) - ChatState: 知识库查询节点 if state[“user_intent”] “product_qa”: query state[“messages”][-1].content # 异步调用向量数据库检索 (假设使用 async client) # results await vector_db.asimilarity_search(query, k3) state[“retrieved_info”] [“检索到的答案片段1”, “片段2”] # 模拟结果 return state async def api_caller_node(state: ChatState) - ChatState: 业务API调用节点 if state[“user_intent”] “query_order”: # 异步调用外部HTTP API # async with aiohttp.ClientSession() as session: # async with session.post(‘ORDER_API_URL’, json{…}) as resp: # state[‘api_result’] await resp.json() state[“api_result”] {“status”: “已发货”, “tracking_no”: “SF123456”} # 模拟结果 return state async def response_generator_node(state: ChatState) - ChatState: 回复生成节点 context_parts [] if state[“retrieved_info”]: context_parts.extend(state[“retrieved_info”]) if state[“api_result”]: context_parts.append(str(state[“api_result”])) user_question state[“messages”][-1].content # 构建LLM提示词将上下文和问题交给LLM生成最终回复 prompt f“”” 基于以下信息以客服的身份友好地回答用户问题。 信息{‘; ‘.join(context_parts) if context_parts else ‘无额外信息’} 用户问题{user_question} 回复 “”” # 异步调用LLM (例如通过 LangChain) # llm_response await llm.ainvoke(prompt) llm_response “根据您的订单号SF123456查询您的包裹已发货正在运输中。” # 模拟回复 # 将助手的回复添加到消息历史中 state[“messages”].append({“role”: “assistant”, “content”: llm_response}) return state # 3. 定义条件边函数控制流程走向 def decide_next_step(state: ChatState) - str: 根据意图决定下一个节点 intent state[“user_intent”] if intent “product_qa”: return “search_knowledge” elif intent “query_order”: return “call_api” else: return “generate_response” # 4. 构建图 builder StateGraph(ChatState) # 添加节点 builder.add_node(“router”, intent_router_node) builder.add_node(“search_knowledge”, knowledge_search_node) builder.add_node(“call_api”, api_caller_node) builder.add_node(“generate_response”, response_generator_node) # 设置入口点 builder.set_entry_point(“router”) # 添加边包括条件边 builder.add_conditional_edges( “router”, decide_next_step, # 这个函数返回下一个节点的名字 { “search_knowledge”: “search_knowledge”, “call_api”: “call_api”, “generate_response”: “generate_response” } ) builder.add_edge(“search_knowledge”, “generate_response”) builder.add_edge(“call_api”, “generate_response”) builder.add_edge(“generate_response”, END) # 5. 配置持久化检查点用于管理对话状态 memory AsyncSqliteSaver.from_conn_string(“:memory:”) # 生产环境换成持久化DB graph builder.compile(checkpointermemory) # 6. 使用图处理对话 async def handle_user_message(session_id: str, user_input: str): # 初始化或加载该会话的状态 initial_state: ChatState {“messages”: [{“role”: “user”, “content”: user_input}], …其他字段默认值} config {“configurable”: {“thread_id”: session_id}} # 异步执行图 final_state await graph.ainvoke(initial_state, configconfig) return final_state[“messages”][-1].content # 返回最新的一条助手回复 # 示例调用 # asyncio.run(handle_user_message(“user_123”, “我的订单到哪里了”))2.3 对话状态管理的实现方案状态管理是智能客服的核心。LangGraph 的Checkpointer机制帮我们优雅地解决了这个问题。检查点 (Checkpoint)在图执行到某个节点后系统可以自动保存当前完整的状态包括所有变量、消息历史。我们选择了AsyncSqliteSaver生产环境可用 PostgreSQL 适配器将会话状态持久化到数据库。会话隔离每个用户对话通过唯一的thread_id即上面的session_id标识。每次调用graph.ainvoke时传入对应的配置LangGraph 会自动加载该会话的上一个状态并在执行后保存新状态。优势容错与恢复如果某次请求失败下次可以基于上一个成功的检查点继续执行不会丢失对话上下文。支持长对话状态被持久化不受服务重启影响可以处理跨越很长时间的多轮对话。便于调试可以查看数据库中保存的任意历史状态方便追踪问题。3. 性能优化让系统跑得更快更稳架构搭好了接下来就是让它经受高并发的考验。我们主要做了以下几方面优化。3.1 负载测试与结果我们使用 JMeter 模拟了从 50 到 1000 的并发用户持续压测 10 分钟。优化前旧系统在 200 并发时平均响应时间RT就超过 2 秒错误率开始攀升。优化后LangGraph 新系统吞吐量 (Throughput)提升了约 3 倍。这主要归功于异步架构I/O 密集型操作网络请求、数据库查询不再阻塞工作线程。响应时间 (P99)99% 的请求响应时间稳定在 500 毫秒以内。我们将耗时长的操作如 LLM 调用、向量检索全部异步化并设置了合理的超时时间。资源利用率CPU 使用更加平稳避免了同步模式下的“锯齿状”高峰。3.2 内存泄漏预防措施在异步和长期运行的服务中内存泄漏是需要警惕的。循环引用检查Python 的异步任务 (asyncio.Task) 或回调函数如果持有对大型对象如对话状态的引用且未正确取消或释放可能导致泄漏。我们定期使用tracemalloc或objgraph等工具进行快照对比排查异常增长的对象。限制上下文长度对话历史 (state[‘messages’]) 会无限增长。我们设置了一个阈值例如最近20轮对话在生成回复或保存检查点前自动截断过旧的历史只保留最近的对话和必要的摘要信息。使用弱引用在某些缓存场景如意图分类模型缓存对于不必须长期存活的对象考虑使用weakref。连接池管理对于数据库、Redis、HTTP 客户端连接确保使用连接池并在任务结束后正确归还连接避免连接数耗尽和内存累积。4. 生产环境避坑指南把系统部署上线才是真正的开始。这里总结几个我们踩过的坑和解决方案。4.1 会话超时处理用户可能中途离开留下大量“僵尸”会话占用资源。实现方案我们启动了一个后台的清理任务定期扫描状态数据库。对于超过一定时间如30分钟没有更新的会话将其状态归档到冷存储如对象存储并从活跃状态表中删除。当用户再次发起请求时如果发现是已归档的会话可以提供一个温和的提示如“您之前的对话已超时我们可以重新开始吗”。4.2 异常流程设计不是所有对话都会按照理想路径进行。节点异常捕获在每个节点函数内部用try…except包裹核心逻辑。发生异常时不是直接崩溃而是将错误信息记录到状态中并跳转到一个专用的“异常处理节点”。这个节点可以尝试重试、降级例如返回一个兜底的通用回复或者明确告知用户“服务暂时不可用请稍后再试”。LLM调用稳定性LLM API 可能不稳定。我们实现了指数退避的重试机制并设置了严格的超时如10秒。如果连续失败则切换到备用方案如返回预定义的常见问题答案。4.3 监控指标配置没有监控的系统就是在“裸奔”。我们配置了以下几类关键指标业务指标每日会话量、意图分布、转人工率、用户满意度如果有评分入口。性能指标各节点平均处理时长、P95/P99响应时间、图执行成功率、异步任务队列深度。系统指标服务内存使用量、数据库连接数、检查点读写延迟。异常告警针对节点执行失败率、LLM API 错误率、响应时间超阈值等情况设置告警以便第一时间介入处理。5. 总结与思考通过这次 LangGraph 的实践我们成功构建了一个响应迅速、易于扩展的智能客服系统。它的“图”抽象非常贴合对话这种多步骤、有状态的业务场景模块化设计也让后续的迭代和维护变得清晰。当然挑战依然存在。一个开放性的问题值得我们继续探索如何处理多轮对话中的“意图漂移”例如用户一开始在咨询产品A的用法聊了几句后突然问“那产品B和它比怎么样”。系统需要能识别这种话题的切换并更新对话状态和后续的流程走向。我们目前的方案是在每一轮都重新进行意图识别并结合对话历史进行加权判断。更复杂的方案可能涉及维护一个更精细的对话主题栈或者利用LLM本身对上下文的理解能力来动态调整图的路径。技术的选择没有银弹LangGraph 为我们提供了强大的基础设施但如何设计出真正智能、流畅的对话流程还需要我们对业务和用户体验有更深的理解。希望这篇笔记能给你带来一些启发欢迎一起交流探讨。