大規模言語モデル API でチャットボットを構築：FastAPI バックエンドとストリーミングフロントエンド

まず小さな図を描く：3 つの役割をそれぞれ担当させる

チャットボットをデリバリー店と想像してください：ブラウザは顧客、あなたのバックエンドは受付、大規模モデル API は厨房です。顧客は厨房に直接入らず、すべての注文は受付が転送します。これが、キーをフロントエンドに置けない理由です。

1 回の会話の流れは次の通りです：ウェブページがメッセージを送信し、バックエンドが保存し、直近の履歴を取得し、system を追加して API にリクエストを送ります。ストリーミングで受信しながらウェブページに転送し、受信完了後に完全な返信も保存します。

なぜフロントエンドの直接接続ではなくバックエンドの転送を選ぶのか？API キーのセキュリティに加え、現実的な理由が 2 つあります。1 つは、転送時に履歴の追加、レート制限、フィルタリングを行う必要があり、これらはバックエンドでしか行えないことです。2 つ目は、将来モデルやサービスを変更する場合、バックエンドの 1 か所のみを変更すればよく、フロントエンドは変更不要だからです。

バックエンド：FastAPI と SQLite

まず依存関係をインストールして起動します：

复制pip install fastapi uvicorn openai
export API_KEY=你的密钥
uvicorn server:app --reload --port 8000

次に server.py を記述します。コードには注目すべき 4 点があります：

复制# server.py
import os
import sqlite3
import uuid

from fastapi import FastAPI
from fastapi.responses import FileResponse, StreamingResponse
from pydantic import BaseModel
from openai import AsyncOpenAI

app = FastAPI()
client = AsyncOpenAI(base_url="https://api.apidamoxing.com/v1", api_key=os.environ["API_KEY"])

SYSTEM = {"role": "system", "content": "你是『小墨』，一个说话简洁、爱用比喻的聊天伙伴。回答控制在 200 字内。"}
KEEP = 20  # 每次只带最近 20 条消息

db = sqlite3.connect("chat.db", check_same_thread=False)
db.execute("create table if not exists msg(id integer primary key autoincrement, sid text, role text, content text)")

def load(sid, n=KEEP):
    rows = db.execute("select role, content from msg where sid=? order by id desc limit ?", (sid, n)).fetchall()
    return [{"role": r, "content": c} for r, c in reversed(rows)]

def save(sid, role, content):
    db.execute("insert into msg(sid, role, content) values (?,?,?)", (sid, role, content))
    db.commit()

class ChatIn(BaseModel):
    session_id: str
    message: str

@app.get("/")
def index():
    return FileResponse("index.html")

@app.post("/session")
def new_session():
    return {"session_id": uuid.uuid4().hex}

@app.post("/chat")
async def chat(body: ChatIn):
    save(body.session_id, "user", body.message[:4000])
    messages = [SYSTEM] + load(body.session_id)

    async def gen():
        parts = []
        try:
            stream = await client.chat.completions.create(
                model="uncensored", messages=messages,
                stream=True, max_tokens=800, temperature=0.8,
            )
            async for chunk in stream:
                if not chunk.choices:        # 末尾的 usage 块没有 choices
                    continue
                delta = chunk.choices[0].delta.content
                if delta:
                    parts.append(delta)
                    yield delta
        except Exception as e:
            yield f"\n[请求失败：{type(e).__name__}]"
        finally:
            if parts:
                save(body.session_id, "assistant", "".join(parts))

    return StreamingResponse(gen(), media_type="text/plain; charset=utf-8")

1. load() は直近 20 件のみ取得し、会話が長くなりすぎて 100,000 トークンのコンテキストウィンドウを超過しないようにします。
2. gen()は非同期ジェネレーターで、断片を受信するたびにyieldし、フロントエンドで逐次表示します。
3. ストリームの末尾には使用量のみを含むデータブロックがあり、choices がないため、空チェックしてスキップします。
4. finally で完全な応答を保存し、途中でエラーが発生しても、生成済みの部分は失われません。

データベースをまだ使いたくない場合は、load と save を辞書の読み書きに置き換えるだけで、他のコードは変更不要です。逆に、アクセスが増えた場合、SQLite はお好みのデータベースに置き換えられ、これらの関数の形状を保てば API は維持されます。また、例では同期の sqlite3 を使用していますが、書き込みは高速でプロトタイプには十分です。高同時並列リクエストのケースでは、非同期ドライバーを検討してください。

フロントエンド：fetch でストリームを読む

ブラウザにはライブラリ不要です。resp.body.getReader() でリーダーを取得し、read() をループで実行し、取得したバイトブロックを TextDecoder でデコードしてページに追加します。以下の内容を index.html として保存し、server.py と同じディレクトリに配置してください：

复制<!doctype html>
<meta charset="utf-8">
<title>小墨</title>
<div id="gate">
  <label><input type="checkbox" id="adult"> 我已年满 18 周岁</label>
  <button id="enter">进入</button>
</div>
<div id="app" hidden>
  <div id="log" style="white-space:pre-wrap;min-height:300px"></div>
  <input id="box" placeholder="说点什么"> <button id="send">发送</button>
</div>
<script>
let sid = localStorage.getItem("sid");
const log = document.getElementById("log");

