SSEでAIの応答をストリーミングする：実践ガイド

ストリーミングがあるかないかで、キャラクターがそこにいると感じられるか、ただのスピナーになるかが決まります。SSEの仕組みはシンプルですが、バグはすべて細部に潜んでいます。このガイドで扱うのは、その細部です。

ワイヤーフォーマット

stream: trueを指定すると、/v1/chat/completionsはtext/event-streamで応答します。各フレームはdata:の行と空行からなり、差分（デルタ）は標準的なチャンクの形で届きます。

data: {"id":"cmpl_…","object":"chat.completion.chunk","choices":[{"delta":{"content":"That "}}]}

data: {"id":"cmpl_…","object":"chat.completion.chunk","choices":[{"delta":{"content":"noise"}}]}

data: {"object":"chat.completion.usage","usage":{"credits_spent":3,"credits_remaining":997}}

data: [DONE]

知っておきたいeroq独自のフレームが2つあります。[DONE]の直前に届くusageイベントには利用量が入っています。そしてerrorイベント（{"error":{…}}）は、本来なら接続の切断から推測するしかないクラッシュを、明示的に知らせてくれます。コンテンツが届く前にストリームがエラーになった場合、その呼び出しはすでに自動で返金されています。

定番のバグを避けるパース

定番のバグは、ネットワークのチャンクを毎回完全なフレームとして扱ってしまうことです。TCPは改行の位置など気にしません。1つのフレームが、複数回の読み取りにまたがって分割されて届くこともあります。バッファに溜め、\n\nで分割し、残りは取っておきましょう。

const reader = res.body.getReader()
const decoder = new TextDecoder()
let buffer = ''
while (true) {
  const { done, value } = await reader.read()
  if (done) break
  buffer += decoder.decode(value, { stream: true })   // stream: true matters for UTF-8
  const frames = buffer.split('\n\n')
  buffer = frames.pop() ?? ''                          // last piece may be incomplete
  for (const frame of frames) {
    const data = frame.replace(/^data: /, '')
    if (data === '[DONE]') continue
    const parsed = JSON.parse(data)
    if (parsed.error) throw new Error(parsed.error.message)
    const delta = parsed.choices?.[0]?.delta?.content
    if (delta) render(delta)
  }
}

decoder.decodeの{ stream: true }は飾りではありません。これがないと、チャンクをまたいで分割されたマルチバイト文字が文字化けします。絵文字の多いロールプレイや日本語のテキストなら、1時間もしないうちにこのバグに出くわします。

バックエンド経由で中継する

APIキーは、絶対にブラウザに渡さないでください。キーとキャラクターシートをすでに管理しているバックエンドを通して、ストリームを中継しましょう。中継部分は薄いものですが、成否を分けるポイントが2つあります。

// Node/Express-style relay
app.post('/chat', async (req, res) => {
  res.setHeader('Content-Type', 'text/event-stream')
  res.setHeader('Cache-Control', 'no-cache')
  res.flushHeaders()                                   // 1. headers out immediately

  const upstream = await fetch('https://eroq.ai/v1/chat/completions', { /* … */ })
  for await (const chunk of upstream.body) {
    res.write(chunk)                                   // 2. relay bytes, re-frame nothing
  }
  res.end()
})

1. ヘッダーはすぐにフラッシュする。そうしないと、リバースプロキシがレスポンス全体をバッファしてから一度に届けてしまうことがあります。ひと塊で届くストリーミングです。nginxでは、X-Accel-Buffering: noも設定してください。
2. バイトはそのまま中継する。中継部分でフレームをパースして再シリアライズしても、何の価値もないまま、バグの入り込む余地が倍になるだけです。パースは、レンダリングを行うクライアント側で行いましょう。

良いUXと素晴らしいUXを分ける細部

• 小さなタイマー（30〜50ms）ごとに描画する。デルタごとではありません。トークンごとにDOMへ書き込むと、モバイルでカクつきます。
• 最初のトークンは素早く表示し、あとは流れに任せる。体感の遅延は、ほぼすべて最初のトークンが出るまでの時間で決まります。
• ストリームが途中で失敗したら、途中までのテキストを残し、再試行の手段を用意しましょう。消えてしまう返答より、半分でも残る返答のほうがましです。しかも、最初の出力を過ぎてから中断されたストリームも生成自体は済んでいるので、テキストを残すことは、支払った分を尊重することにもなります。
• ユーザーが中断できるようにする。中継しているレスポンスを閉じたら、上流のリクエストも閉じるべきです。放置されたストリームを読み続けるのは、誰も見ていない生成にお金を払い続けることです。

よくある質問

eroq APIのSSEチャンクを正しくパースするには？

バイトをバッファに溜め、\n\nで分割し、末尾の断片は次の読み取りまで取っておきます。1つのフレームが、複数のネットワークチャンクに分割されて届くことがあるからです。マルチバイト文字が分割されても壊れないようにTextDecoder.decodeに{ stream: true }を渡し、JSONをパースする前に[DONE]の番兵をスキップします。

ストリーミングは通常のcompletionより高くなりますか？

なりません。クレジットはトークン単位ではなくcompletion単位で課金されるので、ストリーミングした返答もバッファした返答も、RP miniなら同じ1クレジット、RP+なら3クレジットです。返答が50トークンでも、上限の1,200トークンでも変わりません。料金の面で、stream: falseのままにしておく理由はありません。

ストリームの最後にあるusageイベントとは何ですか？

data: [DONE]の直前に届く、credits_spentとcredits_remainingを含むフレームです。これを使えば、APIをもう一度呼ばなくても残高の表示を更新できます。{"error":{…}}の形のerrorフレームは、本来なら推測するしかない接続の切断を明示的に知らせてくれます。また、コンテンツが届く前に失敗したストリームは、すでに自動で返金されています。

ストリームがトークンごとではなく、一度にまとめて届くのはなぜですか？

ほとんどの場合、プロキシがレスポンスをバッファしているのが原因です。最初の書き込みの前にヘッダーをフラッシュし、nginxではX-Accel-Buffering: noを設定してください。次によくある原因は、中継部分でフレームを再シリアライズしていることです。バイトはそのまま中継し、パースはレンダリングを行うクライアント側で行いましょう。

定額の料金体系には、ここでもうひとつ利点があります。ストリーミングのコストは、バッファした場合とまったく同じ1クレジットまたは3クレジットです。ストリーミングしない理由はありません。だからこそ、ドキュメントの例はすべてストリーミングを使っています。