0Pricing
AI Agents · レッスン

レスポンスのストリーミング(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フィードバックを取得できます。ローカル設定は不要です。

このコースのすべてのレッスン

  1. OpenAI APIの呼び出し:chat.completions
  2. Anthropic APIの呼び出し:messages
  3. レスポンスのストリーミング(SSE)
  4. コストを意識する:トークン数のカウントと予算
← AI Agentsに戻る