Потоковые ответы

SSE, stream: true, примеры curl и Python, отмена потока и обработка ошибок.

Чтобы получать текст от AI Gateway по мере генерации, отправьте POST на /v1/chat/completions с полем "stream": true и обрабатывайте приходящие части ответа последовательно. Это сокращает время до появления первого текста в интерфейсе: пользователю не нужно ждать, пока модель сформирует ответ целиком.

Поток передаётся как SSE — обычное HTTP-соединение остаётся открытым, а сервер отправляет в нём небольшие события одно за другим. Для Chat Completions полезный текст обычно находится в choices[0].delta.content; клиент должен собирать эти фрагменты в итоговую строку. Если вы впервые подключаете API, сначала пройдите быстрый старт, а описание базового адреса и полей запроса смотрите в справочнике API.

Что понадобится

API-ключ в переменной окружения AI_GATEWAY_API_KEY;
базовый URL https://route.smartaipack.ru/v1;
точный model ID модели, которая поддерживает /v1/chat/completions, из раздела Модели;
HTTP-клиент, который не буферизует ответ целиком, или пакет openai для Python.

В примерах используется gpt-5.5. Если такого ID нет в вашем каталоге, замените его на доступный идентификатор без сокращений и переименований.

Как работает потоковая передача

Без stream приложение получает один JSON после завершения генерации. С "stream": true ответ приходит частями в том же HTTP-соединении. В SSE каждое событие представлено строкой данных, после которой идёт пустая строка. OpenAI-совместимый клиент разбирает эти события сам и отдаёт приложению отдельные объекты chunk.

Начальные части могут не содержать текста: в них встречаются роль ассистента или служебные поля. Поэтому проверяйте choices, а затем delta.content, не печатая None. Последняя часть может содержать finish_reason. Набор дополнительных полей, счётчики usage и точное разбиение текста на фрагменты могут различаться у моделей и поставщиков.

Пример с curl

Сначала задайте ключ в текущей сессии терминала:

export AI_GATEWAY_API_KEY="<ВАШ_API-КЛЮЧ>"

Затем отправьте запрос. Флаг --no-buffer просит curl выводить поступившие данные сразу, а не накапливать их перед показом:

curl --no-buffer https://route.smartaipack.ru/v1/chat/completions \
  -H "Authorization: Bearer $AI_GATEWAY_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "gpt-5.5",
    "messages": [
      {
        "role": "user",
        "content": "Объясни простыми словами, зачем нужен потоковый ответ."
      }
    ],
    "stream": true
  }'

В терминале будут появляться отдельные SSE-события. Не связывайте один фрагмент с одним словом: границы delta.content определяются моделью и поставщиком, поэтому часть может содержать символ, несколько слов или не содержать текста.

Пример с Python OpenAI SDK

OpenAI-совместимый пакет openai скрывает разбор строк SSE. Приложение получает итератор и обрабатывает каждый chunk по мере поступления:

import os

from openai import OpenAI

client = OpenAI(
    base_url="https://route.smartaipack.ru/v1",
    api_key=os.environ["AI_GATEWAY_API_KEY"],
)

stream = client.chat.completions.create(
    model="gpt-5.5",
    messages=[
        {
            "role": "user",
            "content": "Объясни простыми словами, зачем нужен потоковый ответ.",
        }
    ],
    stream=True,
)

parts: list[str] = []

try:
    for chunk in stream:
        if not chunk.choices:
            continue

        text = chunk.choices[0].delta.content
        if text:
            parts.append(text)
            print(text, end="", flush=True)
finally:
    stream.close()

full_text = "".join(parts)
print()

flush=True нужен, чтобы терминал или серверный процесс не задерживал уже полученные символы в собственном буфере. Список parts позволяет одновременно показывать ответ и собрать итоговый текст для дальнейшей обработки в приложении.

Дополнительная информация

Для интерфейса полезно разделить три состояния: запрос отправлен, получен первый текстовый фрагмент, поток завершён. Индикатор загрузки можно убрать после первого непустого delta.content, но кнопку отмены стоит оставить доступной до завершения итератора.

Не запускайте тяжёлую операцию рендеринга на каждый фрагмент. На практике удобнее накапливать текст и обновлять интерфейс небольшими порциями. При этом сохраняйте порядок получения: параллельная обработка отдельных chunk может переставить части местами.

Если вам важна причина завершения, сохраните последний известный finish_reason. Не считайте закрытие соединения доказательством полного ответа: поток мог завершиться штатно, оборваться в сети или быть остановлен клиентом.

Отмена потока

В curl поток можно прервать сочетанием Ctrl+C. В Python прекратите итерацию и закройте объект потока:

