APIとSDKガイド
Python でのストリーミング応答
公式の openai パッケージを使用して、モデルの回答を生成されると同時に出力します — 同期と非同期の例、推論出力、トークン使用量、エラー処理。
ストリーミングを使うと、モデルがまだ生成している間に回答が少しずつ届くため、ユーザーは完全な応答を待たずにすぐテキストを見ることができます。API は標準の OpenAI ストリーミング形式（サーバー送信イベント）を使用しており、公式の openai Python パッケージがそれを処理します。
以下のすべての例は完全です。ファイルにコピーして実行してください。例は Python 3.10.12 と openai 3.31.0 で、API のストリーミング形式を再現するローカルテストサーバーに対して確認されています（構文と実行）。
セットアップ#
pip install openai
export CHEAPTOKENS_API_KEY="paste-your-key-here"最小限の例#
import os
from openai import OpenAI
client = OpenAI(
base_url="https://api.cheaptokens.dev/v1",
api_key=os.environ["CHEAPTOKENS_API_KEY"],
)
stream = client.chat.completions.create(
model="deepseek-ai/DeepSeek-V4-Flash-0731",
messages=[{"role": "user", "content": "Write a haiku about latency"}],
stream=True,
)
for chunk in stream:
if not chunk.choices: # service chunk without text, e.g. the final usage chunk
continue
text = chunk.choices[0].delta.content
if text:
print(text, end="", flush=True)
print()2 つの詳細が重要です：
choicesのないチャンクはスキップする。 一部のチャンクにはテキストが含まれません — 例えば、トークン使用量を含む最後のチャンクです。delta.contentは空になることがあります。 最初のチャンクはしばしばロールのみを含み、最後のチャンクはfinish_reasonのみを含みます。テキストがある場合にのみ出力してください。
推論モデル#
zai- は回答する前に推論します。推論はデルタの別のフィールド reasoning_ に含まれ、回答自体は content に含まれます。SDK は未知のフィールドを保持するため、getattr で読み取ることができます。これは独自のクライアントと独自のストリームを持つ完全な例です：
import os
import sys
from openai import OpenAI
client = OpenAI(
base_url="https://api.cheaptokens.dev/v1",
api_key=os.environ["CHEAPTOKENS_API_KEY"],
)
stream = client.chat.completions.create(
model="zai-org/GLM-5.3-Flash",
messages=[{"role": "user", "content": "Is 1013 a prime number?"}],
stream=True,
)
for chunk in stream:
if not chunk.choices:
continue
delta = chunk.choices[0].delta
thinking = getattr(delta, "reasoning_content", None)
if thinking:
print(thinking, end="", flush=True, file=sys.stderr) # reasoning → stderr
if delta.content:
print(delta.content, end="", flush=True) # answer → stdout
print()推論は回答の最初の単語が現れるまでに時間がかかることがあります — これもストリーミングする理由の一つです。
トークン使用量#
stream_options で使用量を要求します。それは最後のチャンクに含まれ、choices はありません：
import os
from openai import OpenAI
client = OpenAI(
base_url="https://api.cheaptokens.dev/v1",
api_key=os.environ["CHEAPTOKENS_API_KEY"],
)
stream = client.chat.completions.create(
model="deepseek-ai/DeepSeek-V4-Flash-0731",
messages=[{"role": "user", "content": "Explain idempotency in one paragraph"}],
stream=True,
stream_options={"include_usage": True},
)
usage = None
for chunk in stream:
if chunk.usage:
usage = chunk.usage
if chunk.choices and chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="", flush=True)
if usage:
print(f"\ninput: {usage.prompt_tokens}, output: {usage.completion_tokens}")これらの数値にモデルの料金を掛けると、リクエストのコストが得られます — AI エージェントのコストを見積もる方法 を参照してください。
asyncio#
非同期コードでは AsyncOpenAI と async for を使用します。クライアントには独自の名前 async_client があるため、他の例の同期 client と衝突しません：
import asyncio
import os
from openai import AsyncOpenAI
async_client = AsyncOpenAI(
base_url="https://api.cheaptokens.dev/v1",
api_key=os.environ["CHEAPTOKENS_API_KEY"],
)
async def main() -> None:
stream = await async_client.chat.completions.create(
model="deepseek-ai/DeepSeek-V4-Flash-0731",
messages=[{"role": "user", "content": "Name three uses of a message queue"}],
stream=True,
)
async for chunk in stream:
if chunk.choices and chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="", flush=True)
print()
asyncio.run(main())早期停止とタイムアウト#
ループを抜けると生成が停止します。with でストリームを開くと、ブロックが終了したときに接続が閉じられます。クライアントのタイムアウトは明示的に設定されています。SDK は一部の失敗したリクエストを独自に再試行します（max_retries、デフォルトは 2）。
import os
from openai import OpenAI
client = OpenAI(
base_url="https://api.cheaptokens.dev/v1",
api_key=os.environ["CHEAPTOKENS_API_KEY"],
timeout=120, # seconds; long answers take time
)
received = ""
with client.chat.completions.create(
model="deepseek-ai/DeepSeek-V4-Flash-0731",
messages=[{"role": "user", "content": "Write a long story about a lighthouse"}],
stream=True,
) as stream:
for chunk in stream:
if chunk.choices and chunk.choices[0].delta.content:
received += chunk.choices[0].delta.content
if len(received) > 300: # stop early: enough text received
break
print(received)エラー#
最初のチャンクの前に発生するエラー — 間違ったキー、残高不足、不明なモデル — は、ストリームを作成するときに例外として発生します：
import os
import openai
from openai import OpenAI
client = OpenAI(
base_url="https://api.cheaptokens.dev/v1",
api_key=os.environ["CHEAPTOKENS_API_KEY"],
)
try:
stream = client.chat.completions.create(
model="deepseek-ai/DeepSeek-V4-Flash-0731",
messages=[{"role": "user", "content": "Hello"}],
stream=True,
)
for chunk in stream:
if chunk.choices and chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="", flush=True)
print()
except openai.AuthenticationError: # 401: wrong or rotated key
print("Check the API key")
except openai.BadRequestError as e: # 400: e.g. unknown model ID
print("Bad request:", e.message)
except openai.APIStatusError as e: # other statuses, e.g. 402 when the balance is used up
print(e.status_code, e.message)ステータスコードと修正方法の完全なリストは 一般的な接続エラーとその修正方法 にあります。
ワイヤー上で何が起きているか#
デバッグするには、curl で生のストリームを確認します（-N はバッファリングをオフにします）：
curl -N https://api.cheaptokens.dev/v1/chat/completions \
-H "Authorization: Bearer $CHEAPTOKENS_API_KEY" \
-H "Content-Type: application/json" \
-d '{"model": "deepseek-ai/DeepSeek-V4-Flash-0731", "stream": true,
"messages": [{"role": "user", "content": "Hi"}]}'各イベントは JSON チャンクを含む data: {…} 行で、ストリームは data: [DONE] で終了します。コロンで始まる行は、長い一時停止中に送信されるキープアライブコメントです — クライアントはそれらを無視します。
