1つのAPIキーでAIチャットボットを構築する：ゼロから本番環境へ30分で到達する方法

AIチャットボットが1つの返信をストリーミングする仕組みは、午後があれば構築できます。しかし、最初の429エラーを乗り越えて稼働させ続けることこそが、真の作業です。ここでは、SSEストリーミング、サーバーサイドの会話メモリ、モデル許可リストを備えた小さなFastAPIサービスを、1つのTokenLabキーで構築します。目標は、実際の製品の裏側に配置でき、統合コードを書き直すことなくモデルを変更できるバックエンドです。まだベースURLの切り替えを行っていない場合は、先に移行ガイドを読んでください。負荷時のバックオフが主な懸念事項である場合は、このガイドとAI APIレート制限ガイドを併せて確認してください。

重要なポイント

• 本番環境対応のチャットボットには、同期エンドポイント、ストリーミングエンドポイント、サーバーサイドの会話状態、モデル許可リスト、適切なエラーハンドリング、ストレージのアップグレードパスという6つの要素が必要です。
• キーはすべてのプロセスでTOKENLAB_API_KEYから読み込み、サーバー内に保持してください。認証ドキュメント（2026年10月3日確認）では、クライアント側のコードにキーを公開したり、コミットしたりしないよう強く推奨しています。
• GET /v1/modelsは公開されているため、成功したからといってキーが機能している証明にはなりません。実際のチャットリクエストを1回実行して証明してください（クイックスタート、2026年10月3日確認）。
• SSEはほとんどのチャット製品をカバーします。出力が届いた後に途切れたストリームは、完了した回答ではなく、不完全なものとして扱ってください。
• このガイドの許可リストでは、階層をdeepseek-v4-flash、gpt-5.5、claude-sonnet-5、gemini-3.5-flashにマッピングしています。それぞれがopenai_chat_completionsを受け入れ可能な形式としてリストしています（2026年10月3日確認）。
• 以下の価格からの概算では、入力2,000トークン、出力500トークンの1ターンあたり、モデルに応じて約0.0006ドルから0.0075ドルのコストがかかります。

AIチャットボットの構築：セットアップとコアサービス

完成したサービスには、以下の6つの可動パーツがあります：

1. スモークテスト用の同期型/chatエンドポイント。
2. 実際のUI用のストリーミング型/chat/streamエンドポイント。
3. conversation_idで管理される会話状態。
4. フロントエンドが任意のIDを要求できないようにするためのモデル許可リスト。
5. 最初の429エラーで崩壊しないエラーハンドリング。
6. メモリ内プロトタイプからRedisやPostgreSQLへ移行するためのパス。

これだけで、サポートボット、社内アシスタント、または埋め込みチャットウィジェットの最初のバージョンとしては十分です。以下の各コードブロックは、示された順序で1つのファイルmain.pyに追加していきます。

ステップ1：キーの設定とスタックのインストール

ワークスペースのAPI Keysからキーを作成します。完全な値は一度しか表示されず、キーリスト内のプレフィックスではリクエストを認証できません。すべてのTokenLabキーはsk-で始まります。最初の最初のリクエストはモデルの現在の価格で請求されるため、Billingで残高も確認してください（クイックスタート、2026年10月3日確認）。

キーはソースコードではなく、環境変数に保存してください。以下のプロンプトは、コマンドラインやシェル履歴に残らないようにし、現在のターミナルとその子プロセスにのみ適用されます。デプロイされたアプリにはシークレットマネージャーを使用してください。クイックスタートにはPowerShell版もあります。

printf 'TokenLab API key: '
read -r -s TOKENLAB_API_KEY
printf '\n'
export TOKENLAB_API_KEY

pip install fastapi uvicorn openai pydantic

TokenLab専用のSDKは不要です。キーとベースURLを変更すれば、公式のOpenAI SDKがそのまま動作します（SDKガイド、2026年10月3日確認）。

ステップ2：スキーマ、許可リスト、メモリの定義

