最近在做一个客服智能体的项目发现要把一个“聪明”的对话系统真正用起来挑战还真不小。用户的问题千奇百怪聊着聊着上下文就丢了高峰期响应还慢……这些问题不解决智能体就只是个“智障体”。经过一番折腾我们摸索出了一套基于大语言模型LLM和RAG检索增强生成的解决方案效果提升很明显。今天就来分享一下我们的实战经验希望能帮到有类似需求的同学。1. 背景痛点传统客服系统为何“不够聪明”在引入AI之前我们和很多团队一样也尝试过各种方案但总感觉差那么点意思。意图识别不准答非所问早期的规则引擎或者简单的关键词匹配太死板了。用户问“我的订单怎么还没到”系统可能只匹配到“订单”就回复查询入口而识别不到用户的核心意图是“催单”或“查询物流异常”。稍微复杂点的句式或者口语化表达系统就懵了。多轮对话困难上下文说丢就丢这是最头疼的问题。用户先问“推荐一款手机”系统推荐了A型号用户接着问“那它的电池续航怎么样”。传统系统很难把“它”和上文的“A型号手机”关联起来要么要求用户重复输入要么就给出一个通用回答体验非常割裂。知识更新慢回答“过时”产品信息、活动规则、政策条款经常变。基于规则或需要重新训练的传统NLP模型更新知识库成本高、周期长经常出现回答内容与实际不符的情况。响应延迟高体验打折尤其是在用户咨询高峰期复杂的NLP模型推理或海量知识库检索会导致响应时间变长用户等待不耐烦体验直线下降。正是这些痛点促使我们去寻找更灵活、更强大的技术方案。2. 技术选型规则、传统NLP还是大模型在构建新系统前我们对几种主流技术路径做了对比规则引擎优点逻辑清晰可控性强对于简单、固定的问答如“营业时间”实现快。缺点维护成本随着规则数量指数级增长无法处理未预定义的句式泛化能力几乎为零。不适合复杂、开放的客服场景。传统NLP模型如BERT、TextCNN等优点在特定任务如意图分类、实体识别上经过充分标注数据训练后准确率可以很高。比规则引擎灵活。缺点严重依赖大量高质量的标注数据。每个新意图或新领域都需要重新收集数据、训练模型冷启动成本高。模型能力有上限对于需要深度理解和生成的复杂对话依然乏力。大语言模型LLM优点强大的语言理解和生成能力通过提示词Prompt工程就能完成多种任务泛化能力极强。结合RAG技术可以方便地利用外部知识库如产品文档、FAQ实现知识实时更新。缺点推理成本较高可能存在“幻觉”生成不准确信息对提示词设计敏感。综合来看以LLM为核心结合RAG和必要的微调是目前构建高可用客服智能体最具性价比和可行性的方案。它既拥有了大模型的“聪明大脑”又通过RAG装上了“最新的知识库”还能通过微调让它更懂我们的业务。3. 架构设计高可用客服智能体的核心骨架我们的系统架构主要分为四层目标是实现高准确、低延迟、可扩展。接入与调度层接收来自网页、APP、微信等渠道的用户请求进行统一的鉴权、限流和负载均衡然后将请求分发给下游的对话引擎。对话引擎层核心意图识别模块这里我们采用“轻量级微调模型 LLM Few-shot校准”的策略。先用一个在业务数据上微调过的轻量模型如BERT-small做快速初筛对于低置信度的结果再用包含几个示例的Prompt交给LLM做二次判断兼顾了速度和精度。对话状态管理这是实现多轮对话的关键。我们使用Redis来存储和管理对话状态Dialog State。每个会话Session对应一个唯一的KeyValue中结构化地存储了当前对话的意图、已提取的实体、历史消息摘要等。这样无论请求被哪个服务实例处理都能获取到正确的上下文。知识检索RAG当用户问题涉及具体产品、政策时触发。系统将用户问题结合上下文转化为查询向量在向量数据库如Chroma、Milvus中检索最相关的知识片段如FAQ条目、产品手册段落。LLM合成与生成将用户问题、检索到的知识、当前的对话状态以及精心设计的系统提示词Prompt组合发送给LLM如通过API调用GPT-4、文心一言或部署开源模型如Qwen、ChatGLM生成最终的自然语言回复。数据与知识层包括向量数据库存储知识库的嵌入向量、关系型数据库存储结构化业务数据如订单信息和知识库文档源。有一个独立的“知识库嵌入管道”负责将更新的文档切片、向量化并存入向量数据库。运维与监控层包含日志收集、性能监控响应时间、Token消耗、异常告警以及一个对话复盘平台用于持续分析bad case优化模型和提示词。4. 核心实现关键代码片段解析4.1 意图识别微调Python示例我们并不直接微调巨大的LLM而是微调一个轻量级的文本分类模型作为意图识别的主力。import pandas as pd from transformers import AutoTokenizer, AutoModelForSequenceClassification, Trainer, TrainingArguments from sklearn.model_selection import train_test_split import torch # 1. 数据准备假设我们有一个CSV文件包含text和intent_label两列 df pd.read_csv(intent_training_data.csv) texts df[text].tolist() labels df[intent_label].tolist() # 将标签转换为数字ID label2id {label: idx for idx, label in enumerate(set(labels))} id2label {idx: label for label, idx in label2id.items()} numeric_labels [label2id[l] for l in labels] # 划分训练集和验证集 train_texts, val_texts, train_labels, val_labels train_test_split( texts, numeric_labels, test_size0.2, random_state42 ) # 2. 加载Tokenizer和模型这里以BERT为例 model_name bert-base-chinese # 可根据需要选择更小的模型如bert-tiny tokenizer AutoTokenizer.from_pretrained(model_name) model AutoModelForSequenceClassification.from_pretrained( model_name, num_labelslen(label2id), id2labelid2label, label2idlabel2id ) # 3. 数据预处理Tokenization def encode_texts(text_list, label_list): encodings tokenizer(text_list, truncationTrue, paddingTrue, max_length128) encodings[labels] label_list return encodings train_encodings encode_texts(train_texts, train_labels) val_encodings encode_texts(val_texts, val_labels) # 转换为PyTorch Dataset class IntentDataset(torch.utils.data.Dataset): def __init__(self, encodings): self.encodings encodings def __getitem__(self, idx): item {key: torch.tensor(val[idx]) for key, val in self.encodings.items()} return item def __len__(self): return len(self.encodings[input_ids]) train_dataset IntentDataset(train_encodings) val_dataset IntentDataset(val_encodings) # 4. 配置训练参数 training_args TrainingArguments( output_dir./intent_model_results, # 输出目录 num_train_epochs5, # 训练轮数 per_device_train_batch_size16, # 训练批次大小 per_device_eval_batch_size64, # 评估批次大小 warmup_steps500, # 预热步数 weight_decay0.01, # 权重衰减 logging_dir./logs, # 日志目录 logging_steps50, evaluation_strategyepoch, # 每个epoch评估一次 save_strategyepoch, load_best_model_at_endTrue, # 训练结束后加载最佳模型 ) # 5. 创建Trainer并开始训练 trainer Trainer( modelmodel, argstraining_args, train_datasettrain_dataset, eval_datasetval_dataset, ) trainer.train() # 6. 保存模型和tokenizer用于后续推理 model.save_pretrained(./saved_intent_model) tokenizer.save_pretrained(./saved_intent_model)4.2 基于Redis的对话状态管理我们用Redis存储结构化的对话状态确保多轮对话的连贯性。import redis import json import uuid from datetime import timedelta class DialogStateManager: def __init__(self, redis_hostlocalhost, redis_port6379, session_ttl1800): # 连接Redissession_ttl为会话过期时间秒例如30分钟 self.redis_client redis.Redis(hostredis_host, portredis_port, decode_responsesTrue) self.session_ttl session_ttl def create_or_get_session(self, session_idNone): 创建新会话或获取现有会话。如果未提供session_id则生成一个。 if not session_id: session_id str(uuid.uuid4()) key fdialog_state:{session_id} # 获取现有状态如果不存在则初始化一个空状态 state_json self.redis_client.get(key) if state_json: state json.loads(state_json) else: state { current_intent: None, entities: {}, # 存储提取的实体如 {product_name: 手机A, order_id: 12345} history_summary: , # 历史对话的摘要避免存储全部历史 turn_count: 0 } self._save_state(key, state) return session_id, state def update_state(self, session_id, updates): 更新指定会话的状态。 key fdialog_state:{session_id} state_json self.redis_client.get(key) if not state_json: # 会话可能已过期可以选择重新创建或报错 raise ValueError(fSession {session_id} not found or expired.) state json.loads(state_json) state.update(updates) # 用updates字典更新状态 state[turn_count] 1 self._save_state(key, state) return state def _save_state(self, key, state): 内部方法保存状态到Redis并设置TTL。 self.redis_client.setex(key, self.session_ttl, json.dumps(state, ensure_asciiFalse)) def clear_state(self, session_id): 主动清除某个会话的状态。 key fdialog_state:{session_id} self.redis_client.delete(key) # 使用示例 dsm DialogStateManager() session_id, current_state dsm.create_or_get_session() # 新用户来访 print(f新会话ID: {session_id}, 初始状态: {current_state}) # 用户第一轮对话后识别出意图和实体 updates_round1 { current_intent: query_product, entities: {product_name: 手机A} } new_state dsm.update_state(session_id, updates_round1) print(f第一轮后状态: {new_state}) # 用户第二轮对话追问电池信息意图识别模块会结合历史状态进行理解 # 假设系统识别出意图为“query_spec”并补充实体 updates_round2 { current_intent: query_spec, entities: {product_name: 手机A, spec_item: 电池续航} } new_state dsm.update_state(session_id, updates_round2) print(f第二轮后状态: {new_state})5. 性能优化让智能体又快又稳在高并发场景下性能优化至关重要。并发与异步处理异步框架使用FastAPI或Sanic等异步Web框架构建API服务利用async/await处理I/O密集型操作如调用LLM API、查询向量数据库。并行化将意图识别、知识检索等可以并行执行的任务使用asyncio.gather并发执行缩短整体响应时间。# 伪代码示例并行执行多个任务 import asyncio async def process_user_query(query, session_id): # 并行执行意图识别和实体提取 intent_task asyncio.create_task(detect_intent(query)) entity_task asyncio.create_task(extract_entities(query)) intent, entities await asyncio.gather(intent_task, entity_task) # ... 后续处理多级缓存策略LLM响应缓存对于高频、确定的通用问题如“你好”、“谢谢”或者相同用户短时间内重复的相同问题将其Prompt和回复的映射关系缓存起来可用Redis。下次命中时直接返回省去LLM调用开销。向量检索缓存对常见的查询向量及其Top-K检索结果进行缓存。对话状态缓存这部分已经在使用Redis了本身就是一种缓存。降级与熔断方案LLM服务降级当主要LLM API如GPT-4响应超时或不可用时自动切换到备用LLM如本地部署的较小模型或返回预定义的、基于检索的模板答案。功能降级当意图识别模型置信度极低时可以降级为直接进行知识检索并让LLM基于检索内容回答或者引导用户重新描述问题。熔断机制对LLM API、向量数据库等外部依赖设置熔断器如使用pybreaker库防止因某个依赖服务故障导致整个系统雪崩。6. 避坑指南生产环境中的那些“坑”冷启动问题问题系统上线初期标注数据少意图识别模型效果差知识库内容不足RAG检索效果不佳。应对意图识别采用“Few-shot Prompt LLM”作为冷启动期的主要意图识别手段同时积极收集真实对话数据进行人工标注迭代训练专用模型。知识库优先导入结构清晰、质量高的官方文档和FAQ。上线后通过日志分析“未命中”或“低置信度”的问题快速补充相关知识片段。数据漂移概念漂移问题用户提问的方式、关注的焦点会随着时间、活动而变化导致之前训练的模型或构建的索引效果下降。应对建立监控指标持续监控意图识别的准确率、召回率以及用户对回答的满意度如有埋点。定期迭代建立数据闭环。定期如每月用新产生的对话数据重新评估模型效果下降则启动重新训练或微调。知识库也需要定期审查和更新。A/B测试对模型或策略的重大更新通过A/B测试验证效果后再全量上线。LLM的“幻觉”问题问题LLM可能会生成看似合理但不符合事实或知识库内容的回答。应对强化RAG确保回答严格基于检索到的知识片段在Prompt中明确指令“仅根据提供的上下文信息回答”。引用溯源在回复中注明信息来源如“根据《产品手册》第X章...”增加可信度也方便后续核查。后处理校验对于关键信息如价格、日期、政策条款可以设计规则或小模型进行二次校验。成本控制问题直接使用商用LLM APIToken消耗费用可能很高。应对优化Prompt精简Prompt去除不必要的指令和示例。缓存如第5点所述充分利用缓存减少重复调用。混合模型策略简单任务用小型本地模型复杂任务再用大模型。考虑在流量低峰期使用更强大的模型如GPT-4高峰期使用速度更快、成本更低的模型如GPT-3.5-Turbo。结尾与思考通过这套组合拳我们的客服智能体在准确率、响应速度和用户体验上都有了质的飞跃。当然系统永远有优化空间。最后留一个我们正在思考的开放性问题对于客服场景中大量出现的领域专业术语、缩写和行话例如“5G SA组网”、“固件OTA升级”如何让RAG检索和LLM理解更精准是构建专门的领域词表进行查询扩展还是对嵌入模型进行领域相关的继续预训练Continue Pre-training如果你有好的想法或实践经验欢迎一起探讨。构建一个高可用的AI系统就像搭积木需要把合适的组件放在正确的位置并时刻关注它们的稳定性和效率。希望这篇笔记能为你提供一些有用的“积木块”和搭建思路。