从零到一搭建智能客服系统（LangGraph + FastAPI + 智谱AI 实战）

一、这个项目是做什么的？

「π域」是一个快递行业的 AI 智能客服系统。它的核心价值是：用 AI Agent 替代 80% 的重复性人工客服工作，实现 7×24 小时秒级响应。

具体来说，它能做这几件事：

• FAQ 问答：用户问“运费怎么算？”、“寄到北京要多久？”——系统基于知识库自动回答
• 订单查询：用户输入运单号，系统调用快递鸟 API 返回真实物流轨迹
• 投诉工单：用户说“包裹破损了”，系统提取信息，自动生成工单编号
• 地址修改：用户说“改地址”，系统引导用户提供新地址
• 转人工：用户说“转人工”，系统通过 WebSocket 排队，客服接单后实时对话

二、技术栈选型（为什么是这些？）

三、系统架构（一张图看懂）

用户输入 │ ▼ FastAPI /chat 接口 │ ▼ ───────────────────────────────────────────────────────── │ LangGraph 多 Agent 编排 │ │ │ │ ──────────── ──────────── │ │ │ 意图识别 │ → │ 条件路由 │ │ │ │ (Intent) │ │ (Router) │ │ │ ──────────── ─────────── │ │ │ │ │ ───────────────────────────────── │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ ──────── ──────── ──────── │ │ │ FAQ │ │ 订单 │ │ 投诉/转人工│ │ │ │ Agent │ │ Agent │ │ Agent │ │ │ ──────── ──────── ──────── │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ Chroma检索 快递鸟API WebSocket排队 │ ───────────────────────────────────────────────────────── │ ▼ 返回最终回答

四、核心功能详细实现

1. 意图识别（Few-shot + 上下文记忆）

问题

在零样本场景下，模型对“寄到广州要几天”这类边缘问题容易误判为 other。

解决方案

构建 Few-shot 示例库，运行时动态检索最相似的 3 个示例注入 Prompt。

示例库结构（data/intent_examples.json）：

[{"question":"怎么算运费","intent":"faq"},{"question":"寄到北京要多久","intent":"faq"},{"question":"查一下我的快递","intent":"order"},{"question":"转人工","intent":"human"},{"question":"改地址","intent":"change_address"}]

核心代码：

defintent_agent(state:AgentState):question=state.get("user_question")examples=load_examples()similar=find_similar_examples(question,examples,top_k=3)few_shot_text="\n".join([f"用户：{ex['question']}\n输出：{{\"intent\": \"{ex['intent']}\"}}"forexinsimilar])prompt=f""" 你是一个快递客服意图识别专家。判断用户问题属于以下哪一类： - faq: 咨询常见问题 - order: 查询订单/物流 - complaint: 投诉或理赔 - human: 转人工 - change_address: 修改地址 - other: 其他 以下是一些参考示例：{few_shot_text}用户问题：{question}"""

多轮上下文记忆

维护 context_summary 字段，将最近 2 轮对话摘要传入意图识别 Prompt，解决指代消解问题：

用户：查一下我的快递 AI：请提供运单号 用户：YT3762892935155 ✅ 系统能理解这是在补充运单号

2. RAG 知识库（FAQ 问答）

技术方案

• 分块策略：chunk_size=512，重叠 50 字符
• Embedding 模型：智谱 embedding-2
• 向量库：Chroma（本地持久化）
• 检索策略：Top-3 相似片段

FAQ 加载代码：

defload_faq():chunks=split_faq("data/faq_knowledge.md",chunk_size=512,overlap=50)collection=get_chroma_collection()ids=[f"faq_{i}"foriinrange(len(chunks))]collection.add(documents=chunks,ids=ids)

检索 + 生成代码：

deffaq_agent(state:AgentState):query=state.get("user_question")collection=get_chroma_collection()results=collection.query(query_texts=[query],n_results=3)context="\n\n".join(results['documents'][0])prompt=f"基于以下知识回答用户问题：\n{context}\n问题：{query}"return{"final_answer":call_llm(prompt)}

