Project Overview

This tutorial will guide you through building a complete AI-powered customer service system with the following features:

  • Real-time conversation (streaming output)
  • Multi-turn conversation memory
  • Knowledge base retrieval augmentation (RAG)
  • Sentiment analysis and automatic transfer to human agent
  • Conversation history management

Technical Architecture

User Browser → Nginx → Flask/FastAPI → DeepSeek API
                    ↓
               ChromaDB (Knowledge Base Vector Store)
                    ↓
              SQLite/PostgreSQL (Conversation Records)

Step 1: Project Initialization

# Create project directory
mkdir smart-cs && cd smart-cs
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# Install dependencies
pip install flask openai chromadb python-dotenv

Step 2: DeepSeek Client Wrapper

# 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):
        """Streaming chat generator"""
        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

Step 3: RAG Knowledge Base

# 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 []

Step 4: Flask API Service

# 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()

# Store conversation history (use Redis in production)
sessions = {}

SYSTEM_PROMPT = """You are a professional and friendly intelligent customer service assistant.

Rules:
1. Reply in Chinese
2. Keep answers concise and clear (within 200 characters)
3. If unable to resolve, guide the user to contact human customer service
4. Refer to the provided knowledge base content when answering questions"""

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

    # Get conversation history
    if session_id not in sessions:
        sessions[session_id] = [{"role": "system", "content": SYSTEM_PROMPT}]

    messages = sessions[session_id].copy()

    # Knowledge base retrieval
    docs = kb.search(user_msg)
    if docs:
        context = '\n---\n'.join(docs)
        messages.append({
            "role": "system",
            "content": f"Relevant knowledge base content:\n{context}"
        })

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

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

        # Save conversation history
        sessions[session_id].append({"role": "user", "content": user_msg})
        sessions[session_id].append({"role": "assistant", "content": full_response})

        # Limit history length (keep last 20 turns)
        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)

Step 5: Frontend Interface

Step 6: Deployment

Recommended deployment with Gunicorn + Nginx:

# Start Gunicorn (4 workers)
gunicorn -w 4 -b 127.0.0.1:5000 app:app

# Nginx configuration
location / {
    proxy_pass http://127.0.0.1:5000;
    proxy_buffering off;  # Disable buffering to support SSE
    proxy_cache off;
}
location /chat {
    proxy_pass http://127.0.0.1:5000/chat;
    proxy_buffering off;
    proxy_read_timeout 300s;  # Long connection timeout
}

Cost Estimation

Based on 1000 conversations per day: each averaging 5 turns, 300 tokens output per turn, approximately ¥6/day (Flash cache hit), i.e., ¥180/month. Compared to a GPT-5 solution with similar functionality (approximately ¥1500/month), this saves 88%.