Python:LangChain 集成 DeepSeek 搭建智能客服系统

1、简述

随着大语言模型能力的成熟，智能客服已从「规则匹配」升级为「知识驱动 + 多轮理解」。 本文将基于：

• DeepSeek LLM（高性价比、中文能力强）
• LangChain（LLM 应用编排框架）
• RAG（检索增强生成）

搭建一个 企业级智能客服原型。

2、架构设计

.
├── enhanced_customer_service.py      # 增强版客服系统（集成所有核心能力）
├── knowledge_base_rag.py            # RAG 知识库模块（ChromaDB 版）
├── knowledge_db.py                 # PostgreSQL 知识库模块 🆕
├── agent_system.py                 # Agent 和工具调用系统
├── web_service.py                 # Web 服务接口（RESTful API）
├── start.py                      # 统一启动脚本 ⭐
├── init_db.py                    # 数据库初始化脚本 🆕
├── customer_service.py            # GLM 版命令行客服系统（旧版）
├── web_customer_service.py        # GLM 版 Web 客服系统（旧版）
├── deepseek_customer_service.py   # DeepSeek 版命令行客服系统（旧版）
├── deepseek_web_customer_service.py # DeepSeek 版 Web 客服系统（旧版）
├── requirements.txt               # Python 依赖
├── .env.example                 # 环境变量模板
├── .gitignore                  # Git 忽略文件
├── README.md                   # 项目说明文档
├── README_CORE_CAPABILITIES.md  # 核心能力详细文档
├── DEPLOYMENT_GUIDE.md        # 部署指南
└── templates/
    ├── index.html            # Web 界面模板（旧版）
    └── unified_with_kb.html  # 统一 Web 界面（带知识库管理）🆕

核心能力

✅ 多轮上下文对话 - 智能记住对话历史，保持上下文连贯性

✅ 企业知识库问答 - 基于 RAG 架构，从企业知识库中精准检索信息

✅ 可扩展 Agent/工具调用 - 灵活的 Agent 框架，支持 7 种内置工具和自定义工具扩展

✅ 私有化部署 - 完全自主可控，支持本地部署，数据安全不出企业网络

3、环境准备

下载开源项目智能客服：

git clone https://gitee.com/lhdxhl/intelligent.git
cd intelligent

3.1 安装依赖

# 1. 安装依赖
pip install -r requirements.txt

# 2. 配置环境变量
cp .env.example .env
# 编辑 .env 文件，填入 API 密钥和数据库连接字符串

# 3. 初始化数据库（可选）
python init_db.py

# 4. 启动系统
python start.py

# 5. 选择 1 - 启动 Web 服务
# 然后在浏览器中访问: http://localhost:5000/ui

★

DeepSeek API 兼容 OpenAI 接口协议，可直接复用 ChatOpenAI

3.2 API 配置

# 模型选择（选择其中一个）
AI_PROVIDER=deepseek          # 或 zhipuai

# GLM 模型配置
ZHIPUAI_API_KEY=
ZHIPUAI_MODEL=glm-4# 模型名称

# DeepSeek 模型配置
DEEPSEEK_API_KEY=
DEEPSEEK_MODEL=deepseek-chat  # 模型名称

# 通用配置
TEMPERATURE=0.7# 温度参数 (0-1)
MAX_TOKENS=2000# 最大生成 token 数

# Web 服务配置
FLASK_HOST=0.0.0.0
FLASK_PORT=5000
FLASK_DEBUG=False

# PostgreSQL 数据库配置 🆕
# 格式: postgresql://用户名:密码@主机:端口/数据库名
DATABASE_URL=postgresql://postgres:password@localhost:5432/intelligent_kb

四、模型设计

4.1 使用 DeepSeek 作为 LangChain LLM

# 获取API密钥
self.api_key = api_key or os.getenv("DEEPSEEK_API_KEY")
ifnot self.api_key:
raise ValueError("请设置 DEEPSEEK_API_KEY 环境变量或传入 api_key 参数")

