クイックスタート：5分でチャットエンドポイントを作る
まずは最速でプロンプトを受け取りChatGPTから回答を返すバックエンドを作ってみます。なぜもっと複雑にする必要があるのかは、その後で説明します。
ライブラリのインストール：
pip install flask openai python-dotenv flask-cors.env ファイルを作成：
OPENAI_API_KEY=sk-proj-xxxxxx最小構成の app.py：
from flask import Flask, request, jsonify
from flask_cors import CORS
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()
app = Flask(__name__)
CORS(app)
client = OpenAI()
@app.route("/chat", methods=["POST"])
def chat():
user_message = request.json.get("message", "")
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": user_message}]
)
return jsonify({"reply": response.choices[0].message.content})
if __name__ == "__main__":
app.run(debug=True)curlで動作確認：
curl -X POST http://localhost:5000/chat \
-H "Content-Type: application/json" \
-d '{"message": "Docker とは何かを説明します。"}'これで動きます。ただし、この状態で本番環境に上げると、少なくとも3つの深刻な問題があります。
実際に解決すべき3つの問題
問題1：レスポンスが遅い — UXが悪い
ChatGPTは長い回答を生成するのに通常3〜10秒かかります。その間ユーザーは真っ白な画面を見続けることになり、アプリがフリーズしているように感じます。ストリーミングはこの問題を解決します。モデルが生成するたびにテキストのチャンクを受け取って順次表示することで、ChatGPT.comのUIと同じような体験を実現できます。
問題2：チャットボットがコンテキストを記憶しない
OpenAI APIは完全にステートレスです。各リクエストは独立しており、前のリクエストとは無関係です。チャットボットに会話を「覚えさせる」には、APIを呼び出すたびに会話履歴全体を送信する必要があります。
問題3：フロントエンドから直接呼び出すとAPIキーが漏洩する
ブラウザのJavaScriptからOpenAIを直接呼び出すと、DevToolsを開けば誰でもAPIキーを見ることができます。すべてのリクエストはバックエンド経由にする — これはAI APIを使う際の絶対ルール第1条です。
完全な解決策：ストリーミング + 会話履歴
バックエンドには Server-Sent Events（SSE） を使います。これはサーバーがレスポンスの完了を待たずに、テキストを少しずつブラウザにプッシュできる仕組みです。この方式を本番環境で数ヶ月運用していますが、モデルの生成時間は変わらなくても、ユーザーは5〜8秒間真っ白な画面を見る代わりにテキストが徐々に表示されるのを体験できます。
ストリーミング対応のFlaskバックエンド
from flask import Flask, request, jsonify, Response, stream_with_context
from flask_cors import CORS
from openai import OpenAI
from dotenv import load_dotenv
import json
load_dotenv()
app = Flask(__name__)
CORS(app)
client = OpenAI()
# 会話履歴をメモリに保存
# 本番環境ではRedisまたはデータベースを使用すること
conversations = {}
SYSTEM_PROMPT = """あなたはLinuxとDevOpsを専門とするITエキスパートです。
ベトナム語で簡潔にご返信ください。必要に応じてコード例もご提示ください。
不明な点がある場合は、ごまかさずに直接ご説明ください。"""
MAX_HISTORY = 20 # 最大20件のメッセージを保持
def trim_history(history):
system_msg = [m for m in history if m["role"] == "system"]
other_msgs = [m for m in history if m["role"] != "system"]
if len(other_msgs) > MAX_HISTORY:
other_msgs = other_msgs[-MAX_HISTORY:]
return system_msg + other_msgs
@app.route("/chat/stream", methods=["POST"])
def chat_stream():
data = request.json
session_id = data.get("session_id", "default")
user_message = data.get("message", "")
if session_id not in conversations:
conversations[session_id] = [{"role": "system", "content": SYSTEM_PROMPT}]
conversations[session_id].append({"role": "user", "content": user_message})
conversations[session_id] = trim_history(conversations[session_id])
def generate():
full_response = ""
stream = client.chat.completions.create(
model="gpt-4o-mini",
messages=conversations[session_id],
stream=True,
max_tokens=1000
)
for chunk in stream:
if chunk.choices[0].delta.content:
text = chunk.choices[0].delta.content
full_response += text
yield f"data: {json.dumps({'text': text})}\n\n"
conversations[session_id].append({
"role": "assistant",
"content": full_response
})
yield f"data: {json.dumps({'done': True})}\n\n"
return Response(
stream_with_context(generate()),
mimetype="text/event-stream",
headers={
"Cache-Control": "no-cache",
"X-Accel-Buffering": "no" # Nginxを使う場合は必須
}
)
@app.route("/chat/reset", methods=["POST"])
def reset():
session_id = request.json.get("session_id", "default")
conversations.pop(session_id, None)
return jsonify({"status": "ok"})フロントエンドJavaScript：ストリームを受け取ってリアルタイム表示する
ReadableStream を使ったFetch API — 追加ライブラリは不要です：
const sessionId = Math.random().toString(36).substr(2, 9);
async function sendMessage(message) {
const assistantDiv = createMessageDiv("assistant", "");
const response = await fetch("/chat/stream", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ session_id: sessionId, message })
});
const reader = response.body.getReader();
const decoder = new TextDecoder();
while (true) {
const { done, value } = await reader.read();
if (done) break;
const lines = decoder.decode(value).split("\n");
for (const line of lines) {
if (!line.startsWith("data: ")) continue;
try {
const data = JSON.parse(line.slice(6));
if (data.text) {
assistantDiv.textContent += data.text;
}
} catch (e) {}
}
}
}応用：レート制限とコスト管理
身をもって学んだ教訓：レート制限がないと、ボットや制御不能なユーザーによって一晩でOpenAIの請求が$30〜50に跳ね上がることがあります。flask-limiter を追加しましょう：
pip install flask-limiterfrom flask_limiter import Limiter
from flask_limiter.util import get_remote_address
import openai
limiter = Limiter(
app=app,
key_func=get_remote_address,
default_limits=["200 per day", "50 per hour"]
)
@app.route("/chat/stream", methods=["POST"])
@limiter.limit("10 per minute")
def chat_stream():
try:
# ... 上記と同じコード
pass
except openai.RateLimitError:
return jsonify({"error": "システムが混雑しています。30秒後に再試行してください"}), 429
except openai.APIConnectionError:
return jsonify({"error": "OpenAI APIに接続できません"}), 503
except openai.AuthenticationError:
return jsonify({"error": "APIキーが無効です"}), 401本番デプロイの実践Tips
SSEにはNginxの特別な設定が必要
Nginxはデフォルトでレスポンス全体をバッファリングしてからクライアントに送ります。SSEではこれが原因でストリームが「止まって」しまい、テキストがリクエスト終了時に一度にまとめて表示されるという事態になります。ストリーミングの意味がまったくなくなってしまいます。バッファリングを無効にしてください：
location /chat/ {
proxy_pass http://127.0.0.1:5000;
proxy_buffering off; # SSEには必須
proxy_cache off;
proxy_read_timeout 300s; # 長いレスポンスに十分な時間を確保
add_header X-Accel-Buffering no;
}ユースケースに適したモデルを選ぶ
- gpt-4o-mini — 高速かつ低コスト（gpt-4oの約30分の1）、一般的なサポートチャットボットに十分
- gpt-4o — 複雑な分析や高品質なコード生成が必要な場合に使用
- gpt-4o with structured outputs — 固定スキーマのJSONレスポンスが必要で、フォーマットのハルシネーションを防ぎたい場合
システムプロンプトが品質を決める
システムプロンプトを空にするのは最もよくある失敗です。役割、返答スタイル、トピックの範囲を明確に書いておくと、チャットボットの「でたらめ」が減り、トーンも格段に安定します。チャットボットが話題を外れたり口調がおかしくなったりしたとき、真っ先に確認するのはいつもシステムプロンプトです。
会話履歴を無限に保持しない
会話履歴が長くなる = トークン数が増える = リクエストごとのコストが上がる。上記の trim_history() では直近20件のメッセージを保持しています。内容にもよりますが10,000〜15,000トークン程度に相当し、コンテキストを維持しながらも予算を食い潰しません。ユーザー名や設定情報など長期的に記憶すべき情報がある場合は、履歴をそのまま保持するのではなく、要約してシステムプロンプトに注入する方法をとりましょう。
