用 Dify 和 DeepSeek-V3 搭建智能客服：从需求到上线的完整流程

2026-09-16 13:19 1 阅读

需求从哪儿来？先看企业真痛点

做智能客服不能只盯着技术炫酷，得先搞清楚企业到底被什么问题卡住。我们调研过几家不同行业的公司，发现高频咨询压垮人力是普遍现象。

比如一家 3C 电商，在大促期间单日咨询量能冲到 15 万条以上，其中七成都是“能不能用券”“什么时候发货”这类重复问题。人工客服疲于应付，客户排队等半天，最后可能直接关页面走人——他们内部测算，这种流失大概占潜在订单的 12%。

不同行业的需求重点也不一样：

电商最关心响应速度和活动规则解读，直接影响下单转化；
金融机构需要准确解释产品条款，同时严防合规风险；
教育机构则希望根据学生情况推荐课程，把通用回答变成个性化建议。

基于这些实际场景，我们给智能客服定了四个硬指标：能听懂各种领域的问题、回答要准、对话要连贯、还能自己学新东西。

架构怎么搭？四层结构各司其职

系统不是堆功能，得有清晰的层次。我们用 Dify 平台配合 DeepSeek-V3 模型，搭了一套四层架构：

数据层：答案从哪儿来

所有智能都依赖高质量数据。我们用 Elasticsearch 存产品知识库，5000 多条商品信息能做到毫秒级检索。另外单独建了个用户反馈库，专门记下哪些回答让用户不满意——后来发现很多人嫌答案太简略，就据此调整了生成策略。

模型层：脑子装在哪儿

模型跑在华为云 Flexus 上，选 DeepSeek-V3 是因为它的推理速度和准确率在客服场景表现不错。针对金融客户，我们额外喂了 10 万条金融术语和问答做微调，让模型快速适应专业语境。

接口层：怎么跟外面打交道

对外提供两套接口：一套是 APP SDK，带完整文档和示例，合作方三天就能接通；另一套是标准 API，用来和 CRM、订单系统打通，这样客服能直接查用户订单状态，不用让用户反复报单号。

应用层：具体干哪些活

这里包含几个关键模块：

在这里插入图片描述

意图识别：用规则+BERT 混合判断用户想问什么。比如“买二送一是不是拍两件自动减价”，光靠关键词匹配会漏，加了语义模板后准确率提到 93%以上；
对话状态跟踪（DST）：记住上下文。用户连续问“支持无线充电吗？充多久？能玩多久？”，系统知道这三个问题都关于同一款手机；
数据分析：用 Tableau 把咨询热点、解决率、满意度做成看板，方便运营团队快速发现问题。

在这里插入图片描述

代码怎么写？关键部分拆解

在这里插入图片描述

光说不练假把式，下面看核心代码怎么实现。

在这里插入图片描述

环境准备

在这里插入图片描述

先在华为云注册账号，完成实名认证。Dify 的推理服务按流量计费，建议账户里至少充 10 元避免中断。创建实例时，中小型企业选 4 核 8GB 内存、500GB 存储基本够用，安全组端口按最小权限开。

在这里插入图片描述

智能客服主类

import re
import jieba
import json
from typing import Dict, List
import requests

class SmartCustomerService:
    def __init__(self, knowledge_base_path: str):
        with open(knowledge_base_path, 'r', encoding='utf-8') as f:
            self.knowledge_base = json.load(f)
        self.sensitive_words = self._load_sensitive_words()

def _load_sensitive_words(self) -> List[str]:
        with open("sensitive_words.txt", 'r', encoding='utf-8') as f:
            return [line.strip() for line in f.readlines()]

def preprocess_query(self, query: str) -> str:
        # 清洗特殊字符
        query = re.sub(r'[^\u4e00-\u9fa5a-zA-Z0-9\s]', '', query)
        # 中文分词
        words = jieba.lcut(query)
        # 过滤敏感词
        processed_words = [word for word in words if word not in self.sensitive_words]
        return ' '.join(processed_words)

def postprocess_answer(self, answer: str) -> str:
        # 标点规范化
        answer = re.sub(r'([。！？!?；;])', r'\1\n', answer)
        # 敏感信息脱敏
        answer = re.sub(r'(\d{3})\d{4}(\d{4})', r'\1****\2', answer)  # 手机号
        answer = re.sub(r'(\d{6})\d{8}(\d{4})', r'\1********\2', answer)  # 身份证
        if not answer.endswith(('.', '。')):
            answer += '。'
        return answer.strip()

def process_query(self, query: str) -> str:
        processed_query = self.preprocess_query(query)
        prompt = f"根据以下知识库内容，回答用户的问题：{json.dumps(self.knowledge_base, ensure_ascii=False)}\n用户的问题是：{processed_query}"
        try:
            response = dify_client.generate(
                prompt=prompt, 
                model="deepseek-v3", 
                temperature=0.7, 
                max_tokens=200
            )
            return self.postprocess_answer(response['answer'])
        except requests.exceptions.ConnectionError:
            return "网络连接失败，请检查网络后重试。"
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 401:
                return "API密钥认证失败，请重新获取密钥。"
            return f"请求出错，错误码：{e.response.status_code}，请稍后再试。"
        except Exception as e:
            print(f"调用Dify平台API出错: {e}")
            return "很抱歉，暂时无法回答您的问题，请稍后再试。"

这段代码做了三件事：

预处理：去掉乱码、分词、过滤敏感词，让模型更容易理解用户意图；
调用模型：把清洗后的问题和知识库一起塞给 DeepSeek-V3，控制温度值 0.7 平衡创造力和稳定性；
后处理：规范标点、脱敏隐私信息，确保输出合规。

对话管理

多轮对话需要记住历史，但也不能无限记下去，否则提示词太长影响性能。

class ConversationManager:
    def __init__(self, conversation_history_path: str):
        self.conversation_history_path = conversation_history_path
        try:
            with open(self.conversation_history_path, 'r', encoding='utf-8') as f:
                self.conversation_history = json.load(f)
        except FileNotFoundError:
            self.conversation_history = {}

def add_conversation(self, user_id: str, query: str, answer: str):
        if user_id not in self.conversation_history:
            self.conversation_history[user_id] = []
        self.conversation_history[user_id].append({"query": query, "answer": answer})
        self._save_conversation_history()

def truncate_conversation_history(self, user_id: str, max_length: int = 10):
        if user_id in self.conversation_history:
            # 只保留最近10轮对话
            self.conversation_history[user_id] = self.conversation_history[user_id][-max_length:]
            self._save_conversation_history()

每次对话后存记录，但超过 10 轮就截断，既保证上下文连贯，又控制成本。

流式响应优化

用户不想干等，边生成边显示体验更好。Dify 支持流式输出：

def process_query_stream(self, query: str) -> str:
    processed_query = self.preprocess_query(query)
    prompt = f"根据以下知识库内容，回答用户的问题：{json.dumps(self.knowledge_base, ensure_ascii=False)}\n用户的问题是：{processed_query}"
    try:
        answer = ""
        for part in dify_client.generate_stream(
            prompt=prompt, 
            model="deepseek-v3", 
            temperature=0.7
        ):
            answer += part
            # 这里可以实时推送给前端
        return self.postprocess_answer(answer)
    except Exception as e:
        print(f"流式响应出错: {e}")
        return "很抱歉，暂时无法回答您的问题，请稍后再试。"