共有パーツから始めます。リクエストスキーマでは生のmodel文字列ではなくtierを使用し、このガイドの後のcurl例はすべて同じフィールドを送信します。

最初のチャットボットデモでは、通常、ブラウザにトランスクリプトを保持してターンごとに再送信します。これはプロトタイプには適していますが、再試行、再開可能なセッション、サーバーサイドのツールが必要になると煩雑になります。そのため、最初からサーバー側で履歴を管理します。

import json
import logging
import os
import time
import uuid
from collections import defaultdict

from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from openai import APIConnectionError, APIStatusError, OpenAI, RateLimitError
from pydantic import BaseModel, Field

log = logging.getLogger("chatbot")
app = FastAPI()

client = OpenAI(
    api_key=os.environ["TOKENLAB_API_KEY"],
    base_url="https://api.tokenlab.sh/v1",
    timeout=30.0,
    max_retries=0,
)

ALLOWED_MODELS = {
    "default": "deepseek-v4-flash",
    "flagship": "gpt-5.5",
    "balanced": "claude-sonnet-5",
    "budget": "gemini-3.5-flash",
}
SYSTEM_PROMPT = "You are a helpful assistant. Be concise and direct."
MAX_HISTORY = 20          # 会話ごとに保持するメッセージ数（任意）
MAX_REPLY_TOKENS = 1000   # ターンごとの出力上限（任意）

class ChatRequest(BaseModel):
    message: str = Field(min_length=1, max_length=4000)
    tier: str = "default"
    conversation_id: str | None = None

def resolve_model(tier: str) -> str:
    model = ALLOWED_MODELS.get(tier)
    if model is None:
        raise HTTPException(status_code=400, detail=f"unknown tier: {tier}")
    return model

conversations: dict[str, list[dict]] = defaultdict(list)

def build_messages(conv_id: str, user_msg: str) -> list[dict]:
    history = conversations[conv_id][-MAX_HISTORY:]
    return [
        {"role": "system", "content": SYSTEM_PROMPT},
        *history,
        {"role": "user", "content": user_msg},
    ]

def remember(conv_id: str, user_msg: str, reply: str) -> None:
    conversations[conv_id].append({"role": "user", "content": user_msg})
    conversations[conv_id].append({"role": "assistant", "content": reply})

クライアントはクイックスタートに合わせてmax_retries=0と30秒のタイムアウトを設定しています。ストリーミングされたターンでSDKが盲目的に再試行すると二重課金される可能性があるため、再試行の判断は自分たちで行います。

ステップ3：エラーのマッピングと同期エンドポイントの追加

SDKの例外をステータスコードとユーザーに安全なメッセージに変換する小さな関数を1つ保持します。詳細はユーザーではなくログに出力します。

def describe_error(exc: Exception) -> tuple[int, str]:
    if isinstance(exc, RateLimitError):  # 最初を確認：APIStatusErrorを継承
        return 429, "The assistant is busy. Please retry shortly."
    if isinstance(exc, APIConnectionError):
        return 502, "Connection problem. Please retry."
    if isinstance(exc, APIStatusError):
        log.error("upstream status=%s code=%s", exc.status_code, exc.code)
        if exc.status_code in (401, 402, 403):  # キー、残高、権限：運用上の問題
            return 503, "The assistant is unavailable right now."
        if exc.status_code in (400, 404):  # フィールドやモデルIDの誤り：こちらのバグ
            return 500, "Something went wrong on our side."
        return 502, "The assistant is unavailable right now."
    return 500, "Something went wrong."

def retry_delay(exc: RateLimitError) -> float:
    try:
        return min(float(exc.response.headers.get("retry-after", "2")), 10.0)
    except ValueError:
        return 2.0

def create_with_one_retry(**kwargs):
    try:
        return client.chat.completions.create(**kwargs)
    except RateLimitError as exc:
        time.sleep(retry_delay(exc))
        return client.chat.completions.create(**kwargs)

