最近在做一个AI智能客服的项目从零开始踩了不少坑也积累了一些实战经验。今天就把整个搭建过程、核心架构以及生产环境里那些“坑”整理出来希望能帮到同样想入手的开发者朋友们。传统客服系统尤其是基于规则引擎的大家应该都接触过。它们在小范围、标准问题上还行但一旦遇到用户问法千奇百怪的长尾问题或者需要记住上下文的复杂多轮对话就立刻捉襟见肘了。维护成千上万条规则是个噩梦而且对话流特别容易断裂用户体验很糟糕。这正是我们转向基于大语言模型LLM的智能客服的核心原因。一、技术选型开源框架、云服务还是自研LLM在启动项目前我们仔细对比了几条主流路径开源框架如Rasa优点是生态成熟对话管理Dialogue Management和实体识别NER开箱即用适合快速验证。但它的NLU自然语言理解模块在面对复杂、开放的语义时泛化能力有限且整体架构相对固定深度定制成本高。云服务如Dialogflow部署简单无需关心底层资源前期开发速度极快。但问题也很明显数据隐私性、服务绑定的长期成本、以及功能定制受平台限制。对于有特定业务逻辑和性能要求的企业级应用可能后期会感到掣肘。自研LLM方案这是我们最终选择的方向。它的核心优势在于极强的灵活性和可控性。你可以根据业务数据微调模型使其更懂你的领域可以自主设计整个系统架构无缝对接内部系统并且在数据安全和合规方面拥有绝对主动权。虽然初期投入较大但从长期发展和核心能力建设角度看价值更高。选择自研意味着我们需要自己搭建从接口到推理的完整链条。下面就来聊聊核心的架构设计。二、核心架构设计一个可扩展的分层系统我们的系统采用了清晰的分层设计便于各模块独立开发和迭代。从上到下大致分为四层接入层使用FastAPI提供RESTful API处理来自网页、APP或第三方系统的对话请求负责基础的鉴权、限流和日志。业务逻辑层这是大脑所在。包含请求路由初步判断请求类型例如是纯文本问答、文件上传还是查询订单。意图识别模块使用一个轻量级的分类模型如基于BERT微调快速判断用户意图咨询、投诉、查订单等为后续流程导航。对话状态管理DST维护整个会话的上下文记录当前对话轮次、已填写的槽位Slots信息如订单号、日期这是实现连贯多轮对话的关键。策略模块根据意图和对话状态决定下一步动作调用知识库、询问澄清、转人工等。LLM推理层接收策略模块整理好的上下文和指令调用底层的大模型如ChatGLM、Qwen等生成最终回复。这里会涉及Prompt工程、生成参数如Temperature, Top-p调优等。数据与支撑层包括向量知识库用于精准检索、缓存Redis、持久化数据库以及监控告警系统。这样的分层结构使得意图识别模型可以单独优化对话管理规则可以灵活调整而LLM模型也可以根据需要升级替换耦合度很低。三、关键代码实现从接口到上下文管理接下来用代码展示几个核心模块的实现。我们使用Python的FastAPI和Hugging Face生态系统。1. 基于FastAPI的异步对话接口异步处理对于高并发的客服场景至关重要。from fastapi import FastAPI, HTTPException, BackgroundTasks from pydantic import BaseModel from typing import Optional import asyncio import uuid app FastAPI(titleLLM智能客服API) class ChatRequest(BaseModel): 对话请求体 session_id: Optional[str] None # 为空则创建新会话 user_input: str user_id: Optional[str] None # 用户标识用于个性化 class ChatResponse(BaseModel): 对话响应体 session_id: str bot_response: str status: str # 如success, need_clarification, transfer_human # 模拟一个异步处理对话的核心函数 async def process_dialogue(session_id: str, user_input: str) - dict: 处理对话逻辑包括意图识别、状态管理、LLM调用 # 这里模拟一个耗时操作 await asyncio.sleep(0.1) # 实际项目中这里会调用意图分类器、对话状态跟踪器最后调用LLM return { response: f已收到您的消息{user_input}。这是模拟回复。, new_state: updated } app.post(/v1/chat, response_modelChatResponse) async def chat_endpoint(request: ChatRequest, background_tasks: BackgroundTasks): 核心对话接口 # 1. 会话管理获取或创建session_id session_id request.session_id or fsession_{uuid.uuid4().hex[:8]} # 2. 异步处理用户输入避免阻塞 # 在实际应用中可以将任务放入消息队列这里简化处理 try: process_result await process_dialogue(session_id, request.user_input) except Exception as e: raise HTTPException(status_code500, detailf对话处理失败: {str(e)}) # 3. 记录对话日志可以放入后台任务 background_tasks.add_task(log_conversation, session_id, request.user_input, process_result[response]) # 4. 构造返回 return ChatResponse( session_idsession_id, bot_responseprocess_result[response], statussuccess ) async def log_conversation(session_id: str, query: str, response: str): 异步记录对话日志这里可以接入ELK或数据库 # 实现脱敏和存储逻辑 print(f[LOG] Session: {session_id}, Q: {query}, A: {response})2. 带缓存的上下文管理类多轮对话的核心是记住之前说了什么。我们用一个类来管理会话上下文并加入缓存和超时机制。import time from typing import Dict, List, Any import redis # 需要安装redis-py from collections import deque class DialogueContextManager: 对话上下文管理器负责维护会话状态和历史 def __init__(self, redis_clientNone, max_history_len10, session_ttl1800): 初始化管理器 :param redis_client: Redis客户端用于分布式缓存。为None则使用内存字典。 :param max_history_len: 内存中保存的最大对话轮次 :param session_ttl: 会话超时时间秒默认30分钟无活动则清除 self.redis_client redis_client self.max_history_len max_history_len self.session_ttl session_ttl # 本地内存缓存减少Redis访问适用于单机或会话粘滞场景 self.local_cache: Dict[str, Dict] {} def _get_redis_key(self, session_id: str) - str: 生成Redis中存储会话的key return fchat_context:{session_id} def get_context(self, session_id: str) - Dict[str, Any]: 获取指定会话的上下文 context None # 首先检查本地缓存 if session_id in self.local_cache: context self.local_cache.get(session_id) # 检查本地缓存是否过期简单时间戳检查 if time.time() - context.get(_last_active, 0) self.session_ttl: del self.local_cache[session_id] context None # 本地没有则尝试从Redis获取 if context is None and self.redis_client: redis_key self._get_redis_key(session_id) cached self.redis_client.get(redis_key) if cached: import json context json.loads(cached) # 取回后也放入本地缓存 self.local_cache[session_id] context # 如果还是没有则创建新的上下文 if context is None: context { session_id: session_id, history: deque(maxlenself.max_history_len), # 使用双端队列限制长度 slots: {}, # 存储填写的槽位信息如 {“城市”: “北京”} intent: None, # 当前识别出的意图 _created_at: time.time(), _last_active: time.time() } # 更新最后活跃时间 context[_last_active] time.time() # 确保history是deque从Redis反序列化后会是list if not isinstance(context[history], deque): context[history] deque(context[history], maxlenself.max_history_len) # 保存回缓存更新过期时间 self.save_context(session_id, context) return context def save_context(self, session_id: str, context: Dict[str, Any]): 保存上下文到缓存 context[_last_active] time.time() # 更新本地缓存 self.local_cache[session_id] context # 如果配置了Redis则同步保存 if self.redis_client: redis_key self._get_redis_key(session_id) import json # 将deque转为list以便序列化 context_to_save context.copy() context_to_save[history] list(context[history]) self.redis_client.setex(redis_key, self.session_ttl, json.dumps(context_to_save)) def add_utterance(self, session_id: str, role: str, text: str): 向会话历史中添加一条话语用户或AI context self.get_context(session_id) context[history].append({role: role, text: text, time: time.time()}) self.save_context(session_id, context) def get_formatted_history(self, session_id: str) - str: 获取格式化的对话历史用于构造LLM的Prompt context self.get_context(session_id) history_text for turn in context[history]: prefix 用户 if turn[role] user else 助手 history_text f{prefix}: {turn[text]}\n return history_text.strip() def clear_expired_sessions(self): 清理过期的会话可以定时任务调用 now time.time() expired_sessions [] for sid, ctx in self.local_cache.items(): if now - ctx.get(_last_active, 0) self.session_ttl: expired_sessions.append(sid) for sid in expired_sessions: del self.local_cache[sid] if self.redis_client: self.redis_client.delete(self._get_redis_key(sid))3. 使用HuggingFace Pipeline的意图分类器在请求到达LLM之前先用一个轻量级模型快速分类意图可以大幅提升效率并实现精准路由。from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification import torch class IntentClassifier: 基于微调Transformer模型的意图分类器 def __init__(self, model_name_or_path: str): 初始化分类器 :param model_name_or_path: 本地模型路径或HuggingFace模型ID # 使用pipeline简化调用device_map“auto”可自动分配GPU/CPU self.classifier pipeline( text-classification, modelmodel_name_or_path, tokenizermodel_name_or_path, device0 if torch.cuda.is_available() else -1, truncationTrue, max_length128 ) # 加载模型和分词器本身用于更细粒度的控制可选 self.tokenizer AutoTokenizer.from_pretrained(model_name_or_path) self.model AutoModelForSequenceClassification.from_pretrained(model_name_or_path) if torch.cuda.is_available(): self.model.cuda() # 假设我们有一个意图标签列表 self.id2label {0: 咨询产品, 1: 投诉建议, 2: 查询订单, 3: 闲聊, 4: 其他} def predict(self, text: str, top_k: int 1): 预测用户输入的意图 :param text: 用户输入文本 :param top_k: 返回最可能的K个意图 :return: 预测结果列表每个元素包含label和score results self.classifier(text, top_ktop_k) # 将结果映射到可读的标签 for res in results: # pipeline返回的label可能是字符串ID也可能是数字ID这里统一处理 label_id int(res[label].split(_)[-1]) if _ in res[label] else int(res[label]) res[label_name] self.id2label.get(label_id, 未知) return results def predict_batch(self, texts: list, batch_size: int 8): 批量预测提高吞吐量 # 使用pipeline的批量处理 results self.classifier(texts, batch_sizebatch_size, truncationTrue) # 批量结果处理逻辑... return results # 使用示例 if __name__ __main__: # 假设我们有一个在客服语料上微调过的BERT模型 classifier IntentClassifier(./models/my_finetuned_bert) user_query 我昨天买的手机什么时候能送到 intent classifier.predict(user_query) print(f查询{user_query}) print(f预测意图{intent[0][label_name]} (置信度{intent[0][score]:.3f}))四、性能优化实战让系统跑得更快更稳直接部署原始LLM模型响应速度和资源消耗都是问题。我们做了以下优化模型量化使用bitsandbytes库进行8位或4位量化能将模型体积减少50%甚至更多同时推理速度提升明显精度损失在可接受范围内。这对降低部署成本和提升QPS至关重要。高频问答缓存很多客服问题是重复的。我们使用Redis缓存“标准问题-答案”对。当新问题进来时先用一个简单的语义相似度计算如Sentence-BERT编码后计算余弦相似度去缓存里匹配如果找到高度相似的历史问题直接返回缓存答案完全绕过LLM推理。这在高并发场景下对性能提升是数量级的。五、生产环境避坑指南敏感词过滤这是红线。我们实现了一个多级过滤方案本地词库匹配维护一个敏感词库使用高效的AC自动机算法进行快速过滤。模型识别训练一个文本分类模型识别隐含的敏感或不当内容。LLM自身审查在Prompt中明确加入安全指令要求模型不生成有害内容。任何一级触发都会将回复替换为安全话术或转人工。对话日志脱敏所有对话日志在存储前必须脱敏。我们编写了正则规则和命名实体识别NER模型自动将日志中的手机号、身份证号、邮箱等个人信息替换为[REDACTED]标记确保数据安全合规。六、延伸思考未来还能怎么改进系统上线稳定运行后我们还在规划下一步的优化方向结合知识图谱LLM有时会“胡言乱语”。可以引入领域知识图谱当用户查询具体产品参数、政策条款等事实性内容时优先从图谱中检索精准答案再用LLM进行润色和整合大幅提升事实准确性。个性化对话目前的客服还是“千人一面”。下一步希望根据用户历史行为、订单信息让LLM生成更个性化的回复比如推荐相关产品、使用尊称等提升用户体验和转化率。多模态交互支持用户上传图片如故障设备照片、文档让AI客服能“看”会“读”处理更复杂的服务场景。从零搭建一个LLM智能客服系统确实是个不小的工程涉及到算法、工程、运维多个方面。但看到它最终能流畅地理解用户问题并给出有用回复时成就感也是满满的。希望这篇笔记里的架构思路、代码片段和踩坑经验能为你提供一些参考。技术迭代很快我们一起保持学习持续优化。