从0到1：基于LLM搭建智能客服系统的架构设计与工程实践-平芜编程栈

背景痛点：传统客服为什么总“答非所问”

过去两年，我先后接手过三套“上一代”客服系统：一套基于正则+关键词，两套用 Bert+CRF 做意图分类。上线初期都跑得挺欢，可一旦对话超过三轮，用户就开始吐槽“机器人失忆”。总结下来，老架构的硬伤集中在四点：

多轮断层：对话状态靠 slot 填充，上下文长度超过 256 token 就丢槽位。
长尾失效：训练语料 80% 集中在 Top 200 意图，冷门问题（如“发票抬头写错了还能改吗？”）全部被归为“其它”。
知识更新慢：业务规则改一行，就要重训整个模型，发版窗口至少 24h。
响应延迟高：Bert 意图+FAQ 召回+规则排序，整条链路 1.2s，用户体验“ppt 式聊天”。

痛定思痛，团队决定把底座直接换成大模型，用生成式方案一次性解决“理解+回答+多轮”。

技术选型：微调还是 RAG？先算一笔账

在 8×A100（40G）集群上，我们对比了两种路线，数据如下（均值，单卡 FP16）：

方案	显存占用	首 token 延迟	单卡 QPS	更新成本
全参微调 13B	26G	320ms	6	高（需重训）
LoRA 13B	16G	300ms	7	中（只训 Adapter）
RAG + 6B 底座	10G	180ms	12	低（只更新知识库）

结论很直观：

业务知识每天变 → RAG 更轻；
对话风格强定制 → LoRA 更稳；
预算卡脖子 → 6B+RAG 是性价比之王。

我们最终采用“6B 底座 + RAG + 轻量 Prompt 微调”的混合方案：通用对话让模型直接答，企业知识用召回补充，风格示例靠 5% 数据 LoRA 一把梭。

核心实现：30 行代码跑通对话链

下面给出最小可运行骨架，Python 3.10，依赖：langchain==0.1.0、fastchat、fastapi。

1. 对话链封装（带记忆 & 异常兜底）

from typing import List from langchain import ConversationChain, PromptTemplate from langchain.memory import ConversationTokenBufferMemory from langchain.chat_models import ChatOpenAI class CSChain: def __init__(self, model_path: str, max_token: int = 4096, memory_key: str = "history"): llm = ChatOpenAI( openai_api_base="http://localhost:8000/v1", # 本地 FastChat model_name=model_path, max_tokens=512, temperature=0.3, request_timeout=60, ) template = """你是客服助手，请根据上下文和知识库回答问题。 知识库：{context} 对话历史：{history} 用户：{input} 客服：""" prompt = PromptTemplate( input_variables=["context", "history", "input"], template=template, ) memory = ConversationTokenBufferMemory( llm=llm, max_token_limit=max_token, memory_key=memory_key ) self.chain = ConversationChain( llm=llm, prompt=prompt, memory=memory, verbose=False ) async def answer(self, query: str, context: str) -> str: try: return await self.chain.arun(input=query, context=context) except Exception as e: # 异常兜底：返回静态文案 + 人工入口 return f"系统繁忙，已转人工客服，稍等～（trace: {e})"

2. RAG 召回器（基于 ES + 向量双路）

class Retriever: def __init__(self, es_index: str, emb_model: str): self.es = Elasticsearch() self.encoder = SentenceTransformer(emb_model) async def search(self, q: str, top_k: int = 3) -> List[str]: vec = self.encoder.encode(q) # 1. 向量检索 knn = {"field": "vector", "query_vector": vec, "k": top_k} # 2. 关键词检索 match = {"match": {"title": q}} res = self.es.search( index=self.es_index, knn=knn, query=match, size=top_k, rank="rrf", ) return [h["_source"]["text"] for h in res["hits"]["hits"]]

3. FastAPI 异步入口（带限流 & 负载均衡）

app = FastAPI() chain = CSChain(model_path="cs-6b") retriever = Retriever(es_index="cs_kb", emb_model="bge-small") @app.post("/ask") async def ask(req: AskRequest, background: BackgroundTasks): # 1. 限流：单用户 10 qps if not limiter.allow(req.user_id): raise HTTPException(429, "请求过快") # 2. 召回知识 context = "\n".join(await retriever.search(req.query)) # 3. 异步生成 answer = await chain.answer(req.query, context) # 4. 后台记录 background.add_task(log_conv, req.user_id, req.query, answer) return {"answer": answer}

部署时，起 4 个 replica，Nginx 轮询，单卡 QPS 12，整体 48，满足 2w 峰值并发。

性能优化：把 3090 榨出 50% 富余

1. Batch 大小 vs TPS

实验在 3090（24G）+ 6B 模型上完成，输入 512、输出 128 token：

batch	显存	首 token	TPS
1	10G	180ms	12
4	16G	220ms	38
8	OOM	—	—

结论：在线服务 batch=4 是甜点，再大就 OOM；若离线批处理，可降到 FP8 量化再上 16。

2. 注意力窗口裁剪

我们将 RoPE 基频从 10k 提到 40k，再把窗口限制 2048 实验，结果：

知识型问答（单跳事实）准确率 92% → 90%，几乎无损；
多轮闲聊（>6 轮）准确率 85% → 76%，掉点明显。

因此，知识型场景可大胆裁剪窗口，节省 18% 显存；多轮场景建议开全窗或 4k 滑动。

3. KV Cache 预分配

提前根据最大长度 malloc 显存池，避免torch.cat动态拼接，延迟抖动从 ±60ms 降到 ±15ms，P99 毛刺消失。

避坑指南：上线前必须踩的三颗雷

对话状态丢失
症状：用户问“那我呢？”机器人答“您指什么？”
根因：
- 记忆模块用ConversationSummaryMemory，摘要算法把指代信息压丢；
- 高并发下InMemory被不同线程覆写。
  解法：
- 换成ConversationTokenBufferMemory，保留原始 token；
- 用 Redis 存session_id→ 历史，API 无状态化。
敏感词“漏杀”
大模型天生会“创造性”输出，曾把“退款”说成“退妈”。
最佳实践：
- 双层过滤：①模型输出 ②正则+DFA 白名单；
- 业务侧维护动态敏感词表，5 分钟热更新；
- 对金融/医疗场景，加外部合规 API 二次校验。
知识库“幻觉”
症状：模型把过时的“7 天无理由”说成“15 天”。
解法：
- 召回片段打时间戳标签，Prompt 里声明“仅使用 2024 版政策”；
- 对答案做“可溯源”标记，点击展开引用原文，降低投诉率 30%。