# 获取配置
self.model = model or os.getenv("DEEPSEEK_MODEL", "deepseek-chat")
self.temperature = float(os.getenv("TEMPERATURE", "0.7"))
self.max_tokens = int(os.getenv("MAX_TOKENS", "2000"))

# 初始化 DeepSeek 客户端
# DeepSeek API 兼容 OpenAI API 格式
self.client = OpenAI(
    api_key=self.api_key,
    base_url="https://api.deepseek.com"
)

★

客服场景建议：

• temperature 低一点（更稳定）
• 回答偏「事实 + 引导」

4.2 客服 Prompt 设计

# 系统提示
    system_prompt = """你是一个专业的智能客服助手。你的职责是：

1. 友好、耐心地回答用户的问题
2. 提供准确、有用的信息
3. 如果遇到不知道的问题，诚实地说明并建议用户联系人工客服
4. 保持专业和礼貌的态度
5. 尽量用简洁明了的语言回答
6. 记住对话历史，保持上下文连贯性

请以专业、友好的态度为用户提供帮助。"""

# 添加系统消息
        self.history.append({
"role": "system",
"content": system_prompt
        })

★

Prompt 决定客服专业度，务必明确：

• 角色
• 回答边界
• 行为约束

5、构建知识库

5.1 准备企业知识文档

docs/
 ├── faq.txt
 ├── product.txt
 └── policy.txt

5.2 文档加载与切分

from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter

loader = TextLoader("docs/faq.txt", encoding="utf-8")
documents = loader.load()

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50
)

docs = splitter.split_documents(documents)

5.3 向量化并构建检索器

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

