最近在做一个智能客服项目选型时被各种大模型和开源框架搞得眼花缭乱。传统基于规则或小模型的客服系统在意图识别Intent Recognition和上下文保持Context Keeping上总是差强人意用户稍微换个说法可能就识别不了多聊几句就忘了前面说过啥。直到接触了沙丘智库这类大模型API才发现事情可以简单很多。今天就来分享一下如何从零开始用沙丘智库大模型搭建一个能上线使用的智能客服系统重点聊聊工程化落地时那些“坑”。1. 为什么选择大模型先看清传统方案的“天花板”在决定用沙丘智库之前我们团队也评估过传统方案。核心痛点非常明显意图识别僵化基于关键词或传统机器学习模型如用BERT微调的意图识别需要大量的标注数据并且泛化能力有限。用户问“怎么取消订单”和“我不想买了怎么办”在模型眼里可能就是两个不同的意图需要大量相似问法去“喂”才能识别准确。上下文记忆短暂传统的对话管理Dialogue Management多采用有限状态机FSM或基于规则的槽位填充Slot Filling。一旦对话轮次变多或者用户中途跳转话题状态管理就会变得极其复杂很难实现流畅的自由多轮对话。开发维护成本高每新增一个业务场景比如从“查物流”扩展到“开发票”都需要重新收集数据、训练模型、调整对话流程周期长响应业务变化慢。大模型的出现尤其是像沙丘智库这样提供了对话API的模型从根本上改变了这个局面。它通过海量数据预训练具备了强大的语言理解和生成能力能够直接理解用户的自然语言表达并基于整个对话历史生成连贯、准确的回复。这意味着我们不再需要单独维护一个意图识别模型和一个对话状态跟踪器很多复杂性被大模型自身的能力吸收了。2. 技术选型沙丘智库API vs. 开源方案如Rasa确定了要用大模型接下来就是选型。我们主要对比了直接调用沙丘智库这类云API和采用Rasa等开源框架自建两种路径。准确率与智能程度沙丘智库API优势在于模型本身能力强。由于是超大规模预训练模型在开放域对话、语义理解深度上通常优于我们自己在有限数据上微调的小模型。对于客服中常见的同义表述、模糊查询理解更精准。Rasa自建NLU其NLU核心通常是像DIET或BERT这样的轻量级模型。在垂直领域、固定意图集上如果有充足且高质量的标注数据可以达到很高的准确率。但对于训练数据未覆盖的长尾问题泛化能力不如大模型。响应速度与性能沙丘智库API响应时间取决于网络和API服务端的性能一般在几百毫秒到一两秒之间。作为服务调用方我们无需关心模型推理的硬件优化。Rasa本地部署响应速度极快通常在几十毫秒内。但前提是需要有足够的GPU/CPU资源进行本地推理并且要自行优化服务化部署如用Redis做Tracker Store。成本与工程复杂度沙丘智库API按调用量付费无前期模型训练和服务器成本。工程重点在于API的稳定调用、对话状态管理和业务逻辑集成相对轻量。Rasa零API调用费用但需要投入人力进行数据标注、模型训练、Pipeline调优、服务器运维和弹性伸缩设计。对于中小团队全链路的工程复杂度不低。综合来看如果追求快速上线、应对复杂语言灵活、且希望初期投入可控沙丘智库API是更优选择。如果业务场景非常固定、对话流程标准化程度高、且对数据隐私和响应延迟有极致要求自建Rasa方案值得考虑。我们项目属于前者所以选择了沙丘智库。3. 核心实现封装API与对话状态管理选型定了就开始动手写代码。第一步是把API调用封装得健壮一些。3.1 稳健的API客户端封装直接裸调requests可不行网络波动、API限流、服务暂时不可用都是生产环境常客。下面是一个包含异常重试、限流和基础日志的封装类import time import logging from typing import Optional, Dict, Any from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type import requests from requests.exceptions import RequestException, Timeout class ShaqiuClient: 沙丘智库API客户端封装 def __init__(self, api_key: str, base_url: str https://api.shaqiu.com/v1): self.api_key api_key self.base_url base_url self.session requests.Session() self.session.headers.update({Authorization: fBearer {api_key}}) self.logger logging.getLogger(__name__) retry( stopstop_after_attempt(3), # 最多重试3次 waitwait_exponential(multiplier1, min2, max10), # 指数退避等待 retryretry_if_exception_type((RequestException, Timeout)), # 仅对网络异常重试 reraiseTrue ) def chat_completion(self, messages: list[Dict[str, str]], model: str shaqiu-pro, temperature: float 0.7, timeout: int 30) - Optional[str]: 调用对话补全API :param messages: 对话历史消息列表格式 [{role: user, content: 你好}] :param model: 使用的模型名称 :param temperature: 生成多样性0-1 :param timeout: 请求超时时间秒 :return: 模型生成的回复内容失败返回None url f{self.base_url}/chat/completions payload { model: model, messages: messages, temperature: temperature } try: response self.session.post(url, jsonpayload, timeouttimeout) response.raise_for_status() # 检查HTTP状态码 result response.json() return result[choices][0][message][content] except (RequestException, Timeout) as e: self.logger.error(fAPI请求失败: {e}, 消息: {messages[-1][content][:50]}...) raise # 触发重试 except KeyError as e: self.logger.error(f解析API响应失败响应结构异常: {e}, 原始响应: {response.text}) return None except Exception as e: self.logger.error(f未知错误: {e}) return None # 使用示例 if __name__ __main__: client ShaqiuClient(api_keyyour_api_key_here) history [{role: user, content: 你们公司的退货政策是什么}] reply client.chat_completion(messageshistory) print(fAI回复: {reply})这个封装里用了tenacity库实现优雅重试用指数退避避免加重服务压力。同时区分了网络异常重试和业务逻辑异常如响应格式错误不重试。3.2 多轮对话状态管理Redis实现客服对话不是一问一答需要记住上下文。我们用Redis来存储和管理会话状态关键点是会话隔离和自动过期。import json import uuid from datetime import timedelta from typing import List, Dict, Any, Optional import redis # 需要 pip install redis class DialogueStateManager: 基于Redis的多轮对话状态管理器 def __init__(self, redis_client: redis.Redis, ttl_seconds: int 1800): :param redis_client: Redis连接客户端 :param ttl_seconds: 会话存活时间秒默认30分钟 self.redis redis_client self.ttl ttl_seconds self.session_prefix chat_session: def create_session(self, user_id: Optional[str] None) - str: 创建新会话返回会话ID session_id user_id if user_id else str(uuid.uuid4()) key self.session_prefix session_id # 初始化一个空的对话历史 initial_state { message_history: [], # 存储完整的messages列表 created_at: time.time() } self.redis.setex(key, self.ttl, json.dumps(initial_state)) return session_id def get_session_history(self, session_id: str) - List[Dict[str, str]]: 获取指定会话的完整对话历史 key self.session_prefix session_id data self.redis.get(key) if not data: return [] # 会话已过期或不存在 state json.loads(data) return state.get(message_history, []) def add_message_to_session(self, session_id: str, role: str, content: str) - bool: 向指定会话添加一条消息并刷新TTL :return: 操作是否成功 key self.session_prefix session_id data self.redis.get(key) if not data: return False # 会话不存在 state json.loads(data) history state.get(message_history, []) history.append({role: role, content: content}) # 可选控制历史记录长度防止无限增长例如只保留最近10轮 if len(history) 20: # 10轮对话user和assistant交替 history history[-20:] state[message_history] history # 重新设置键值并刷新过期时间 self.redis.setex(key, self.ttl, json.dumps(state)) return True def clear_session(self, session_id: str) - None: 清除指定会话 key self.session_prefix session_id self.redis.delete(key) # 使用示例 if __name__ __main__: # 连接Redis假设运行在本地 r redis.Redis(hostlocalhost, port6379, db0, decode_responsesTrue) manager DialogueStateManager(r, ttl_seconds1200) # 20分钟过期 # 用户开始对话 session_id manager.create_session(user_iduser_123) # 用户提问 manager.add_message_to_session(session_id, user, 我想查询订单状态) # 获取历史用于调用API history manager.get_session_history(session_id) # 这里将history传给上面的ShaqiuClient... # ai_reply client.chat_completion(history) # 假设AI回复了 ai_reply_content 好的请提供您的订单号。 manager.add_message_to_session(session_id, assistant, ai_reply_content) # 下一轮用户回复 manager.add_message_to_session(session_id, user, 订单号是 ABC123) # 再次获取更新后的历史进行下一轮调用...这里每个用户或每个对话有一个独立的session_id作为Redis键的一部分实现了数据隔离。setex操作保证了会话在指定时间TTL无活动后自动清除避免内存泄漏。我们还简单实现了历史记录长度截断防止上下文过长导致API令牌Token超限或成本过高。4. 性能优化降低成本与提升响应速度直接调用大模型API成本和延迟是两大关注点。我们做了两方面的优化。4.1 请求批处理Batch Processing如果客服系统同时收到多个用户相似的问题比如高峰期都问“发货时间”我们可以将这些问题批量发送给API减少调用次数。沙丘智库的API可能本身支持批量请求如果不支持我们可以在应用层对短时间内相似的问题进行聚合但要注意不能影响用户体验适用于异步或非实时场景。一个简单的思路是维护一个短暂的问题缓冲区from collections import defaultdict import asyncio from typing import List, Tuple class BatchProcessor: 简单的问答批处理器示例概念 def __init__(self, batch_window: float 0.5): # 0.5秒的批处理窗口 self.batch_window batch_window self.buffer defaultdict(list) # key可以是问题特征或意图value是(question, callback)列表 async def submit_question(self, question: str, question_key: str, callback): 提交问题 :param question_key: 用于判断是否可批量处理的键如问题Embedding的聚类ID或意图 self.buffer[question_key].append((question, callback)) # 如果是该批次第一个问题启动一个延迟任务来处理整个批次 if len(self.buffer[question_key]) 1: asyncio.get_event_loop().call_later( self.batch_window, self._process_batch, question_key ) def _process_batch(self, key: str): 处理一个批次的问题 if key not in self.buffer: return batch_items self.buffer.pop(key) if not batch_items: return # 假设所有问题相同或相似只调用一次API这里需要根据实际API能力调整 sample_question batch_items[0][0] # 调用大模型API获取答案 # answer client.chat_completion(...) answer f这是对于{sample_question}的批量回答。 # 将答案分发给所有请求 for _, callback in batch_items: callback(answer) # 通过回调函数返回结果4.2 LRU缓存高频问答对对于非常常见、答案固定的问题如“公司地址”、“客服电话”每次调用大模型既浪费钱又慢。我们可以用LRU最近最少使用缓存来存储这些问答对。from functools import lru_cache import hashlib class FAQCache: 高频问答缓存 def __init__(self, maxsize: int 100): # 使用LRU缓存最多缓存100个问答对 self._get_answer_cached lru_cache(maxsizemaxsize)(self._get_answer_uncached) def _get_answer_uncached(self, question_hash: str) - str: 未命中缓存时的实际处理函数。 在实际应用中这里可能查询本地知识库或调用大模型。 本例中我们模拟一个慢速的“后备源”。 # 模拟一个耗时的操作比如查询数据库或调用API time.sleep(0.1) # 这里应该是实际的逻辑例如 # answer client.chat_completion([{role: user, content: question}]) return f这是对于哈希值{question_hash}对应问题的答案。 def get_answer(self, question: str) - str: 获取答案优先从缓存读取。 使用问题的哈希值作为缓存键。 # 对问题内容进行哈希作为缓存键。也可以结合意图识别结果。 question_hash hashlib.md5(question.strip().encode(utf-8)).hexdigest() return self._get_answer_cached(question_hash) # 使用示例 cache FAQCache(maxsize50) # 第一次询问会调用 _get_answer_uncached answer1 cache.get_answer(你们的营业时间) print(answer1) # 短时间内再次询问相同问题直接从缓存返回 answer2 cache.get_answer(你们的营业时间) print(answer2) # 这次会非常快时间复杂度分析lru_cache的存取操作时间复杂度近似O(1)。哈希计算是O(n)n为字符串长度但相对于网络IO或模型推理可以忽略。5. 避坑指南生产环境必须考虑的几点5.1 对话日志脱敏记录日志用于分析很重要但用户隐私数据手机号、身份证、订单号绝不能明文存储。import re def desensitize_text(text: str) - str: 简单脱敏函数示例 # 脱敏手机号简单正则实际需要更严谨 text re.sub(r(\d{3})\d{4}(\d{4}), r\1****\2, text) # 脱敏身份证号18位或15位 text re.sub(r(\d{6})\d{8}(\w{4}), r\1********\2, text) # 脱敏邮箱保留前第一位和最后一位中间用*代替 text re.sub(r(\w)[^]*, r\1****, text) return text # 在记录日志前调用 log_message f用户提问: {desensitize_text(user_input)}5.2 冷启动与默认回复策略在系统刚启动或大模型API不可用时需要有降级方案。class FallbackResponder: 回退响应器 def __init__(self, default_responses: List[str]): self.default_responses default_responses self.index 0 def get_fallback_response(self, user_input: str) - str: 获取回退回复。可以基于规则或轮询默认回复列表。 # 策略1: 简单轮询默认回复 response self.default_responses[self.index % len(self.default_responses)] self.index 1 return response # 策略2: 基于关键词的简单规则匹配可选 # if 价格 in user_input: # return 关于价格的具体信息请您稍后咨询人工客服。 # return 您的问题我已收到正在努力理解中请稍后再试或联系人工客服。5.3 并发控制令牌桶限流防止意外代码bug或突发流量导致API调用超限引发额外费用或服务被封。import threading import time class TokenBucket: 简单的令牌桶限流器 def __init__(self, capacity: int, fill_rate: float): :param capacity: 桶容量最大令牌数 :param fill_rate: 每秒填充的令牌数 self.capacity float(capacity) self._tokens float(capacity) self.fill_rate fill_rate self.last_time time.time() self.lock threading.Lock() def consume(self, tokens: int 1) - bool: 消费指定数量的令牌。 :return: 如果成功消费返回True否则返回False被限流 with self.lock: now time.time() # 计算从上一次到现在应该填充的令牌数 delta self.fill_rate * (now - self.last_time) self._tokens min(self.capacity, self._tokens delta) self.last_time now if self._tokens tokens: self._tokens - tokens return True return False # 在API客户端调用前使用 bucket TokenBucket(capacity10, fill_rate2) # 桶容量10每秒补2个令牌 def make_api_call_with_limit(): if bucket.consume(1): # 执行API调用 # client.chat_completion(...) pass else: # 被限流执行降级策略 # fallback_responder.get_fallback_response(...) pass6. 延伸思考基于用户反馈的模型优化系统上线后收集用户反馈如“点赞”、“点踩”或转人工是持续优化的关键。我们可以利用这些数据对模型进行微调Fine-tuning使其更贴合我们的业务场景。一个简单的微调数据准备流程设计数据收集在对话界面提供“有帮助/无帮助”的反馈按钮。当用户给出负面反馈或转人工时自动将最近几轮对话用户问题AI回复保存为待优化样本。数据清洗与标注定期从待优化样本中筛选出有代表性的案例。对于AI回复不佳的样本由业务专家标注出期望的正确回复或修改思路。构建微调数据集将清洗标注后的数据按照大模型微调要求的格式例如对于对话模型构造{messages: [{role: user, content: ...}, {role: assistant, content: ...}]}的序列进行整理。模型微调与评估使用沙丘智库等平台提供的微调API在小规模标注数据上训练一个专属模型版本。之后通过A/B测试对比微调模型和基础模型在相同业务场景下的表现如满意度、问题解决率。迭代循环将效果更好的模型版本部署上线继续收集反馈形成“数据收集 - 标注 - 微调 - 评估 - 部署”的闭环迭代。写在最后从零搭建基于大模型的智能客服技术选型上云API能极大降低起步门槛。核心在于做好工程化封装一个健壮的API客户端、一个可靠的状态管理中间件、以及针对成本和性能的优化策略。生产环境中日志脱敏、降级策略和限流这些“非功能性需求”往往比模型本身更能决定系统的稳定性。上面分享的代码和思路是我们项目实践中的一些总结。实际应用中还需要结合具体的业务逻辑比如如何从大模型的回复中提取结构化信息来查询订单数据库进行填充。希望这篇笔记能帮你避开我们踩过的一些坑更快地把智能客服系统跑起来。大模型技术迭代很快保持学习持续从用户反馈中优化才是王道。