llm-streaming-response-handler

LLM ストリーミングレスポンスハンドラー

即座に反応するように感じられる、LLM レスポンス向けの本番品質のストリーミングインターフェース構築に精通しています。

使用するタイミング

✅ 次の場合に使用します：

• タイピングアニメーションを備えたチャットインターフェース
• リアルタイム AI アシスタント
• ライブプレビューを備えたコード生成
• 段階的に表示するドキュメント要約
• ユーザーが LLM からの即時応答を期待するあらゆる UI

❌ 次の場合は使用しません：

• バッチでのドキュメント処理（ユーザーが見ていない場合）
• ストリーミングに対応していない API
• WebSocket ベースの双方向チャット（Socket.IO を使用してください）
• シンプルなリクエスト／レスポンス（fetch で十分です）

クイック判断ツリー

Does your LLM interaction:
├── Need immediate visual feedback? → Streaming
├── Display long-form content (>100 words)? → Streaming
├── User expects typewriter effect? → Streaming
├── Short response (<50 words)? → Regular fetch
└── Background processing? → Regular fetch

---

技術の選択

Server-Sent Events（SSE） - 推奨

LLM ストリーミングで WebSocket より SSE が優れている理由：

• シンプルさ：HTTP ベースで、既存のインフラストラクチャで動作します
• 自動再接続：再接続ロジックが組み込まれています
• ファイアウォールとの親和性：プロキシ経由での接続は WebSocket より簡単です
• 一方向通信に最適：LLM はサーバーからクライアントへのみストリーミングします

年表：

• 2015～2020 年：あらゆる用途に WebSocket を使用
• 2020 年：ストリーミング API で SSE の採用が進む
• 2023 年以降：LLM ストリーミングの標準として SSE が定着（OpenAI、Anthropic）
• 2024 年：Vercel AI SDK が SSE のパターンを普及させる

ストリーミング API

プロバイダー	ストリーミング方式	レスポンス形式
OpenAI	SSE	data: {"choices":[{"delta":{"content":"token"}}]}
Anthropic	SSE	data: {"type":"content_block_delta","delta":{"text":"token"}}
Claude (API)	SSE	data: {"delta":{"text":"token"}}
Vercel AI SDK	SSE	プロバイダー間で正規化

---

よくあるアンチパターン

アンチパターン 1：表示前にバッファリングする

初心者の考え方：「すべてのトークンを収集してから、レスポンス全体を表示する」

問題：ストリーミングの目的そのものを損ないます。

誤った方法：

// ❌ Waits for entire response before showing anything
const response = await fetch('/api/chat', { method: 'POST', body: prompt });
const fullText = await response.text();
setMessage(fullText); // User sees nothing until done

正しい方法：

// ✅ Display tokens as they arrive
const response = await fetch('/api/chat', {
  method: 'POST',
  body: JSON.stringify({ prompt })
});

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

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

  const chunk = decoder.decode(value);
  const lines = chunk.split('\n').filter(line => line.trim());

  for (const line of lines) {
    if (line.startsWith('data: ')) {
      const data = JSON.parse(line.slice(6));
      setMessage(prev => prev + data.content); // Update immediately
    }
  }
}

年表：

• 2023 年以前：多くのアプリがレスポンス全体をバッファリング
• 2023 年以降：トークンごとの表示が期待されるようになる

---

アンチパターン 2：ストリームをキャンセルできない

問題：ユーザーが生成を停止できず、トークンと費用が無駄になります。

症状：「停止」ボタンが機能しない、または存在しません。

正しい方法：

// ✅ AbortController for cancellation
const [abortController, setAbortController] = useState<AbortController | null>(null);

const streamResponse = async () => {
  const controller = new AbortController();
  setAbortController(controller);

  try {
    const response = await fetch('/api/chat', {
      signal: controller.signal,
      method: 'POST',
      body: JSON.stringify({ prompt })
    });

    // Stream handling...
  } catch (error) {
    if (error.name === 'AbortError') {
      console.log('Stream cancelled by user');
    }
  } finally {
    setAbortController(null);
  }
};