3. 真实订单查询（快递鸟 API）

对接步骤

1. 注册快递鸟账号，获取 EBusinessID 和 APIKey
2. 封装签名算法（MD5 + Base64）
3. 实现智能识别快递公司编码（根据运单号前缀）

核心代码：

defquery_order(order_id:str):# 1. 智能识别快递公司shipper_code=recognize_express(order_id)# SF/YTO/ZTO...ifnotshipper_code:return{"code":-1,"msg":"无法识别该运单号所属快递公司"}# 2. 构造请求request_data=f'{{"LogisticCode":"{order_id}","ShipperCode":"{shipper_code}"}}'params={"EBusinessID":CUSTOMER_CODE,"RequestType":"8002","RequestData":request_data,"DataSign":encrypt(request_data,APP_KEY),"DataType":2}# 3. 发送请求并解析resp=requests.post("https://api.kdniao.com/api/dist",data=params)result=resp.json()ifresult.get("Success"):return{"code":0,"data":{"traces":result.get("Traces",[])}}return{"code":-1,"msg":result.get("Reason","查询失败")}

4. 转人工闭环（WebSocket）

架构设计

用户端 ws──→ WebSocket 服务器 ←──ws── 客服端 │ ├── 排队队列 ├── 活跃会话管理 ─ 消息路由

消息类型

WebSocket 消息路由核心代码：

asyncdefhandle_message(ws,message):data=json.loads(message)msg_type=data.get("type")client_id=data.get("client_id")ifmsg_type=="join_queue":waiting_queue.append({"user_id":client_id,"ws":ws})awaitws.send(json.dumps({"type":"queue_status","position":len(waiting_queue)}))elifmsg_type=="agent_take":user_info=waiting_queue.pop(0)active_sessions[user_info["user_id"]]={"user_ws":user_info["ws"],"agent_ws":ws}awaituser_info["ws"].send(json.dumps({"type":"assigned"}))awaitws.send(json.dumps({"type":"assigned"}))elifmsg_type=="chat":target=data.get("target")content=data.get("content")iftarget=="agent":awaitactive_sessions[client_id]["agent_ws"].send(...)eliftarget=="user":foruid,sessioninactive_sessions.items():ifsession["agent_ws"]==ws:awaitsession["user_ws"].send(...)

五、踩坑记录（真实经验）

六、项目成果

功能完成度

代码结构

pisphere/ ── main.py # FastAPI 入口 ── state.py # AgentState 定义 ── graph.py # LangGraph 图构建 ── websocket_server.py # WebSocket 服务器 ── agents/ │ ├── intent.py # 意图识别 │ ├── faq.py # FAQ 检索 │ ├── order.py # 订单查询 │ ├── complaint.py # 投诉工单 │ ├── change_address.py # 地址修改 │ ├── handoff.py # 转人工 │ ─ fallback.py # 兜底 ── rag/ │ ├── chroma_client.py # Chroma 客户端 │ ─ faq_loader.py # FAQ 加载器 ── web/ │ ├── index.html # 用户端 │ ─ customer_service.html # 客服端 ── data/ ─ intent_examples.json # Few-shot 示例库

七、后续优化方向

八、总结

从零到一，我用 3 周时间 完成了「π域」智能客服系统的开发。这个项目让我深入理解了：

• RAG 完整流程：分块 → 向量化 → 存储 → 检索 → 生成
• LangGraph 多 Agent 编排：状态管理、条件路由、节点协作
• WebSocket 实时通信：排队、接单、消息转发
• Prompt Engineering 实战：Few-shot、上下文注入、置信度阈值

更重要的是，这个项目验证了 “Java 后端开发者可以快速转型 AI Agent 开发” ——你不需要成为算法专家，也能构建出可用的 AI 产品。