最近在做一个企业级的AI智能问答客服项目从零到一搞下来踩了不少坑也积累了一些实战经验。今天就来聊聊怎么把一个听起来很“AI”的客服系统实实在在地做出来并部署上线。这不仅仅是调个API那么简单涉及到架构选型、模型优化、高并发处理和生产环境的各种“惊喜”。传统客服系统不管是规则引擎还是早期的简单机器学习模型痛点都很明显。意图识别稍微复杂点就抓瞎用户说“我想订一张明天下午去北京的机票”和“帮我看看后天飞北京有啥航班”可能就被识别成两个意图。多轮对话更是噩梦经常聊着聊着上下文就丢了用户还得重复说。一旦访问量上来系统响应变慢都是小事直接宕机也不稀奇。所以我们这次的目标很明确要高准确率、要能记住上下文、还要扛得住压力。技术选型是第一步市面上方案很多。我们重点对比了Rasa、DialogFlow和自建BERT模型。Rasa开源定制灵活对话管理Tracker设计得不错。但它的NLU自然语言理解部分如果不用自己的DIETDual Intent and Entity Transformer模型而用开源预训练模型意图识别的准确率在垂直领域上往往需要大量数据来喂且QPS每秒查询率在高并发下是个考验资源消耗不小。DialogFlow谷歌家的上手快意图和实体配置可视化。对于通用场景和快速原型非常友好。但黑盒化严重定制化能力弱数据要上传到云端对数据隐私要求高的企业是个硬伤而且成本会随着调用量线性增长。自建BERT微调模型这条路最“硬核”也最贴合我们的需求。优势是模型完全自主可控可以根据我们的客服日志数据做深度领域适配准确率提升潜力最大。性能上通过模型蒸馏、量化、使用更高效的推理框架如ONNX Runtime, TensorRTQPS可以优化到很高。缺点是初期开发成本高需要机器学习相关的工程能力。综合考虑可控性、成本长期和性能天花板我们选择了自建基于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_tokensTrue, max_lengthself.max_len, paddingmax_length, truncationTrue, return_attention_maskTrue, return_tensorspt, ) return { input_ids: encoding[input_ids].flatten(), attention_mask: encoding[attention_mask].flatten(), labels: torch.tensor(label, dtypetorch.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_idsinput_ids, attention_maskattention_mask, labelslabels) loss outputs.loss total_loss loss.item() loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm1.0) # 梯度裁剪防止爆炸 optimizer.step() return total_loss / len(data_loader) # 数据清洗和领域适配技巧 # 1. 去重去除完全相同的用户query。 # 2. 纠错利用规则或简单模型如symspell纠正明显错别字如“素服”-“客服”。 # 3. 领域词典将产品名、专业术语等加入分词器或作为特殊token提升实体识别。 # 4. 数据增强对现有语料进行同义词替换、随机删除、回译等增加数据多样性。模型准备好后对话状态管理是另一个核心。我们采用Redis作为中心化的状态存储架构很简单但很有效每个对话会话Session用一个唯一的UUID作为KeyValue是一个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, jsonpayload, catch_responseTrue) as response: if response.status_code 200: response.success() else: response.failure(fStatus code: {response.status_code})生产环境还必须考虑安全与合规。我们实现了敏感词过滤模块和完整的审计日志。敏感词过滤使用DFADeterministic Finite Automaton算法构建敏感词树对用户输入和机器人输出进行双向过滤。匹配到的词会替换为***并记录到审计日志。这个模块必须高效不能成为性能瓶颈。审计日志所有对话请求和响应连同用户ID、时间戳、IP、会话ID、识别出的意图/实体、以及是否触发敏感词过滤都以结构化的格式如JSON写入到Elasticsearch或时序数据库。这既满足了合规审查要求也为后续分析模型效果、挖掘用户问题提供了数据金矿。在实际部署中我们遇到了两个典型的“坑”。第一个是BERT模型冷启动问题。新业务没有标注数据模型效果很差。我们尝试了以下几种方法无监督预训练在领域相关的纯文本如产品手册、历史工单上继续预训练BERTContinue 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_tensorspt, paddingTrue, truncationTrue, max_length128) 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(fIntent recognition failed for query: {query}. Error: {e}) raise RuntimeError(Model inference error) from e最后还有一些延伸思考。当前模型在有新业务、新意图加入时需要重新标注数据、重新训练整个模型成本很高。小样本学习Few-shot Learning能否让我们只用几个例子就教会模型一个新意图增量训练Incremental Training又该如何设计才能让模型在不遗忘旧知识的前提下高效地学习新知识这些都是下一步值得深入探索的方向也是让AI客服系统真正具备持续进化能力的关键。整个项目从架构设计到部署上线的过程让我深刻体会到构建一个生产可用的AI系统算法模型只占一部分更多的挑战来自于工程实现、性能优化、稳定性和安全性保障。希望这篇笔记里的这些实战经验和踩坑记录能对正在或打算做类似项目的朋友有所帮助。