最近在做一个智能客服的项目，之前用规则引擎和传统NLP模型效果总是不理想，要么答非所问，要么上下文接不上。正好看到通义千问的API开放了，就想着用它来重构一下。折腾了一周多，从零到一搭了个能用的生产级服务，踩了不少坑，也总结了一些经验，在这里分享给大家，希望能帮到有同样需求的同学。

1. 为什么选择大模型做客服？先聊聊背景和痛点

以前我们用的客服系统，核心是“关键词匹配+人工配置话术”。用户问“怎么退款”，系统就去匹配“退款”这个关键词，然后返回预设好的答案。这种模式有几个硬伤：

• 响应死板：稍微换个说法，比如“我想把钱退回来”，可能就匹配不上了，用户体验很差。
• 维护成本高：业务一有变动，比如上新了活动规则，运营同学就得去后台一条条添加和修改关键词与话术，非常繁琐。
• 没有记忆：每次对话都是独立的，用户多问几句“上一步说的那个活动，具体几点开始？”，系统就懵了，因为它不记得“上一步”说了啥。

大模型的出现，尤其是像通义千问这样的对话模型，简直就是为客服场景量身定做的。它能理解自然语言的意图，能进行多轮对话，还能根据上下文生成连贯、准确的回复。我们的目标，就是用一个稳定、高效的技术方案，把大模型的能力“嫁接”到我们的业务系统中。

2. 技术选型：为什么是通义千问？

市面上可选的大模型API不少，我们当时主要对比了通义千问、GPT系列和一些开源的国产模型。最终选择通义千问，主要是基于以下几点考虑：

• 效果与成本平衡：通义千问在中文理解和生成上效果非常出色，尤其在客服这种对事实准确性要求较高的场景，它的回答相对严谨。同时，它的API定价在国内模型中很有竞争力，对于需要频繁调用的客服系统来说，长期成本可控。
• 稳定与易用性：作为国内大厂推出的服务，API的稳定性和文档的完备性都很好。提供了标准的HTTP接口和SDK，接入门槛低，调试方便。
• 合规与数据安全：数据在国内处理，对于企业级应用来说，在数据安全和合规方面更让人放心，避免了潜在的数据跨境风险。

当然，它也不是完美的。比如，在特定垂直领域的专业知识上，可能不如专门微调过的模型。这就需要我们后面提到的“知识库集成”来补强。

3. 核心实现：三步搭建智能客服骨架

整个系统的核心可以抽象为三个模块：可靠的API调用层、有状态的对话管理器和必不可少的安全过滤器。

3.1 API调用封装：稳定性的基石

直接裸调API是不可靠的。网络波动、服务端限流都可能导致失败。一个健壮的封装层必须包含重试和限流。

import logging
import time
from typing import Optional, Dict, Any
import backoff
from openai import OpenAI  # 这里使用官方SDK，需安装 `openai` 包

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class QwenChatClient:
    """通义千问API客户端封装"""
    
    def __init__(self, api_key: str, base_url: str = "https://dashscope.aliyuncs.com/compatible-mode/v1"):
        self.client = OpenAI(
            api_key=api_key,
            base_url=base_url
        )
        # 简单的令牌桶限流器参数（示例）
        self.last_call_time = 0
        self.min_call_interval = 0.1  # 最小调用间隔100ms，约10 QPS
        
    def _rate_limit(self):
        """简单的限流：确保调用间隔"""
        elapsed = time.time() - self.last_call_time
        if elapsed < self.min_call_interval:
            time.sleep(self.min_call_interval - elapsed)
        self.last_call_time = time.time()
    
    @backoff.on_exception(
        backoff.expo,
        (Exception,),  # 可根据需要指定更具体的异常，如连接错误、服务器错误等
        max_tries=3,
        jitter=backoff.full_jitter
    )
    def chat_completion(self, messages: list, model: str = "qwen-max", **kwargs) -> Optional[Dict[str, Any]]:
        """
        发送聊天补全请求，包含重试机制
        :param messages: 对话消息历史，格式 [{"role": "user", "content": "你好"}]
        :param model: 使用的模型名称
        :return: API响应字典，失败返回None
        """
        self._rate_limit()  # 调用前限流
        try:
            response = self.client.chat.completions.create(
                model=model,
                messages=messages,
                **kwargs
            )
            # 适配SDK返回格式
            return {
                "choices": [{
                    "message": {
                        "role": response.choices[0].message.role,
                        "content": response.choices[0].message.content
                    }
                }]
            }
        except Exception as e:
            logger.error(f"调用通义千问API失败: {e}", exc_info=True)
            raise  # 抛出异常以便backoff进行重试
            
    def get_reply(self, messages: list) -> str:
        """简化接口，直接获取回复文本"""
        resp = self.chat_completion(messages)
        if resp and resp["choices"]:
            return resp["choices"][0]["message"]["content"].strip()
        return "抱歉，服务暂时不可用，请稍后再试。"

