最近在做一个智能客服系统的升级项目，客户对传统客服的响应速度和理解能力很不满意。调研了一圈，最终决定用 DeepSeek 来试试，效果确实不错。今天就把整个搭建过程中的技术实现和优化心得整理一下，希望能给有类似需求的同学一些参考。

1. 为什么传统方案在客服场景里总感觉“差点意思”？

最开始我们用的是规则引擎加一些简单的 NLP 模型，比如用正则表达式匹配关键词，或者用传统的分类算法做意图识别。运行一段时间后，问题就暴露出来了：

• 意图识别率上不去：用户问“我的订单怎么还没到？”和“快递到哪了？”，在规则引擎里可能需要写两条规则，但模型应该能理解这是同一个意图（查询物流）。传统模型对这类语义相似但表述不同的句子泛化能力弱。
• 多轮对话是硬伤：用户先问“推荐一款手机”，客服回答后，用户接着问“那它的电池续航怎么样？”。这里的“它”指代上一轮提到的手机，传统方案很难维持这种上下文关联，对话容易断片。
• 冷启动和维护成本高：每增加一个新业务（比如退换货政策变了），就需要人工梳理大量问法，去补充规则或标注数据，非常耗时。

这些痛点让我们开始寻找更“智能”的解决方案。

2. 技术选型：为什么是DeepSeek？

当时主要对比了 Rasa（开源）、Dialogflow（Google）和 DeepSeek。简单说说我的看法：

• Rasa：功能强大，完全开源可控，适合对数据隐私要求高、且有较强算法团队的公司。但它的 NLU（自然语言理解）和对话管理需要自己训练和调优，初始搭建和持续优化的技术门槛不低。
• Dialogflow：谷歌出品，上手快，对话管理设计得很直观。但它是云服务，数据要出境，很多国内项目有合规顾虑。而且定制化程度深了之后，成本会上升得比较快。
• DeepSeek：吸引我们的点有几个。第一，它的基座模型在中文理解和生成上表现非常出色，这在客服场景（尤其是中文为主）是巨大优势。第二，API 调用方式简单，让我们能快速集成和验证效果，把精力更多放在业务逻辑和工程优化上。第三，在成本可控的前提下，提供了足够强大的上下文理解和多轮对话能力。

综合来看，对于想快速构建一个高性能、且以中文场景为核心的智能客服团队，DeepSeek 是一个平衡了效果、速度和成本的选项。

3. 核心实现：从API调用到对话引擎

3.1 搭建基础的对话引擎

首先，我们需要一个核心的 DialogueEngine 类来处理用户输入。这里的关键是维护好对话历史（context）。

import json
import time
from typing import List, Dict, Any
import requests

