本チュートリアルでは、FastAPI、SSEストリーミング、会話メモリ、モデル切り替え機能を備えた、小規模ながらもプロダクション環境に対応したチャットボットサービスを構築します。これは単なるデモ用のおもちゃではありません。実際の製品のバックエンドとして組み込み、モデルを変更するたびに統合コードを書き直すことなく、安全にイテレーションを回せる設計を目指します。
すでにOpenAI互換のSDKをTokenLabに向けている場合は、この記事の続きから進めてください。まだベースURLの切り替えを行っていない場合は、先に移行ガイドをお読みください。負荷時のリクエスト制御やバックオフが主な懸念事項である場合は、本ガイドと合わせてAI APIレート制限ガイドを参照してください。
重要なポイント
- プロダクション対応のチャットボットには、同期エンドポイント、ストリーミングエンドポイント、サーバーサイドの会話状態管理、モデルの許可リスト、適切なエラーハンドリング、そしてストレージのアップグレードパスという6つの要素が必要です。
- ストリーミングやメモリ、ツール呼び出しを実装する前に、まずは1つの小さなチャットエンドポイントで、APIキー、ベースURL、ルーティングが正しく機能することを証明してください。
- SSEストリーミングはほとんどのチャット製品をカバーでき、Websocketよりも運用負荷が低くなります。
- モデルは自由入力フィールドではなく、バックエンドの許可リストを通じて公開してください。これにより、フロントエンドから任意のモデルIDや廃止されたモデルIDを要求されることを防げます。
- モデルの可用性やラインナップは頻繁に変更されます。許可リストをプロダクション環境に固定する前に、TokenLabのモデルディレクトリ（2026年7月7日時点）を確認してください。
構築するもの
完成するサービスには、以下の6つの可動パーツが含まれます：
- スモークテスト用の同期型
/chatエンドポイント。 - 実際のUI用のストリーミング型
/chat/streamエンドポイント。 conversation_idで管理される会話状態。- フロントエンドが勝手なIDを要求できないようにするためのモデル許可リスト。
- 429エラーが発生しても即座に崩壊しないエラーハンドリング。
- インメモリのプロトタイプからRedisやPostgreSQLへ移行するための明確なパス。
これだけで、サポートボット、社内アシスタント、あるいは組み込みチャットウィジェットの最初のバージョンを動かすには十分です。
最小限のスタックをインストールする
pip install fastapi uvicorn openai pydantic redis
最初の段階では redis をスキップしても構いませんが、今のうちにインポートを組み込んでおけば、後のアップグレードがリファクタリングではなく単なる設定変更で済むようになります。
ステップ1：小さく退屈なチャットエンドポイントから始める
チャットボット開発で迷子になる一番の近道は、基本的なリクエストパスが安定する前にWebsocketやツール利用、エージェントオーケストレーションから始めることです。まずは、キー、ベースURL、モデルルーティングが機能することを証明する小さなエンドポイントから始めましょう。
from fastapi import FastAPI
from openai import OpenAI
from pydantic import BaseModel
app = FastAPI()
client = OpenAI(
api_key="sk-tokenlab-xxx",
base_url="https://api.tokenlab.sh/v1"
)
class ChatRequest(BaseModel):
message: str
model: str = "deepseek-v4-flash"
conversation_id: str | None = None
@app.post("/chat")
async def chat(req: ChatRequest):
response = client.chat.completions.create(
model=req.model,
messages=[{"role": "user", "content": req.message}]
)
return {"reply": response.choices[0].message.content}
スモークテストを1回実行してください。これが失敗する場合、その上に構築を続けてはいけません。
ステップ2：ストリーミングを追加する（ユーザーは測定する前に遅延を感じるため）
多くのチャットボット製品が遅く感じられるのは、モデルの速度ではなく、完全な応答が届くまでUIが空白のままだからです。SSEはほとんどのチャット製品にとって十分であり、Websocketよりも運用負荷が低くなります。
from fastapi.responses import StreamingResponse
@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
def generate():
stream = client.chat.completions.create(
model=req.model,
messages=[{"role": "user", "content": req.message}],
stream=True
)
for chunk in stream:
delta = chunk.choices[0].delta
if delta.content:
yield f"data: {delta.content}\n\n"
yield "data: [DONE]\n\n"
return StreamingResponse(generate(), media_type="text/event-stream")
フロントエンドでは、標準的なfetchリーダーで十分です：
async function sendMessage(payload) {
const response = await fetch('/chat/stream', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(payload),
});
const reader = response.body.getReader();
const decoder = new TextDecoder();
while (true) {
const { done, value } = await reader.read();
if (done) break;
const chunk = decoder.decode(value, { stream: true });
console.log(chunk);
}
}
製品がすでに標準的なHTTP上でブラウザクライアントを実行している場合、SSEを使えばWebsocket層よりもアーキテクチャをシンプルに保てます。
ステップ3：会話状態をリクエストボディから分離する
最初のチャットボットデモでは、完全なトランスクリプトをブラウザに保持し、ターンごとに再送信することがよくあります。これはプロトタイプには適していますが、リトライやセッションの再開、サーバーサイドのツール利用が必要になるとすぐに複雑化します。
最初はインメモリのストアで十分です：
from collections import defaultdict
import uuid
conversations: dict[str, list] = defaultdict(list)
SYSTEM_PROMPT = "You are a helpful assistant. Be concise and direct."
def build_messages(conv_id: str, user_msg: str) -> list:
messages = [{"role": "system", "content": SYSTEM_PROMPT}]
history = conversations[conv_id][-20:]
messages.extend(history)
messages.append({"role": "user", "content": user_msg})
conversations[conv_id].append({"role": "user", "content": user_msg})
return messages
Redisへのアップグレードパスは、ロジックの変更ではなく、主にストレージの配線作業です：
import json
import redis
redis_client = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)
def load_history(conv_id: str) -> list:
raw = redis_client.get(f"chat:{conv_id}")
return json.loads(raw) if raw else []
def save_history(conv_id: str, history: list) -> None:
redis_client.setex(f"chat:{conv_id}", 60 * 60 * 24, json.dumps(history))
会話にTTL（有効期限）、再開可能性、マルチインスタンスデプロイが必要になったらRedisを検討してください。トランスクリプト自体が後でクエリ、監査、レポートを行う必要がある製品データである場合は、PostgreSQLを検討してください。
ステップ4：エラーを例外ではなく製品の動作として扱う
顧客向けのチャットボットでは、成功ルートと同じくらい失敗ルートが重要です。ユーザーは、失敗の原因がレート制限、残高不足、あるいは上流のモデル障害であるかを気にしません。彼らが気にするのは、UIがフリーズするかどうかです。
from openai import APIConnectionError, APIError, RateLimitError
@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
conv_id = req.conversation_id or str(uuid.uuid4())
messages = build_messages(conv_id, req.message)
def generate():
full_response = []
try:
stream = client.chat.completions.create(
model=req.model,
messages=messages,
stream=True
)
for chunk in stream:
delta = chunk.choices[0].delta
if delta.content:
full_response.append(delta.content)
yield f"data: {delta.content}\n\n"
except RateLimitError:
yield "data: [error: rate limited, please retry shortly]\n\n"
except APIConnectionError:
yield "data: [error: connection issue, please retry]\n\n"
except APIError:
yield "data: [error: something went wrong on our end]\n\n"
finally:
if full_response:
conversations[conv_id].append(
{"role": "assistant", "content": "".join(full_response)}
)
yield "data: [DONE]\n\n"
return StreamingResponse(generate(), media_type="text/event-stream")
ステップ5：フロントエンドが要求できるモデルを制限する
フロントエンドから任意のモデル文字列を直接APIに渡させてはいけません。自由入力フィールドは、廃止されたモデルやタイポ、あるいは公開するつもりのなかったモデルへのリクエストを招きます。代わりに、バックエンドの許可リストを経由させてください。
ALLOWED_MODELS = {
"default": "deepseek-v4-flash",
"flagship": "gpt-5.5",
"balanced": "claude-sonnet-5",
}
class ChatRequest(BaseModel):
message: str
tier: str = "default"
conversation_id: str | None = None
@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
model = ALLOWED_MODELS.get(req.tier, ALLOWED_MODELS["default"])
# ストリーミングロジックの残りは、クライアントが提供した生の文字列ではなく `model` を使用する
これにより、プロバイダーがモデルを廃止した際にも、フロントエンドのコードを触ったりクライアントを更新したりすることなく、一箇所でモデルを切り替えることができます。
ステップ6：ハッピーパスだけでなく、プロダクションの残りの部分を処理する
チャットボットのバックエンドは、コアとなるチャット呼び出しが賢いときではなく、周囲のエッジケースが処理されたときに初めてプロダクション対応と見なされます。
チェックリストは以下の通りです：
- フロントエンドの失敗をバックエンドのログと紐付けられるよう、リクエストIDを追加する
- ユーザーごとの同時実行数とリクエストサイズに上限を設ける
- トークン予算を使い切る前に、長い履歴をトリミングする
- モデル、レイテンシ、入力サイズ、終了理由をログに記録する
- ユーザーに見せるエラーメッセージと内部のエラー詳細を分離する
- 最初の実際の障害が発生する前に、代替モデルをテストしてフォールバックが機能することを確認する
履歴のトリミングはシンプルに保てます：
def trim_history(messages: list, max_tokens: int = 8000) -> list:
system = messages[0]
history = messages[1:]
total_chars = len(system["content"])
trimmed = []
for msg in reversed(history):
msg_chars = len(msg["content"])
if total_chars + msg_chars > max_tokens * 4:
break
trimmed.insert(0, msg)
total_chars += msg_chars
return [system] + trimmed
目的はトークンを完璧に計算することではなく、請求額やレイテンシに影響が出る前に、明らかなコンテキストの肥大化を止めることです。
デモから製品へ
このバックエンドが安定したら、次のアップグレードは「AIの強化」ではなく、退屈なインフラストラクチャであることがほとんどです：
- あるユーザーが他のユーザーの会話を読めないようにするための認証
- デプロイ後もセッションが維持されるための永続化
- 騒がしいユーザーがクォータを使い果たさないためのレート制限
- チャットボットが顧客向けの場合の課金や使用量の紐付け
- 会話に長期記憶が必要な場合のバックグラウンドでの要約
統合ゲートウェイは、これらのほとんどに役立ちます。ベースURLの移行さえ済ませてしまえば、モデルの変更はプラットフォームの書き直しではなく、設定の編集になります。
スモークテスト
uvicorn main:app --reload --port 8000
curl -N -X POST http://localhost:8000/chat/stream \
-H "Content-Type: application/json" \
-d '{"message": "Hello!", "model": "deepseek-v4-flash"}'
1ターンのストリーミング、1つの会話の保持、そして強制的な失敗に対するクリーンなエラー返却ができれば、正しい基盤ができています。
コスト見積もり
TokenLabでAPIキーを作成し、OpenAI SDKを https://api.tokenlab.sh/v1 に向けるだけで、プロバイダーごとに別々のアカウントを管理することなく、チャットボットの最初のプロダクションバージョンをリリースできます。
| モデル | 一般的なティア | 備考 |
|---|---|---|
| DeepSeek V4 Flash | 高速 / デフォルト | 高頻度・低レイテンシのターンに適したデフォルト |
| GPT-5.5 | フラッグシップ | より強力な推論が必要なターンに使用 |
| Claude Sonnet 5 | バランス型 | コーディングやレビュー形式の返信に最適 |
| Gemini 3.5 Flash | 予算 / 高速代替 | 高頻度ルーティングのための低コスト・高速な代替 |
正確なトークン単価はプロバイダー間で頻繁に変更されるため、ここでは固定値として記載していません。予算を立てる前に、モデルディレクトリ（2026年7月7日時点）で現在の料金を確認してください。実際には、ほとんどの会話をDeepSeek V4 FlashやGemini 3.5 Flashのような高速/デフォルトティアにルーティングし、GPT-5.5やClaude Sonnet 5を必要なターンに限定することで、ほとんどのアプリケーションの月額料金を低く抑えることができます。ただし、予算を確定する前に、アカウントの実際の100万トークンあたりの料金を確認してください。
FAQ
AIチャットボットを構築するのにWebsocketは必要ですか？ いいえ。ステップ2で示したSSEストリーミングで、ほとんどのチャット製品をカバーできます。Websocketは、ライブコラボレーションやサーバー主導のイベントなど、リクエスト/レスポンス以外の双方向プッシュが必要な場合に真価を発揮します。標準的なチャットUIであれば、SSEの方がデプロイ、デバッグ、スケーリングが容易です。
どのモデルをデフォルトにすべきかどうやって判断しますか？ デフォルトティアにはDeepSeek V4 FlashやGemini 3.5 Flashのような高速で低コストなモデルから始め、ステップ5で示した許可リストの背後にClaude Sonnet 5やGPT-5.5によるバランス型または推論型ティアを追加してください。新しいモデルのリリースや古いモデルの廃止は制御できないスケジュールで行われるため、最新のオプションについてはモデルディレクトリ（2026年7月7日時点）を確認してください。
チャットボットがデモから実際のトラフィックに移行する際、最初に壊れるのは何ですか？ ほぼ間違いなく、ハッピーパスではなくエラーパスです。無制限のリトライ、ユーザーごとの同時実行制限の欠如、無制限の会話履歴は、実際の負荷の下でチャットボットのバックエンドがダウンする最も一般的な3つの原因です。上記のステップ4と6は、これら3つすべてに直接対処しています。
APIキーで始めましょう：tokenlab.shは1つのエンドポイントを通じて300以上のモデルを提供します。構築開始用に1ドルの無料クレジットが付与されます。
出典
価格確認日 2026-07-07
- TokenLab model directory2026-07-07 時点で確認
