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()두 가지 세부 사항이 중요합니다:
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}")이 숫자에 모델의 요금을 곱하면 요청 비용을 구할 수 있습니다 — How to estimate the cost of an AI agent를 참고하세요.
asyncio#
비동기 코드에서는 AsyncOpenAI와 async for를 사용하세요. 클라이언트는 다른 예제의 동기 client와 충돌하지 않도록 자체 이름인 async_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)상태 코드와 해결 방법의 전체 목록은 Common connection errors and how to fix them에 있습니다.
전송되는 내용#
디버깅하려면 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]로 끝납니다. 콜론으로 시작하는 줄은 긴 일시 중지 동안 전송되는 연결 유지 주석입니다 — 클라이언트는 이를 무시합니다.
