DeepSeek APIゼロから本番まで：完全プロジェクト実践

プロジェクト概要

このチュートリアルでは、以下の機能を備えた完全なAIカスタマーサービスシステムを構築します。

• リアルタイム対話（ストリーミング出力）
• マルチターン対話メモリ
• 知識ベース検索拡張（RAG）
• 感情分析と自動オペレーター転送
• 対話履歴管理

技術アーキテクチャ

ユーザーブラウザ → Nginx → Flask/FastAPI → DeepSeek API
                    ↓
               ChromaDB（知識ベースベクトルストア）
                    ↓
              SQLite/PostgreSQL（対話記録）

ステップ1：プロジェクト初期化

# プロジェクトディレクトリの作成
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

ステップ2：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

ステップ3：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 []

ステップ4：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)

ステップ5：フロントエンドインターフェース

送信

ステップ6：デプロイ

Gunicorn + Nginxでのデプロイを推奨します：

# Gunicornの起動（4ワーカー）
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;  # 長時間接続のタイムアウト
}

コスト見積もり

1日1000回の対話を想定：各対話平均5ターン、各ターン300トークン出力、約¥6/日（Flashキャッシュヒット）、つまり月額¥180。同機能のGPT-5ソリューション（約¥1500/月）と比較して、88%のコスト削減になります。