レスポンスのストリーミング(SSE)
Server-Sent Eventsを介してLLMの出力をトークン単位でストリーミングし、快適なUIと進捗表示を実現します。
「レスポンスのストリーミング(SSE)」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン3/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。
なぜストリーミングするのですか
ストリーミングを使わない場合、何も表示する前に応答全体を待つ必要があります。50トークンの応答に5秒かかるなら、ユーザーは5秒間何も見られません。
ストリーミングでは、到着したトークンを順次表示します。合計時間は同じでも、UXは瞬時に感じられます。
Server-Sent Events(SSE)
OpenAIとAnthropicはSSEでストリーミングします。これは、サーバーがdata: {...}\n\nという形式でイベントをプッシュする一方向のHTTPプロトコルです。
ほとんどのSDKでは、SSEの処理をイテレーターインターフェースの背後に隠しています。
OpenAIでのストリーミング
stream=Trueを設定し、応答を反復処理します。
stream = client.chat.completions.create(
model='gpt-4o-mini',
messages=messages,
stream=True,
)
for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
print(delta, end='', flush=True)完全な応答の収集
デルタを蓄積して、最終的な文字列を取得します。
buf = []
for chunk in stream:
delta = chunk.choices[0].delta.content or ''
buf.append(delta)
print(delta, end='', flush=True)
full_text = ''.join(buf)
messages.append({'role': 'assistant', 'content': full_text})Anthropicでのストリーミング
同じパターンですが、APIが少し異なります。
with client.messages.stream(
model='claude-sonnet-4-5',
max_tokens=1024,
messages=messages,
) as stream:
for text in stream.text_stream:
print(text, end='', flush=True)
final = stream.get_final_message()
print('\ndone, tokens:', final.usage.output_tokens)ツール呼び出しのストリーミング
ツール呼び出しもストリーミングされます。OpenAIでは、関数名と引数が分割されて届きます。
for chunk in stream:
delta = chunk.choices[0].delta
if delta.tool_calls:
for tc in delta.tool_calls:
# tc.function.name and tc.function.arguments arrive as partial strings
passバックプレッシャーとキャンセル
ユーザーがページを閉じた場合は、トークンを消費し続けないようにストリームを中止します。
try:
for chunk in stream:
if request_was_cancelled():
stream.close()
break
finally:
stream.close()Webサーバー経由のストリーミング
FastAPIでは、StreamingResponseを使用します。
from fastapi.responses import StreamingResponse
def token_generator():
for chunk in stream:
delta = chunk.choices[0].delta.content or ''
yield f'data: {delta}\n\n'
return StreamingResponse(token_generator(), media_type='text/event-stream')文単位のバッファリング
音声合成や分割表示では、文が終わるまでバッファリングします。
buf = ''
for chunk in stream:
buf += chunk.choices[0].delta.content or ''
while '. ' in buf:
sentence, buf = buf.split('. ', 1)
speak(sentence + '.')ストリーミングされたJSONの解析
JSON出力はストリームの途中では解析できません。選択肢は次のとおりです。
- 出力全体をバッファリングしてから一度だけ解析する
- ストリーミングJSONパーサー(
ijson)を使い、フィールドが完成するたびに出力する
レイテンシー指標:TTFTとTPS
- TTFT — 最初のトークンまでの時間(体感上の応答性)
- TPS — 1秒あたりのトークン数(スループット)
ストリーミングで最適化されるのは合計時間ではなくTTFTです。チャットUXでは、TTFTを500ms未満にすることを目指します。
なぜストリーミングするのですか
ストリーミングの主なメリットは何でしょうか。
まとめ
ストリーミングは速度ではなくUXを向上させます。人間とリアルタイムに対話するエージェントには、必ず実装してください。
よくある質問
「レスポンスのストリーミング(SSE)」レッスンは無料ですか?
はい。「レスポンスのストリーミング(SSE)」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し(組み込みコードエディタと24時間対応のAIチューター)、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。
「レスポンスのストリーミング(SSE)」で何を学びますか?
Server-Sent Eventsを介してLLMの出力をトークン単位でストリーミングし、快適なUIと進捗表示を実現します。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Agentsを始めるのに経験は必要ですか?
事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン3/4です。
「レスポンスのストリーミング(SSE)」レッスンにはどのくらい時間がかかりますか?
ほとんどのCoddyKitレッスンは約5~10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Agentsレッスンでコードを書いて実行できますか?
はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。