Powrót do bloga

API i SDKPrzewodnik

Strumieniowanie odpowiedzi w Pythonie

Wypisuj odpowiedź modelu w miarę jej generowania za pomocą oficjalnego pakietu openai — przykłady synchroniczne i asynchroniczne, dane wyjściowe rozumowania, zużycie tokenów i obsługa błędów.

Autor: CheapTokens TeamOpublikowano 3 min czytania

Kod w Pythonie z pętlą for po fragmentach strumienia i tekstem pojawiającym się token po tokenie
Na tej stronie
  1. Konfiguracja
  2. Minimalny przykład
  3. Modele rozumujące
  4. Zużycie tokenów
  5. asyncio
  6. Wczesne zatrzymanie i limity czasu
  7. Błędy
  8. Co jest przesyłane

Dzięki strumieniowaniu odpowiedź dociera kawałek po kawałku, gdy model wciąż ją generuje, więc użytkownicy widzą tekst od razu, zamiast czekać na całą odpowiedź. API używa standardowego formatu strumieniowania OpenAI (server-sent events), a oficjalny pakiet openai Python obsługuje to za Ciebie.

Każdy poniższy przykład jest kompletny: skopiuj go do pliku i uruchom. Przykłady zostały sprawdzone z Python 3.10.12 i openai 3.31.0 na lokalnym serwerze testowym, który odtwarza format strumieniowania API (składnia i wykonanie).

Konfiguracja#

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

Minimalny przykład#

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

Dwa szczegóły mają znaczenie:

  • Pomijaj fragmenty bez choices. Niektóre fragmenty nie zawierają tekstu — na przykład końcowy fragment ze zużyciem tokenów.
  • delta.content może być pusty. Pierwszy fragment często zawiera tylko rolę, a ostatni tylko finish_reason. Wypisuj tekst tylko wtedy, gdy jest obecny.

Modele rozumujące#

zai-org/GLM-5.3-Flash rozumuje, zanim odpowie. Rozumowanie przychodzi w osobnym polu delty, reasoning_content, a sama odpowiedź w content. SDK zachowuje nieznane pola, więc możesz je odczytać za pomocą getattr. To kompletny przykład z własnym klientem i własnym strumieniem:

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

Rozumowanie może zająć trochę czasu, zanim pojawi się pierwsze słowo odpowiedzi — kolejny powód, by strumieniować.

Zużycie tokenów#

Poproś o zużycie za pomocą stream_options; przychodzi w ostatnim fragmencie, który nie ma 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}")

Pomnóż te liczby przez stawki modelu, aby otrzymać koszt żądania — zobacz Jak oszacować koszt agenta AI.

asyncio#

W kodzie asynchronicznym używaj AsyncOpenAI i async for. Klient ma własną nazwę, async_client, więc nie koliduje z synchronicznym client z pozostałych przykładów:

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

Wczesne zatrzymanie i limity czasu#

Opuść pętlę, aby zatrzymać generowanie. Otwarcie strumienia za pomocą with zamyka połączenie, gdy blok się kończy. Limit czasu klienta jest ustawiany jawnie; SDK samo ponawia też niektóre nieudane żądania (max_retries, domyślnie 2).

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)

Błędy#

Błędy, które występują przed pierwszym fragmentem — zły klucz, puste saldo, nieznany model — są zgłaszane jako wyjątki podczas tworzenia strumienia:

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)

Pełna lista kodów statusu i poprawek znajduje się w Typowe błędy połączenia i jak je naprawić.

Co jest przesyłane#

Aby debugować, spójrz na surowy strumień za pomocą curl (-N wyłącza buforowanie):

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

Każde zdarzenie to linia data: {…} z fragmentem JSON; strumień kończy się data: [DONE]. Linie zaczynające się od dwukropka to komentarze keep-alive wysyłane podczas długich przerw — klienci je ignorują.

Tagi: