Назад в блог

API и SDKИнструкция

Потоковые ответы через Python

Выводим ответ модели по мере генерации с официальным пакетом openai — синхронный и асинхронный примеры, рассуждения модели, учёт токенов и обработка ошибок.

Автор: CheapTokens TeamОпубликовано 3 мин чтения

Код на Python с циклом по частям потока и текстом, который появляется по токенам
Содержание
  1. Подготовка
  2. Минимальный пример
  3. Рассуждающие модели
  4. Число токенов
  5. asyncio
  6. Досрочная остановка и тайм-ауты
  7. Ошибки
  8. Что передаётся по сети

В потоковом режиме ответ приходит частями, пока модель его ещё генерирует: пользователь сразу видит текст, а не ждёт ответ целиком. API использует стандартный потоковый формат OpenAI (server-sent events), а официальный пакет openai для Python всё разбирает сам.

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

Подготовка#

bash
pip install openai
export CHEAPTOKENS_API_KEY="вставьте-ваш-ключ"

Минимальный пример#

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": "Напиши хокку о задержках сети"}],
    stream=True,
)
for chunk in stream:
    if not chunk.choices:  # служебная часть без текста, например последняя — с числом токенов
        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": "Является ли 1013 простым числом?"}],
    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)  # рассуждение → stderr
    if delta.content:
        print(delta.content, end="", flush=True)              # ответ → 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": "Объясни идемпотентность одним абзацем"}],
    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": "Назови три применения очереди сообщений"}],
        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,  # секунды; длинным ответам нужно время
)

received = ""
with client.chat.completions.create(
    model="deepseek-ai/DeepSeek-V4-Flash-0731",
    messages=[{"role": "user", "content": "Напиши длинный рассказ о маяке"}],
    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:  # досрочная остановка: текста достаточно
            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": "Привет"}],
        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: неверный или заменённый ключ
    print("Проверьте API-ключ")
except openai.BadRequestError as e:         # 400: например, неизвестный ID модели
    print("Ошибка в запросе:", e.message)
except openai.APIStatusError as e:          # другие коды, например 402, когда закончился баланс
    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]. Строки, начинающиеся с двоеточия, — служебные комментарии, которые приходят во время долгих пауз, чтобы соединение не обрывалось; клиенты их пропускают.

Теги: