Gemini APIを使ってFastAPIのStreaming Responseについて深掘りしてみた

はじめに

こんにちは、レアゾン・ホールディングスでエンジニアをしているYuto Moriです。

今回はAIチャットボットなどで使用されているストリーミングという技術について深掘りしてみます。

ストリーミングとは

本記事で扱うストリーミングは、特定の生成物がすべてを完成してからユーザーに見えるのではなく、生成された内容を順番に見せることを指します。

GeminiやChatGPTといった生成AIチャットボットは文章を先頭から生成していきます。そのため、最初の一文が生成されてから最後の一文が生成されるまでの時間が大きく変化します。また、ユーザー側としては最初のレスポンスが返ってくるまでの時間がユーザー体験に大きく影響を及ぼします。そのため、ストリーミングがあることによって、ユーザーとしての体感的な待ち時間が劇的に短縮されます。

FastAPIの StreamingResponse

StreamingResponseは、FastAPIが提供するクラスで、レスポンスボディ全体を一度にメモリにロードするのではなく、段階的に（チャンクとして）送信することを可能にします。

最もシンプルなコードは以下のようになります。

main.py

import asyncio
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

async def non_blocking_streamer():
    """非同期ジェネレータ"""
    for i in range(10):
        yield f"Chunk {i}\n"
        # ノンブロッキングの待機。イベントループは他のタスクを実行できる
        await asyncio.sleep(0.5)

@app.get("/stream-async")
async def stream_async():
    """非同期ジェネレータをストリーミングで返すエンドポイント"""
    return StreamingResponse(non_blocking_streamer(), media_type="text/plain")

上記のコードをサーバーで起動し、curlで動作を確認すると0.5秒ごとに行が表示されることを確認できます。

# FastAPIをdevモードで起動
$ uv run fastapi dev 

# 非同期ストリームのテスト（0.5秒ごとに行が表示される）
$ curl -N http://127.0.0.1:8000/stream-async

Gemini API (google-genai)によるストリーミングレスポンス

GeminiAPIでストリーミングレスポンスを生成する場合はclient.models.generate_content_streamを使用します。最もシンプルなコードは以下の通りです。

text_generation.py

# https://github.com/google-gemini/api-examples/blob/9f5adb78a77820ef2d4f2a040d698481803e8214/python/text_generation.py#L37-L45
from google import genai

client = genai.Client()
response = client.models.generate_content_stream(
    model="gemini-2.5-flash", contents="Write a story about a magic backpack."
)
for chunk in response:
    print(chunk.text)
    print("_" * 80)

以前のライブラリ (google-generativeai) のGemini APIを使用している場合は、バージョンアップをしてください。
generativeai版では、generate_content("質問内容", stream=True)でStreamingResponseを生成していました。

ストリーミングをweb側（Next.js）で受け取る方法

本記事ではNext.jsで受け取ることを想定としています。受け取るためのapiのロジックは以下の通りです。

api.ts

export async function chatStream(
  message: string,
  onChunk: (text: string) => void
): Promise<void> {
  const response = await fetch(`${API_BASE_URL}/chat_stream`, {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
    },
    body: JSON.stringify({ message } as ChatRequest),
  });

  if (!response.ok) {
    throw new Error(`API error: ${response.statusText}`);
  }

  const reader = response.body?.getReader();
  const decoder = new TextDecoder();

  if (!reader) {
    throw new Error('No response body');
  }

  try {
    while (true) {
      const { done, value } = await reader.read();
      
      if (done) break;
      
      const text = decoder.decode(value, { stream: true });
      onChunk(text);
    }
  } finally {
    reader.releaseLock();
  }
}

fetch の戻り値 response.body から ReadableStreamDefaultReader を取得し、reader.read() をループで呼び出すことで、サーバーから届くデータを「塊（チャンク）」単位で受け取っています。受け取った Uint8Array のバイト列を TextDecoder で文字列に復号し、その文字列を onChunk　コールバックに渡します。
StreamingResponse はサーバー側でチャンク単位に yield されているため、ブラウザ側もそれに合わせて少しずつテキストが増えていきます。
ブラウザが「レスポンスの完了」を待たずにストリームを読み進められるのは、fetch が ReadableStream をサポートしているためです。

サンプルコード

Python (FastAPI) + Next.jsで書いた通常のレスポンスとストリーミングレスポンスを比較するサンプルを作成しました。
https://github.com/yuto-mori-re/streaming-response-practice

デモ動画で使用しているモデルはGemini-2.5-proです。Streamingの出力はチャンクごとに-------を挿入してわかりやすくしています。このように通常（Regular）のレスポンスは最初の出力が遅く、生成に時間がかかっている印象があります。生成物に時間がかかる内容をユーザ側に出力する際は重要となる機能の一つであると言えるでしょう。

終わりに

生成AIを活用したプロダクトにおいて、ユーザーの体感待ち時間を短縮するストリーミング技術は、UX向上のための必須要件となりつつあります。FastAPIとNext.jsを組み合わせれば、今回のように比較的シンプルなコードで実装でき、その効果は絶大です。ぜひ公開したサンプルコードを実際に動かして挙動の違いを体感し、ご自身の開発にも取り入れてみてください。

関連記事

▼新卒エンジニア研修のご紹介

レアゾン・ホールディングスでは、2025年新卒エンジニア研修にて「個のスキル」と「チーム開発力」の両立を重視した育成に取り組んでいます。 実際の研修の様子や、若手エンジニアの成長ストーリーは以下の記事で詳しくご紹介していますので、ぜひご覧ください！

▼採用情報

レアゾン・ホールディングスは、「世界一の企業へ」というビジョンを掲げ、「新しい"当たり前"を作り続ける」というミッションを推進しています。 現在、エンジニア採用を積極的に行っておりますので、ご興味をお持ちいただけましたら、ぜひ下記リンクからご応募ください。