当前位置：首页 > news >正文

基于大模型的智能客服知识库架构设计与实战优化

news 2026/3/27 2:29:21

背景痛点：传统客服系统到底卡在哪？

过去两年，我先后接手过三套“祖传”客服后台，它们共同的症状可以总结成一句话：答非所问、越聊越懵、一改就崩。

意图识别靠关键词+正则，用户换一种说法就“已读不回”；
多轮对话没有状态机，用户问完“我的订单呢？”再补一句“算了取消吧”，系统直接失忆；
知识库更新靠 DBA 手动导 SQL，从业务提需求到上线平均 3 天，热点活动早已结束；
高峰期并发一上来，Elasticsearch 的multi-match查询 RT 飙到 2 s，客服页面集体转圈圈。

一句话，传统架构在语义理解、状态维护、动态更新三个核心环节全面失守，才让我们有了“上大模型”的冲动。

技术选型：RAG vs 全量微调，为什么最后“全都要”？

团队最初也纠结过两条路线：

A. 全量微调：拿 ChatGLM-6B 做 LoRA，数据标了 50 W 条，效果确实猛，意图准确率 94%。但痛点也明显：
- 每新增一篇产品白皮书就要重训，GPU 训练 4 小时，回滚窗口太长；
- 幻觉不可控，一旦答错就是“一本正经地胡说”；
- 模型体积 12 G，推理 RT 300 ms，还要占一张 A10，成本扛不住。
B. RAG（检索增强生成）：把知识切片→向量库→Prompt 拼接，知识更新分钟级， hallucination 靠检索结果约束。可缺点是：
- 召回率看 embedding 质量，冷门长尾问题容易“检索为空”；
- 多轮上下文长度受限于 LLM 的 max_length，闲聊两轮就“爆显存”。

综合业务场景——80% 高频 FAQ + 20% 长尾工单，我们最终采用“RAG 为主、微调兜底”的混合架构：

高频问题走 RAG，毫秒级更新；
长尾工单走微调小模型（3 B 参数），牺牲一定 RT 换准确率；
二者结果由置信度仲裁器统一排序输出。

核心实现：LangChain 向量化流水线 + FastAPI 异步接口

1. 知识库向量化流水线

下面这段脚本跑在ARM 64 核 ECS上，单台 QPS 800，CPU 就能玩，成本直接腰斩。

# ingest.py 符合 PEP8，类型标注齐全 from pathlib import Path from typing import List import langchain, fitz, httpx from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.vectorstores import Chroma from sentence_transformers import SentenceTransformer EMB_MODEL = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2" CHUNK_SIZE, CHUNK_OVERLAP = 384, 64 def load_pdf(path: Path) -> List[str]: """逐页提取文本，异常页跳过""" doc = fitz.open(path) pages = [] for page in doc: try: pages.append(page.get_text()) except Exception as e: print(f"[WARN] skip page {page.number}: {e}") return pages def build_vector_db(pdf_dir: Path, persist_dir: Path) -> None: splitter = RecursiveCharacterTextSplitter( chunk_size=CHUNK_SIZE, chunk_overlap=CHUNK_OVERLAP, separators=["\n\n", "\n", "。"] ) emb = SentenceTransformer(EMB_MODEL) all_texts, metas = [], [] for pdf in pdf_dir.glob("*.pdf"): pages = load_pdf(pdf) chunks = splitter.split_text("\n".join(pages)) all_texts.extend(chunks) metas.extend([{"source": pdf.name} for _ in chunks]) Chroma.from_texts( texts=all_texts, embedding=emb, metadatas=metas, persist_directory=str(persist_dir) ) if __name__ == "__main__": build_vector_db(Path("./docs"), Path("./chroma_db"))

跑完脚本会在本地生成chroma_db目录，直接拷贝到生产容器即可，无需再跑一次。

2. FastAPI 异步查询接口

接口层我们坚持“无阻塞 + 连接池”两条红线：

# service.py from typing import List import fastapi, uvicorn from langchain import RetrievalQA from langchain.llms import AsyncOpenAI from langchain.callbacks import AsyncIteratorCallbackHandler app = fastapi.FastAPI(title="SmartQA") # 连接池：全局单例，减少重复握手 llm = AsyncOpenAI( model_name="gpt-3.5-turbo", max_tokens=512, temperature=0.1, openai_api_base="http://127.0.0.1:8001/v1", # 自托管转发 max_retries=3 ) qa = RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", retriever=Chroma( persist_directory="./chroma_db", embedding_function=SentenceTransformer(EMB_MODEL) ).as_retriever(search_kwargs={"k": 5}) ) @app.post("/ask") async def ask(query: str) -> dict: """异步入口，RT 中位数 180 ms""" try: ans = await qa.acall(query) return {"answer": ans["result"], "source": ans["source_documents"]} except Exception as e: # 异常直接抛给网关统一熔断 raise fastapi.HTTPException(status_code=500, detail=str(e)) if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000, loop="uvloop")

要点：

用AsyncOpenAI+uvloop，单实例 1 k 并发CPU 占用 < 30%；
向量库as_retriever传search_kwargs={"k": 5}，召回 5 段再塞 Prompt，显存占用恒定。

性能优化：在 ARM 服务器上榨干最后一点算力

Embedding 模型横评
在 80 核 ARM 上，用ab -c 50 -n 10000压测，数据如下：
模型 QPS p99 (ms)
bge-base-zh 420 180
text2vec-base 380 210
paraphrase-MiniLM 800 95
结论：MiniLM 参数少、量化友好，在 CPU 场景性价比最高。
GPU 显存 vs 并发
实测 A10 24 G 上，gpt-3.5-turbo 自托管版本：
- 并发 10 → 显存 9 G
- 并发 30 → 显存 18 G
- 并发 40 → OOM
经验公式：每并发 ≈ 0.6 G，留 20 % buffer 比较安全。

模型	QPS	p99 (ms)
bge-base-zh	420	180
text2vec-base	380	210
paraphrase-MiniLM	800	95

避坑指南：幻觉、冷启动与版本回滚

Prompt Engineering 防幻觉
在stuff模板里加“若检索结果为空，请直接回复‘暂无答案’，切勿编造”，幻觉率从 15 % 降到 3 %。
再加“请用中文回答，字数 < 150”，平均长度下降 40 %，RT 再省 30 ms。
知识库冷启动预热
上线前先把近 30 天工单日志做离线聚类，挑出 Top 5 k 问题人工审核后灌库，首日命中率即可 72%，避免“空库上线”的尴尬。
版本回滚
Chroma 的persist_directory按天打 Tar 包存对象存储；
模型层用灰度标签路由，10% 流量先切新库，核心指标（命中率、满意度）下跌即一键回滚。