关键点：

1. 重试机制：使用了 backoff 库，在遇到异常时进行指数退避重试，最多3次。这能有效应对短暂的网络问题。
2. 限流：实现了简单的令牌桶逻辑（这里用睡眠模拟），防止突发流量超过API的QPS限制，避免被限流。
3. 错误处理与日志：捕获所有异常并记录详细日志，便于排查问题。即使最终失败，也返回一个友好的默认回复。

3.2 对话状态管理：让AI拥有“记忆”

客服对话是多轮的。我们需要把每次的对话历史保存下来，下次提问时连同历史一起发给模型。Redis是存储这种会话状态的理想选择，因为它快，并且支持设置过期时间。

import json
import uuid
from datetime import timedelta
import redis  # 需安装 `redis` 包

class DialogueManager:
    """基于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
        
    def _get_key(self, session_id: str) -> str:
        return f"chat:session:{session_id}"
    
    def initialize_session(self, system_prompt: str = "你是一个专业的客服助手。") -> str:
        """初始化一个新会话，返回会话ID"""
        session_id = str(uuid.uuid4())
        initial_history = [{"role": "system", "content": system_prompt}]
        self.redis.setex(
            self._get_key(session_id),
            self.ttl,
            json.dumps(initial_history, ensure_ascii=False)
        )
        return session_id
    
    def add_message(self, session_id: str, role: str, content: str):
        """向指定会话添加一条消息"""
        key = self._get_key(session_id)
        history_json = self.redis.get(key)
        
        if not history_json:
            # 会话已过期或不存在，重新初始化
            history = [{"role": "system", "content": "你是一个专业的客服助手。"}]
        else:
            history = json.loads(history_json)
            
        history.append({"role": role, "content": content})
        
        # 关键点：控制上下文长度，防止超出模型限制和成本激增
        max_turns = 10  # 保留最近10轮对话（含系统提示词）
        if len(history) > max_turns * 2 + 1:  # 每轮包含user和assistant两条消息
            # 保留系统提示词和最近的N轮对话
            history = [history[0]] + history[-(max_turns * 2):]
            
        self.redis.setex(key, self.ttl, json.dumps(history, ensure_ascii=False))
        
    def get_history(self, session_id: str) -> list:
        """获取指定会话的完整历史"""
        key = self._get_key(session_id)
        history_json = self.redis.get(key)
        if history_json:
            return json.loads(history_json)
        return []
    
    def clear_session(self, session_id: str):
        """清除指定会话"""
        self.redis.delete(self._get_key(session_id))

关键点：

1. 会话隔离：每个用户或对话线程拥有唯一的 session_id，数据互不干扰。
2. 自动过期：利用Redis的 SETEX 命令，自动清理不活跃的会话，节省内存。
3. 上下文窗口管理：这是最重要的部分！大模型的API通常按输入+输出的总token数计费，并且有长度限制。我们不能无限制地保存历史。这里采取了一个策略：始终保留系统提示词，然后只保留最近N轮（例如10轮）的对话。这样既能维持短期记忆，又控制了成本和长度。

3.3 敏感词过滤：安全防护网

AI生成的内容不可控，必须加一层过滤，防止输出不当内容。

import re
from typing import List

class ContentFilter:
    """简单的敏感词过滤模块"""
    
    def __init__(self, sensitive_words_path: str = "sensitive_words.txt"):
        # 从文件加载敏感词库，每行一个词
        with open(sensitive_words_path, 'r', encoding='utf-8') as f:
            self.sensitive_words = [line.strip() for line in f if line.strip()]
            
        # 构建正则表达式，匹配任意敏感词
        pattern = '|'.join(map(re.escape, self.sensitive_words))
        self.regex = re.compile(pattern)
        
    def filter_text(self, text: str, replace_char: str = "*") -> str:
        """过滤文本中的敏感词"""
        if not self.sensitive_words:
            return text
        return self.regex.sub(replace_char, text)
    
    def check_safe(self, text: str) -> bool:
        """检查文本是否安全（不含敏感词）"""
        if not self.sensitive_words:
            return True
        return not self.regex.search(text)
    
# 集成到回复流程中
def get_safe_reply(qwen_client, dialogue_manager, session_id, user_input):
    # 1. 先过滤用户输入（防止用户输入攻击性内容影响AI）
    filter_tool = ContentFilter()
    clean_input = filter_tool.filter_text(user_input)
    
    # 2. 更新对话历史
    dialogue_manager.add_message(session_id, "user", clean_input)
    history = dialogue_manager.get_history(session_id)
    
    # 3. 获取AI回复
    raw_reply = qwen_client.get_reply(history)
    
    # 4. 过滤AI回复
    safe_reply = filter_tool.filter_text(raw_reply)
    
    # 5. 将安全的回复存入历史
    dialogue_manager.add_message(session_id, "assistant", safe_reply)
    
    return safe_reply

关键点：

1. 双向过滤：既过滤用户的输入（防止诱导AI），也过滤AI的输出（最终安全屏障）。
2. 可配置词库：敏感词库放在外部文件，方便运营同学随时更新，无需重启服务。
3. 灵活处理：filter_text 方法可以替换为更复杂的逻辑，比如调用第三方审核API。

4. 性能考量与优化：让服务又快又稳

上线前，性能测试必不可少。我们用 locust 做了压力测试。

• QPS与延迟：在4核8G的标准云服务器上，核心服务（含Redis）的极限QPS大约在120左右，平均响应时间（RT）在800ms左右。其中，网络IO和模型API调用占了95%以上的时间。
• 优化方案：
  1. 异步化改造：使用 aiohttp 和 asyncio 将整个请求处理链路异步化，可以大幅提升并发能力，在IO密集型场景下，QPS有望提升数倍。
  2. 上下文缓存：对于常见、标准的问题（如“工作时间”、“联系方式”），可以将AI的回复缓存起来（例如缓存1小时），下次用户再问时直接返回，减少API调用。
  3. 模型选型：通义千问提供不同规格的模型（如 qwen-turbo, qwen-plus, qwen-max）。qwen-turbo 速度最快，成本最低，在多数客服场景下精度足够，是平衡性能与效果的优选。

5. 避坑指南：我踩过的那些“坑”

1. 上下文长度爆炸：这是最大的坑。如果不加控制，对话历史会越来越长，导致API调用费用暴涨，甚至触发模型的长度限制而报错。务必像上面代码一样，实现一个滑动窗口，只保留最近N轮对话。 对于需要长期记忆的场景，可以考虑用向量数据库摘要历史信息，而不是全量发送。

2. API计费优化：大模型API按Token计费。除了控制上下文长度，还可以：

   • 设置 max_tokens：在调用时明确限制AI回复的最大长度，避免它“长篇大论”。
   • 使用流式响应：如果前端支持，使用SSE（Server-Sent Events）流式输出回复。这不仅能提升用户体验（打字机效果），还能在AI生成到一半发现内容不对时提前中断，节省Token。

3. 异常流处理：网络超时、API限流、内容审核不通过……异常情况很多。

   • 分级降级：设计好降级策略。例如，第一次调用失败重试，重试失败后，可以尝试换一个更轻量的模型（如 qwen-turbo），或者直接返回预置的兜底话术。
   • 监控与告警：对API调用错误率、平均响应时间、Token消耗量设置监控指标，异常时及时告警。

6. 扩展思考：从“通用客服”到“专家客服”

基本的智能客服搭建好了，但它只能回答通用问题。一旦用户问到“我的订单123456的退款进度到哪里了？”或者“你们最新发布的XX产品支持哪些特性？”，它就无能为力了。这就需要集成知识库和业务系统。

一个常见的架构是 RAG（检索增强生成）：

1. 知识库构建：将产品文档、订单系统数据库表结构、常见问题解答（FAQ）等文本资料，通过嵌入模型（Embedding Model）转换成向量，存入向量数据库（如Milvus, Pinecone）。
2. 用户提问时：先将用户的问题转换成向量，去向量数据库中检索出最相关的几段知识。
3. 增强提示：将检索到的知识作为上下文，和用户问题一起拼接成一个新的提示（Prompt），发送给大模型。例如：“请根据以下信息回答问题：[检索到的知识]。用户问题：[用户原始问题]”。
4. 模型生成：大模型基于你提供的“知识”进行生成，这样回答的准确性和专业性会极大提高。

这样一来，你的客服机器人就不仅能聊天，还能真正查询业务数据，成为业务专家了。这是智能客服价值最大化的方向。

写在最后

通过上面这套组合拳，我们用一个多星期的时间，就搭建起了一个效果不错、运行稳定、具备生产部署能力的AI智能客服原型。通义千问的API确实大大降低了技术门槛。当然，这只是起点。真正要让这个系统创造业务价值，后续还需要在知识库构建、多轮对话策略优化、用户满意度评估等方面持续迭代。

最后，留三个问题给大家思考，也是我们团队正在探索的方向：

1. 在多租户的SaaS客服场景下，如何高效地管理和隔离不同客户的知识库与对话数据？
2. 当客服机器人无法确定答案时，如何设计一个平滑、智能的转人工流程？
3. 如何利用用户与AI客服的真实对话日志，自动化地发现知识盲区，并补充到知识库中？

希望这篇笔记对你有帮助。搭建过程中有任何问题，欢迎一起交流。