ストーリー

Server-Sent Events (SSE)

SSEの仕組み

sequenceDiagram
    participant Client as クライアント
    participant Server as サーバー

    Client->>Server: POST /api/chat
(Accept: text/event-stream)
    Note right of Server: LLMストリーミング開始
    Server-->>Client: data: type=start
    Server-->>Client: data: type=token, text=こ
    Server-->>Client: data: type=token, text=ん
    Server-->>Client: data: type=token, text=に
    Note over Client,Server: ...（トークンが逐次送信される）
    Server-->>Client: data: type=done, usage={}

サーバー側の実装

import { Hono } from 'hono';
import { streamSSE } from 'hono/streaming';

const app = new Hono();

// ストリーミングイベントの型定義
type StreamEvent =
  | { type: 'start'; requestId: string }
  | { type: 'token'; text: string }
  | { type: 'source'; sources: { id: number; title: string }[] }
  | { type: 'done'; usage: { inputTokens: number; outputTokens: number } }
  | { type: 'error'; message: string; code: string };

app.post('/api/chat', async (c) => {
  const { query } = await c.req.json<{ query: string }>();
  const requestId = crypto.randomUUID();

  return streamSSE(c, async (stream) => {
    try {
      // 1. 開始イベント
      await stream.writeSSE({
        data: JSON.stringify({ type: 'start', requestId }),
      });

      // 2. 検索実行（ストリーミング前に完了）
      const contexts = await ragPipeline.retrieve(query);

      // 出典情報を先行送信
      await stream.writeSSE({
        data: JSON.stringify({
          type: 'source',
          sources: contexts.map((c, i) => ({
            id: i + 1,
            title: c.chunk.metadata.documentTitle,
          })),
        }),
      });

      // 3. LLMストリーミング生成
      const messages = buildPrompt(query, contexts);
      const tokenStream = llmProvider.stream(messages, {
        temperature: 0,
        maxTokens: 1000,
      });

      let totalOutputTokens = 0;
      for await (const chunk of tokenStream) {
        if (chunk.type === 'content') {
          totalOutputTokens++;
          await stream.writeSSE({
            data: JSON.stringify({ type: 'token', text: chunk.content }),
          });
        }
      }

      // 4. 完了イベント
      await stream.writeSSE({
        data: JSON.stringify({
          type: 'done',
          usage: {
            inputTokens: messages.reduce(
              (sum, m) => sum + llmProvider.countTokens(m.content), 0,
            ),
            outputTokens: totalOutputTokens,
          },
        }),
      });
    } catch (error) {
      // エラーイベント
      await stream.writeSSE({
        data: JSON.stringify({
          type: 'error',
          message: 'サーバーエラーが発生しました。再度お試しください。',
          code: 'STREAM_ERROR',
        }),
      });
    }
  });
});

クライアント側の実装

EventSourceベースの受信

class AIStreamClient {
  private abortController: AbortController | null = null;

  async streamChat(
    query: string,
    callbacks: {
      onStart?: (requestId: string) => void;
      onToken?: (text: string) => void;
      onSource?: (sources: { id: number; title: string }[]) => void;
      onDone?: (usage: { inputTokens: number; outputTokens: number }) => void;
      onError?: (error: { message: string; code: string }) => void;
    },
  ): Promise<void> {
    this.abortController = new AbortController();

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

    if (!response.ok) {
      callbacks.onError?.({
        message: `HTTP ${response.status}`,
        code: 'HTTP_ERROR',
      });
      return;
    }

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

    if (!reader) return;

    let buffer = '';

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

      buffer += decoder.decode(value, { stream: true });
      const lines = buffer.split('\n');
      buffer = lines.pop() ?? '';

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

          switch (data.type) {
            case 'start':
              callbacks.onStart?.(data.requestId);
              break;
            case 'token':
              callbacks.onToken?.(data.text);
              break;
            case 'source':
              callbacks.onSource?.(data.sources);
              break;
            case 'done':
              callbacks.onDone?.(data.usage);
              break;
            case 'error':
              callbacks.onError?.(data);
              break;
          }
        }
      }
    }
  }

  cancel(): void {
    this.abortController?.abort();
    this.abortController = null;
  }
}

Progressive Rendering

Reactコンポーネントでのストリーミング表示

import { useState, useCallback, useRef } from 'react';

interface ChatMessage {
  id: string;
  role: 'user' | 'assistant';
  content: string;
  sources?: { id: number; title: string }[];
  isStreaming: boolean;
}

function useStreamingChat() {
  const [messages, setMessages] = useState<ChatMessage[]>([]);
  const [isLoading, setIsLoading] = useState(false);
  const clientRef = useRef(new AIStreamClient());

  const sendMessage = useCallback(async (query: string) => {
    // ユーザーメッセージを追加
    const userMessage: ChatMessage = {
      id: crypto.randomUUID(),
      role: 'user',
      content: query,
      isStreaming: false,
    };

    const assistantId = crypto.randomUUID();
    const assistantMessage: ChatMessage = {
      id: assistantId,
      role: 'assistant',
      content: '',
      isStreaming: true,
    };

    setMessages(prev => [...prev, userMessage, assistantMessage]);
    setIsLoading(true);

    await clientRef.current.streamChat(query, {
      onToken: (text) => {
        setMessages(prev =>
          prev.map(m =>
            m.id === assistantId
              ? { ...m, content: m.content + text }
              : m,
          ),
        );
      },

      onSource: (sources) => {
        setMessages(prev =>
          prev.map(m =>
            m.id === assistantId
              ? { ...m, sources }
              : m,
          ),
        );
      },

      onDone: () => {
        setMessages(prev =>
          prev.map(m =>
            m.id === assistantId
              ? { ...m, isStreaming: false }
              : m,
          ),
        );
        setIsLoading(false);
      },

      onError: (error) => {
        setMessages(prev =>
          prev.map(m =>
            m.id === assistantId
              ? {
                  ...m,
                  content: `エラーが発生しました: ${error.message}`,
                  isStreaming: false,
                }
              : m,
          ),
        );
        setIsLoading(false);
      },
    });
  }, []);

  const cancel = useCallback(() => {
    clientRef.current.cancel();
    setIsLoading(false);
    setMessages(prev =>
      prev.map(m => m.isStreaming ? { ...m, isStreaming: false } : m),
    );
  }, []);

  return { messages, isLoading, sendMessage, cancel };
}

ストリーミング中のUI表示

function StreamingMessage({ message }: { message: ChatMessage }) {
  return (
    <div className="flex flex-col gap-2">
      {/* メッセージ本文 */}
      <div className="prose prose-sm max-w-none">
        <ReactMarkdown>{message.content}</ReactMarkdown>
        {message.isStreaming && (
          <span className="inline-block w-2 h-4 bg-blue-500 animate-pulse ml-1" />
        )}
      </div>

      {/* 出典（ストリーミング中でも表示） */}
      {message.sources && message.sources.length > 0 && (
        <div className="flex flex-wrap gap-1 mt-2">
          {message.sources.map(source => (
            <span
              key={source.id}
              className="text-xs bg-gray-100 rounded px-2 py-1"
            >
              [{source.id}] {source.title}
            </span>
          ))}
        </div>
      )}
    </div>
  );
}

UX最適化のポイント

TTFB の最適化

ローディング状態の段階表示

type LoadingPhase =
  | 'thinking'    // 検索中
  | 'searching'   // ベクトル検索実行中
  | 'generating'  // LLM生成中
  | 'complete';   // 完了

function LoadingIndicator({ phase }: { phase: LoadingPhase }) {
  const phaseLabels: Record<LoadingPhase, string> = {
    thinking: '質問を分析しています...',
    searching: 'ナレッジベースを検索しています...',
    generating: '回答を生成しています...',
    complete: '',
  };

  if (phase === 'complete') return null;

  return (
    <div className="flex items-center gap-2 text-gray-500 text-sm">
      <div className="flex gap-1">
        <span className="w-1.5 h-1.5 bg-blue-500 rounded-full animate-bounce" />
        <span className="w-1.5 h-1.5 bg-blue-500 rounded-full animate-bounce delay-100" />
        <span className="w-1.5 h-1.5 bg-blue-500 rounded-full animate-bounce delay-200" />
      </div>
      <span>{phaseLabels[phase]}</span>
    </div>
  );
}

キャンセル機能の重要性

// ストリーミング中のキャンセル対応
function ChatInput({
  onSend,
  onCancel,
  isLoading,
}: {
  onSend: (query: string) => void;
  onCancel: () => void;
  isLoading: boolean;
}) {
  const [input, setInput] = useState('');

  return (
    <div className="flex gap-2">
      <input
        value={input}
        onChange={(e) => setInput(e.target.value)}
        placeholder="質問を入力してください"
        disabled={isLoading}
        className="flex-1 rounded border px-3 py-2"
        onKeyDown={(e) => {
          if (e.key === 'Enter' && !e.shiftKey && !isLoading) {
            onSend(input);
            setInput('');
          }
        }}
      />
      {isLoading ? (
        <button
          onClick={onCancel}
          className="rounded bg-red-500 px-4 py-2 text-white"
        >
          停止
        </button>
      ) : (
        <button
          onClick={() => { onSend(input); setInput(''); }}
          disabled={!input.trim()}
          className="rounded bg-blue-500 px-4 py-2 text-white disabled:opacity-50"
        >
          送信
        </button>
      )}
    </div>
  );
}

まとめ

チェックリスト

• SSEの仕組みとサーバー実装を理解した
• クライアント側のストリーミング受信と表示を理解した
• Reactでのストリーミング状態管理を理解した
• TTFB最適化とUXパターンを理解した

次のステップへ

ストリーミングとUXの最適化を学びました。次のセクションでは、LLMシステムのコスト最適化について学びます。

ユーザーが感じる速度はTTFBで決まる。合計レイテンシが同じでも、ストリーミングでフィードバックを即座に返すことが最良のUXだ。

推定読了時間: 40分