1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@aidocs

FastAPIとReactで作るストリーミング対応AIチャットUI

1
Last updated at Posted at 2026-10-04

AIチャットでは、回答がすべて生成されてから画面に表示すると、短い質問でも待ち時間が長く感じられます。そこで今回は、FastAPIから回答を少しずつ送信し、React側で受信した文字列を逐次描画するチャットUIを作ります。

通信にはSSE（Server-Sent Events）形式を使います。実装を一通り動かして確認したところ、fetchとReadableStreamを組み合わせることで、POSTリクエストを使ったチャットでもストリーミング表示ができました。

全体構成

処理全体の流れを図にすると次の通りです。

今回の構成は次のとおりです。

  • React：メッセージ入力とストリーミング表示
  • FastAPI：チャットAPIとSSEレスポンス
  • AIモデル：FastAPI内の生成関数から呼び出す
  • 通信形式：text/event-stream

通常のAPIでは、次のように回答全体をJSONで返します。

{
  "answer": "回答全文..."
}

一方、ストリーミングでは次のようなイベントを複数回送信します。

data: {"token":"FastAPI"}

data: {"token":"と"}

data: {"token":"React"}

data: {"done":true}

この形式なら、生成された部分から画面へ反映できます。

FastAPIでストリーミングAPIを作る

まずバックエンドを作成します。

mkdir ai-chat
cd ai-chat

python -m venv .venv
.venv\Scripts\activate

pip install fastapi uvicorn

main.pyを作成し、以下のコードを書きます。

import asyncio
import json
from typing import AsyncIterator

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
from pydantic import BaseModel


app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:5173"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


class Message(BaseModel):
    role: str
    content: str


class ChatRequest(BaseModel):
    messages: list[Message]


async def generate_reply(messages: list[Message]) -> AsyncIterator[str]:
    question = messages[-1].content

    # 実際にはここでAIモデルのストリーミングAPIを呼び出す
    reply = (
        f"「{question}」について回答します。"
        "これはストリーミング表示のサンプルです。"
        "生成された文字列を少しずつクライアントへ送信しています。"
    )

    for word in reply:
        await asyncio.sleep(0.03)
        yield word


async def event_stream(
    request: Request,
    messages: list[Message],
) -> AsyncIterator[str]:
    async for token in generate_reply(messages):
        if await request.is_disconnected():
            break

        data = json.dumps(
            {"token": token},
            ensure_ascii=False,
        )
        yield f"data: {data}\n\n"

    yield 'data: {"done": true}\n\n'


@app.post("/api/chat")
async def chat(request: Request, body: ChatRequest):
    return StreamingResponse(
        event_stream(request, body.messages),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
        },
    )

起動します。

uvicorn main:app --reload --port 8000

curlで確認する場合は、次のコマンドを実行します。

curl -N -X POST http://localhost:8000/api/chat ^
  -H "Content-Type: application/json" ^
  -d "{\"messages\":[{\"role\":\"user\",\"content\":\"SSEとは何ですか？\"}]}"

-Nを付けるのがポイントです。付けない場合、curl側でバッファリングされ、まとめて表示されることがあります。

React側でイベントを受信する

Reactプロジェクトを作成します。

npm create vite@latest frontend -- --template react-ts
cd frontend
npm install
npm run dev

src/App.tsxでは、fetchのレスポンスを読み込みます。SSEには専用のEventSourceもありますが、今回はPOSTでメッセージ履歴を送る必要があるため、ReadableStreamを使いました。

import { useState } from "react";

type Message = {
  role: "user" | "assistant";
  content: string;
};