vectorstore = FAISS.from_documents(docs, embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

6、构建WEB API

"""
增强版智能客服系统的 Web 服务接口
提供 RESTful API 支持私有化部署
"""
from flask import Flask, request, jsonify, render_template
from flask_cors import CORS
from werkzeug.utils import secure_filename
import os
from datetime import datetime
from enhanced_customer_service import EnhancedCustomerServiceBot
from knowledge_db import get_knowledge_base

app = Flask(__name__)
CORS(app)  # 允许跨域请求

# 配置文件上传
UPLOAD_FOLDER = 'uploads'
ALLOWED_EXTENSIONS = {'txt', 'md', 'pdf', 'doc', 'docx'}
app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER
app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024# 16MB

# 确保上传目录存在
os.makedirs(UPLOAD_FOLDER, exist_ok=True)

# 初始化客服机器人（预加载模式）
bot = None
kb = None

defpreload_services():
"""预加载所有服务（应用启动时调用）"""
global bot, kb

    print("\n开始预加载服务...")

# 1. 初始化知识库（这一步最慢，因为需要加载嵌入模型）
try:
        print("- 正在初始化知识库（首次加载需要下载模型，可能需要几分钟）...")
        kb = get_knowledge_base()
        print("  ✓ 知识库已初始化")
except Exception as e:
        print(f"  ✗ 知识库初始化失败: {e}")
        print("  ⚠ 知识库功能将不可用")
        kb = None

# 2. 初始化客服机器人
try:
        print("- 正在初始化客服机器人...")
        provider = os.getenv("AI_PROVIDER", "deepseek")
        bot = EnhancedCustomerServiceBot(provider=provider)
        print(f"  ✓ 客服机器人已初始化 (使用 {provider.upper()} 模型)")
except Exception as e:
        print(f"  ✗ 客服机器人初始化失败: {e}")
        print("  ⚠ 聊天功能将不可用")
        bot = None

    print("\n✓ 服务预加载完成！")
    print()


definit_bot():
"""初始化客服机器人（懒加载模式）"""
global bot
if bot isNone:
        provider = os.getenv("AI_PROVIDER", "deepseek")
try:
            bot = EnhancedCustomerServiceBot(provider=provider)
            print(f"✓ 客服机器人已初始化 (使用 {provider.upper()} 模型)")
except Exception as e:
            print(f"✗ 初始化失败: {e}")
            bot = None
return bot


definit_knowledge_base():
"""初始化知识库（懒加载模式）"""
global kb
if kb isNone:
try:
            kb = get_knowledge_base()
            print("✓ 知识库已初始化")
except Exception as e:
            print(f"✗ 知识库初始化失败: {e}")
            kb = None
return kb


defallowed_file(filename):
"""检查文件类型是否允许"""
return'.'in filename and \
           filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS


defextract_text_from_file(filepath, filename):
"""从文件中提取文本"""
    ext = filename.rsplit('.', 1)[1].lower()

try:
if ext in ['txt', 'md']:
with open(filepath, 'r', encoding='utf-8') as f:
return f.read()
elif ext == 'pdf':
# PDF 文件处理（简化版，实际生产环境需要 pdfplumber 或 PyPDF2）
returnf"[PDF 文件内容: {filename}]"
elif ext in ['doc', 'docx']:
# Word 文件处理（简化版）
returnf"[Word 文件内容: {filename}]"
else:
returnf"不支持的文件类型: {ext}"
except Exception as e:
returnf"文件读取失败: {str(e)}"


@app.route('/chat', methods=['POST'])
defchat():
"""
    发送消息并获取回复

    请求体:
    {
        "message": "用户消息"
    }

    返回:
    {
        "response": "AI回复",
        "session_id": "会话ID"
    }
    """
try:
        data = request.json
ifnot data:
return jsonify({'error': '请提供请求体'}), 400

        user_input = data.get('message', '').strip()
ifnot user_input:
return jsonify({'error': '消息不能为空'}), 400

# 初始化机器人
if bot isNone:
            init_bot()

# 获取回复
        response = bot.chat(user_input)
        session_info = bot.get_session_info()

return jsonify({
'response': response,
'session_id': session_info['session_id'],
'provider': session_info['provider']
        })

except ValueError as e:
return jsonify({'error': str(e)}), 400
except Exception as e:
return jsonify({'error': f'服务器错误: {str(e)}'}), 500

@app.route('/info', methods=['GET'])
defget_info():
"""
    获取会话信息

    返回:
    {
        "session_id": "会话ID",
        "provider": "模型提供商",
        "model": "模型名称",
        "message_count": 消息数量,
        "available_tools": ["工具1", "工具2", ...],
        "knowledge_base_categories": ["分类1", "分类2", ...]
    }
    """
try:
if bot isNone:
return jsonify({
'message': '机器人尚未初始化',
'status': 'not_initialized'
            })

        info = bot.get_session_info()
return jsonify(info)

except Exception as e:
return jsonify({'error': f'服务器错误: {str(e)}'}), 500


@app.route('/knowledge/search', methods=['POST'])
defsearch_knowledge():
"""
    搜索知识库

    请求体:
    {
        "query": "搜索关键词",
        "top_k": 5,
        "category": "products"  # 可选
    }

    返回:
    {
        "query": "搜索关键词",
        "results": [
            {
                "id": 1,
                "title": "文档标题",
                "content": "文档内容",
                "similarity": 0.95
            },
            ...
        ],
        "count": 1
    }
    """
try:
# 初始化知识库
        init_knowledge_base()

        data = request.json
ifnot data:
return jsonify({'error': '请提供请求体'}), 400

        query = data.get('query', '').strip()
ifnot query:
return jsonify({'error': '搜索关键词不能为空'}), 400

        top_k = data.get('top_k', 5)
        category = data.get('category')

# 使用 PostgreSQL 知识库搜索
        results = kb.search(query, top_k=top_k, category=category)

return jsonify({
'query': query,
'results': results,
'count': len(results)
        })

except Exception as e:
return jsonify({'error': f'服务器错误: {str(e)}'}), 500


@app.route('/knowledge/documents', methods=['GET'])
defget_documents():
"""
    获取知识库文档列表

    参数:
        category: 分类筛选（可选）

    返回:
    {
        "documents": [...],
        "count": 10
    }
    """
try:
        init_knowledge_base()

        category = request.args.get('category')
        documents = kb.get_all_documents(category=category)

return jsonify({
'documents': documents,
'count': len(documents)
        })

except Exception as e:
return jsonify({'error': f'服务器错误: {str(e)}'}), 500


@app.route('/knowledge/documents/<int:doc_id>', methods=['GET'])
defget_document(doc_id):
"""
    获取单个文档

    返回:
    {
        "id": 1,
        "title": "文档标题",
        "content": "文档内容",
        ...
    }
    """
try:
        init_knowledge_base()

        doc = kb.get_document(doc_id)
ifnot doc:
return jsonify({'error': '文档不存在'}), 404

return jsonify(doc)

except Exception as e:
return jsonify({'error': f'服务器错误: {str(e)}'}), 500


@app.route('/knowledge/documents', methods=['POST'])
defadd_document():
"""
    添加文本到知识库

    请求体:
    {
        "title": "文档标题",
        "content": "文档内容",
        "category": "general"
    }

    返回:
    {
        "success": true,
        "document_id": 1,
        "message": "文档添加成功"
    }
    """
try:
        init_knowledge_base()

        data = request.json
ifnot data:
return jsonify({'error': '请提供请求体'}), 400

        title = data.get('title', '未命名文档')
        content = data.get('content', '')
        category = data.get('category', 'general')

ifnot content.strip():
return jsonify({'error': '文档内容不能为空'}), 400

# 添加到知识库
        doc_id = kb.add_document(
            title=title,
            content=content,
            category=category
        )

return jsonify({
'success': True,
'document_id': doc_id,
'message': '文档添加成功'
        })

except Exception as e:
return jsonify({'error': f'添加失败: {str(e)}'}), 500




@app.route('/', methods=['GET'])
defindex():
"""首页 - API 文档"""
    api_docs = {
'service': '智能客服系统 Web API',
'version': '2.0.0',
'endpoints': {
'GET /': 'API 文档',
'GET /ui': '统一 Web 界面（推荐）',
'GET /health': '健康检查',
'POST /chat': '发送消息并获取回复',
'POST /clear': '清空对话历史',
'GET /history': '获取对话历史',
'GET /info': '获取会话信息',
'GET /tools': '获取可用工具列表',
'POST /knowledge/search': '搜索知识库'
        },
'example': {
'chat': {
'endpoint': 'POST /chat',
'body': {
'message': '你好，我想了解一下你们的产品'
                }
            }
        },
'note': '所有 API 返回 JSON 格式数据',
'ui_url': '访问 Web 界面: http://localhost:5000/ui'
    }

return jsonify(api_docs)


@app.route('/ui', methods=['GET'])
defui():
"""统一 Web 界面（带知识库管理）"""
return render_template('unified_with_kb.html')


if __name__ == '__main__':
# 从环境变量获取配置
    host = os.getenv('FLASK_HOST', '0.0.0.0')
    port = int(os.getenv('FLASK_PORT', 5000))
    debug = os.getenv('FLASK_DEBUG', 'False').lower() == 'true'

    print("=" * 70)
    print("智能客服系统 Web 服务 v2.0.0")
    print("=" * 70)
    print(f"服务地址: http://{host}:{port}")
    print(f"调试模式: {debug}")
    print(f"API 文档: http://{host}:{port}/")
    print(f"Web 界面: http://{host}:{port}/ui")
    print("=" * 70)

# 预加载所有服务（加快首次访问速度）
    preload_services()

    print("=" * 70)
    print("✓ 服务已启动，等待请求...")
    print("=" * 70)
    print()

# 启动 Flask 应用
    app.run(host=host, port=port, debug=debug)

启动当前服务：

python start.py

7、总结

LangChain + DeepSeek 是目前性价比极高的智能客服组合方案：

• DeepSeek：强中文、低成本
• LangChain：工程化 + 可扩展
• RAG：安全、可控、可解释

非常适合：

• 企业内部客服
• SaaS 产品帮助中心
• 私有化 AI 助手