Voltar ao blog

API e SDKGuia

Respostas em streaming em Python

Imprima a resposta do modelo conforme ela é gerada com o pacote oficial openai — exemplos síncronos e assíncronos, saída de raciocínio, uso de tokens e tratamento de erros.

Por CheapTokens TeamPublicado em 3 min de leitura

Código Python com um loop for sobre chunks do stream e texto aparecendo token a token
Nesta página
  1. Configuração
  2. Exemplo mínimo
  3. Modelos de raciocínio
  4. Uso de tokens
  5. asyncio
  6. Parar cedo e timeouts
  7. Erros
  8. O que trafega na rede

Com streaming, a resposta chega pedaço por pedaço enquanto o modelo ainda a está gerando, então seus usuários veem o texto imediatamente em vez de esperar pela resposta completa. A API usa o formato de streaming padrão da OpenAI (server-sent events), e o pacote oficial openai Python cuida disso para você.

Todos os exemplos abaixo são completos: copie para um arquivo e execute. Os exemplos foram verificados com Python 3.10.12 e openai 3.31.0 contra um servidor de teste local que reproduz o formato de streaming da API (sintaxe e execução).

Configuração#

bash
pip install openai
export CHEAPTOKENS_API_KEY="paste-your-key-here"

Exemplo mínimo#

python
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()

Dois detalhes importam:

  • Ignore chunks sem choices. Alguns chunks não carregam texto — por exemplo, o chunk final com o uso de tokens.
  • delta.content pode estar vazio. O primeiro chunk geralmente carrega apenas o role, e o último apenas finish_reason. Imprima texto somente quando ele estiver presente.

Modelos de raciocínio#

zai-org/GLM-5.3-Flash raciocina antes de responder. O raciocínio vem em um campo separado do delta, reasoning_content, e a resposta em si em content. O SDK mantém campos desconhecidos, então você pode lê-lo com getattr. Este é um exemplo completo com seu próprio client e seu próprio stream:

python
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()

O raciocínio pode demorar um pouco antes que a primeira palavra da resposta apareça — mais um motivo para usar streaming.

Uso de tokens#

Peça o uso com stream_options; ele vem no último chunk, que não tem choices:

python
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}")

Multiplique esses números pelas tarifas do modelo para obter o custo da requisição — veja How to estimate the cost of an AI agent.

asyncio#

Em código assíncrono use AsyncOpenAI e async for. O client tem seu próprio nome, async_client, para não colidir com o síncrono client dos outros exemplos:

python
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())

Parar cedo e timeouts#

Saia do loop para interromper a geração. Abrir o stream com with fecha a conexão quando o bloco termina. O timeout do client é definido explicitamente; o SDK também tenta novamente algumas requisições com falha por conta própria (max_retries, 2 por padrão).

python
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)

Erros#

Erros que acontecem antes do primeiro chunk — chave errada, saldo vazio, modelo desconhecido — são lançados como exceções quando você cria o stream:

python
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)

Uma lista completa de códigos de status e correções está em Common connection errors and how to fix them.

O que trafega na rede#

Para depurar, veja o stream bruto com curl (-N desativa o buffering):

bash
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"}]}'

Cada evento é uma linha data: {…} com um chunk JSON; o stream termina com data: [DONE]. Linhas que começam com dois-pontos são comentários de keep-alive enviados durante longas pausas — os clients as ignoram.

Tags: