企业官网建设流程全解析

简介：这是一套面向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。

请求的完整路径是这样的：前端把用户消息发给 FastAPI，FastAPI 从 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 关键词检索，然后把两路结果用 RRF（Reciprocal Rank Fusion）做融合排序。实测下来的召回准确率提升非常明显。BM25 用 Elasticsearch 的multi_match可以相对简单跑起来，如果不想引入 ES，用rank_bm25库在内存里做也行，数据量不大的情况下性能可以接受。

第二，对切块策略做了和业务对齐的调整。原来用固定 500 字符切块，经常把“七天无理由退货的适用条件”和“不适用场景”切成两半。后来我们改成按语义段落切分，同时做了重叠窗口：切块大小 300 token，重叠 50 token。更重要的是，每一块从入库时就把“商品ID”“类目”“适用政策类型”作为 metadata 存进向量库，检索时根据用户的上下文用 metadata 过滤做前置缩小，经实测相关性显著提升。这就是网上很多人提的“结构化 RAG”或“元数据过滤 RAG”，本质是让检索器只在你关心的范围内搜。

第三，Prompt 模板里要求 LLM 回答时必须标注引用的知识来源。我们预先给每条知识加了一个knowledge_id，Prompt 告诉模型“如果回答依据是检索到的片段，请在回答末尾用 [来源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, k=5, filter={"user_id": user_id} # 按用户维度过滤，提高相关性 ) # 2. BM25 关键词检索 bm25_results = await es_search(question, index="product_knowledge", size=5) # 3. RRF 融合 fused_results = reciprocal_rank_fusion(vector_results, bm25_results, k=60) # 4. 组装上下文 context = "\n\n".join([f"[来源{i+1}] {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 嵌套发挥威力的地方

退货流程是整个系统最复杂的部分，因为它不是一个单轮问答，而是多轮状态机式的交互，一个典型的流程是：

1. 用户说“我要退货”
2. Agent 询问订单号
3. 用户提供订单号
4. Agent 查询订单系统，判断是否在退货期内
5. 如果在，询问退货原因
6. 用户提供原因
7. Agent 校验原因是否合规，创建退货单
8. 返回退货单号和后续操作指引

这个流程里有信息收集、外部系统校验、条件分支，还有可能某个环节用户直接放弃或者情绪升级转人工。如果用一棵大的 LangGraph 图来做，节点之间的状态管理会变得非常混乱。所以我们把退货流程单独抽成一个子图，然后在主图里用add_node把它作为一个节点调用，LangGraph 原生支持这种节点嵌套的方式。

子图内部有自己的 State，我们用了一个单独的ReturnState：

class 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(host="localhost", port=6379, decode_responses=True) def save_state(session_id: str, state: dict, ttl_seconds: int = 1800): r.setex(f"agent_state:{session_id}", ttl_seconds, json.dumps(state, ensure_ascii=False)) def load_state(session_id: str) -> Optional[dict]: data = r.get(f"agent_state:{session_id}") if data: return json.loads(data) return None def clear_state(session_id: str): r.delete(f"agent_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 的 SSE（Server-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_mode="updates"): 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 f"data: {json.dumps({'type': 'token', 'content': delta}, ensure_ascii=False)}\n\n" # 流结束，保存完整状态 state["messages"].append({"role": "assistant", "content": full_answer}) save_state(session_id, state) yield f"data: {json.dumps({'type': 'done'})}\n\n" return StreamingResponse(event_generator(), media_type="text/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-stream，Swagger UI 里没法直接预览这种格式，需要额外做一个“非流式版”的调试接口或者在前端自己用代码测。这个项目开发过程中，我们内部留了一个debug开关，请求参数里stream=false时走非流式接口，方便在 Swagger 里调试业务逻辑，这是一个非常实用的细节。

接口响应格式方面，我见过很多项目一个接口一个返回结构，前端接得想骂人。我们的统一约定如下：

{ "code": 0, "message": "success", "data": { ... } }

错误码统一用整数，0表示成功，非 0 表示各类错误，message是人类可读的错误描述。FastAPI 中可以用response_model来约束返回结构，但因为流式接口的特殊性，注意不要给流式接口加response_model，否则会破坏 SSE 输出。

鉴权方面，客服 Agent 一般是服务端到服务端的调用，不直接暴露公网。我们用了一个简单的 API Key 机制：请求头带X-API-Key，FastAPI 依赖里校验，如果 key 不存在或过期直接返回 401。生产环境如果对接的是前端 App，还可以把用户 token 换成 JWT，在get_current_user依赖中解析用户 ID 并注入 Request 对象，这个逻辑跟普通 FastAPI 项目完全一致，Agent 层不需要关心用户的身份从哪来。

5. 状态记忆与上下文管理：让 Agent 在多轮对话中不“失忆”

电商客服对话的显著特点就是话题跳跃性强，用户可能上一轮在问 A 商品的库存，下一轮就跳到退货政策。如果没有有效的记忆机制，Agent 很容易出现前后矛盾的回答。

LangGraph 的messages字段用add_messages可以无脑追加历史消息，但我们不能真的把所有消息都留给模型看，原因前面提过：一方面是 token 限制，另一方面是性能。我们的方案是“近 6 轮全量 + 早期摘要 + 业务字段覆盖”，具体来说：

1. 近 6 轮完整消息保留，用于模型理解当前语境。
2. 超过 6 轮的部分，在每轮结束时用一次独立的 LLM 调用，生成一段不超过 200 字的摘要，摘要是按业务维度组织的，比如“用户已确认购买 A 商品，对 B 商品的价格有疑问；尚未提及退货意向”。
3. 业务关键字段（如订单号、退款单号、当前退货状态）独立存放在 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 服务都在内网）下，测试结果如下：

这个数据是可接受的，但有一个性能杀手必须注意：如果同一个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）的消息去重同步，这些是我们下一步要啃的硬骨头，后面有进展了我再来分享。

本文还有配套的精品资源，点击获取