function App() {
  const [input, setInput] = useState("");
  const [messages, setMessages] = useState<Message[]>([]);
  const [loading, setLoading] = useState(false);

  async function sendMessage() {
    if (!input.trim() || loading) return;

    const userMessage: Message = {
      role: "user",
      content: input,
    };

    const nextMessages = [...messages, userMessage];

    setMessages([
      ...nextMessages,
      { role: "assistant", content: "" },
    ]);
    setInput("");
    setLoading(true);

    try {
      const response = await fetch("http://localhost:8000/api/chat", {
        method: "POST",
        headers: {
          "Content-Type": "application/json",
        },
        body: JSON.stringify({
          messages: nextMessages,
        }),
      });

      if (!response.ok || !response.body) {
        throw new Error("通信に失敗しました");
      }

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

      while (true) {
        const { value, done } = await reader.read();
        if (done) break;

        buffer += decoder.decode(value, { stream: true });

        const events = buffer.split("\n\n");
        buffer = events.pop() ?? "";

        for (const event of events) {
          const line = event
            .split("\n")
            .find((item) => item.startsWith("data:"));

          if (!line) continue;

          const data = JSON.parse(line.slice(5).trim());

          if (data.done) continue;

          setMessages((current) => {
            const copied = [...current];
            const last = copied.length - 1;

            copied[last] = {
              ...copied[last],
              content: copied[last].content + data.token,
            };

            return copied;
          });
        }
      }
    } catch (error) {
      setMessages((current) => {
        const copied = [...current];
        const last = copied.length - 1;

        copied[last] = {
          ...copied[last],
          content: "エラーが発生しました。",
        };

        return copied;
      });
    } finally {
      setLoading(false);
    }
  }

  return (
    <main style={{ maxWidth: 720, margin: "40px auto" }}>
      <h1>AI Chat</h1>

      <section>
        {messages.map((message, index) => (
          <p key={index}>
            <strong>{message.role}:</strong> {message.content}
          </p>
        ))}
      </section>

      <textarea
        value={input}
        onChange={(event) => setInput(event.target.value)}
        disabled={loading}
        rows={4}
        style={{ width: "100%" }}
      />

      <button onClick={sendMessage} disabled={loading}>
        {loading ? "生成中..." : "送信"}
      </button>
    </main>
  );
}

export default App;

実際に動かすと、assistantの空メッセージに1文字ずつ内容が追加されていきます。レスポンスのチャンクは任意の位置で分割されるため、bufferに一度ためてから\n\n単位で処理することが重要です。

実際のAIモデルへ置き換える

今回のgenerate_replyは動作確認用の関数です。実際には、利用するAI SDKのストリーミング処理に置き換えます。

大切なのは、モデルから届いた差分をそのままyieldすることです。

async def generate_reply(messages: list[Message]):
    async for chunk in model.stream(messages):
        token = chunk.get("text", "")

        if token:
            yield token

モデル固有の形式をFastAPI側で吸収しておくと、React側は常に{"token": "..."}だけを扱えばよくなります。

実装時に注意した点

まず、CORSの許可オリジンは開発環境用に限定しました。本番ではallow_origins=["*"]のままにせず、実際のフロントエンドURLを指定します。

また、ユーザーが画面を閉じた場合に備えて、request.is_disconnected()を確認しています。これにより、切断後も不要な生成処理を続ける可能性を減らせます。

リバースプロキシを使う場合は、レスポンスのバッファリングにも注意が必要です。Nginxなどでバッファリングされると、バックエンドが逐次送信していても、ブラウザにはまとめて届くことがあります。

認証、レート制限、入力文字数の制限も、本番運用では追加する必要があります。特に会話履歴を毎回すべて送る設計では、履歴が長くなるほどリクエストサイズとモデルの入力トークン数が増えます。

まとめ

FastAPIのStreamingResponseとReactのReadableStreamを使うことで、AIの回答を生成途中から表示できました。

実装の要点は次の3つです。

  • FastAPIでtext/event-streamを返す
  • SSEイベントの区切りを考慮してクライアント側でバッファリングする
  • AIモデル固有のレスポンス形式をバックエンドで吸収する

ストリーミング対応は、AIモデルの種類に依存しない構成にできます。まずは今回のサンプルで通信の流れを確認し、その後に実際のモデル呼び出し、キャンセル処理、認証を追加していくのがおすすめです。

1
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?