コアガイド
ストリーミング
リアルタイムのストリーミング応答を実装します
概要
ストリーミングは出力を順次返します。モデルの accepted_request_formats に openai_responses があれば Responses を使えます。既存の Chat Completions クライアントは同じ形式を維持できます。
推奨: Responses Streaming
curl https://api.tokenlab.sh/v1/responses \
-H "Authorization: Bearer sk-your-api-key" \
-H "Content-Type: application/json" \
-d '{
"model": "gpt-5.6-terra",
"input": "Write a short poem.",
"stream": true
}'Responses と Gemini の streaming 境界
Responses SSE は公開されたイベント名、順序、フィールドを保持します。出力後に切断された応答は未完了で、自動的に再開されません。
Responses WebSocket は response.create を使い、常にストリーミングします。background と response.cancel は利用できません。接続ごとに一度に一つの応答を処理し、上限は 60 分です。generate: false は継続用 ID を作成し、モデル出力やモデル料金を発生させません。
Gemini SSE はネイティブのチャンクを返します。途中のイベントには finishReason がない場合があり、Chat の [DONE] なしで自然に終了できます。
Chat Completions ストリーミング
フレームワークが依然として /v1/chat/completions からの SSE チャンクを想定している場合でも、これも動作します。
import os
from openai import OpenAI
with OpenAI(
api_key=os.environ["TOKENLAB_API_KEY"],
base_url="https://api.tokenlab.sh/v1",
timeout=30.0,
max_retries=0,
) as client:
finish_reason = None
with client.chat.completions.create(
model="gpt-5.6-terra",
messages=[{"role": "user", "content": "Write a short poem."}],
stream=True,
stream_options={"include_usage": True},
) as stream:
for chunk in stream:
if not chunk.choices:
continue
choice = chunk.choices[0]
if choice.delta.content:
print(choice.delta.content, end="", flush=True)
if choice.finish_reason:
finish_reason = choice.finish_reason
if finish_reason != "stop":
raise RuntimeError(f"Stream ended without a complete text answer: {finish_reason}")ストリーム終了条件
一般的な完了条件:
- Responses API ストリームでは
response.completed - Chat Completions ストリームでは
finish_reason: "stop" - token 制限に達した場合は
finish_reason: "length" - モデルがツールを使用したい場合の tool/function call イベント
Web アプリのパターン
SDK のストリーム処理はサーバーで実行してください。ブラウザーは認証済みの自前バックエンドを呼び出し、TokenLab キーはサーバー環境に保管します。テキスト差分をブラウザーへ転送し、停止や切断時には上流ストリームをキャンセルします。SSE の解析には SDK を使ってください。ネットワークのチャンク境界とイベント境界は一致しません。