AIのストリーミングレスポンスで「累積型」と「差分型」を正しく扱う — SSEクライアント実装の実践
はじめに
生成AIのストリーミングレスポンスをブラウザに表示する機能を実装したとき、テキストが二重に表示される問題が発生しました。
原因を調べると、使用していたAI APIのSSE（Server-Sent Events）が累積型——つまり各イベントに「これまでの全テキスト」を含むフォーマットだったのです。差分（デルタ）だけを送ってくるAPIと同じ処理をしていたため、毎回全文が追加されてしまっていました。
この記事では、累積型SSEレスポンスから差分を正しく抽出する方法と、ブラウザでのSSEパースの実装を紹介します。
前提・環境
- TypeScript / SvelteKit
- Server-Sent Events (SSE)
- ReadableStream API（Fetch API）
累積型 vs 差分型
AI APIのストリーミングには大きく2つのフォーマットがあります。
差分型（Delta） — OpenAI ChatCompletions APIなど：
event: data
data: {"choices": [{"delta": {"content": "こん"}}]}
event: data
data: {"choices": [{"delta": {"content": "にち"}}]}
event: data
data: {"choices": [{"delta": {"content": "は"}}]}
各イベントに新しく生成された部分だけが含まれます。クライアントは受け取った内容をそのまま追加すればOKです。
累積型（Cumulative） — 一部のエンタープライズAI APIなど：
event: data
data: [{"type": "ai", "content": "こん"}]
event: data
data: [{"type": "ai", "content": "こんにち"}]
event: data
data: [{"type": "ai", "content": "こんにちは"}]
各イベントにこれまでに生成された全テキストが含まれます。クライアントは前回との差分を自分で計算する必要があります。
差分型のつもりで累積型を処理すると：
表示結果: こん → こんこんにち → こんこんにちこんにちは
期待結果: こん → こんにち → こんにちは
差分抽出の実装
コアロジック
let fullContent = "";
// SSEイベントを処理
for (const message of parsedData) {
if (message.type === "ai" && message.content) {
const newContent = message.content;
// 差分 = 今回の全文 - 前回までの全文
const incrementalContent = newContent.substring(fullContent.length);
// 全文を更新
fullContent = newContent;
// UIには差分だけを送る
if (incrementalContent) {
onStream?.({
content: incrementalContent,
isIncremental: true
});
}
}
}
substring(fullContent.length) が差分抽出のキモです。累積型レスポンスでは、前回のテキストが今回のテキストのプレフィックスになっているため、前回の長さ以降を切り出せば差分が得られます。
SSEストリーム全体の処理
ブラウザでSSEを処理する完全な実装です。
static async generateResponse(
prompt: string,
name: string,
instruction: string,
webSearch: boolean,
onStream?: StreamCallback
): Promise<GptResponse> {
const response = await fetch("/api/gpt/generate", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ prompt, name, instruction, webSearch })
});
if (!response.body) {
throw new Error("No response body received");
}
const reader = response.body.getReader();
const decoder = new TextDecoder();
let fullContent = "";
try {
while (true) {
const { value, done } = await reader.read();
if (done) {
onStream?.({ content: fullContent, done: true });
break;
}
const chunk = decoder.decode(value);
// チャンクを個別のSSEイベントに分割
const events = chunk.split("\n\n").filter((event) => event.trim() !== "");
for (const event of events) {
const lines = event.split("\n");
let eventType = "";
let eventData = "";
for (const line of lines) {
if (line.startsWith("event: ")) {
eventType = line.substring(7).trim();
} else if (line.startsWith("data: ")) {
eventData = line.substring(6).trim();
}
}
// metadataやendイベントはスキップ
if (eventType === "metadata" || eventType === "end") continue;
if (eventType === "data" && eventData) {
try {
const parsedData = JSON.parse(eventData);
if (Array.isArray(parsedData)) {
for (const message of parsedData) {
if (message.type === "ai" && message.content) {
const newContent = message.content;
const incrementalContent = newContent.substring(fullContent.length);
fullContent = newContent;
if (incrementalContent) {
onStream?.({ content: incrementalContent, isIncremental: true });
}
}
}
}
} catch (e) {
console.error("Error parsing event data:", e);
}
}
}
}
return { content: fullContent };
} finally {
reader.releaseLock();
}
}
なぜEventSourceではなくfetch + ReadableStreamを使うのか
ブラウザにはEventSource APIがありますが、今回は使っていません。理由：
| 項目 | EventSource | fetch + ReadableStream |
|---|---|---|
| HTTPメソッド | GETのみ | POST対応 |
| リクエストボディ | 送れない | JSON等を送れる |
| カスタムヘッダー | 不可 | Authorization等を設定可能 |
| 接続管理 | 自動再接続 | 手動管理 |
AI APIへのリクエストはPOSTでプロンプトを送る必要があるため、EventSourceは使えません。
SSEパースの落とし穴
チャンク境界の問題
reader.read()で受け取るチャンクは、SSEイベントの境界と一致する保証がありません。
// こうなることもある
チャンク1: "event: data\ndata: {\"type\": \"ai\", \"cont"
チャンク2: "ent\": \"こんにちは\"}\n\n"
上記の実装ではイベントを\n\nで分割していますが、チャンク境界をまたぐイベントが存在すると不完全なJSONをパースしようとしてcatchブロックに落ちます。今回のケースでは、次のイベントで全文が再送されるため（累積型なので）、1つのイベントを落としても表示に大きな問題は起きません。
より堅牢な実装が必要な場合は、バッファリングを導入します。
let buffer = "";
while (true) {
const { value, done } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
// 完全なイベント（\n\nで終わるもの）だけ処理
const parts = buffer.split("\n\n");
buffer = parts.pop() || ""; // 最後の不完全な部分をバッファに残す
for (const part of parts) {
if (part.trim()) processEvent(part);
}
}
コールバックの設計：差分 vs 全文
UIコンポーネントに渡すコールバックでは、差分だけでなく「差分であること」をフラグで伝えています。
interface GptResponse {
content: string;
error?: string;
done?: boolean;
isIncremental?: boolean; // trueなら差分、falseなら全文
}
UI側では：
// 差分ならテキストを追加
if (response.isIncremental) {
output += response.content;
}
// 完了なら全文を最終確認用にセット
if (response.done) {
output = response.content;
}
完了時に全文を改めてセットすることで、万が一チャンク落ちがあっても最終的な表示は正しくなります。
サーバーサイド：SSEレスポンスのプロキシ
SvelteKitのAPIルートでAI APIのストリームをそのままクライアントに返しています。
export const POST: RequestHandler = async ({ request }) => {
const { prompt, name, instruction, webSearch } = await request.json();
const response = await aiClient.chat(name, instruction, prompt, !!webSearch);
return new Response(response.data, {
headers: {
"Content-Type": "text/event-stream",
"Cache-Control": "no-cache",
"Connection": "keep-alive"
}
});
};
ストリームの中身を変換せず、そのままResponseに渡しています。累積型→差分型への変換はクライアント側で行うことで、サーバーサイドの処理をシンプルに保っています。
まとめ
AI APIのストリーミングレスポンスを扱う際は、まずそのAPIが差分型か累積型かを確認してください。
| 確認ポイント | 差分型 | 累積型 |
|---|---|---|
| 各イベントの内容 | 新しい部分のみ | これまでの全文 |
| クライアントの処理 | そのまま追加 | 前回との差分を計算 |
| チャンク落ちの影響 | テキストが欠損 | 次のイベントで復元 |
| 例 | OpenAI API | 一部のエンタープライズAPI |
差分抽出のコア処理はたった1行です：
const incrementalContent = newContent.substring(fullContent.length);
簡単な処理ですが、これに気づかないと「テキストが二重に表示される」バグに悩まされます。