const cancelStream = () => {
  abortController?.abort();
};

return (
  <button onClick={cancelStream} disabled={!abortController}>
    Stop Generating
  </button>
);

---

アンチパターン 3：エラーから復旧できない

問題：レスポンスの途中でストリームが失敗すると、ユーザーには失敗を知らせる表示もなく、テキストの一部だけが表示されます。

正しい方法：

// ✅ Error states and recovery
const [streamState, setStreamState] = useState<'idle' | 'streaming' | 'error' | 'complete'>('idle');
const [errorMessage, setErrorMessage] = useState<string | null>(null);

try {
  setStreamState('streaming');

  // Streaming logic...

  setStreamState('complete');
} catch (error) {
  setStreamState('error');

  if (error.name === 'AbortError') {
    setErrorMessage('Generation stopped');
  } else if (error.message.includes('429')) {
    setErrorMessage('Rate limit exceeded. Try again in a moment.');
  } else {
    setErrorMessage('Something went wrong. Please retry.');
  }
}

// UI feedback
{streamState === 'error' && (
  <div className="error-banner">
    {errorMessage}
    <button onClick={retryStream}>Retry</button>
  </div>
)}

---

アンチパターン 4：ストリームを閉じないことによるメモリリーク

問題：ストリームがクリーンアップされず、メモリリークが発生します。

症状：複数回リクエストした後、ブラウザーの動作が遅くなります。

正しい方法：

// ✅ Cleanup with useEffect
useEffect(() => {
  let reader: ReadableStreamDefaultReader | null = null;

  const streamResponse = async () => {
    const response = await fetch('/api/chat', { ... });
    reader = response.body.getReader();

    // Streaming...
  };

  streamResponse();

  // Cleanup on unmount
  return () => {
    reader?.cancel();
  };
}, [prompt]);

---

アンチパターン 5：トークン間にタイピングインジケーターがない

問題：トークンの出力が遅いと、UI がフリーズしたように感じられます。

正しい方法：

// ✅ Animated cursor during generation
<div className="message">
  {content}
  {isStreaming && <span className="typing-cursor">▊</span>}
</div>

.typing-cursor {
  animation: blink 1s step-end infinite;
}

@keyframes blink {
  50% { opacity: 0; }
}

---

実装パターン

パターン 1：基本的な SSE ストリームハンドラー

async function* streamCompletion(prompt: string) {
  const response = await fetch('/api/chat', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ prompt })
  });

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

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

    const chunk = decoder.decode(value);
    const lines = chunk.split('\n');

    for (const line of lines) {
      if (line.startsWith('data: ')) {
        const data = JSON.parse(line.slice(6));

        if (data.content) {
          yield data.content;
        }

        if (data.done) {
          return;
        }
      }
    }
  }
}

// Usage
for await (const token of streamCompletion('Hello')) {
  console.log(token);
}

パターン 2：ストリーミング用 React フック

import { useState, useCallback } from 'react';

interface UseStreamingOptions {
  onToken?: (token: string) => void;
  onComplete?: (fullText: string) => void;
  onError?: (error: Error) => void;
}

export function useStreaming(options: UseStreamingOptions = {}) {
  const [content, setContent] = useState('');
  const [isStreaming, setIsStreaming] = useState(false);
  const [error, setError] = useState<Error | null>(null);
  const [abortController, setAbortController] = useState<AbortController | null>(null);

  const stream = useCallback(async (prompt: string) => {
    const controller = new AbortController();
    setAbortController(controller);
    setIsStreaming(true);
    setError(null);
    setContent('');

    try {
      const response = await fetch('/api/chat', {
        method: 'POST',
        signal: controller.signal,
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ prompt })
      });

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

      let accumulated = '';

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

        const chunk = decoder.decode(value);
        const lines = chunk.split('\n').filter(line => line.trim());

        for (const line of lines) {
          if (line.startsWith('data: ')) {
            const data = JSON.parse(line.slice(6));

            if (data.content) {
              accumulated += data.content;
              setContent(accumulated);
              options.onToken?.(data.content);
            }
          }
        }
      }

      options.onComplete?.(accumulated);
    } catch (err) {
      if (err.name !== 'AbortError') {
        setError(err as Error);
        options.onError?.(err as Error);
      }
    } finally {
      setIsStreaming(false);
      setAbortController(null);
    }
  }, [options]);

  const cancel = useCallback(() => {
    abortController?.abort();
  }, [abortController]);

  return { content, isStreaming, error, stream, cancel };
}

