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

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

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

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

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

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

组件	选型	选型理由
后端框架	FastAPI	轻量、异步、自动生成 Swagger 文档，开发效率极高
多 Agent 编排	LangGraph	支持状态管理和条件路由，比 LangChain 更灵活可控
大模型	智谱 GLM-4-Flash	性价比极高，响应速度快，中文能力优秀
向量数据库	Chroma	轻量级、本地持久化、零配置，无需额外部署
实时通信	WebSocket	双向实时通信，天然适合排队 + 聊天场景
前端	原生 HTML + CSS + JS	无框架依赖，轻量快速，酷黑主题

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

用户输入 │ ▼ 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）

对接步骤

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

核心代码：

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── 客服端 │ ├── 排队队列 ├── 活跃会话管理 ─ 消息路由

消息类型

类型	方向	说明
join_queue	用户 → 服务器	加入排队
agent_ready	客服 → 服务器	客服上线
agent_take	客服 → 服务器	接单
chat	双方 → 服务器	聊天消息转发
agent_offline	客服 → 服务器	客服下线
end_session	客服 → 服务器	结束会话

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(...)

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

问题	原因	解决方案
KeyError: ‘“intent”’	Prompt 中 JSON 示例的花括号被 str.format() 误解析	将 {“intent”: “faq”} 改为 {{“intent”: “faq”}}
LLM 返回 ```json {…} ```	模型有时会输出 Markdown 代码块	用正则 r’```json\s*({.?})\s```’ 提取纯 JSON
KeyError: ‘user_question’	LangGraph 状态传递丢失字段	使用 state.get(“user_question”, “”) 安全取值
快递鸟返回"没有可用套餐"	账号未开通服务或套餐未生效	切换沙箱环境测试，或联系客服开通免费套餐
WebSocket 客服消息用户收不到	路由逻辑错误，未正确映射 agent_ws 到 user_id	在 active_sessions 中双向存储
排队列表不刷新	renderQueue 动态修改 h3 导致 DOM 引用丢失	预置 refreshSpinner，只更新内容不重建 DOM

六、项目成果

功能完成度

功能	状态
FAQ 问答	✅
订单查询（真实 API）	✅
投诉工单生成	✅
地址修改	✅
转人工闭环（WebSocket）	✅
用户退出人工	✅
客服结束会话	✅
前端酷黑主题	✅

代码结构

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 示例库

七、后续优化方向

优先级	优化项	说明
高	环境变量配置	API Key 等敏感信息移入 .env
高	Docker 容器化	便于部署和迁移
中	工单存储升级	JSON → SQLite/PostgreSQL
中	日志结构化	print → logging 模块
低	Embedding 模型对比	测试 bge-large-zh 等模型对检索效果的影响