背景痛点：传统客服系统的三座大山
客服系统早已不是“能回答就行”的年代，业务方对准确率、响应时间、可维护性都提出了更高要求。传统方案普遍采用“规则引擎 + 关键词匹配”的组合拳，痛点集中体现在三点：
- 规则膨胀：每上线一个新产品就要追加几十条正则，半年下来几千条规则交织在一起，牵一行而动全身，维护成本指数级上升。
- 意图漂移：用户口语千变万化，规则只能覆盖字面，稍有同义词或语序变化就识别失败，准确率常年徘徊在 70% 左右。
- 多轮断层：缺乏对话状态机，一旦用户追问“那第二个套餐呢？”，系统只能从头再来，体验感瞬间拉垮。
大模型虽然火，但直接拿来问答仍面临幻觉、延迟、成本三大挑战。AI 辅助开发的核心思路是“让模型做它最擅长的事”，把生成、泛化、推理交给大模型，把精准、可控、合规留给工程层。
技术选型：微调 vs. RAG 的拉锯战
| 维度 | 微调（Fine-tuning） | RAG（Retrieval-Augmented Generation） |
|---|---|---|
| 数据要求 | 需要千级高质量“问答对”，标注成本高 | 只需万级文档，无需人工标注 |
| 知识更新 | 重新训练 ≥30 min，版本管理复杂 | 分钟级增量更新，向量库即插即用 |
| 幻觉风险 | 仍有，尤其长尾知识 | 检索结果先过滤，再生成，幻觉显著降低 |
| 响应延迟 | 纯生成，平均 800 ms | 检索+生成，平均 1.2 s（可缓存） |
| 成本 | 训练 GPU+推理 GPU 双份 | 仅推理 GPU，训练零成本 |
在客服场景，业务知识半月一小改、季度一大改，RAG 的“热更新”优势直接击中痛点；同时，检索环节能把企业私有知识先过滤一遍，显著降低大模型信口开河的概率。综合维护性与可控性，最终采用RAG 为主、微调兜底的混合方案：高频标准问答走检索，冷门长尾问题微调模型生成通用回答，再辅以人工抽检。
系统架构：一张图看清数据流
核心流程分四段：
- 对话接入层：统一封装微信、网页、App 三端消息，做鉴权、限流、敏感词过滤。
- 语义理解层：BERT+SimCSE 做意图初筛，命中置信度≥0.85 直接返回；否则走大模型。
- 知识检索层：Milvus 向量库 + Elasticsearch 混合检索，Top5 块给 LLM 做上下文。
- 答案生成层：LangChain 负责组装 Prompt、调用大模型、解析答案、回填会话状态。
核心实现：LangChain 与语义相似度的组合拳
1. LangChain 对话管理框架
LangChain 把“链”抽象成可插拔的组件，方便在检索、生成、记忆之间自由拼装。下面给出最简可运行代码，已含异常捕获与日志：
# chain_customer_service.py import logging from typing import List from langchain.chat_models import ChatOpenAI from langchain.schema import HumanMessage, AIMessage from langchain.memory import ConversationBufferWindowMemory from langchain.chains import ConversationalRetrievalChain from retriever.milvus_retriever import MilvusRetriever # 自封装 logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s") class CustomerServiceChain: def __init__(self, llm_model: str = "gpt-3.5-turbo", retriever_top_k: int = 5, memory_window: int = 3): self.llm = ChatOpenAI(model_name=llm_model, temperature=0.2, max_tokens=512) self.retriever = MilvusRetriever(top_k=retriever_top_k) self.memory = ConversationBufferWindowMemory(k=memory_window) self.chain = ConversationalRetrievalChain.from_llm( llm=self.llm, retriever=self.retriever, memory=self.memory, return_source_documents=True ) def answer(self, query: str, user_id: str) -> str: try: resp = self.chain({"question": query, "chat_history": self.memory.buffer}) logging.info(f"[{user_id}] Q:{query} A:{resp['answer']}") return resp["answer"] except Exception as e: logging.exception("chain invoke failed") return "系统繁忙，请稍后再试"2. 基于 SimCSE 的语义相似度初筛
如果向量检索每次都要把全库 200 万条向量做暴力比对，延迟扛不住。先用轻量 SimCSE 模型把用户问题映射到 512 维向量，再与“高频标准问”向量库做 Faiss 快速比对，置信度高于阈值即可直接返回答案，节省一次大模型调用。
# semantic_cache.py import torch import numpy as np from transformers import AutoTokenizer, AutoModel from sklearn.metrics.pairwise import cosine_similarity import logging logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(message)s") class SemanticCache: def __init__(self, model_path: str = "shibing624-text2vec-base-chinese"): self.tokenizer = AutoTokenizer.from_pretrained(model_path) self.model = AutoModel.from_pretrained(model_path) self.cache = {} # key:向量 value:答案 self.threshold = 0.85 def _encode(self, text: str) -> np.ndarray: inputs = self.tokenizer(text, return_tensors="pt", truncation=True, max_length=64) with torch.no_grad(): vec = self.model(**inputs).last_hidden_state.mean(dim=1) return vec.numpy() def search(self, query: str) -> str | None: q_vec = self._encode(query) for cached_vec, answer in self.cache.items(): sim = cosine_similarity(q_vec, cached_vec)[0][0] if sim >= self.threshold: logging.info(f"hit cache, sim={sim:.3f}") return answer return None def add(self, query: str, answer: str): self.cache[self._encode(query)] = answer3. 异常处理与日志规范
- 任何外部调用（数据库、向量库、大模型）均包 try/except，异常信息写入日志并返回统一兜底文案，避免把堆栈抛给前端。
- 日志字段统一：时间、用户 ID、耗时、答案长度、是否命中缓存，方便后续做离线评估与效果归因。
性能优化：把 1.2 s 压缩到 300 ms
1. 多级缓存策略
- L1 语义缓存：SimCSE 命中即返回，P99 延迟 30 ms。
- L2 向量缓存：对近 7 天热门问题预热，启动时批量导入 Faiss，降低 Milvus 查询 60%。
- L3 答案缓存：同一问题 md5 直接读 Redis，TTL 10 min，适合集中促销高峰。
2. 异步化改造
I/O 密集型场景用同步等于自废武功。把“用户问题 -> 向量检索 -> 大模型生成”三步拆成异步协程，并发度拉到 200，CPU 利用率从 35% 提到 75%。
# async_handler.py import asyncio from chain_customer_service import CustomerServiceChain chain = CustomerServiceChain() async def async_answer(query: str, user_id: str) -> str: loop = asyncio.get_event_loop() # 把同步函数丢到线程池，防止阻塞主事件循环 return await loop.run_in_executor(None, chain.answer, query, user_id)3. 负载测试数据
使用 locust 压测 5 分钟，4 核 8 G 容器：
- 峰值 QPS：280
- 平均延迟：320 ms
- P99 延迟：580 ms
- 缓存命中率：42%
- 大模型调用占比：14%
符合业务方“并发 200、延迟 <600 ms”的 SLA。
避坑指南：那些踩过的坑
1. 对话状态管理常见错误
把 history 直接拼字符串当上下文，导致超长被截断、关键信息丢失。
解决：用结构化 memory，按轮次存储，超出窗口自动丢弃最早一轮，保证首尾完整。多轮变量未清空，用户 A 的手机号跑到用户 B 的会话。
解决：memory 实例按 user_id 隔离，会话结束立即 del。
2. 大模型 API 限流应对方案
- 采用“令牌桶 + 退避重试”双保险：单账号限流 30 QPS，触发 429 后指数退避，最多 3 次。
- 夜间低峰期提前预热，把热门问答刷进缓存，削峰填谷。
3. 敏感信息过滤实现
- 正则初筛：身份证、银行卡、手机号统一打码。
- 模型二次过滤：用 bert-base-chinese 微调二分类，召回率 96%，确保日志侧无敏感数据。
延伸思考：下一步还能卷什么
- 知识图谱 + RAG：把向量检索的“扁平”知识升级为“图谱”结构，支持多跳查询，例如“套餐 A 的宽带速率是多少 -> 套餐 A 包含哪些宽带 -> 速率分别是多少”。
- 强化学习微调：用真实用户反馈（点赞 / 点踩）做 reward，对生成模型做轻量级 PPO，每周迭代一次，持续提升回答满意度。
- 端侧小型化：把 6B 小模型蒸馏到 1.3B，部署在边缘节点，离线场景也能跑，降低 40% 云成本。
落地大模型客服系统，本质是把“生成自由”与“工程严谨”拼在一起：RAG 负责热更新与可控，缓存负责性能，异步负责吞吐，监控负责兜底。只要在这四条跑道上持续调优，客服就能从“能回答”进化到“答得快、答得准、答得稳”。