// Usage in component
function ChatInterface() {
  const { content, isStreaming, stream, cancel } = useStreaming({
    onToken: (token) => console.log('New token:', token),
    onComplete: (text) => console.log('Done:', text)
  });

  return (
    <div>
      <div className="message">
        {content}
        {isStreaming && <span className="cursor">▊</span>}
      </div>

      <button onClick={() => stream('Tell me a story')} disabled={isStreaming}>
        Generate
      </button>

      {isStreaming && <button onClick={cancel}>Stop</button>}
    </div>
  );
}

パターン 3：サーバーサイドストリーミング（Next.js）

// app/api/chat/route.ts
import { OpenAI } from 'openai';

export const runtime = 'edge'; // Required for streaming

export async function POST(req: Request) {
  const { prompt } = await req.json();

  const openai = new OpenAI({
    apiKey: process.env.OPENAI_API_KEY
  });

  const stream = await openai.chat.completions.create({
    model: 'gpt-4',
    messages: [{ role: 'user', content: prompt }],
    stream: true
  });

  // Convert OpenAI stream to SSE format
  const encoder = new TextEncoder();

  const readable = new ReadableStream({
    async start(controller) {
      try {
        for await (const chunk of stream) {
          const content = chunk.choices[0]?.delta?.content;

          if (content) {
            const sseMessage = `data: ${JSON.stringify({ content })}\n\n`;
            controller.enqueue(encoder.encode(sseMessage));
          }
        }

        // Send completion signal
        controller.enqueue(encoder.encode('data: {"done":true}\n\n'));
        controller.close();
      } catch (error) {
        controller.error(error);
      }
    }
  });

  return new Response(readable, {
    headers: {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
      'Connection': 'keep-alive'
    }
  });
}

---

本番環境向けチェックリスト

□ AbortController for cancellation
□ Error states with retry capability
□ Typing indicator during generation
□ Cleanup on component unmount
□ Rate limiting on API route
□ Token usage tracking
□ Streaming fallback (if API fails)
□ Accessibility (screen reader announces updates)
□ Mobile-friendly (touch targets for stop button)
□ Network error recovery (auto-retry on disconnect)
□ Max response length enforcement
□ Cost estimation before generation

---

使用する場合と避ける場合

シナリオ	ストリーミングを使用する？
チャットインターフェース	✅ はい
長文コンテンツの生成	✅ はい
プレビュー付きコード生成	✅ はい
短い補完（<50 words）	❌ いいえ - 通常の fetch を使用
バックグラウンドジョブ	❌ いいえ - ジョブキューを使用
双方向チャット	⚠️ 代わりに WebSocket を使用

---

技術の比較

機能	SSE	WebSockets	Long Polling
複雑さ	低	中	高
自動再接続	✅	❌	❌
双方向通信	❌	✅	❌
ファイアウォールとの親和性	✅	⚠️	✅
ブラウザーのサポート	✅ すべてのモダンブラウザー	✅ すべてのモダンブラウザー	✅ すべてのブラウザー
LLM API のサポート	✅ 標準	❌ まれ	❌ 使用されない

---

参考資料

• /references/sse-protocol.md - Server-Sent Events 仕様の詳細
• /references/vercel-ai-sdk.md - Vercel AI SDK の統合パターン
• /references/error-recovery.md - ストリームのエラー処理戦略

スクリプト

• scripts/stream_tester.ts - SSE エンドポイントをローカルでテストします
• scripts/token_counter.ts - 生成前にコストを見積もります

---

このスキルの対象: LLM のストリーミング実装 | SSE プロトコル | リアルタイム UI 更新 | キャンセル | エラーからの復旧 | トークンごとの表示