@app.post("/chat")
def chat(req: ChatRequest):
    model = resolve_model(req.tier)
    conv_id = req.conversation_id or str(uuid.uuid4())
    messages = build_messages(conv_id, req.message)
    try:
        response = create_with_one_retry(
            model=model, messages=messages, max_tokens=MAX_REPLY_TOKENS
        )
    except (APIConnectionError, APIStatusError) as exc:
        status, message = describe_error(exc)
        raise HTTPException(status_code=status, detail=message)
    reply = response.choices[0].message.content or ""
    log.info("model=%s usage=%s", model, response.usage)
    remember(conv_id, req.message, reply)
    return {"conversation_id": conv_id, "reply": reply}

他の機能を構築する前に、1回リクエストを実行してください。これが失敗する場合、その上に構築を続けないでください。

返信をストリーミングし、AIチャットボットの応答性を維持する

ほとんどのチャットボット製品が遅く感じるのは、モデルが遅いからではなく、完全な応答が届くまでUIが空白のままだからです。SSEを採用したのは、ほとんどのチャット製品をカバーし、Websocketよりも運用負荷が低いためです。製品がすでにプレーンHTTPでブラウザクライアントを実行している場合、SSEの方がアーキテクチャをシンプルに保てます。

ステップ4：ストリーミングエンドポイントの追加

ストリーミングドキュメント（ストリーミングガイド、2026年10月3日確認）では、/v1/chat/completionsからのSSEチャンクをすでに期待しているフレームワークに対して、Chat Completionsのストリーミングを有効に保っています。すべての許可リストモデルがこれを受け入れるため、このパスを維持します。各イベントをJSONとしてエンコードすることで、返信内の改行がフレームを壊さないようにします。

def sse(payload: dict) -> str:
    return f"data: {json.dumps(payload)}\n\n"

@app.post("/chat/stream")
def chat_stream(req: ChatRequest):
    model = resolve_model(req.tier)
    conv_id = req.conversation_id or str(uuid.uuid4())
    messages = build_messages(conv_id, req.message)

    def generate():
        parts: list[str] = []
        finish_reason = None
        try:
            with client.chat.completions.create(
                model=model,
                messages=messages,
                max_tokens=MAX_REPLY_TOKENS,
                stream=True,
                stream_options={"include_usage": True},
            ) as stream:
                for chunk in stream:
                    if chunk.usage:
                        log.info("model=%s usage=%s", model, chunk.usage)
                    if not chunk.choices:
                        continue
                    choice = chunk.choices[0]
                    if choice.delta.content:
                        parts.append(choice.delta.content)
                        yield sse({"delta": choice.delta.content})
                    if choice.finish_reason:
                        finish_reason = choice.finish_reason
        except (APIConnectionError, APIStatusError) as exc:
            _, message = describe_error(exc)
            yield sse({"error": message})
            return
        if finish_reason != "stop":
            yield sse({"error": f"Reply incomplete ({finish_reason})."})
            return
        remember(conv_id, req.message, "".join(parts))
        yield sse({"done": True, "conversation_id": conv_id})

    return StreamingResponse(generate(), media_type="text/event-stream")

ここで重要な詳細は2つあります。第一に、include_usage: trueはストリームチャンクにトークン使用量を追加しますが、最後の使用量チャンクにはchoicesがないため、ループはそれをスキップします。第二に、finish_reasonがstopの場合のみターンを保存します。lengthでの終了はトークン制限により返信が途切れたことを意味し、ドキュメントでは両方が正常な終了条件としてリストされています。

エラーはジェネレーター内で発生するため、失敗した時点でHTTPステータスはすでに200です。クライアントはerrorイベントを読み取る必要があります。withブロックは、ジェネレーターが終了したときにアップストリームのストリームを閉じます。ドキュメントでは、ユーザーが停止または切断したときにアップストリームをキャンセルすることを推奨しているため、切断時にASGIサーバーが実際にジェネレーターを停止することを確認してください。

ステップ5：ブラウザでストリームを読み取る

