最近在做一个智能客服项目，选型时被各种大模型和开源框架搞得眼花缭乱。传统基于规则或小模型的客服系统，在意图识别（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": f"Bearer {api_key}"})
        self.logger = logging.getLogger(__name__)
        
    @retry(
        stop=stop_after_attempt(3), # 最多重试3次
        wait=wait_exponential(multiplier=1, min=2, max=10), # 指数退避等待
        retry=retry_if_exception_type((RequestException, Timeout)), # 仅对网络异常重试
        reraise=True
    )
    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, json=payload, timeout=timeout)
            response.raise_for_status() # 检查HTTP状态码
            result = response.json()
            return result["choices"][0]["message"]["content"]
        except (RequestException, Timeout) as e:
            self.logger.error(f"API请求失败: {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_key="your_api_key_here")
    history = [{"role": "user", "content": "你们公司的退货政策是什么？"}]
    reply = client.chat_completion(messages=history)
    print(f"AI回复: {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(host='localhost', port=6379, db=0, decode_responses=True)
    manager = DialogueStateManager(r, ttl_seconds=1200) # 20分钟过期
    
    # 用户开始对话
    session_id = manager.create_session(user_id="user_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(maxsize=maxsize)(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(maxsize=50)
# 第一次询问，会调用 _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(capacity=10, fill_rate=2) # 桶容量10，每秒补2个令牌
def make_api_call_with_limit():
    if bucket.consume(1):
        # 执行API调用
        # client.chat_completion(...)
        pass
    else:
        # 被限流，执行降级策略
        # fallback_responder.get_fallback_response(...)
        pass

6. 延伸思考：基于用户反馈的模型优化

系统上线后，收集用户反馈（如“点赞”、“点踩”或转人工）是持续优化的关键。我们可以利用这些数据对模型进行微调（Fine-tuning），使其更贴合我们的业务场景。

一个简单的微调数据准备流程设计：

1. 数据收集：在对话界面提供“有帮助/无帮助”的反馈按钮。当用户给出负面反馈或转人工时，自动将最近几轮对话（用户问题+AI回复）保存为待优化样本。
2. 数据清洗与标注：定期从待优化样本中筛选出有代表性的案例。对于AI回复不佳的样本，由业务专家标注出期望的正确回复或修改思路。
3. 构建微调数据集：将清洗标注后的数据，按照大模型微调要求的格式（例如，对于对话模型，构造{"messages": [{"role": "user", "content": "..."}, {"role": "assistant", "content": "..."}]}的序列）进行整理。
4. 模型微调与评估：使用沙丘智库等平台提供的微调API，在小规模标注数据上训练一个专属模型版本。之后通过A/B测试，对比微调模型和基础模型在相同业务场景下的表现（如满意度、问题解决率）。
5. 迭代循环：将效果更好的模型版本部署上线，继续收集反馈，形成“数据收集 -> 标注 -> 微调 -> 评估 -> 部署”的闭环迭代。

写在最后

从零搭建基于大模型的智能客服，技术选型上云API能极大降低起步门槛。核心在于做好工程化封装：一个健壮的API客户端、一个可靠的状态管理中间件、以及针对成本和性能的优化策略。生产环境中，日志脱敏、降级策略和限流这些“非功能性需求”往往比模型本身更能决定系统的稳定性。

上面分享的代码和思路，是我们项目实践中的一些总结。实际应用中，还需要结合具体的业务逻辑（比如如何从大模型的回复中提取结构化信息来查询订单数据库）进行填充。希望这篇笔记能帮你避开我们踩过的一些坑，更快地把智能客服系统跑起来。大模型技术迭代很快，保持学习，持续从用户反馈中优化，才是王道。