RAGチャットに「今何してるか」が見えるSSE進捗バブルを実装した
はじめに
RAG（Retrieval-Augmented Generation）ベースのチャットUIを運用していると、ユーザーの問い合わせ送信から回答表示まで数秒〜十数秒の待ち時間が発生します。この間、単なるローディングスピナーだけでは「本当に動いているのか？」「あとどれくらいかかるのか？」がわかりません。
そこで、バックエンドの処理パイプラインの各ステップをリアルタイムに可視化する進捗ステータスバブルを実装しました。
この記事では、FastAPI + Next.js構成のRAGチャットアプリケーションにSSE（Server-Sent Events）ベースの進捗表示を段階的に構築した過程を紹介します。
完成イメージ
進捗バブルは以下の4ステップをリアルタイムに表示します：
- クエリを解析中 — ユーザーの問い合わせを分析
- ナレッジベースを検索中 — KB検索クエリの生成と実行
- チャンクを取得中 — 検索結果のチャンクを取得・処理
- 回答を生成中 — LLMによる回答生成（トークンストリーミング）
各ステップは pending → active（シマーアニメーション付き） → done（チェックマーク）と遷移します。バブルはクリックで展開でき、実際に使われた検索クエリや取得チャンクの詳細（タイトル・スコア・プレビュー）も確認できます。
前提・環境
| 項目 | バージョン |
|---|---|
| Next.js | 16 |
| React | 19 |
| FastAPI | 0.115+ |
| Amazon Bedrock (Claude) | Sonnet 4.5 |
| Python | 3.13 |
| Node.js | 24 LTS |
アーキテクチャは以下の通りです：
Step 1: バックエンドのSSEイベント設計
SSEイベントの種類を定義する
まず、フロントエンドに送るSSEイベントの種類を設計します。進捗表示だけでなく、トークンストリーミングやエラー通知も含めた統一的なイベント体系を作りました。
# SSEイベントレジストリ — 全イベントの定義を一箇所に集約
SSE_EVENTS: dict[str, tuple[type, str]] = {
"progress": (SseProgressData, "処理ステップの進捗"),
"kb_query": (SseKbQueryData, "KB検索に使用したクエリ文字列"),
"kb_chunks": (SseKbChunksData, "KBから取得したチャンクのプレビュー一覧"),
"token": (SseTokenData, "AIが生成するテキストチャンク"),
"result": (ChatResponse, "最終結果"),
"error": (SseErrorData, "エラー発生時の詳細"),
}
このレジストリは Single Source of Truth として機能し、OpenAPI仕様書の x-sse-events セクションも自動生成されます。新しいイベントを追加する場合、ここに1行足すだけで済みます。
Pydanticモデルでイベントデータを型定義する
各SSEイベントのペイロードをPydanticモデルで定義します：
from pydantic import BaseModel, Field
from typing import Literal
class SseProgressData(BaseModel):
step: Literal["analyzing", "retrieving", "fetching", "generating"]
class SseKbQueryData(BaseModel):
query: str
class SseKbChunkPreview(BaseModel):
title: str = Field(..., description="ソースドキュメントのタイトル")
score: float = Field(..., description="関連スコア（0-1）")
preview: str = Field(..., description="チャンク内容の先頭120文字")
class SseKbChunksData(BaseModel):
chunks: list[SseKbChunkPreview]
class SseErrorData(BaseModel):
detail: str
Literal 型でステップ名を制約することで、バックエンドとフロントエンドの間で不正なステップ名が流れるのを防ぎます。
ストリームジェネレーターでイベントを送出する
FastAPIの StreamingResponse を使い、処理パイプラインの各ポイントでSSEイベントを送出します：
def _sse(event: str, data: object) -> str:
return f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n"
async def _chat_stream(request: ChatRequest) -> AsyncGenerator[str]:
try:
# Step 1: 解析開始
yield _sse("progress", {"step": "analyzing"})
# Step 2: KBクエリ生成と問題分類を並行実行
yield _sse("progress", {"step": "fetching"})
kb_query, problem_ctx = await asyncio.gather(
generate_kb_query(request.messages),
classify_problem(request.messages[0].content),
)
# Step 3: KB検索実行
yield _sse("progress", {"step": "retrieving"})
chunks = await retrieve_from_kb(kb_query, top_k=request.top_k)
# 検索メタデータを送出（展開表示用）
yield _sse("kb_query", {"query": kb_query})
yield _sse("kb_chunks", {"chunks": _make_chunk_previews(chunks)})
# Step 4: LLM回答生成（トークンストリーミング）
yield _sse("progress", {"step": "generating"})
async for token_or_result in generate_chat_answer(request.messages, chunks):
if isinstance(token_or_result, str):
yield _sse("token", {"content": token_or_result})
else:
yield _sse("result", token_or_result)
except Exception as e:
yield _sse("error", {"detail": str(e)})
ポイントは asyncio.gather によるKBクエリ生成と問題分類の並行実行です。これにより、2つのLLM呼び出しのレイテンシを重ねずに済みます。
Step 2: フロントエンドの型定義
バックエンドのSSEイベントに対応するTypeScript型を定義します：
export type KbChunk = {
title: string;
score: number;
preview: string;
};
export type ProgressStep = {
id: "analyzing" | "retrieving" | "fetching" | "generating";
label: string;
status: "pending" | "active" | "done";
kbQuery?: string;
kbChunks?: KbChunk[];
reasoning?: string;
};
export type ChatBubble =
| { type: "user"; text: string }
| { type: "streaming"; general_advice: string; queryId?: number }
| { type: "progress"; steps: ProgressStep[]; collapsed: boolean; queryId?: number }
| { type: "progress-done"; steps: ProgressStep[]; collapsed: boolean; queryId?: number }
| { type: "results"; results: SearchResult[]; general_advice: string; /* ... */ }
| { type: "error"; text: string };
ChatBubble はUnion型で、チャットスレッド内のあらゆるバブル種別を表現します。progress と progress-done を分けているのは、後述するマルチターン分離のためです。
Step 3: SSEストリームのパース処理
ReadableStreamでSSEを受信する
fetch API の ReadableStream を使い、SSEイベントを1つずつパースしてReactの状態を更新します：
async function handleSend(text: string) {
const queryId = ++queryIdRef.current;
const STEP_LABELS: Record<ProgressStep["id"], string> = {
analyzing: "クエリを解析中",
fetching: "チャンクを取得中",
retrieving: "ナレッジベースを検索中",
generating: "回答を生成中",
};
function makeStep(id: ProgressStep["id"]): ProgressStep {
return { id, label: STEP_LABELS[id], status: "active" };
}
// 進捗バブルを初期配置
setBubbles((prev) => [
...prev,
{ type: "user", text },
{ type: "progress", steps: [], collapsed: false, queryId },
]);
const res = await fetch("/api/chat", { method: "POST", body: JSON.stringify({ messages }) });
const reader = res.body!.getReader();
const decoder = new TextDecoder();
// ... SSEパースループ
}
queryIdRef はクロージャで各ターンのIDを捕捉します。これにより、2回目の問い合わせのSSEイベントが1回目の進捗バブルを書き換えてしまう問題を防ぎます。
イベントハンドラーの実装
各SSEイベント種別に対応するハンドラーを実装します：
if (eventType === "progress") {
const newStepId = data.step as ProgressStep["id"];
setBubbles((prev) =>
prev.map((b) => {
if (b.type !== "progress" || b.queryId !== queryId) return b;
// 現在activeなステップをdoneに、新ステップをactiveに
const updated = b.steps.map((s) =>
s.status === "active" ? { ...s, status: "done" as const } : s,
);
return { ...b, steps: [...updated, makeStep(newStepId)] };
}),
);
} else if (eventType === "kb_query") {
setBubbles((prev) =>
prev.map((b) => {
if (b.type !== "progress" || b.queryId !== queryId) return b;
return {
...b,
steps: b.steps.map((s) =>
s.id === "retrieving" ? { ...s, kbQuery: data.query } : s
),
};
}),
);
} else if (eventType === "kb_chunks") {
setBubbles((prev) =>
prev.map((b) => {
if (b.type !== "progress" || b.queryId !== queryId) return b;
return {
...b,
steps: b.steps.map((s) =>
s.id === "retrieving" ? { ...s, kbChunks: data.chunks } : s
),
};
}),
);
}
kb_query と kb_chunks イベントは、retrieving ステップのオブジェクト内にメタデータを注入します。これにより、ステップを展開したときに「どんな検索クエリが使われ、どんなチャンクが取得されたか」を確認できます。
バブルの状態遷移
SSEストリームの最後に result イベントが届くと、進捗バブルをファイナライズします：
// result イベント受信時
setBubbles((prev) => {
// progress → progress-done に遷移（全ステップをdoneに）
const withProgress = prev.map((b) => {
if (b.type !== "progress" || b.queryId !== queryId) return b;
return {
...b,
type: "progress-done" as const,
collapsed: true,
steps: b.steps
.map((s) => s.status === "active" ? { ...s, status: "done" as const } : s)
.map((s) => s.id === "generating" ? { ...s, reasoning } : s),
};
});
// streamingバブルを除去し、resultsバブルに統合
return withProgress.filter(
(b) => !(b.type === "streaming" && b.queryId === queryId),
);
});
progress → progress-done への型遷移がここでのポイントです。progress-done に変わることで、以降のSSEイベントハンドラーが b.type !== "progress" のガードで自動的にスキップされます。
Step 4: UIコンポーネントの実装
ProgressTimeline — 折りたたみ可能なタイムライン
進捗バブル全体を管理するコンポーネントです：
function ProgressTimeline({ bubble }: {
bubble: Extract<ChatBubble, { type: "progress" | "progress-done" }>;
}) {
const [expanded, setExpanded] = useState(false);
const activeStep = [...bubble.steps].reverse().find((s) => s.status === "active");
const doneCount = bubble.steps.filter((s) => s.status === "done").length;
const allDone = bubble.collapsed;
return (
<div className="rounded-lg border border-gray-100 bg-gray-50">
{/* 常に表示される1行サマリー */}
<button onClick={() => setExpanded((v) => !v)} className="flex w-full items-center gap-2 px-4 py-2.5">
{/* ステータスアイコン */}
{allDone ? (
<CheckIcon className="text-emerald-500" />
) : (
<span className="animate-spin rounded-full border-2 border-blue-400 border-t-transparent" />
)}
{/* 現在のステップ名 or 完了メッセージ */}
<span className={allDone ? "text-gray-500" : "shimmer-text font-medium"}>
{allDone ? `${doneCount} ステップ完了` : (activeStep?.label ?? "...")}
</span>
</button>
{/* 展開時のステップリスト */}
{expanded && (
<ol className="space-y-3">
{bubble.steps.map((step) => (
<ProgressStepRow key={step.id} step={step} expanded={expanded} />
))}
</ol>
)}
</div>
);
}
折りたたみ時は1行のサマリー表示（アクティブなステップ名 or 完了数）のみ。展開すると全ステップの詳細が見えるデザインです。
ProgressStepRow — 個別ステップの表示
function ProgressStepRow({ step, expanded }: {
step: ProgressStep;
expanded: boolean;
}) {
return (
<li className="animate-fade-in-up">
<div className="flex items-center gap-2">
{/* done=チェック, active=スピナー, pending=空丸 */}
{step.status === "done" ? (
<CheckIcon className="text-emerald-500" />
) : step.status === "active" ? (
<span className="animate-spin rounded-full border-2 border-blue-400 border-t-transparent" />
) : (
<span className="rounded-full border border-gray-300" />
)}
<span className={step.status === "active" ? "shimmer-text font-medium" : "text-gray-600"}>
{step.label}
</span>
</div>
{/* 展開時: 検索クエリ・取得チャンク・推論の詳細 */}
{expanded && (
<div className="mt-2 ml-5 space-y-2 text-xs">
{step.kbQuery && (
<div>
<span className="text-gray-400">検索クエリ:</span>
<span className="rounded bg-gray-100 px-1.5 py-0.5 font-mono">{step.kbQuery}</span>
</div>
)}
{step.kbChunks?.map((chunk, idx) => (
<div key={idx} className="flex gap-2 rounded border bg-gray-50 px-2 py-1.5">
<span className="font-medium text-blue-600">#{chunk.title}</span>
<span className="text-gray-400">({chunk.score.toFixed(3)})</span>
<span className="truncate text-gray-500">{chunk.preview}</span>
</div>
))}
</div>
)}
</li>
);
}
Step 5: CSSアニメーション
進捗バブルの見た目を仕上げるために、2つのアニメーションを定義しました。
フェードイン + スライドアップ
新しいステップが追加されたときの入場アニメーションです：
@keyframes fadeInUp {
from {
opacity: 0;
transform: translateY(8px);
}
to {
opacity: 1;
transform: translateY(0);
}
}
.animate-fade-in-up {
animation: fadeInUp 300ms ease-out;
}
シマーテキスト
アクティブなステップのラベルに流れるような光沢アニメーションを適用します：
@keyframes shimmer-sweep {
0% { background-position: -200% center; }
100% { background-position: 200% center; }
}
.shimmer-text {
background: linear-gradient(90deg, #374151 35%, #93c5fd 50%, #374151 65%);
background-size: 300% auto;
-webkit-background-clip: text;
-webkit-text-fill-color: transparent;
background-clip: text;
animation: shimmer-sweep 1.8s linear infinite;
}
このシマー効果により、アクティブなステップが静的なテキストと明確に区別され、「処理中」であることが直感的に伝わります。
苦労した点: マルチターンでの状態分離
問題
実装初期は progress バブルの型が1種類だけでした。ユーザーが2回目の問い合わせを送ると、2回目のSSEイベントハンドラーが b.type === "progress" で1回目のバブルもマッチしてしまい、1回目の完了済みバブルのステップが上書きされるバグが発生しました。
解決: queryId + progress-done型
2つのメカニズムで解決しました：
1. queryIdによるスコーピング
各問い合わせにインクリメンタルなIDを付与し、SSEハンドラー内で b.queryId !== queryId のガードを入れます：
const queryId = ++queryIdRef.current;
// すべてのsetBubbles内で:
if (b.type !== "progress" || b.queryId !== queryId) return b;
2. progress → progress-done 型遷移
result イベント受信時に progress を progress-done に書き換えます。以降のSSEハンドラーは b.type !== "progress" で自動スキップされるため、完了済みバブルが不変であることを型レベルで保証できます。
// ChatBubble union型の抜粋
| { type: "progress"; steps: ProgressStep[]; collapsed: boolean; queryId?: number }
| { type: "progress-done"; steps: ProgressStep[]; collapsed: boolean; queryId?: number }
この二重ガードにより、何回問い合わせを重ねても各ターンの進捗バブルが独立して動作します。
データフロー全体図
最終的なデータフローをまとめます：
まとめ
SSE進捗バブルの実装を通じて得られた知見をまとめます。
タイマーではなくリアルイベントで駆動する: setTimeout で疑似的な進捗を作るのではなく、バックエンドの実際の処理完了タイミングでイベントを送出します。これにより、処理が速いときは一瞬で全ステップが完了し、遅いときはアクティブなステップで待つ、という自然な挙動になります。
SSEイベントレジストリで一元管理する: イベント定義を SSE_EVENTS dictに集約することで、バックエンドのコード・Pydanticモデル・OpenAPI仕様書が常に同期します。
Union型でバブル遷移を型安全にする: progress → progress-done → results といった状態遷移をTypeScriptのUnion型で表現することで、各SSEハンドラーが正しい型のバブルだけを操作することをコンパイル時に保証できます。
queryIdで複数ターンを分離する: クロージャでキャプチャしたIDとバブルの型の二重ガードにより、マルチターン会話でも各ターンの進捗バブルが独立して動作します。
バックエンドの処理パイプラインが複雑になるほど、こうした進捗の可視化はユーザー体験に大きな差を生みます。RAGアプリケーションに限らず、複数ステップの非同期処理をUIに反映する場合の参考になれば幸いです。