ブラウザは独自の認証済みバックエンドを呼び出すべきであり、TokenLabを直接呼び出してはいけません。ネットワークチャンクが必ずしも完全なSSEイベントとは限らないため、イベント境界で解析してください。

async function sendMessage(payload, onDelta) {
  const response = await fetch('/chat/stream', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify(payload), // { message, tier, conversation_id }
  });
  if (!response.ok || !response.body) throw new Error(`HTTP ${response.status}`);

  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  let buffer = '';

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    buffer += decoder.decode(value, { stream: true });
    const events = buffer.split('\n\n');
    buffer = events.pop();
    for (const raw of events) {
      if (!raw.startsWith('data: ')) continue;
      const event = JSON.parse(raw.slice(6));
      if (event.error) throw new Error(event.error);
      if (event.delta) onDelta(event.delta);
    }
  }
}

Redisへのメモリのアップグレード

メモリ内辞書は1プロセスには適しています。会話にTTL、再開可能性、または複数インスタンスが必要になったらRedisを使用してください。トランスクリプトがクエリ、監査、レポートが必要な製品データになったらPostgreSQLを使用してください。Redisへの変更はストレージの配管であり、ロジックではありません。pip install redisを実行し、build_messagesとrememberが以下のヘルパーを通して読み書きするようにします。

import redis

