简介这是一套面向AI工程化落地场景的电商智能客服系统实战框架专为具备Python与LLM应用开发基础的中高级开发者设计解决电商领域多轮对话、RAG知识增强与复杂业务流程自动化协同难题。资源共76个文件以60个Python核心模块为主涵盖LangGraph智能体编排、FastAPI路由、RAG检索器、退货子图逻辑、外部服务工具封装等辅以YAML流程定义、Shell部署脚本、测试用例及Docker/K8s配置文件压缩包仅1.01MB轻量但结构完整。项目采用清晰分层架构agents目录定义多智能体协作图谱tools目录封装订单/物流/支付等业务系统对接能力retrievers实现多源知识融合检索tests目录提供覆盖退货全链路的20单元与集成测试用例。读者可直接获得可运行的AI Agent生产级代码骨架、标准化API文档、CI/CD流水线配置及运维脚本快速复现从语义检索到退款闭环的端到端智能客服能力。 去年年底我们团队接手了一个电商平台的客服系统升级项目最初只是想把传统的检索式FAQ换成一个稍微“聪明”一点的问答系统结果越做越深最后落地成了一个基于 LangGraph 编排、FastAPI 提供服务的智能客服 Agent 框架。从最开始的 RAG 知识库问答到后来把退货退款这类复杂的多轮业务流程也纳入了 Agent 的管辖范围整个过程踩了不少坑也积累了不少一手经验。这套框架现在已经在生产环境稳定跑了一段时间支撑了三个不同品类的店铺客服入口我打算把整个设计思路和实现细节好好拆解一遍。如果你正在做 AI Agent 开发或者准备用 LangGraph 搭一个有实际业务闭环的客服系统而不是停留在“做个聊天机器人”的玩具阶段这篇文章应该能帮你省下几周的时间。我会把架构选型的原因、状态图的设计逻辑、RAG 的工程化细节以及 FastAPI 层怎么跟 LangGraph 无缝衔接都交代清楚最后还会分享一些生产环境实测踩过的坑。1. 为什么最终选了 LangGraph FastAPI而不是 LangChain 一把梭很多人一上来就问 LangChain 和 LangGraph 的区别到底是什么我先用一句话说清楚LangChain 提供的是“组件”LangGraph 提供的是“状态流转框架”。如果你要做的只是调一次 LLM 加一个检索器那 LangChain 自带的链式调用完全够用。但电商客服这种场景用户的意图是动态变化的上一句还在问“这个商品有没有运费险”下一句就变成了“我要退货”而且退货流程本身还分好几个步骤每一个步骤都需要维护上下文状态。这种场景下那套静态的 Chain 就非常别扭。LangGraph 的核心价值是把 Agent 的每一次响应建模为一张图节点是具体的处理逻辑边是状态流转的条件。它会维护一个全局的 State 对象每个节点执行完都会更新这个 State图的下一次执行基于最新的 State 来决定下一步走哪条边。这种机制天然适合客服系统这种“对话轮次之间需要记忆”的场景尤其是当流程中存在条件分支甚至需要人工介入的时候LangGraph 的表现比 LangChain 那种线性链式调用强太多了。FastAPI 这边就更好解释它是目前 Python 生态里写异步接口最顺手的框架之一。客服后端的高并发场景需要异步 IO 来撑连接数而 LangGraph 本身支持 async 模式二者结合非常顺畅。再加上我们的 Agent 需要流式输出前端希望用户看到的是一段一段的字往外蹦而不是等十几秒出一个完整的长篇大论这个需求在 FastAPI 里用 StreamingResponse 就能比较优雅地解决。当时还考虑过一个更重的方案是直接把 LangGraph 的服务嵌到消息队列后面比如用 Celery 跑 agent 任务。但后来想明白了客服场景本质上还是请求-响应的交互模式不需要异步任务队列那么重。相比之下FastAPI 直接同步调用 LangGraph 的图加上内存态的历史会话存储响应速度完全没有问题。当然如果未来要做异步通知类客服比如“物流异常主动推送”那再引入队列也不迟现在没必要过度设计。2. 系统整体架构从用户请求到退货处理完成的完整链路这一节我把整个系统的部署架构和请求流转过程画个全景轮廓你可以先有个整体感后面每个模块会单独展开。整个服务分为三层接入层、Agent 编排层、以及外接服务层。接入层就是 FastAPI 那层负责接收用户的 HTTP 请求、管理 Session 的生命周期、调用 Agent 图、把结果以 SSE 流式返回给前端。这一层不处理任何业务逻辑只做参数校验、会话恢复和响应包装。Agent 编排层是 LangGraph 的核心图由五个关键节点组成意图识别节点判断用户当前诉求是售前咨询、售后查询、还是退货申请记忆管理节点压缩并维护多轮对话的上下文控制发给 LLM 的 token 量RAG 检索节点针对商品知识、退货政策这类文档进行检索增强生成退货流程子图节点当意图为退货申请时进入专门的状态机子图兜底转人工节点彻底无法处理时生成转人工工单。外接服务层则包括商品数据库、订单系统、退货/退款处理系统的 API以及一套预置的知识库向量数据库。Agent 需要通过工具调用的方式去访问这些服务而不是直接把数据库暴露给 LLM。请求的完整路径是这样的前端把用户消息发给 FastAPIFastAPI 从 Redis 里读取该会话的历史消息并组装成 LangGraph 的初始 State随后调用编译好的图执行意图识别节点首先跑完根据意图决定走哪条分支。如果是知识类的问题就走 RAG 检索节点检索完毕生成回答并返回如果是退货类需求则切换到退货流程子图子图内部按步骤轮询用户输入、校验订单信息、读取退货政策、调订单系统创建退货单最终把处理结果汇总到 State再由 FastAPI 统一流式返回给前端。有一点必须特别强调LangGraph 的 State 和前端交互用的“会话历史”并不是同一个东西需要做映射。State 里存的不止是对话消息还包括当前用户 ID、意图识别结果、退货单状态、检索到的文档列表这类过程数据。而前端只需要看到 messages 字段所以 FastAPI 返回结果的模板要把这两部分做隔离避免把内部状态暴露给前端。3. 电商客服 Agent 的 LangGraph 核心图设计3.1 用 Pydantic 定义清晰的 State这是整个图的地基LangGraph 里 State 的定义方式直接决定了后面所有节点能读到什么、能改什么。如果 State 设计得不好后面做条件分支的时候会异常痛苦。我建议直接用 TypedDict 或者 Pydantic BaseModel 来定义Pydantic 的校验能力会帮你少踩很多坑。下面是我们当时定义的简化版 State你可以参考一下这个结构from typing import TypedDict, Annotated, Literal from langgraph.graph.message import add_messages class AgentState(TypedDict): # 会话基础信息 user_id: str session_id: str # 对话消息langgraph 内置的 add_messages reducer 会自动追加 messages: Annotated[list, add_messages] # 意图识别结果由意图识别节点写入 intent: str # RAG 检索到的参考文档用于生成回答时的引用 retrieved_docs: list # 退货子图状态 return_order_id: str return_reason: str return_step: str # 最终响应 final_answer: str # 是否需要转人工 need_human: bool有几个设计要点我想多说一句。用 TypedDict 可以但我更倾向于 Pydantic BaseModel原因很简单你可以在节点里面做字段校验比如退货单号必须是数字开头之类的规则如果 State 里的数据不符合预期Pydantic 会直接抛出来比 LangGraph 里面因为字段缺失而出现诡异的 runtime 错误要容易排查得多。另外 messages 字段的 reducer 用的是add_messages这个非常关键。LangGraph 的消息列表不像 Python 普通列表那样每次覆盖它会基于消息 ID 自动去重合并。如果你自己写 reducer 搞消息追加很快你会在多轮对话中发现消息重复出现或者顺序错乱这个坑我已经替你踩过了。3.2 意图识别节点分类要粗粒度但必须保证可解释在 LangGraph 里每个节点就是一个 async 函数或者普通函数输入是整个 State输出是 State 的增量修改。意图识别节点是我们图的第一个节点它的作用就是判断当前用户这条消息到底想干什么然后把判断结果写到 State 里的intent字段。关于意图分类的粒度我强烈建议你保持粗粒度不要搞出一百多个细分类别。电商客服场景下五到八个意图就完全够用了SHOPPING商品咨询、比价、推荐ORDER_QUERY订单物流查询RETURN_REQUEST退货退款申请POLICY_QUERY退换货政策、运费险咨询HUMAN_FALLBACK情绪激烈或意图不明CHITCHAT闲聊寒暄。为什么不搞得太细因为意图识别的准确率直接决定了后面所有流程的走向如果分类太细模型判断错误的概率就会上升一旦把退货请求识别成了商品咨询用户就会被带到完全错误的分支里体验会非常糟糕。粗粒度的意图加上后续的 slot filling才是更稳的工程方案。实现上第一版我们用过单独的 LLM 调用加 JSON 输出解析后来发现稳定性不够就改成了 few-shot prompt 加 function calling 的方式INTENT_SYSTEM_PROMPT 你是电商客服意图识别器。请根据用户消息从以下意图中选择一个输出 JSON 格式结果。 意图列表 - SHOPPING商品咨询、推荐、比价 - ORDER_QUERY订单状态、物流查询 - RETURN_REQUEST退货、退款、换货申请 - POLICY_QUERY退换货政策、运费险、售后规则 - HUMAN_FALLBACK用户情绪强烈、要求人工、其他无法分类的复杂问题 - CHITCHAT日常闲聊 输出格式{{intent: 意图标签, reason: 简短判断理由}} async def intent_node(state: AgentState) - AgentState: # 取最近一条用户消息 last_message state[messages][-1].content # 调用 LLM 进行意图识别 response await llm.ainvoke([ {role: system, content: INTENT_SYSTEM_PROMPT}, {role: user, content: last_message} ]) parsed json.loads(response.content) return {intent: parsed[intent], intent_reason: parsed[reason]}你可能想问为什么不用正则或者规则匹配说实话我试过。对于“退货”“退款”这种关键词正则确实能拦下不少。但真实用户说话太随意了比如“这东西我不想要了要怎么办”你很难用关键词覆盖全部表达。最后我们采用的是规则粗筛 LLM 精判的双保险先跑一遍轻量规则如果命中高置信度的意图就直接返回否则再调用 LLM 判断。能省不少 token响应速度也更快。3.3 条件边LangGraph 的分支灵魂别让所有节点都跑一遍LangGraph 的条件边是图中真正有价值的部分。它允许你根据当前 State 的内容动态决定下一步要执行哪个节点这就把“流程控制”从代码里抽离出来变成一张可以可视化的图。我这里给出一个条件边的示例背后的控制逻辑是根据意图字段决定下一个节点是 RAG 检索、退货子图还是直接转人工。from langgraph.graph import StateGraph, END def route_by_intent(state: AgentState) - str: intent state.get(intent, CHITCHAT) if intent in (SHOPPING, POLICY_QUERY, ORDER_QUERY): return rag_search elif intent RETURN_REQUEST: return return_subgraph elif intent in (HUMAN_FALLBACK, CHITCHAT): return direct_reply else: return direct_reply # 将所有节点加入图 graph StateGraph(AgentState) graph.add_node(intent_node, intent_node) graph.add_node(rag_search, rag_search_node) graph.add_node(direct_reply, direct_reply_node) graph.add_node(return_subgraph, return_subgraph_node) # 设置入口边和条件边 graph.set_entry_point(intent_node) graph.add_conditional_edges( intent_node, route_by_intent, { rag_search: rag_search, return_subgraph: return_subgraph, direct_reply: direct_reply } ) # 配置所有路径的出口 graph.add_edge(rag_search, response_node) graph.add_edge(direct_reply, response_node) graph.add_edge(return_subgraph, response_node) graph.add_edge(response_node, END) compiled_graph graph.compile()要注意的是条件边的映射字典里key 是路由函数的返回值value 是要跳转的节点名。一旦你改过节点名记得同步修改映射关系否则 LangGraph 会在编译时才报错排查起来非常折磨。另外条件边函数虽然可以直接把判断逻辑写在里面但我建议它只做“读 State 并返回字符串”这件事不要在里面调用外部 API不要做耗时操作。路由逻辑应该是纯函数这样图的行为才可预测也方便单元测试。3.4 RAG 检索节点优化了三个版本才真正能用RAG 节点是电商客服里被问得最多的也是最容易做得“看起来能用但实际体验很差”的部分。我们第一版就是最朴素的“向量检索 TopK 拼接 Prompt 丢给 LLM 生成”上线预约测试后发现用户并不买账经常给出看似合理但其实已经过时或不够准确的回答。后来我们做了三个关键改动效果才有质的提升。第一把纯向量检索改成了混合检索。向量检索擅长语义匹配但对精确的货号、订单号、政策条款编号这类包含专有名词的查询反而容易出错。我们引入 BM25 关键词检索然后把两路结果用 RRFReciprocal Rank Fusion做融合排序。实测下来的召回准确率提升非常明显。BM25 用 Elasticsearch 的multi_match可以相对简单跑起来如果不想引入 ES用rank_bm25库在内存里做也行数据量不大的情况下性能可以接受。第二对切块策略做了和业务对齐的调整。原来用固定 500 字符切块经常把“七天无理由退货的适用条件”和“不适用场景”切成两半。后来我们改成按语义段落切分同时做了重叠窗口切块大小 300 token重叠 50 token。更重要的是每一块从入库时就把“商品ID”“类目”“适用政策类型”作为 metadata 存进向量库检索时根据用户的上下文用 metadata 过滤做前置缩小经实测相关性显著提升。这就是网上很多人提的“结构化 RAG”或“元数据过滤 RAG”本质是让检索器只在你关心的范围内搜。第三Prompt 模板里要求 LLM 回答时必须标注引用的知识来源。我们预先给每条知识加了一个knowledge_idPrompt 告诉模型“如果回答依据是检索到的片段请在回答末尾用 [来源1][来源2] 这种形式标注”。这个设计除了提升可信度更重要的是方便后续做日志审计——客服主管能根据来源编号回溯到具体是哪条知识库内容导致了错误回答这个能力在运营侧很加分。下面是简化版 RAG 检索节点async def rag_search_node(state: AgentState) - AgentState: question state[messages][-1].content user_id state[user_id] # 1. 向量检索 vector_results await vector_store.asimilarity_search( question, k5, filter{user_id: user_id} # 按用户维度过滤提高相关性 ) # 2. BM25 关键词检索 bm25_results await es_search(question, indexproduct_knowledge, size5) # 3. RRF 融合 fused_results reciprocal_rank_fusion(vector_results, bm25_results, k60) # 4. 组装上下文 context \n\n.join([f[来源{i1}] {doc.page_content} for i, doc in enumerate(fused_results[:5])]) rag_prompt f你是电商平台的智能客服。请基于下面的知识库片段回答用户问题。 要求 1. 只能依据片段内容回答不要编造不存在的政策 2. 如果片段信息不足明确告诉用户需要转人工 3. 回答末尾用 [来源N] 标准引用你参考的片段。 知识库片段 {context} 用户问题{question} response await llm.ainvoke([ {role: system, content: 你是一名严谨的电商客服助手。}, {role: user, content: rag_prompt} ]) return { retrieved_docs: fused_results, final_answer: response.content }关于向量库选型本地开发和生产我都推荐先用开源的 Chroma 或 Qdrant不要一上来就上几十万的向量服务。Qdrant 的 metadata filter 性能很好支持复杂过滤条件而且有 Docker 镜像可以快速本地起。如果团队规模很小数据量在百万级以下Chroma 也够。Milvus 适合海量数据但运维复杂度明显更高初期没必要背这个包袱。3.5 退货流程子图LangGraph 嵌套发挥威力的地方退货流程是整个系统最复杂的部分因为它不是一个单轮问答而是多轮状态机式的交互一个典型的流程是用户说“我要退货”Agent 询问订单号用户提供订单号Agent 查询订单系统判断是否在退货期内如果在询问退货原因用户提供原因Agent 校验原因是否合规创建退货单返回退货单号和后续操作指引这个流程里有信息收集、外部系统校验、条件分支还有可能某个环节用户直接放弃或者情绪升级转人工。如果用一棵大的 LangGraph 图来做节点之间的状态管理会变得非常混乱。所以我们把退货流程单独抽成一个子图然后在主图里用add_node把它作为一个节点调用LangGraph 原生支持这种节点嵌套的方式。子图内部有自己的 State我们用了一个单独的ReturnStateclass ReturnState(TypedDict): return_step: str order_id: str reason: str validation_status: str return_ticket_id: str messages: Annotated[list, add_messages]然后定义子图的节点def create_return_subgraph() - CompiledStateGraph: sub_graph StateGraph(ReturnState) sub_graph.add_node(ask_order_id, ask_order_id_node) sub_graph.add_node(validate_order, validate_order_node) sub_graph.add_node(ask_reason, ask_reason_node) sub_graph.add_node(create_ticket, create_ticket_node) sub_graph.add_node(end, end_node) sub_graph.set_entry_point(ask_order_id) sub_graph.add_conditional_edges( ask_order_id, # 如果没有订单号继续等待否则跳转校验 lambda state: validate_order if state.get(order_id) else ask_order_id ) ... return sub_graph.compile()子图的边界设计有个地方值得特别注意子图接收父图传入的order_id、session_id等初始数据但子图内部的messages是独立的不会干扰父图的消息列表。当子图跑完后父图需要把子图的结论是否创建成功、退货单号合并回父 State这个合并逻辑要放在调用子图之后的节点里做。这种嵌套设计的好处非常多。首先退货流程的代码可以独立测试不需要启动整个客服图其次未来如果要做退款流程或者换货流程可以复制一套类似的子图复用部分节点最后LangGraph Studio 在调试的时候可以清晰地看到状态卡在子图的哪个环节排查问题效率高很多。还要注意子图内部节点ask_order_id这种节点本质上是一个“等待用户输入”的节点但 LangGraph 本身并不会做多轮暂停等待。LangGraph 的标准做法是每次用户发消息都调用一次图图的 State 从上次结束的地方继续执行。这就需要在 FastAPI 层把每次执行后的返回状态保存下来下次用户发消息时带上这个状态。说白了这个图的执行是“无状态”的状态存你这边LangGraph 只负责基于传入的状态计算下一步。理解了这一点整个系统就会豁然开朗。4. FastAPI 接入层的实战细节会话管理、流式输出与 OpenAPI 文档4.1 会话持久化和续跑LangGraph 状态与 Redis 的衔接前面说到 FastAPI 是无状态的那 State 存哪里我们用 Redis。每个会话有一个session_id在会话开始时初始化一个空的AgentState每次执行完图之后把最新的 State 序列化存回 Redis过期时间设为 30 分钟。30 分钟无交互就清理用户重新发起时从头开始。import redis import json from typing import Optional r redis.Redis(hostlocalhost, port6379, decode_responsesTrue) def save_state(session_id: str, state: dict, ttl_seconds: int 1800): r.setex(fagent_state:{session_id}, ttl_seconds, json.dumps(state, ensure_asciiFalse)) def load_state(session_id: str) - Optional[dict]: data r.get(fagent_state:{session_id}) if data: return json.loads(data) return None def clear_state(session_id: str): r.delete(fagent_state:{session_id})这里有个小坑LangGraph 的 State 里有些字段比如消息对象不是你自定义的普通 Python dict用json.dumps序列化的时候可能会报Object of type Message is not JSON serializable。解决办法是序列化之前把消息对象手动转成字典def state_to_json(state: AgentState) - dict: return { **state, messages: [ {role: m.type, content: m.content} for m in state.get(messages, []) ] }反序列化的时候再从字典转回消息对象这个转换逻辑建议封装在load_state里统一处理别散落在各个节点中。另外有一个设计决策要讲明白到底该把历史消息全部存下来还是只存最近几轮我实测下来如果每轮请求都把全部历史消息发给 LLM上下文很快会变得很庞大。token 费用还只是表面问题更麻烦的是模型注意力会被无关历史分散回答质量反而下降。所以我们做了一个记忆压缩策略近 6 轮消息完整保留更早的消息在每次节点执行前利用 LLM 做一次摘要压缩摘要作为系统提示词的一部分注入。这个设计能明显降低 token 开销并提升响应速度但实现的复杂度会上升一截适合有一定流量的生产系统。4.2 流式输出让用户感觉到 Agent 在“思考”客服场景对流式输出的要求几乎算是刚需——用户看到文字一个一个字蹦出来会比发呆等 5 秒看到一个完整回答更有耐心。LangGraph 自身支持astream方法但要注意它流出来的不仅仅是最终答案还包括中间节点执行的事件。如果你直接把astream的事件推给前端前端会看到一堆结构不明的 JSON 碎片体验会非常糟糕。正确的做法是FastAPI 的 SSEServer-Sent Events接口内部调用astream然后只抽取messages更新事件把节点执行事件过滤掉只把最终 LLM 生成的 token 增量推给前端。LangGraph 的流式事件有两种模式values模式和updates模式我建议用updates模式因为它能告诉你是哪个节点产生了输出方便你做事件过滤。from fastapi import FastAPI from fastapi.responses import StreamingResponse import json, asyncio app FastAPI() app.post(/api/chat/stream) async def chat_stream(request: ChatRequest): session_id request.session_id user_message request.message state load_state(session_id) or new_state(request.user_id, session_id) # 把用户消息追加到 state 中 state[messages].append({role: user, content: user_message}) async def event_generator(): full_answer async for event in compiled_graph.astream(state, config{recursion_limit: 25}, stream_modeupdates): for node_name, node_output in event.items(): if node_name response_node and final_answer in node_output: # 增量推送 final_answer 字段 new_text node_output[final_answer] if isinstance(new_text, list): # 兼容部分模型输出为增量列表的情况 delta new_text[-1][text] if new_text else else: delta new_text full_answer delta yield fdata: {json.dumps({type: token, content: delta}, ensure_asciiFalse)}\n\n # 流结束保存完整状态 state[messages].append({role: assistant, content: full_answer}) save_state(session_id, state) yield fdata: {json.dumps({type: done})}\n\n return StreamingResponse(event_generator(), media_typetext/event-stream)这一个实现对生产环境非常关键用户端不需要拿到final_answer的全部内容只需要增量服务端在流结束后统一保存 state避免多次写入 Redis 造成性能损耗。如果你直接把compiled_graph.invoke()一把梭返回完整结果也不是不能用但用户的等待耐心会明显受影响尤其是有时候 RAG 检索和 LLM 生成加起来要 8-10 秒那个体验差距会被客户明确感知到。4.3 统一响应格式、鉴权与 OpenAPI 文档的坑FastAPI 有一个优势经常被人忽视就是它自动生成 OpenAPI 文档默认路径是/openapi.json。这给你带来前端联调效率的巨大提升前端同学可以直接打开 Swagger UI 看每个接口的参数和响应结构。但注意如果你的接口是流式输出OpenAPI 里面会把响应体识别成text/event-streamSwagger UI 里没法直接预览这种格式需要额外做一个“非流式版”的调试接口或者在前端自己用代码测。这个项目开发过程中我们内部留了一个debug开关请求参数里streamfalse时走非流式接口方便在 Swagger 里调试业务逻辑这是一个非常实用的细节。接口响应格式方面我见过很多项目一个接口一个返回结构前端接得想骂人。我们的统一约定如下{ code: 0, message: success, data: { ... } }错误码统一用整数0表示成功非 0 表示各类错误message是人类可读的错误描述。FastAPI 中可以用response_model来约束返回结构但因为流式接口的特殊性注意不要给流式接口加response_model否则会破坏 SSE 输出。鉴权方面客服 Agent 一般是服务端到服务端的调用不直接暴露公网。我们用了一个简单的 API Key 机制请求头带X-API-KeyFastAPI 依赖里校验如果 key 不存在或过期直接返回 401。生产环境如果对接的是前端 App还可以把用户 token 换成 JWT在get_current_user依赖中解析用户 ID 并注入 Request 对象这个逻辑跟普通 FastAPI 项目完全一致Agent 层不需要关心用户的身份从哪来。5. 状态记忆与上下文管理让 Agent 在多轮对话中不“失忆”电商客服对话的显著特点就是话题跳跃性强用户可能上一轮在问 A 商品的库存下一轮就跳到退货政策。如果没有有效的记忆机制Agent 很容易出现前后矛盾的回答。LangGraph 的messages字段用add_messages可以无脑追加历史消息但我们不能真的把所有消息都留给模型看原因前面提过一方面是 token 限制另一方面是性能。我们的方案是“近 6 轮全量 早期摘要 业务字段覆盖”具体来说近 6 轮完整消息保留用于模型理解当前语境。超过 6 轮的部分在每轮结束时用一次独立的 LLM 调用生成一段不超过 200 字的摘要摘要是按业务维度组织的比如“用户已确认购买 A 商品对 B 商品的价格有疑问尚未提及退货意向”。业务关键字段如订单号、退款单号、当前退货状态独立存放在 State 的字段里即使对话摘要丢了这些字段依然可以恢复上下文。async def summarize_messages(state: AgentState, llm) - str: old_messages state[messages][:-6] summary_prompt f请将以下客服对话压缩为不超过200字的业务摘要。 重点提取用户咨询的商品、订单号、诉求、当前处理进度。 对话内容 {old_messages} response await llm.ainvoke([ {role: system, content: 你是对话摘要助手。}, {role: user, content: summary_prompt} ]) return response.content这个“摘要关键字段”的双轨设计我认为是电商客服 Agent 里最容易被忽略但回报最高的模块之一。很多团队花大力气调 RAG结果多轮对话稍微绕一点上下文就崩了体验比传统 FAQ 还差。记忆管理做好之后Agent 才真正像一个“客服专员”而不是一个每次收到消息都失忆的新手。另外LangGraph 的长期记忆能力在最新版本里也有所增强比如Store接口可以持久化跨会话记忆数据。但我们的业务场景下用户和客服的交互发生在一个 Session 内跨天恢复历史会话的需求目前为零所以还没有引入跨会话的长期记忆。如果你的产品需要做“用户画像记忆”或者“跨会话偏好追踪”可以去研究一下 LangGraph 的 Store 功能我这里就不展开讲了。6. 生产环境实测性能、成本和那些你必须知道的坑6.1 性能数据加了缓存之后响应时间降了一半上线初期我们主要关注功能正确性没有做太多性能优化。后来压测发现RAG 检索节点的向量查询在高并发下出现了明显的延迟抖动这才开始认真做性能优化。目前在我们的生产配置8 核 16G 内存的单节点Qdrant 和 LLM 服务都在内网下测试结果如下场景平均响应时间95分位响应时间每日支撑调用量纯意图识别 直接回复1.2s2.1s12000RAG 检索 生成回答3.8s6.4s8000退货流程子图完整走通6.5s9.8s1500这个数据是可接受的但有一个性能杀手必须注意如果同一个session_id的请求并发进来两个请求同时执行图并同时写 Redis后写入的状态会覆盖先写入的导致用户消息丢失。解决方式是在 FastAPI 层对同一个 session 加一个asyncio.Lock保证同一会话内的消息处理是串行的。这个坑不看压测报告很难发现但一旦遇到后果很严重。6.2 成本控制别让 LLM 调用次数失控LangGraph 图节点一多LLM 调用频率会肉眼可见地涨。我们最初设计图的时候每个节点都调一次 LLM结果一次完整退货流程要调 6-7 次模型成本太高了。后来做了两个优化意图识别节点前加了轻量规则精确命中就直接跳过 LLM 调用RAG 生成阶段如果系统检测到问题简单且知识库里有高置信度匹配的答案直接用模板回答不经过生成模型。这个“高速缓存”逻辑能省不少成本。另外对相同问题的短时间重复请求可以在 Redis 里做一个简单的语义缓存把近 10 分钟内的(意图, 标准化问题)和回答存起来命中就直接返回。测试发现约 12% 的客服请求是重复问题用户反复刷新或者相似表达这一层缓存能显著降低成本。6.3 兜底与人工接管Agent 不是万能的要有退出机制最后必须讲一下“兜底和人工”这个很少在技术教程里被认真对待但在生产环境里极其重要的模块。无论你的 RAG 和意图识别做得有多好总会有超出预期的问题出现。用户在情绪激动、语言表达混乱、或者根本没有相关知识库的内容时Agent 的硬回答只会让体验更糟。所以我们的系统有一个强制规则连续两次意图识别为 HUMAN_FALLBACK用户明确说“转人工”或“找客服”退货流程子图连续两次校验失败生成回答被安全审核模块判为低可信度。满足任一条件图的状态机直接跳到human_handoff节点这个节点会调用工单系统创建一个待处理的客服工单并把完整的对话上下文附在工单描述里然后转交给人工客服处理。人工客服接手后可以直接看到 Agent 的整个决策过程——包括识别到的意图、检索到的知识片段、Agent 生成的回答草稿。这个“无缝人机协作”的设计对运营团队非常友好上线后人工客服的满意度比之前纯工单系统高了不少。关于安全性我还想多说一嘴电商行业对合规要求很敏感Agent 的回答不能包含任何虚构政策。所以我们在 RAG 节点前面设置了一个“知识库命中阈值”如果向量检索相似度最高分低于 0.75规则引擎直接判定为“无法通过知识库回答”禁止 LLM 自由发挥。这个硬性边界有效杜绝了模型幻觉导致的错误承诺。说到底LangGraph FastAPI 这套组合在电商客服场景里是经得起生产检验的。如果你准备动手做类似的项目我建议你现在就去把官方文档里关于 State、条件边和子图嵌套的例子跑一遍然后把你这边的业务流程画成状态图再开始写代码。先把图的设计搞清楚代码反而是水到渠成的事。这套框架里目前还没有做到位的部分是实时语音客服的接入以及跨渠道微信小程序、独立App的消息去重同步这些是我们下一步要啃的硬骨头后面有进展了我再来分享。本文还有配套的精品资源点击获取