从0到1构建智能客服agent：基于LLM的实战架构与避坑指南-育师

从0到1构建智能客服agent：基于LLM的实战架构与避坑指南

背景痛点：规则引擎的“三座大山”

去年我们团队接手某电商售后系统时，老代码里躺着 1.3 万条正则规则，维护人已经离职，留下一句话：“改一条规则，全站回归 3 天”。
痛点总结如下：

冷启动成本高：新场景需要写正则、画流程图、再写单元测试，平均 2 人/周。
泛化能力差：用户一句“我买的白色 42 码鞋子能换成黑色 43 吗？”需要 7 条规则才能覆盖颜色和尺码的排列组合。
上下文断裂：传统槽位填充只能记住上一轮，用户中途问“运费谁出”再回来，状态机直接懵圈。

于是老板拍板：用 LLM 重构，目标 200 并发、P99<800 ms、意图准确率 ≥92%。

技术选型：Rasa vs Dialogflow vs LLM 自研

维度	Rasa 3.x	Dialogflow CX	LLM + 自研框架
中文预训练	需自己训	谷歌通用模型	可接自研/开源 13B
私有部署
多轮状态	有限	可视化画布	代码级灵活
知识更新	重启服务	后台上传	热加载向量库
成本(月)	4 核 8 G ≈ 1 k	0.006 美元/轮	GPU 推理 ≈ 1.2 k

结论：ToB 场景数据不能出机房，LLM 方案胜出。
架构图如下：

核心链路：网关 → 敏感词异步过滤 → 对话状态机（LangChain）→ 向量检索（FAISS）→ LLM → 后处理 → 超时重试 → 回包。

核心实现

1. 用 LangChain 搭状态机

LangChain 的ConversationBufferWindowMemory默认把全历史扔给 LLM，200 轮后 token 爆炸。我们重写了一个CompressedMemory：

from typing import List, Dict from langchain.schema import BaseMemory class CompressedMemory(BaseMemory): """滑动窗口 + 摘要压缩，token 控制在 1k 以内""" def __init__(self, max_token: int = 1024): self.max_token = max_token self.history: List[Dict[str, str]] = [] def save_context(self, inputs: Dict[str, str], outputs: Dict[str, str]) -> None: self.history.append({"in": inputs.get("query"), "out": outputs.get("reply")}) self._compress() def _compress(self) -> None: while self._token_len() > self.max_token: # 弹出最早一轮，保留最近 3 轮 self.history.pop(0) def _token_len(self) -> int: return sum(len(m["in"]) + len(m["out"]) for m in self.history) def load_memory_variables(self, inputs: Dict[str, str]) -> Dict[str, str]: return {"history": "\n".join([f"User:{m['in']}\nBot:{m['out']}" for m in self.history[-5:]])}

把CompressedMemory塞进LLMChain，状态机就瘦身成功。

2. 基于 FAISS 的知识库语义检索

知识库格式：Markdown，每段≤512 字，先拆块再向量化。

import faiss import numpy as np from sentence_transformers import SentenceTransformer class FaissIndex: def __init__(self, model_name: str = "shibing624/text2vec-base-chinese"): self.encoder = SentenceTransformer(model_name) self.index = faiss.IndexFlatIP(768) # 余弦相似度 self.text_map = [] def add_docs(self, docs: List[str]) -> None: embeddings = self.encoder.encode(docs, normalize_embeddings=True) self.index.add(np.array(embeddings, dtype=np.float32)) self.text_map.extend(docs) def search(self, query: str, topk: int = 3) -> List[str]: q = self.encoder.encode([query], normalize_embeddings=True) scores, idx = self.index.search(np.array(q, dtype=np.float32), topk) return [self.text_map[i] for i in idx[0] if i != -1]

实测 4 核 8 G，10 万条向量，平均检索 18 ms，QPS 400 无压力。

3. 对话历史压缩算法与性能对比

方案	平均 token/轮	意图准确率	P99 延迟
全历史	3.2 k	94.1 %	1.3 s
滑动窗口 5 轮	0.9 k	93.8 %	0.7 s
摘要压缩	0.6 k	92.5 %	0.6 s

权衡后选“滑动窗口 5 轮”，准确率掉 0.3 %，延迟降一半。

生产考量

1. 超时重试机制

LLM 推理偶尔 5 s 才回包，不能让前端空等。用tenacity包两层重试：

from tenacity import retry, stop_after_attempt, wait_random_exponential @retry(wait=wait_random_exponential(multiplier=1, max=10), stop=stop_after_attempt(3)) def llm_generate(prompt: str) -> str: resp = openai.ChatCompletion.create( model="gpt-3.5-turbo", messages=[{"role": "user", "content": prompt}], timeout=8 ) return resp.choices[0].message.content

超时阈值 8 s，最多 3 次，失败返回兜底话术“人工客服稍后联系您”。

2. 敏感词过滤异步化

把敏感词检测拆成独立服务，用asyncio并行：

import aiohttp, asyncio async def sensitive_check(text: str) -> bool: async with aiohttp.ClientSession() as session: async with session.post("http://internal-filter/sensitive", json={"q": text}) as resp: result = await resp.json() return result.get("hit", False)

主流程里await asyncio.wait_for(sensitive_check(query), timeout=0.1)，超时就当通过，后续离线审计再处理，保证主链路 P99 不受拖累。

避坑指南

1. 避免 LLM 幻觉的 prompt 技巧

先检索后回答：prompt 里加“仅使用以下上下文回答，若找不到请说‘暂无相关信息’”。
给示例：Few-shot 3 例，把“不知道”的样本也写进去，让模型学会拒绝。
温度 0.1 起步，别迷信 temperature=0，实测 0 反而产出重复废话。

2. 对话上下文窗口滑动实现

上文已给CompressedMemory，记得把系统提示（如“你是客服助手”）固定在最前，不参与滑动，否则模型会“失忆”自己是谁。

3. 知识库更新时的版本兼容

向量库带版本号：kb_v20240618.index，新库先灰度 5 % 流量，观察 30 min 无异常再全量。
兼容老会话：用户已开聊的 session 仍指向旧索引，新 session 才用新索引，避免上下文跳变。

性能数据小结

测试环境：4 核 8 G / 100 并发 / 单卡 A10

QPS：峰值 230
P99：780 ms
意图准确率：93.8 %
幻觉率：2.1 %（人工抽检 500 轮）

开放问题

如何设计降级策略应对 API 限流？
当 LLM 供应商突然返回 429 时，我们除了“抱歉请稍后再试”，还能不能：

本地 6B 小模型接力？
把请求拆成异步工单，后续短信回复？
直接给 FAQ 链接让用户自助？

欢迎留言聊聊你的实战做法。