stream = client.chat.completions.create(
    model="gpt-5.5",
    messages=[{"role": "user", "content": "Напиши длинный рассказ."}],
    stream=True,
)

try:
    for chunk in stream:
        if user_requested_cancel():
            break

        if chunk.choices:
            text = chunk.choices[0].delta.content
            if text:
                print(text, end="", flush=True)
finally:
    stream.close()

Функция user_requested_cancel() здесь обозначает проверку вашего флага отмены; её нужно реализовать в приложении. Закрытие клиентского соединения прекращает получение данных вашим кодом, но не следует считать это гарантией, что поставщик немедленно остановил расчёт или что списание отсутствует. После отмены проверьте фактический запрос и расход в разделе Использование, а сводные данные — в Аналитике.

Ошибки при потоковой передаче

Обработка зависит от момента сбоя. До начала SSE сервер ещё может вернуть обычную HTTP-ошибку. После начала потока HTTP-статус уже отправлен, поэтому проблема может проявиться как исключение клиента, обрыв соединения или отдельное событие поставщика.

Ошибка до первых данных

Если запрос отклонён до открытия успешного потока, проверяйте HTTP-статус и тело ошибки. Частые причины: неверный или истёкший ключ, нулевой баланс, превышение лимита, некорректный JSON, неизвестный model ID или путь, который модель не поддерживает.

Ответ на этом этапе не обязательно имеет форму chunk. Не начинайте разбор choices[0].delta, пока клиентская библиотека не подтвердила успешный ответ. Исправляйте ошибки запроса и авторизации без автоматического повтора; для временных сетевых ошибок и 429 используйте ограниченное число повторов с увеличивающейся задержкой.

Ошибка после начала потока

Если несколько частей уже получены, считайте накопленный текст неполным, пока не увидели штатное завершение. Сохраните его отдельно от законченного ответа и покажите пользователю понятный статус, например «Ответ прерван».

Не рассчитывайте на один точный формат ошибки внутри уже начавшегося потока для всех поставщиков. Клиент должен уметь обработать как событие без ожидаемого delta.content, так и сетевой обрыв или исключение при чтении. Автоматический повтор всего запроса может создать второй ответ и дополнительный расход, поэтому повторяйте его только по явной политике приложения.

Пример обработки ошибок

Следующий пример отдельно обрабатывает отказ до начала потока и сбой во время чтения. Он не пытается склеить результат повторного запроса с уже полученным текстом:

import os
import sys

from openai import APIConnectionError, APIError, APIStatusError, OpenAI

client = OpenAI(
    base_url="https://route.smartaipack.ru/v1",
    api_key=os.environ["AI_GATEWAY_API_KEY"],
    timeout=60.0,
)

try:
    stream = client.chat.completions.create(
        model="gpt-5.5",
        messages=[{"role": "user", "content": "Дай краткий ответ."}],
        stream=True,
    )
except APIStatusError as error:
    print(f"Запрос отклонён: HTTP {error.status_code}", file=sys.stderr)
    raise
except APIConnectionError:
    print("Не удалось открыть поток.", file=sys.stderr)
    raise

parts: list[str] = []
completed = False

try:
    for chunk in stream:
        if not chunk.choices:
            continue

        choice = chunk.choices[0]
        if choice.delta.content:
            parts.append(choice.delta.content)
            print(choice.delta.content, end="", flush=True)

        if choice.finish_reason is not None:
            completed = True
except APIError:
    print("\nПоток прерван; полученный текст неполный.", file=sys.stderr)
finally:
    stream.close()

if not completed:
    partial_text = "".join(parts)
    # Сохраните partial_text как незавершённый результат или отбросьте его.

В рабочей среде задайте ограничение времени ожидания, журналируйте время, путь, модель и доступный идентификатор запроса, но никогда не записывайте полный API-ключ. Если причина неясна, передайте эти данные в поддержку.

Особенности API

Потоковый режим включается полем stream: true в JSON для POST /v1/chat/completions.
Для прямого HTTP-клиента нужно читать SSE последовательно и отключить локальную буферизацию вывода. В текущем маршруте AI Gateway ответ SSE передаётся без буферизации на прокси.
Для Python используйте обычный пакет openai с base_url="https://route.smartaipack.ru/v1".
Текст обычно приходит в choices[0].delta.content, а не в choices[0].message.content, используемом для непотокового ответа.
Не полагайтесь на одинаковые границы фрагментов, наличие usage в каждом событии или единый формат ошибки после начала потока.
После завершения, отмены или сбоя сверяйте фактическое использование в журнале запросов.

Потоковые ответы

На этой странице