class DeepSeekDialogueEngine:
    def __init__(self, api_key: str, base_url: str = "https://api.deepseek.com/v1/chat/completions"):
        self.api_key = api_key
        self.base_url = base_url
        self.conversation_history: List[Dict[str, str]] = []  # 存储对话历史
        self.max_history_turns = 10  # 控制历史记录轮数，防止上下文过长

    def _call_deepseek_api(self, messages: List[Dict]) -> Dict[str, Any]:
        """调用DeepSeek API的核心方法"""
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        payload = {
            "model": "deepseek-chat",  # 根据实际可用模型调整
            "messages": messages,
            "temperature": 0.7,  # 控制回复的随机性，客服场景可以调低
            "max_tokens": 1024
        }
        try:
            response = requests.post(self.base_url, headers=headers, json=payload, timeout=30)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            # 在实际生产中，这里需要更完善的错误处理和重试机制
            print(f"API调用失败: {e}")
            return {"choices": [{"message": {"content": "抱歉，服务暂时不可用，请稍后再试。"}}]}

    def add_to_history(self, role: str, content: str):
        """向对话历史中添加一条记录，并控制总长度"""
        self.conversation_history.append({"role": role, "content": content})
        # 如果历史记录超过限制，移除最老的几轮对话（保留系统提示和最近对话）
        if len(self.conversation_history) > self.max_history_turns * 2:  # 每轮包含user和assistant
            # 通常保留第一条系统消息和最新的对话
            if len(self.conversation_history) > 1 and self.conversation_history[0]["role"] == "system":
                self.conversation_history = [self.conversation_history[0]] + self.conversation_history[-(self.max_history_turns*2-1):]
            else:
                self.conversation_history = self.conversation_history[-(self.max_history_turns*2):]

    def get_response(self, user_input: str, system_prompt: str = None) -> str:
        """处理用户输入并获取回复"""
        # 1. 构建消息列表
        messages = []
        if system_prompt:
            messages.append({"role": "system", "content": system_prompt})
        
        # 2. 添加历史对话（压缩后的）
        compressed_history = self._compress_conversation_history()
        messages.extend(compressed_history)
        
        # 3. 添加当前用户输入
        messages.append({"role": "user", "content": user_input})
        
        # 4. 调用API
        api_response = self._call_deepseek_api(messages)
        
        # 5. 提取助手回复
        assistant_reply = api_response["choices"][0]["message"]["content"]
        
        # 6. 更新历史记录
        self.add_to_history("user", user_input)
        self.add_to_history("assistant", assistant_reply)
        
        return assistant_reply

    def _compress_conversation_history(self) -> List[Dict[str, str]]:
        """对话历史压缩算法（简化版），防止token超限"""
        # 这里是简化逻辑，实际可以更智能，比如总结多轮对话为一个摘要
        # 时间复杂度 O(n)，n为历史记录条数
        if len(self.conversation_history) <= self.max_history_turns * 2:
            return self.conversation_history.copy()
        # 简单截断，保留最近的对话
        return self.conversation_history[-(self.max_history_turns*2):]

3.2 意图识别与实体抽取的增强

虽然 DeepSeek 能端到端地理解用户问题，但在客服场景，我们通常需要明确的结构化信息（比如意图分类、订单号、产品名）来触发后续的业务流程。我们可以设计一个“轻量级模型+DeepSeek兜底”的混合策略。

import re
from enum import Enum

class UserIntent(Enum):
    GREETING = "greeting"
    QUERY_ORDER = "query_order"
    COMPLAINT = "complaint"
    PRODUCT_RECOMMENDATION = "product_recommendation"
    UNKNOWN = "unknown"

class IntentEntityExtractor:
    def __init__(self, deepseek_engine: DeepSeekDialogueEngine):
        self.engine = deepseek_engine
        # 可以预加载一些规则或小模型用于快速匹配
        self.intent_keywords = {
            UserIntent.QUERY_ORDER: ["订单", "物流", "快递", "到哪了", "运单"],
            UserIntent.COMPLAINT: ["投诉", "不满意", "差评", "生气", "问题"],
            UserIntent.PRODUCT_RECOMMENDATION: ["推荐", "哪款好", "有什么", "介绍一下"],
        }
        
    def extract(self, user_input: str) -> Dict[str, Any]:
        """提取意图和实体，优先使用快速规则，失败则调用大模型"""
        result = {
            "intent": UserIntent.UNKNOWN,
            "entities": {},
            "confidence": 0.0
        }
        
        # 1. 快速规则匹配（O(k*m)，k为意图数，m为关键词数，通常很小）
        for intent, keywords in self.intent_keywords.items():
            for keyword in keywords:
                if keyword in user_input:
                    result["intent"] = intent
                    result["confidence"] = 0.7  # 规则匹配赋予中等置信度
                    break
            if result["intent"] != UserIntent.UNKNOWN:
                break
                
        # 2. 如果规则未匹配，或置信度要求高，则调用DeepSeek进行深度理解
        if result["intent"] == UserIntent.UNKNOWN:
            # 构建一个专门用于信息提取的Prompt
            extraction_prompt = f"""
            请分析以下用户对话，并提取结构化信息。
            用户输入：{user_input}
            请以JSON格式返回，包含以下字段：
            - intent: 用户意图，可选值：问候、查询订单、投诉、产品咨询、其他。
            - entities: 实体字典，如订单号、产品名称、日期等。
            - confidence: 你对这个分析的置信度（0-1之间）。
            """
            try:
                # 注意：这里为清晰起见直接调用，实际应避免在提取器中嵌套调用对话引擎，可能造成循环。
                # 更好的做法是直接使用一次DeepSeek调用，同时完成对话和结构化提取。
                analysis_result = self.engine.get_response(extraction_prompt, system_prompt="你是一个精准的信息提取助手，只返回JSON。")
                # 解析返回的JSON字符串
                parsed_result = json.loads(analysis_result)
                # 映射回我们的Intent Enum (这里需要做映射逻辑，代码略)
                result.update(parsed_result)
            except:
                # 解析失败，降级处理
                pass
                
        # 3. 实体抽取（示例：抽订单号）
        order_pattern = r'订单[号|编号]?[:：]?\s*([A-Z0-9]{8,12})'
        match = re.search(order_pattern, user_input)
        if match:
            result["entities"]["order_id"] = match.group(1)
            
        return result

3.3 知识图谱与对话状态的协同

智能客服不能只靠“生成”，还需要“查询”准确的知识。我们的做法是构建一个产品知识图谱（用 Neo4j 或只是简单的 JSON 索引），然后在系统提示（System Prompt）中巧妙地引导模型去利用这个图谱。

1. 知识图谱存储：将产品信息、常见问题解答（FAQ）、政策条款以结构化的方式存储，形成实体（产品、问题）和关系（属于、相关、步骤）的网络。
2. 对话状态管理：维护一个 DialogueState 对象，记录当前对话的意图、已填写的槽位（Slots，如订单号、问题类型）、以及历史路径。
3. 协同机制：
   • 当 IntentEntityExtractor 识别出用户想查询某个产品的参数时，DialogueState 会记录“产品咨询”意图和产品名称实体。
   • 系统提示词会动态包含：“当前用户正在咨询产品A的电池信息。以下是产品A的已知知识：[从知识图谱查询到的电池规格]”。
   • DeepSeek 在生成回复时，就会基于这些精准的知识来组织语言，而不是凭空想象。

这样既保证了信息的准确性，又保留了大模型自然语言组织的优势。

4. 性能优化：让系统扛得住真实流量

直接调用大模型 API，延迟和成本是必须考虑的问题。以下是几个关键的优化点：

4.1 对话上下文压缩算法

随着对话轮数增加，上下文会越来越长，导致后续每次 API 调用的 Token 数激增，响应变慢，费用也更高。我们实现了以下压缩策略：

1. 关键信息摘要：对于较久远的对话，不再传送原始文本，而是用一个小模型（或让大模型自己）生成一段摘要。例如，将前5轮关于“手机选购”的讨论，总结为“用户想购买预算3000-4000元、续航强的手机，已排除A型号”。
2. 丢弃无关轮次：识别并丢弃与当前对话目标明显无关的历史轮次（比如中间的寒暄）。
3. Token 数精确预算：设定一个 Token 上限（如 4000），采用滑动窗口，优先保留最近的对话和系统指令。

# 一个简化的摘要压缩示例（生产环境需要更复杂的实现）
def summarize_dialogue_turns(self, old_turns: List[Dict]) -> str:
    """将多轮旧对话压缩成一段摘要文本"""
    # 这里可以集成一个文本摘要模型，例如用T5-small
    # 为简化，这里演示逻辑
    concatenated_text = " ".join([turn["content"] for turn in old_turns])
    # 模拟摘要过程：取开头、结尾和关键词
    # 实际应用应替换为真实的摘要模型调用
    if len(concatenated_text) > 500:
        summary = concatenated_text[:200] + " ... [中间部分已总结] ... " + concatenated_text[-200:]
    else:
        summary = concatenated_text
    return f"【历史对话摘要】{summary}"

4.2 异步响应与流式输出

对于复杂的用户问题，模型生成可能需要几秒钟。为了不让用户面对空白页面等待，我们采用两种方式：

1. 异步任务队列：将用户的请求放入消息队列（如 Redis + Celery 或 RabbitMQ），立即返回一个“正在思考”的提示。后端 worker 调用 DeepSeek API 得到结果后，再通过 WebSocket 或轮询接口推送给前端。
2. 流式输出（如果API支持）：如果 DeepSeek API 支持 Server-Sent Events (SSE) 或类似的流式响应，就可以实现打字机效果，边生成边返回，用户体验更好。

4.3 负载测试与数据

我们使用 Locust 进行了压力测试，模拟了不同并发用户下的场景（假设API延迟在 1-2 秒左右）：

• 单实例基准：在上下文管理良好的情况下，单服务实例处理简单问答，QPS 大约在 10-15 左右。瓶颈主要在于网络 I/O 和模型 API 的响应时间。
• 响应延迟：P95 延迟控制在 2.5 秒内，P99 延迟在 4 秒内。通过前面提到的异步化和上下文压缩，避免了长尾延迟。
• 扩容策略：由于对话服务本身是无状态的（状态可存入 Redis），水平扩展很容易。通过增加实例和负载均衡，可以线性提升整体 QPS。

5. 避坑指南：安全、合规与冷启动

5.1 敏感词过滤

绝对不能完全依赖大模型的自律。必须在调用模型前和模型返回后都进行过滤。

• 前置过滤：用户输入文本经过敏感词库（AC自动机算法，时间复杂度 O(n)）过滤，命中则直接返回预设的安全回复，不调用模型。
• 后置过滤：对模型生成的内容再次进行敏感词和合规性校验，确保输出安全。
• 模型层提示：在系统提示词中明确加入“你必须遵守法律法规，拒绝回答涉及...的问题”。

5.2 对话日志的脱敏存储

客服对话包含大量用户隐私（电话、地址、订单号）。存储日志用于分析前必须脱敏。

import re

def anonymize_log(text: str) -> str:
    """对日志文本进行脱敏处理"""
    # 脱敏手机号
    text = re.sub(r'1[3-9]\d{9}', '[PHONE]', text)
    # 脱敏身份证号（简易版）
    text = re.sub(r'[1-9]\d{5}(19|20)\d{2}(0[1-9]|1[0-2])(0[1-9]|[12]\d|3[01])\d{3}[\dXx]', '[ID]', text)
    # 脱敏邮箱（简易版）
    text = re.sub(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', '[EMAIL]', text)
    # 自定义规则：脱敏订单号（假设格式为ORDER123456）
    text = re.sub(r'ORDER\d{6}', '[ORDER_ID]', text)
    return text

# 在存储到数据库或日志文件前调用
safe_log = anonymize_log(original_dialogue_text)

5.3 冷启动时的语料准备建议

在系统刚上线、没有真实对话数据时，如何让模型表现更好？

1. 构建高质量的种子问答对（FAQ）：这是最重要的。收集过去客服工单、邮件中的真实问答，进行清洗和整理。数量不在多，而在精和覆盖核心业务场景。
2. 利用系统提示词（Prompt）进行引导：在 Prompt 中详细定义客服的角色、职责、回答风格和边界。提供几个优秀的回答示例（Few-shot Learning）。
3. 模拟对话演练：编写脚本，模拟各种用户（包括刁钻用户）进行多轮对话测试，不断调整 Prompt 和流程逻辑。
4. 设计人工接管与反馈闭环：初期一定要有人工坐席监控，对不满意的回答进行纠正，并将这些纠正后的数据收集起来，作为后续优化模型或 Prompt 的宝贵材料。

6. 写在最后：关于未来的思考

基于 DeepSeek 搭建智能客服核心，让我们在短时间内就看到了效果提升。它强大的语言理解能力，让客服机器人终于听起来不那么“机械”了。

当然，这套系统还有很多可以深化的地方。比如，我们现在主要服务于中文用户。如果业务要拓展到海外，如何优雅地实现多语言支持？是接一个翻译 API 在输入输出时做中转，还是为不同语言训练/微调不同的模型？又或者，有没有可能利用 DeepSeek 本身的多语言能力，通过一个统一的模型来服务所有用户？这里面的成本、效果和架构复杂度该如何权衡？

每个团队的业务场景和技术栈不同，最优解也会不一样。希望我们这次的实践分享，能为你提供一个可行的起点和思路。如果你也在做类似的项目，欢迎一起交流探讨。