Torna al blog

API e SDKGuida

Risposte in streaming in Python

Stampa la risposta del modello mentre viene generata con il pacchetto openai ufficiale — esempi sync e async, output di ragionamento, utilizzo dei token e gestione degli errori.

Di CheapTokens TeamPubblicato il 3 min di lettura

Codice Python con un ciclo for sui chunk dello stream e testo che appare token per token
In questa pagina
  1. Setup
  2. Esempio minimo
  3. Modelli di ragionamento
  4. Utilizzo dei token
  5. asyncio
  6. Interrompere in anticipo e timeout
  7. Errori
  8. Cosa passa sulla rete

Con lo streaming, la risposta arriva pezzo per pezzo mentre il modello la sta ancora generando, così i tuoi utenti vedono il testo subito invece di aspettare l'intera risposta. L'API usa il formato di streaming standard di OpenAI (server-sent events), e il pacchetto Python ufficiale openai se ne occupa per te.

Ogni esempio qui sotto è completo: copialo in un file ed eseguilo. Gli esempi sono stati verificati con Python 3.10.12 e openai 3.31.0 su un server di test locale che riproduce il formato di streaming dell'API (sintassi ed esecuzione).

Setup#

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

Esempio minimo#

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

Due dettagli contano:

  • Salta i chunk senza choices. Alcuni chunk non contengono testo — per esempio il chunk finale con l'utilizzo dei token.
  • delta.content può essere vuoto. Il primo chunk spesso contiene solo il ruolo, e l'ultimo solo finish_reason. Stampa il testo solo quando c'è.

Modelli di ragionamento#

zai-org/GLM-5.3-Flash ragiona prima di rispondere. Il ragionamento arriva in un campo separato del delta, reasoning_content, e la risposta stessa in content. L'SDK conserva i campi sconosciuti, quindi puoi leggerlo con getattr. Questo è un esempio completo con il suo client e il suo 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()

Il ragionamento può richiedere un po' prima che appaia la prima parola della risposta — un altro motivo per usare lo streaming.

Utilizzo dei token#

Richiedi l'utilizzo con stream_options; arriva nell'ultimo chunk, che non ha 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}")

Moltiplica questi numeri per le tariffe del modello per ottenere il costo della richiesta — vedi How to estimate the cost of an AI agent.

asyncio#

Nel codice asincrono usa AsyncOpenAI e async for. Il client ha il suo nome, async_client, così non entra in conflitto con il client sincrono degli altri esempi:

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

Interrompere in anticipo e timeout#

Esci dal ciclo per fermare la generazione. Aprire lo stream con with chiude la connessione quando il blocco termina. Il timeout del client è impostato esplicitamente; l'SDK ritenta anche alcune richieste fallite da solo (max_retries, 2 per impostazione predefinita).

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)

Errori#

Gli errori che si verificano prima del primo chunk — chiave errata, saldo vuoto, modello sconosciuto — vengono sollevati come eccezioni quando crei lo 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)

Un elenco completo dei codici di stato e delle soluzioni è in Common connection errors and how to fix them.

Cosa passa sulla rete#

Per il debug, guarda lo stream grezzo con curl (-N disattiva il 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"}]}'

Ogni evento è una riga data: {…} con un chunk JSON; lo stream termina con data: [DONE]. Le righe che iniziano con i due punti sono commenti keep-alive inviati durante le lunghe pause — i client li ignorano.

Tag: