【FastAPI × Azure OpenAI】LLMの回答を「文字送り」で返すServer-Sent Events (SSE) 実装と、タイムアウト回避の非同期処理設計

はじめに：前回の記事からの発展

前回の記事「NestJS使いがFastAPIを触ったら実家のような安心感だった」では、TypeScriptエンジニアがFastAPIを学ぶ際の概念対応表を解説しました。

今回は、実際にAIアプリケーションを作る上で必須となる「ストリーミング処理」 について、FastAPIとAzure OpenAIを使って実装します。

SNS投稿の感情分析や、チャットボットの開発において、LLM（大規模言語モデル）の生成結果を「一気に返す」のではなく、「文字送り」で返すことは、UX向上のため必須です。

課題：LLMの生成は遅い

Azure OpenAI API（GPT-4）を使ったSNS投稿分析を実装するとき、最初は以下のような同期処理を書きがちです：

# 問題のある実装（同期処理）
@app.post("/analyze")
def analyze_sync(request: AnalyzeRequest):
    # この処理には5〜10秒かかる！
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": request.text}]
    )
    return {"result": response.choices.message.content}

問題点：

• ユーザーは5〜10秒間「ローディング中」を見続ける必要がある
• サーバーの接続がタイムアウトするリスク（30秒以上のレスポンスは切断されることが多い）
• サーバーリソースを占有し、他のリクエストを処理できなくなる

解決策：Server-Sent Events (SSE) とは

Server-Sent Events (SSE) は、サーバーからクライアントに対して、イベントストリームをプッシュする技術です。WebSocketとは異なり、**片方向（サーバー→クライアント）**の通信に特化しており、HTTPプロトコル上で動作します。

NestJS/Next.js開発者向けの説明

技術	用途	FastAPIでの実装
WebSocket	双方向通信（チャットリアルタイム更新）	WebSocketクラス
SSE	片方向ストリーミング（AI生成結果の逐次表示）	StreamingResponse
通常HTTP	リクエスト/レスポンス（JSON返却）	JSONResponse

NestJSでいうところの、@Sse()デコレータに相当します。

// NestJSの場合
@Sse('events')
events(): Observable<MessageEvent> {
  return interval(1000).pipe(map((_) => ({ data: 'hello' })));
}

FastAPIでは、StreamingResponseを使って同様の実装ができます。

実装：FastAPI + Azure OpenAI + SSE

前提条件

# 必要なパッケージ
pip install fastapi uvicorn openai python-dotenv

完全な実装例

import os
import asyncio
from typing import AsyncGenerator
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from openai import AsyncAzureOpenAI
from dotenv import load_dotenv

# 環境変数の読み込み
load_dotenv()

app = FastAPI(title="SNS Analysis API with Streaming")

# CORS設定（フロントエンドからの呼び出し用）
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 本番環境では適切に制限すること
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Azure OpenAIクライアントの初期化（非同期用）
client = AsyncAzureOpenAI(
    api_key=os.getenv("AZURE_OPENAI_API_KEY"),
    api_version=os.getenv("AZURE_OPENAI_API_VERSION", "2024-02-15-preview"),
    azure_endpoint=os.getenv("AZURE_OPENAI_ENDPOINT")
)

# リクエスト/レスポンスの型定義
class AnalyzeRequest(BaseModel):
    text: str
    max_tokens: int = 500

class AnalysisResult(BaseModel):
    score: int  # 0-100の危険度スコア
    reasoning: str
    keywords: list[str]

