最近在做一个企业级的AI智能问答客服项目，从零到一搞下来，踩了不少坑，也积累了一些实战经验。今天就来聊聊，怎么把一个听起来很“AI”的客服系统，实实在在地做出来并部署上线。这不仅仅是调个API那么简单，涉及到架构选型、模型优化、高并发处理和生产环境的各种“惊喜”。
传统客服系统，不管是规则引擎还是早期的简单机器学习模型，痛点都很明显。意图识别稍微复杂点就抓瞎，用户说“我想订一张明天下午去北京的机票”和“帮我看看后天飞北京有啥航班”，可能就被识别成两个意图。多轮对话更是噩梦，经常聊着聊着上下文就丢了，用户还得重复说。一旦访问量上来，系统响应变慢都是小事，直接宕机也不稀奇。所以，我们这次的目标很明确：要高准确率、要能记住上下文、还要扛得住压力。
技术选型是第一步，市面上方案很多。我们重点对比了Rasa、DialogFlow和自建BERT模型。
- Rasa：开源，定制灵活，对话管理（Tracker）设计得不错。但它的NLU（自然语言理解）部分，如果不用自己的DIET（Dual Intent and Entity Transformer）模型而用开源预训练模型，意图识别的准确率在垂直领域上往往需要大量数据来喂，且QPS（每秒查询率）在高并发下是个考验，资源消耗不小。
- DialogFlow：谷歌家的，上手快，意图和实体配置可视化。对于通用场景和快速原型非常友好。但黑盒化严重，定制化能力弱，数据要上传到云端，对数据隐私要求高的企业是个硬伤，而且成本会随着调用量线性增长。
- 自建BERT微调模型：这条路最“硬核”，也最贴合我们的需求。优势是模型完全自主可控，可以根据我们的客服日志数据做深度领域适配，准确率提升潜力最大。性能上，通过模型蒸馏、量化、使用更高效的推理框架（如ONNX Runtime, TensorRT），QPS可以优化到很高。缺点是初期开发成本高，需要机器学习相关的工程能力。
综合考虑可控性、成本（长期）和性能天花板，我们选择了自建基于BERT的微调模型作为核心NLU引擎。
确定了方向，接下来就是核心实现。首先是模型的微调。我们用的是PyTorch和transformers库。
import torch from torch.utils.data import Dataset, DataLoader from transformers import BertTokenizer, BertForSequenceClassification, AdamW from typing import List, Tuple import pandas as pd class CustomerServiceDataset(Dataset): """自定义客服数据集类 """ def __init__(self, texts: List[str], labels: List[int], tokenizer: BertTokenizer, max_len: int = 128): self.texts = texts self.labels = labels self.tokenizer = tokenizer self.max_len = max_len def __len__(self): return len(self.texts) def __getitem__(self, idx) -> dict: text = str(self.texts[idx]) label = self.labels[idx] encoding = self.tokenizer.encode_plus( text, add_special_tokens=True, max_length=self.max_len, padding='max_length', truncation=True, return_attention_mask=True, return_tensors='pt', ) return { 'input_ids': encoding['input_ids'].flatten(), 'attention_mask': encoding['attention_mask'].flatten(), 'labels': torch.tensor(label, dtype=torch.long) } def train_epoch(model: BertForSequenceClassification, data_loader: DataLoader, optimizer: AdamW, device: torch.device): """训练一个epoch""" model.train() total_loss = 0 for batch in data_loader: input_ids = batch['input_ids'].to(device) attention_mask = batch['attention_mask'].to(device) labels = batch['labels'].to(device) optimizer.zero_grad() outputs = model(input_ids=input_ids, attention_mask=attention_mask, labels=labels) loss = outputs.loss total_loss += loss.item() loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) # 梯度裁剪，防止爆炸 optimizer.step() return total_loss / len(data_loader) # 数据清洗和领域适配技巧： # 1. 去重：去除完全相同的用户query。 # 2. 纠错：利用规则或简单模型（如symspell）纠正明显错别字，如“素服”->“客服”。 # 3. 领域词典：将产品名、专业术语等加入分词器或作为特殊token，提升实体识别。 # 4. 数据增强：对现有语料进行同义词替换、随机删除、回译等，增加数据多样性。模型准备好后，对话状态管理是另一个核心。我们采用Redis作为中心化的状态存储，架构很简单但很有效：每个对话会话（Session）用一个唯一的UUID作为Key，Value是一个Hash结构，存储当前意图、填槽信息、历史对话轮次等。这样无论用户的请求被负载均衡到哪台后端服务器，都能从Redis中恢复完整的对话上下文，实现无缝的多轮对话。
系统做出来，能不能扛住生产环境的压力是关键。我们使用Locust进行压力测试，模拟用户并发提问。
from locust import HttpUser, task, between import uuid class ChatbotUser(HttpUser): wait_time = between(1, 3) # 用户思考时间 def on_start(self): self.session_id = str(uuid.uuid4()) # 每个虚拟用户一个独立会话 @task def ask_question(self): payload = { "session_id": self.session_id, "query": "你们的退货政策是什么？" # 可以准备一个问题池随机选取 } with self.client.post("/api/chat", json=payload, catch_response=True) as response: if response.status_code == 200: response.success() else: response.failure(f"Status code: {response.status_code}")生产环境还必须考虑安全与合规。我们实现了敏感词过滤模块和完整的审计日志。
- 敏感词过滤：使用DFA（Deterministic Finite Automaton）算法构建敏感词树，对用户输入和机器人输出进行双向过滤。匹配到的词会替换为
***，并记录到审计日志。这个模块必须高效，不能成为性能瓶颈。 - 审计日志：所有对话请求和响应，连同用户ID、时间戳、IP、会话ID、识别出的意图/实体、以及是否触发敏感词过滤，都以结构化的格式（如JSON）写入到Elasticsearch或时序数据库。这既满足了合规审查要求，也为后续分析模型效果、挖掘用户问题提供了数据金矿。
在实际部署中，我们遇到了两个典型的“坑”。
第一个是BERT模型冷启动问题。新业务没有标注数据，模型效果很差。我们尝试了以下几种方法：
- 无监督预训练：在领域相关的纯文本（如产品手册、历史工单）上继续预训练BERT（Continue Pre-training），让模型先熟悉领域语言。
- 远程监督：利用业务规则模板生成大量弱标注数据。
- 少样本学习：采用Prompt Tuning或Pattern-Exploiting Training (PET) 方法，让模型适应极少量的标注样本。
- 知识蒸馏：用一个大模型（如ChatGPT）对无标注数据生成伪标签，然后让小模型（我们微调的BERT）去学习。
- 主动学习：让模型对未标注数据做出预测，筛选出它最“不确定”的样本交给人工标注，用最小的标注成本获得最大效果提升。
第二个是异步响应导致的上下文丢失。为了提高吞吐，我们最初将NLU推理和对话管理做成了异步流水线。但这就可能发生：用户问了A问题，紧接着问了B问题，结果B问题的推理先完成，并错误地更新了对话状态，覆盖了A问题的上下文。解决方案是引入一个基于会话ID的请求序列号或简单的消息队列，确保同一个会话的请求被顺序处理。或者，更简单点，在对话状态更新时采用乐观锁，检查当前状态版本是否与读取时一致。
在整个开发过程中，代码规范是保证可维护性的基础。所有Python代码遵循PEP8，关键函数和类都有详细的类型注解和异常处理，让代码即文档。
from typing import Optional, Dict, Any from pydantic import BaseModel import logging logger = logging.getLogger(__name__) class ChatRequest(BaseModel): """聊天请求数据模型""" session_id: str query: str user_id: Optional[str] = None def intent_recognition(query: str, model, tokenizer) -> Dict[str, Any]: """ 意图识别核心函数 Args: query: 用户输入文本 model: 加载的BERT模型 tokenizer: 分词器 Returns: 包含预测意图和置信度的字典 Raises: RuntimeError: 模型推理失败时抛出 """ try: inputs = tokenizer(query, return_tensors="pt", padding=True, truncation=True, max_length=128) with torch.no_grad(): outputs = model(**inputs) probabilities = torch.nn.functional.softmax(outputs.logits, dim=-1) predicted_class_id = probabilities.argmax().item() confidence = probabilities.max().item() return {"intent": predicted_class_id, "confidence": confidence} except Exception as e: logger.error(f"Intent recognition failed for query: '{query}'. Error: {e}") raise RuntimeError("Model inference error") from e最后，还有一些延伸思考。当前模型在有新业务、新意图加入时，需要重新标注数据、重新训练整个模型，成本很高。小样本学习（Few-shot Learning）能否让我们只用几个例子就教会模型一个新意图？增量训练（Incremental Training）又该如何设计，才能让模型在不遗忘旧知识的前提下，高效地学习新知识？这些都是下一步值得深入探索的方向，也是让AI客服系统真正具备持续进化能力的关键。
整个项目从架构设计到部署上线的过程，让我深刻体会到，构建一个生产可用的AI系统，算法模型只占一部分，更多的挑战来自于工程实现、性能优化、稳定性和安全性保障。希望这篇笔记里的这些实战经验和踩坑记录，能对正在或打算做类似项目的朋友有所帮助。