redis_client = redis.Redis(
    host=os.environ.get("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))

AIチャットボットを本番環境向けに構築する際のエラーハンドリング

顧客向けのユーザーは、失敗がレート制限、残高不足、アップストリームの停止のどれによるものかは気にしません。UIがフリーズするかどうかを気にします。40語でストリームが途切れるサポートウィジェットを想像してください。TokenLabのドキュメントによると、出力が届いた後に中断されたストリームは完全な応答ではなく、繰り返すと異なる出力が生成され、2回目の課金が発生する可能性があります（APIエラーの処理、2026年10月3日確認）。これが、ステップ4で未完了のターンを保存しない理由です。

ステップ6：再試行ルールの適用

エラーは、予告なく変更される可能性のある人間が読めるmessageではなく、HTTPステータスとcodeで処理してください。以下の表は、ドキュメントとこのサービスが何を行うかを組み合わせたものです。

ステータス	意味（ドキュメント準拠）	このサービスの動作
400	フィールド、モデルID、または入力が無効	500を返し、ログに記録し、再試行しない
401	キーの欠落、無効、期限切れ、または取り消し	503を返し、オペレーターに警告
402	残高またはAPIキー制限が低すぎる（insufficient_balance, quota_exceeded）	503を返し、オペレーターに警告
403	キーがこのモデルを使用できない	503を返し、キーのモデルリストを確認
404	リソースまたはモデルが見つからない	500を返し、/v1/modelsとIDを照合
429	リクエスト制限に達した	Retry-Afterを待ち、1回再試行（同期のみ）
500–504	サービス利用不可またはネットワーク障害	retryableがtrueの場合のみ再試行；retry_afterを尊重

ドキュメントからの2つの注意点。503 all_channels_failedは必ずしも一時的な停止を意味しません。retryableがfalseの場合は同じリクエストを繰り返さず、別のモデルを選択する前にGET /v1/modelsを確認してください。また、Chat CompletionsとResponsesはOpenAIスタイルのerrorオブジェクトを使用しますが、Anthropic MessagesとGeminiは独自の形式を保持しています。このサービスはすべての許可リストモデルでChat Completionsを呼び出すため、1つのパーサーで機能します。もし階層をネイティブ形式に移行する場合は、2つ目のパーサーが必要です。

レスポンスヘッダーからリクエストIDをエンドポイント、モデル、時間、独自のユーザーIDまたはジョブIDとともに保存してください。ドキュメントには使用したページにヘッダー名が記載されていないため、ログを配線する前にトラブルシューティングパスで確認してください。

本番環境チェックリストの残りは短いです：

• フロントエンドの失敗をバックエンドのログに接続できるように、独自のリクエストIDを追加してください。
• ユーザーごとの同時実行数とリクエストサイズに上限を設けてください。messageのmax_length=4000は、調整すべき出発点です。
• MAX_HISTORYのように、メッセージ数で履歴に上限を設けてください。課金ドキュメントによると、トークン化はモデル、言語、メディア入力によって異なるため、文字数計算式は信頼できる課金見積もりにはなりません。
• すべてのターンでモデル、レイテンシ、使用量、終了理由をログに記録してください。
• ユーザーに見えるメッセージと内部エラーの詳細を分離してください。
• 最初の実際の停止の前にフォールバックが機能することを確認するため、代替階層に1回テストターンを送信してください。

AIチャットボットのモデル選択とコスト見積もり

フロントエンドに任意のモデル文字列をAPIに渡させないでください。自由記述フィールドは、タイプミス、廃止されたモデル、公開するつもりのなかったモデルを招きます。ALLOWED_MODELSマップは、プロバイダーがモデルを廃止したときにクライアントの更新なしでモデルを交換できる場所を提供します。不明なモデルIDはmodel_not_foundを返し、ドキュメントは現在のIDについて/v1/modelsを参照するように指示しています。許可リストをロックする前にモデルディレクトリを確認してください。

100万トークンあたりの現在の価格

これらは2026年10月3日に確認された、各モデルのライブAPIエントリーに基づくUSDでのTokenLab価格です。価格は2026年10月2日T16:53:30.068Zに最終更新されました。4つのモデルすべてがライフサイクルステージactiveを示していました。

階層	モデルID	入力 / 1M	出力 / 1M	キャッシュ読み取り / 1M	ソース
default	deepseek-v4-flash (オフピーク)	$0.15	$0.60	$0.003	model API
default	deepseek-v4-flash (ピーク)	$0.30	$1.20	$0.006	model API
flagship	gpt-5.5	$1.50	$9.00	$0.15	model API
balanced	claude-sonnet-5	$0.90	$4.50	$0.09	model API
budget	gemini-3.5-flash	$0.75	$4.50	$0.075	model API

deepseek-v4-flashの場合、エントリーには平日のピーク時間帯がリストされており、中国の祝日は除外されます。オフピークには週末とそれらの休日が含まれます。claude-sonnet-5は、100万トークンあたり1.125ドルのキャッシュ書き込み価格もリストしています。

1回の作業リクエスト

入力2,000トークン、出力500トークンの1ターンを想定します。このサイズは測定値ではなく、私たちの仮定です。gpt-5.5の場合：

• 入力: 2,000 × $1.50 / 1,000,000 = $0.0030
• 出力: 500 × $9.00 / 1,000,000 = $0.0045
• 合計: $0.0075 / ターン（見積もり、キャッシュヒットなし）

同じ算術を使用して、見積もりは次のようになります：deepseek-v4-flashはオフピークで$0.0006、ピークで$0.0012、claude-sonnet-5は$0.00405、gemini-3.5-flashは$0.00375。1,000ターン以上では、gpt-5.5で約7.50ドル、deepseek-v4-flashで0.60ドルから1.20ドルです。実際のターンは履歴が蓄積されるにつれて増加するため、このサイズ設定を信頼するのではなく、レスポンスまたは最終ストリームチャンクからusageを読み取ってください。

実際には、ほとんどの会話をデフォルト階層にルーティングし、それが必要なターンのためにgpt-5.5またはclaude-sonnet-5を予約します。課金ドキュメントでは、トークンあたりの最低価格が必ずしもタスク完了あたりの最低コストではないと警告しています。同じプロンプトを2つの階層で実行し、Usageで最終コストを比較してください。

出荷前に2つのチェックを行ってください。課金ドキュメントではコピーした価格表をハードコードしないように指示されているため、アプリがコストを表示する場合はGET /v1/models/:model/pricingを読み取ってください（課金と価格設定、2026年10月3日確認）。また、deepseek-v4-flashエントリーにはjson-mode、prompt-cache、tool-useはリストされていますが、visionはリストされていません。チャットボットが画像を受け入れる場合は、エントリーにvisionがリストされている階層にそれらのターンをルーティングしてください。

スモークテストとAIチャットボットの出荷

ステップ7：エンドツーエンドでの実行

TOKENLAB_API_KEYがエクスポートされているターミナルでサーバーを起動します。curlペイロードは、ChatRequestと同じmessage、tier、conversation_idフィールドを使用します。

uvicorn main:app --reload --port 8000

# 1. 同期エンドポイント：短い返信とconversation_idを期待
curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "Reply only with OK.", "tier": "default"}'

# 2. ストリーミング：data: {"delta": ...} 行、次に {"done": true, ...} を期待
curl -N -X POST http://localhost:8000/chat/stream \
  -H "Content-Type: application/json" \
  -d '{"message": "Hello!", "tier": "default"}'

# 3. メモリ：上記で返されたconversation_idを再利用
curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "What did I just say?", "tier": "default", "conversation_id": "PASTE_ID_FROM_STEP_1"}'

