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

このチュートリアルでは、ストリーミング応答、会話履歴、モデル切り替え、適切なエラーハンドリングを備えた本番対応のAIチャットボットバックエンドを構築します。Python、FastAPI、OpenAI SDKを使用し、APIアグリゲーターを通じて任意のモデルを利用できるようにします。

前提条件

pip install fastapi uvicorn openai

ステップ1：基本的なチャットエンドポイント

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import OpenAI
from pydantic import BaseModel

app = FastAPI()

client = OpenAI(
    api_key="sk-lemon-xxx",
    base_url="https://api.lemondata.cc/v1"
)

class ChatRequest(BaseModel):
    message: str
    model: str = "gpt-4.1-mini"
    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}

これは動作しますが、ストリーミングも履歴もエラーハンドリングもありません。これらを改善しましょう。

ステップ2：ストリーミングの追加

ストリーミングは、完全な応答を待つのではなく、トークンが生成されるたびに送信します。ユーザーはリアルタイムで返信が形成されるのを見られます。

@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"
    )

ステップ3：会話履歴

会話履歴をメモリに保存します（本番環境ではRedisやデータベースに置き換えてください）。

from collections import defaultdict
import uuid

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

SYSTEM_PROMPT = "You are a helpful assistant. Be concise and direct."

@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
    conv_id = req.conversation_id or str(uuid.uuid4())

    # メッセージ履歴の構築
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    messages.extend(conversations[conv_id])
    messages.append({"role": "user", "content": req.message})

    # ユーザーメッセージの保存
    conversations[conv_id].append(
        {"role": "user", "content": req.message}
    )

    def generate():
        full_response = []
        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"

        # アシスタントの応答を保存
        conversations[conv_id].append(
            {"role": "assistant", "content": "".join(full_response)}
        )
        yield f"data: [DONE]\n\n"

    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
        headers={"X-Conversation-ID": conv_id}
    )

ステップ4：エラーハンドリング

AI APIの呼び出しは、レート制限、残高不足、モデルの利用不可などで失敗することがあります。各ケースを処理しましょう：

from openai import (
    APIError,
    RateLimitError,
    APIConnectionError
)

@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():
        try:
            full_response = []
            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"

            conversations[conv_id].append(
                {"role": "assistant", "content": "".join(full_response)}
            )

        except RateLimitError as e:
            yield f"data: [ERROR] レート制限に達しました。しばらくお待ちください。\n\n"
        except APIConnectionError:
            yield f"data: [ERROR] 接続に失敗しました。再試行中...\n\n"
        except APIError as e:
            yield f"data: [ERROR] {e.message}\n\n"

        yield "data: [DONE]\n\n"

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

def build_messages(conv_id: str, user_msg: str) -> list:
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    # コンテキスト長を管理するため直近10ターンを保持
    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

ステップ5：モデル切り替え

会話の途中でユーザーがモデルを切り替えられるようにします。用途に応じて異なるモデルを：

AVAILABLE_MODELS = {
    "fast": "gpt-4.1-mini",
    "smart": "claude-sonnet-4-6",
    "reasoning": "o3",
    "budget": "deepseek-chat",
    "creative": "claude-sonnet-4-6",
}

@app.get("/models")
async def list_models():
    return {"models": AVAILABLE_MODELS}

フロントエンドはこれらを選択肢として提示できます。すべてのモデルはアグリゲーターを通じてOpenAI互換フォーマットを使っているため、切り替えはmodelパラメータを変えるだけです。

ステップ6：コンテキストウィンドウ管理

長い会話はモデルのコンテキスト制限を超えることがあります。スライディングウィンドウを実装しましょう：

def trim_history(messages: list, max_tokens: int = 8000) -> list:
    """システムプロンプト＋最近のメッセージをトークン予算内に保持する"""
    # おおよその目安：1トークン ≈ 4文字
    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

完成したアプリケーション

# 実行例: uvicorn main:app --reload --port 8000
# テスト: curl -N -X POST http://localhost:8000/chat/stream \
#   -H "Content-Type: application/json" \
#   -d '{"message": "Hello!", "model": "gpt-4.1-mini"}'

全コードは100行未満です。ここから以下を追加できます：

• 認証（APIキーやJWT）
• 永続ストレージ（PostgreSQLやRedisで会話を保存）
• ユーザーごとのレート制限
• 利用状況の追跡と課金
• 双方向ストリーミングのためのWebSocket対応
• フロントエンド（React、Vue、またはEventSourceを使ったバニラJS）

コスト見積もり

1日あたり1,000会話（平均5ターン/会話）を処理するチャットボットの場合：

モデル	1日あたりのコスト	1か月あたりのコスト
GPT-4.1-mini	約$2.40	約$72
GPT-4.1	約$12.00	約$360
Claude Sonnet 4.6	約$18.00	約$540
DeepSeek V3	約$1.68	約$50

ほとんどの会話はGPT-4.1-miniを使い、ユーザーが希望した場合のみClaude Sonnet 4.6に切り替えることで、ほとんどのアプリケーションで月額100ドル未満に抑えられます。

---

APIキーを取得：lemondata.ccは300以上のモデルを1つのエンドポイントで提供。$1の無料クレジットで開発を始められます。