async def generate_analysis_stream(text: str) -> AsyncGenerator[str, None]:
    """
    SNS投稿テキストをストリーミング分析するジェネレータ
    SSEフォーマット: data: {content}\n\n
    """
    try:
        # システムプロンプト：SNS分析用
        system_prompt = """あなたはSNS投稿の感情分析エキスパートです。
以下の投稿を分析し、以下の形式でJSONを返してください：
{
  "score": 0-100の危険度スコア（高いほど問題あり）,
  "reasoning": "分析理由",
  "keywords": ["重要キーワード1", "重要キーワード2"]
}

投稿内容を段階的に分析し、リアルタイムで思考プロセスを出力してください。"""

        # Azure OpenAI APIを非同期ストリーミングで呼び出し
        stream = await client.chat.completions.create(
            model=os.getenv("AZURE_OPENAI_DEPLOYMENT_NAME", "gpt-4"),
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": f"分析対象の投稿:\n{text}"}
            ],
            stream=True,  # ストリーミング有効化
            max_tokens=500,
            temperature=0.3  # 安定した出力のため低めに設定
        )
        
        # ストリームから逐次読み出し
        async for chunk in stream:
            if chunk.choices and chunk.choices[0].delta.content:
                content = chunk.choices[0].delta.content
                # SSEフォーマットでエンコード
                yield f"data: {content}\n\n"
                # 小さく区切って送信することで、滑らかな表示を実現
                await asyncio.sleep(0.01)
        
        # ストリーム終了の信号
        yield "data: [DONE]\n\n"
        
    except Exception as e:
        # エラーハンドリング
        yield f"data: {{\"error\": \"{str(e)}\"}}\n\n"
        yield "data: [DONE]\n\n"

@app.post("/analyze-stream")
async def analyze_sns_stream(request: AnalyzeRequest):
    """
    SNS投稿をストリーミング分析するエンドポイント
    
    使用方法:
    const eventSource = new EventSource('/analyze-stream');
    eventSource.onmessage = (event) => {
      if (event.data === '[DONE]') eventSource.close();
      else console.log(event.data); // 逐次表示
    };
    """
    if not request.text or len(request.text.strip()) == 0:
        raise HTTPException(status_code=400, detail="テキストが空です")
    
    return StreamingResponse(
        generate_analysis_stream(request.text),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
        }
    )

@app.get("/health")
async def health_check():
    """ヘルスチェックエンドポイント"""
    return {"status": "ok", "service": "sns-analyzer"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

非同期処理のポイント解説

1. AsyncAzureOpenAI vs AzureOpenAI

# 非同期クライアント（推奨）
client = AsyncAzureOpenAI(...)

# 同期クライアント（非推奨：ブロッキングする）
client = AzureOpenAI(...)

FastAPIでは、async defで定義したエンドポイント内で同期APIを呼び出すと、イベントループがブロックされ、他のリクエストを処理できなくなります。必ず非同期クライアントを使用してください。

2. StreamingResponseの仕組み

async def generate_analysis_stream(...) -> AsyncGenerator[str, None]:
    yield f"data: {content}\n\n"  # ここで送信

• AsyncGeneratorは、値を順次生成し、都度クライアントに送信します
• yieldごとにチャンクが送信されるため、メモリ効率が良い
• SSEフォーマット（data: ...\n\n）を厳守すること

3. タイムアウト回避

FastAPIのデフォルトタイムアウトは設定によって異なりますが、ストリーミングレスポンスではタイムアウトを無効化または延長する必要があります（使用するサーバー/プロキシ次第）。

Azure固有のハマりポイント

1. APIバージョンの指定

Azure OpenAIは、APIバージョンを明示的に指定する必要があります。

client = AsyncAzureOpenAI(
    api_version="2024-02-15-preview",  # 必須！指定しないとエラー
    ...
)

よくあるエラー：

Unsupported api-version: 2024-02-15

対策：Azure Portalで「エンドポイントとキー」ページを確認し、正しいAPIバージョンを使用してください。

2. デプロイメント名の指定

Azure OpenAIでは、モデル名（gpt-4）ではなく、デプロイメント名を指定します。

# 誤り
model="gpt-4"

# 正しい
model=os.getenv("AZURE_OPENAI_DEPLOYMENT_NAME")  # 例: "gpt-4-deployment"

Azure Portalで作成した「デプロイメント名」を環境変数に設定してください。

3. 環境変数の設定例

# .envファイル
AZURE_OPENAI_API_KEY=your-api-key
AZURE_OPENAI_ENDPOINT=https://your-resource.openai.azure.com/
AZURE_OPENAI_API_VERSION=2024-02-15-preview
AZURE_OPENAI_DEPLOYMENT_NAME=gpt-4-deployment

フロントエンド実装（確認用）

以下のHTMLをブラウザで開くと、ストリーミング処理を確認できます：

<!DOCTYPE html>
<html lang="ja">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>SNS分析 - ストリーミングデモ</title>
    <style>
        body { font-family: Arial, sans-serif; max-width: 800px; margin: 0 auto; padding: 20px; }
        textarea { width: 100%; height: 100px; margin-bottom: 10px; }
        button { padding: 10px 20px; background: #007acc; color: white; border: none; cursor: pointer; }
        #result { margin-top: 20px; padding: 15px; background: #f5f5f5; border-radius: 5px; white-space: pre-wrap; }
        .loading { color: #666; }
    </style>
</head>
<body>
    <h1>SNS投稿分析（ストリーミング版）</h1>
    <textarea id="input" placeholder="分析したいSNS投稿を入力してください...">この商品本当に最悪！詐欺だよみんな気をつけて！</textarea>
    <button onclick="analyze()">分析開始</button>
    <div id="result">結果がここに表示されます...</div>

    <script>
        async function analyze() {
            const text = document.getElementById('input').value;
            const resultDiv = document.getElementById('result');
            resultDiv.innerHTML = '<span class="loading">分析中...</span>';
            
            try {
                const response = await fetch('http://localhost:8000/analyze-stream', {
                    method: 'POST',
                    headers: { 'Content-Type': 'application/json' },
                    body: JSON.stringify({ text: text })
                });

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

                while (true) {
                    const { done, value } = await reader.read();
                    if (done) break;
                    
                    const chunk = decoder.decode(value);
                    // SSEフォーマットを解析
                    const lines = chunk.split('\n');
                    for (const line of lines) {
                        if (line.startsWith('data: ')) {
                            const data = line.slice(6);
                            if (data === '[DONE]') {
                                return;
                            }
                            result += data;
                            resultDiv.innerHTML = result + '<span class="loading">▌</span>';
                        }
                    }
                }
            } catch (error) {
                resultDiv.innerHTML = 'エラー: ' + error.message;
            }
        }
    </script>
</body>
</html>

コスト最適化のTips

Azure OpenAIの課金は、入力トークン + 出力トークンの合計で計算されます。ストリーミングを使っても、トークン数自体は変わりませんが、以下の工夫でコストを抑えられます：

1. モデルの選定

# 高コスト（高品質）
model="gpt-4"  # または gpt-4o

# 低コスト（高速）
model="gpt-3.5-turbo"  # または gpt-4o-mini

SNS投稿分析などの構造化された出力が必要なタスクでは、GPT-3.5-turboでも十分な精度が出ることが多いです。

2. max_tokensの制限

await client.chat.completions.create(
    max_tokens=500,  # 上限を設定
    ...
)

出力トークン数に上限を設けることで、予想外の長文生成によるコスト爆発を防ぎます。

3. レスポンス形式の最適化

不要な改行や空白を減らすことで、トークン数を削減できます：

system_prompt = """分析結果をJSON形式で返してください。
余分な空白や改行を省き、コンパクトな形式で出力してください。"""

まとめ

本記事では、FastAPIとAzure OpenAIを使って、LLMの生成結果をストリーミングで返す実装を解説しました。

重要なポイント：

• 非同期クライアント（AsyncAzureOpenAI）を必ず使用する
• StreamingResponseでSSEフォーマットを返す
• Azure特有の設定（APIバージョン、デプロイメント名）に注意
• タイムアウト回避のため、重い処理は非同期ストリーミングで処理する

この実装パターンは、SNS投稿監視システムやチャットボット、リアルタイム分析ダッシュボードなど、AIを組み込んだWebアプリケーションで必須となる技術です。

次回は、「LLM-as-a-Judge」 による自動評価パイプラインの構築について解説します。

---

この記事を書いた人✏️@YushiYamamoto
ITPRODX.com代表 / AIアーキテクト
Next.js / TypeScript / n8nを活用した自律型アーキテクチャ設計を専門としています。
日々の自動化の検証結果や、ビジネス側の視点（ROI等）に関するより深い考察は、以下の公式サイトおよびnoteで発信しています。