EastX API の SSE ストリーミング入門｜Python 実装と中断時の対処

SSE ストリーミングで生成途中のテキストを表示する

Server-Sent Events（SSE）を利用すると、モデルの生成が終わる前から受信したテキストを画面に表示できます。EastX の OpenAI 互換 API では、ストリーミング対応モデルに同じ SDK のコードを使えます。クライアントごとに API キーを分けると、利用量や作業ごとの費用も確認しやすくなります。

最初のテキストまでの時間と応答全体の時間は別の指標です。リクエスト開始、最初の表示、完了の時刻をそれぞれ記録して比較します。

Python の最小例

EastX のキーを環境変数 EASTX_API_KEY、ストリーミング対応モデルの正確な ID を EASTX_MODEL に設定します。SDK をインストールし、次のコードを実行してください。

python -m pip install openai

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["EASTX_API_KEY"],
    base_url="https://api.eastx.ai/v1",
    max_retries=0,
    timeout=60.0,
)
stream = client.chat.completions.create(
    model=os.environ["EASTX_MODEL"],
    messages=[{"role": "user", "content": "SSE を簡単に説明してください。"}],
    stream=True,
)
try:
    for chunk in stream:
        if not chunk.choices:
            continue
        text = chunk.choices[0].delta.content
        if text:
            print(text, end="", flush=True)
finally:
    stream.close()

SDK は SSE のイベント境界を処理します。HTTP を自分で読み取る場合、ネットワークの一つの受信単位が完全な JSON と一致するとは限りません。data: のイベント境界で処理してください。Chat Completions のテキストは choices[0].delta.content に入り、空の差分は表示せず、終了マーカーと finish_reason も確認します。

中断したときの扱い

途中でタイムアウトした場合は、受信済みの文章を保持して中断状態を表示します。再生成するかどうかはユーザーまたはアプリの方針で決めます。再リクエストでは別の文章になり、追加の利用料金が発生する場合があります。Agent がファイルや外部ツールを操作する場合は、再試行する前に実行済みの操作を確認してください。

まとめて表示される場合の確認

最後に全テキストが一度に表示されるときは、クライアントが逐次読み取っているか、プロキシやサーバーが応答をバッファしていないか、タイムアウトが適切か確認します。Messages と Responses はイベント構造が異なるため、Chat Completions 用の読み取り処理をそのまま共用できるとは限りません。

• EastX API ドキュメント
• SDK の初回リクエスト
• SSE ガイド（中国語・英語）