document.getElementById("enter").onclick = async () => {
  if (!document.getElementById("adult").checked) return;
  if (!sid) {
    const r = await fetch("/session", { method: "POST" });
    sid = (await r.json()).session_id;
    localStorage.setItem("sid", sid);
  }
  document.getElementById("gate").hidden = true;
  document.getElementById("app").hidden = false;
};

document.getElementById("send").onclick = async () => {
  const box = document.getElementById("box");
  const text = box.value.trim();
  if (!text) return;
  box.value = "";
  log.textContent += "\n你：" + text + "\n小墨：";
  const resp = await fetch("/chat", {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ session_id: sid, message: text }),
  });
  const reader = resp.body.getReader();
  const decoder = new TextDecoder("utf-8");
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    log.textContent += decoder.decode(value, { stream: true });
  }
  log.textContent += "\n";
};
</script>

2 つの注意点：デコード時に { stream: true } を追加するのは、1 文字のバイトが複数のブロックに分割される可能性があるためであり、追加しないと文字化けが発生します。また、入力欄の送信ロジックはリクエスト中にボタンを無効にするべきで、ユーザーの連打を防ぐためです。例では簡略化のため省略していますが、公開時には必ず実装してください。

実行後、ブラウザでポート 8000 にアクセスし、年齢確認にチェックを入れてメッセージを送信してみてください。文字が逐字ではなく段落として突然表示される場合、中間のプロキシが応答をバッファリングしている可能性があります。まずローカルポートに直接接続して除外してください。Nginx 配下にデプロイする場合は、このパスに対する応答バッファリングを無効にする必要があります。

会話履歴：なぜ自前で保存するのか

チャット API は「ステートレス」です。手渡されたメモしか見ない受付係のようなものです。メモに何があるかによって、彼が知る内容が決まります。つまり「コンテキストの記憶」はあなたの責任であり、具体的な方法は 3 つのレベルがあります：

1. シンプル：メモリ内の辞書にのみ保存します。再起動するとデータは消去され、デバッグ用途に適しています。
2. 一般的：例のように SQLite に保存し、各メッセージを1行として記録します。session_id で最新の N 件のメッセージを照会します。
3. 上級：履歴が長くなりすぎた場合、古いメッセージをモデルに要約させ、その要約を system プロンプトの後に配置します。直近のメッセージは原文のまま保持します。

自前のバックエンドに保存するもう一つの利点は、保持期間を完全に制御できる点です。「チャット履歴を消去」ボタンを提供し、対応する session の記録を本当に削除し、プライバシーポリシーで何を保存しているかを明確に記載することを推奨します。

要約方式のトリガーはシンプルに設定できます。推定される履歴トークンが1万を超えた場合、最も古いメッセージの半分をモデルに要約させ、300字以内の要約文を作成します。それをストレージに書き込み、元のメッセージを削除します。これにより、会話の文脈は維持され、各リクエストの規模も制御可能な範囲に安定します。ただし、要約自体もモデルによって生成されるため、詳細が欠落する可能性があります。ユーザーのニックネームや禁忌トピックなどの重要な設定は、要約に依存するのではなく、system プロンプトに固定して記載するのが適しています。

大人向け製品：入口と境界

このサービスは18歳以上の成人のみを対象としています。あなたのアプリケーションも同様であるべきです。例のフロントエンドには最もシンプルな確認入口が設置されており、より厳格な製品では、より完全な年齢確認プロセスを追加できます。いくつかの実践的な提案は以下の通りです。

• 入口ページで年齢制限を明確に提示し、確認されるまでチャット画面を表示しない。
• 製品を学生や未成年者層にプロモーションせず、彼ら向けのシーンにも配置しない。
• 未成年者に関連する性的なコンテンツは、実在か架空かを問わず、APIインターフェースによってブロックされ403が返されます。バックエンドはこのステータスを認識し、ユーザーに友好的なメッセージを表示し、生のエラーをそのまま表示しないようにします。

プロダクトデザインとして、ユーザーが設定でニックネームやトーンの設定を自分で行えるようにすることもできます。これらは system プロンプトに書き込むだけでよく、会話のパーソナライズが可能になり、モデルに推測させる必要がなくなります。

公開前の強化と拡張

• 入力の制限：例では1メッセージあたり4000文字に切り捨てています。必要に応じて調整してください。
• レート制限：各キーを1分あたり300リクエストに制限します。ユーザーまたはIP単位でバックエンド側でレート制限を適用します。
• エラー表示：バックエンドで例外をキャッチし、フロントエンドに短いメッセージを返します。スタックトレースを公開しないようにします。
• 使用量の監視：トークン数を統計するには、非ストリーミングリクエストで usage を読み取るか、ストリームの最後のチャンクで読み取ります。詳細はトークンと課金を参照してください。
• フレームワークの変更：フロントエンドを Vue や React に変更しても、ストリームを読むロジックは全く同じです。バックエンドを Express に変更した場合も、StreamingResponse を対応するストリーミング処理に置き換えるだけです。

より完全なパラメータ説明が必要な場合はAPI パラメータ詳細をご覧ください。その他の質問はよくある質問をご覧ください。

公開前に最終確認を行います：秘密鍵はサーバー側の環境変数のみに格納されているか。1メッセージの入力と履歴の長さに上限があるか。エラーに友好的なメッセージが表示されるか。「チャット履歴を消去」が実際に記録を削除するか。年齢確認が入口にあるか。これら5項目がすべて満たされていれば、このプロトタイプを実際のユーザーに試してもらうことができます。