# 4. 強制的な失敗：HTTP 400と "unknown tier: nope" を期待
curl -i -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "Hello!", "tier": "nope"}'

アップストリームの失敗を強制するには、間違ったTOKENLAB_API_KEYでサーバーを再起動します。クライアントはクリーンな503メッセージを受け取り、ログにはinvalid_api_keyを含むアップストリームの401が表示されるはずです。次にRequestsを開き、成功した呼び出しのモデル、ステータス、使用量を確認します。1ターンをストリーミングし、1つの会話を保持し、強制的な失敗時にクリーンなエラーを返すことができれば、正しい基盤ができています。

その後、次のアップグレードはAIそのものよりも退屈なインフラストラクチャです：

• 認証：あるユーザーが他のユーザーの会話を読めないようにするため。
• 永続化：デプロイをまたいでセッションを存続させるため。
• ユーザーごとのレート制限：騒がしいユーザーがクォータを使い果たさないようにするため。
• 使用量の帰属：チャットボットが顧客向けの場合。
• バックグラウンド要約：会話に長期記憶が必要な場合。

ベースURL移行が完了すれば、モデルの変更はALLOWED_MODELSへの設定編集のみで行えます。

FAQ

AIチャットボットを構築するためにWebsocketは必要ですか？

いいえ。ステップ4で示したSSEは、ほとんどのチャット製品をカバーします。Websocketは、ライブコラボレーションやサーバー主導のイベントなど、リクエストとレスポンス以外の双方向プッシュが必要な場合に価値を発揮します。標準的なチャットUIの場合、SSEの方がデプロイとデバッグが簡単です。

チャットボットはデフォルトでどのモデルを使用すべきですか？

deepseek-v4-flashのような低価格の階層から始め、より難しいターンのために許可リストの背後にclaude-sonnet-5やgpt-5.5を追加してください。2026年10月3日に確認された上記の価格は、その差を示しています。次に、独自のプロンプトを2つの階層でテストし、Usageで最終コストを比較してください。コミットする前にモデルディレクトリを確認してください。ラインナップは変更されるためです。

なぜチャットボットが402や429を返すのですか？

402は、アカウント残高またはキー自体の使用制限が低すぎることを意味します。コードはinsufficient_balanceとquota_exceededであり、リクエストを繰り返しても解決しません。429はリクエスト制限に達したことを意味するため、Retry-Afterを待ってから再試行してください。同期エンドポイントは1回再試行し、ストリーミングエンドポイントはエラーをクライアントに報告します。

ブラウザはTokenLabを直接呼び出せますか？

安全ではありません。認証ドキュメントでは、クライアント側のコードにAPIキーを公開しないように指示されています。フロントエンドのサンプルが示すように、キーはサーバー環境に保持し、ブラウザには独自の認証済みバックエンドを呼び出させてください。

---

tokenlab.shでAPIキーを作成し、資金を追加して、上記のスモークテストを実行してください。