DeepSeek API 从零到上线：完整项目实战

项目概述

本教程将带你构建一个完整的 AI 智能客服系统，具备以下功能：

• 实时对话（流式输出）
• 多轮对话记忆
• 知识库检索增强（RAG）
• 情感分析与自动转人工
• 对话历史管理

技术架构

用户浏览器 → Nginx → Flask/FastAPI → DeepSeek API
                    ↓
               ChromaDB（知识库向量存储）
                    ↓
              SQLite/PostgreSQL（对话记录）

第一步：项目初始化

# 创建项目目录
mkdir smart-cs && cd smart-cs
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# 安装依赖
pip install flask openai chromadb python-dotenv

第二步：DeepSeek 客户端封装

# deepseek_client.py
from openai import OpenAI
import os
from dotenv import load_dotenv

load_dotenv()

class DeepSeekClient:
    def __init__(self):
        self.client = OpenAI(
            api_key=os.getenv('DEEPSEEK_API_KEY'),
            base_url='https://api.deepseek.com'
        )
        self.model = 'deepseek-v4-flash'

    def chat(self, messages, stream=False):
        return self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            stream=stream
        )

    def chat_stream(self, messages):
        """流式聊天生成器"""
        stream = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            stream=True
        )
        for chunk in stream:
            if chunk.choices[0].delta.content:
                yield chunk.choices[0].delta.content

第三步：RAG 知识库

# knowledge_base.py
import chromadb
from chromadb.utils import embedding_functions

class KnowledgeBase:
    def __init__(self):
        self.client = chromadb.PersistentClient(path="./chroma_db")
        self.ef = embedding_functions.OpenAIEmbeddingFunction(
            api_key=os.getenv('DEEPSEEK_API_KEY'),
            api_base='https://api.deepseek.com',
            model_name='text-embedding-3-small'
        )
        self.collection = self.client.get_or_create_collection(
            name="faq",
            embedding_function=self.ef
        )

    def add_documents(self, texts, metadatas=None):
        ids = [f"doc_{i}" for i in range(len(texts))]
        self.collection.add(documents=texts, ids=ids, metadatas=metadatas)

    def search(self, query, k=3):
        results = self.collection.query(query_texts=[query], n_results=k)
        return results['documents'][0] if results['documents'] else []

第四步：Flask API 服务

# app.py
from flask import Flask, request, Response, jsonify
from deepseek_client import DeepSeekClient
from knowledge_base import KnowledgeBase
import json

app = Flask(__name__)
ds = DeepSeekClient()
kb = KnowledgeBase()

# 存储对话历史（生产环境应使用 Redis）
sessions = {}

SYSTEM_PROMPT = """你是一个专业友好的智能客服助手。

规则：
1. 使用中文回复
2. 回答简短清晰（200字内）
3. 如无法解决，引导用户联系人工客服
4. 参考提供的知识库内容回答问题"""

@app.route('/chat', methods=['POST'])
def chat():
    data = request.json
    session_id = data.get('session_id', 'default')
    user_msg = data.get('message', '')

    # 获取对话历史
    if session_id not in sessions:
        sessions[session_id] = [{"role": "system", "content": SYSTEM_PROMPT}]

    messages = sessions[session_id].copy()

    # 知识库检索
    docs = kb.search(user_msg)
    if docs:
        context = '\n---\n'.join(docs)
        messages.append({
            "role": "system",
            "content": f"相关知识库内容：\n{context}"
        })

    messages.append({"role": "user", "content": user_msg})

    # 流式返回
    def generate():
        full_response = ''
        for token in ds.chat_stream(messages):
            full_response += token
            yield f"data: {json.dumps({'token': token})}\n\n"

        # 保存对话历史
        sessions[session_id].append({"role": "user", "content": user_msg})
        sessions[session_id].append({"role": "assistant", "content": full_response})

        # 限制历史长度（保留最近 20 轮）
        if len(sessions[session_id]) > 42:
            sessions[session_id] = [sessions[session_id][0]] + sessions[session_id][-40:]

        yield "data: [DONE]\n\n"

    return Response(generate(), mimetype='text/event-stream')

if __name__ == '__main__':
    app.run(debug=True, port=5000)

第五步：前端界面

发送

第六步：部署上线

推荐使用 Gunicorn + Nginx 部署：

# 启动 Gunicorn（4 workers）
gunicorn -w 4 -b 127.0.0.1:5000 app:app

# Nginx 配置
location / {
    proxy_pass http://127.0.0.1:5000;
    proxy_buffering off;  # 关闭缓冲以支持 SSE
    proxy_cache off;
}
location /chat {
    proxy_pass http://127.0.0.1:5000/chat;
    proxy_buffering off;
    proxy_read_timeout 300s;  # 长连接超时
}

成本估算

按每天 1000 次对话计算：每次平均 5 轮，每轮 300 tokens 输出，约 ¥6/天（Flash 缓存命中），即 ¥180/月。对比同功能 GPT-5 方案（约 ¥1500/月），节省 88%。