API та SDKПосібник

Стрімінгові відповіді в Python

Друкуйте відповідь моделі в міру її генерації за допомогою офіційного пакета openai — синхронні та асинхронні приклади, виведення міркувань, використання токенів та обробка помилок.

Автор: CheapTokens TeamОпубліковано 15 жовт. 2026 р.3 хв читання

Код Python із циклом for по чанках потоку та текстом, що з'являється токен за токеном

На цій сторінці

Налаштування
Мінімальний приклад
Моделі з міркуванням
Використання токенів
asyncio
Дострокова зупинка та тайм-аути
Помилки
Що передається мережею

Зі стрімінгом відповідь надходить частинами, поки модель ще її генерує, тож ваші користувачі бачать текст одразу, а не чекають на всю відповідь. API використовує стандартний формат стрімінгу OpenAI (server-sent events), а офіційний Python-пакет openai обробляє його за вас.

Кожен приклад нижче є повним: скопіюйте його у файл і запустіть. Приклади перевірено з Python 3.10.12 та openai 3.31.0 на локальному тестовому сервері, який відтворює формат стрімінгу API (синтаксис і виконання).

Налаштування#

bash

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

Мінімальний приклад#

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

Дві деталі мають значення:

Пропускайте чанки без choices. Деякі чанки не містять тексту — наприклад, фінальний чанк із використанням токенів.
delta.content може бути порожнім. Перший чанк часто містить лише роль, а останній — лише finish_reason. Друкуйте текст, тільки коли він є.

Моделі з міркуванням#

zai-org/GLM-5.3-Flash міркує перед тим, як відповісти. Міркування надходять в окремому полі дельти, reasoning_content, а сама відповідь — у content. SDK зберігає невідомі поля, тож ви можете прочитати його за допомогою getattr. Це повний приклад із власним клієнтом і власним потоком:

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

Міркування можуть тривати деякий час, перш ніж з'явиться перше слово відповіді — ще одна причина використовувати стрімінг.

Використання токенів#

Запросіть використання за допомогою stream_options; воно надходить в останньому чанку, який не має 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}")

Помножте ці числа на тарифи моделі, щоб отримати вартість запиту — див. Як оцінити вартість AI-агента.

asyncio#

В асинхронному коді використовуйте AsyncOpenAI та async for. Клієнт має власну назву, async_client, щоб не конфліктувати із синхронним client з інших прикладів:

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

Дострокова зупинка та тайм-аути#

Вийдіть із циклу, щоб зупинити генерацію. Відкриття потоку з with закриває з'єднання, коли блок завершується. Тайм-аут клієнта встановлюється явно; SDK також самостійно повторює деякі невдалі запити (max_retries, 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)

Помилки#

Помилки, що виникають до першого чанку — неправильний ключ, порожній баланс, невідома модель — виникають як винятки, коли ви створюєте потік:

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)

Повний список кодів стану та виправлень наведено в Поширені помилки з'єднання та як їх виправити.

Що передається мережею#

Для налагодження перегляньте необроблений потік за допомогою curl (-N вимикає буферизацію):

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

Кожна подія — це рядок data: {…} із JSON-чанком; потік завершується data: [DONE]. Рядки, що починаються з двокрапки, — це коментарі keep-alive, які надсилаються під час довгих пауз — клієнти їх ігнорують.

Теги:

Налаштування#

Мінімальний приклад#

Моделі з міркуванням#

Використання токенів#

asyncio#

Дострокова зупинка та тайм-аути#

Помилки#

Що передається мережею#

Схожі статті

Поширені помилки підключення та як їх виправити

Що таке токен в AI? Пояснення простою мовою

Ваш перший API-запит із CheapTokens

Ціни API DeepSeek у 2026 році — скільки ви насправді платите