Grok API

Streaming API для Grok

View as Markdown

Обновлено:

Стриминг Grok используют, чтобы показывать ответ пользователю сразу по мере генерации, а не ждать завершения всего запроса. Streaming API передаёт текст и служебные события через Server-Sent Events (SSE).

Если нужен один готовый объект ответа, используйте Messages API.

Как работает стриминг

HTTP-соединение остаётся открытым и передаёт упорядоченные события до завершения ответа или отмены клиентом.

Включите стриминг в теле запроса"stream": true
01

Транспорт

SSE работает внутри одного HTTP-соединения.

02

Данные

Каждое событие потока отражает этап жизненного цикла генерации.

03

Результат

Стриминг позволяет показывать процесс генерации в реальном времени.

Запуск потокового ответа

curl -N https://api.llm-gate.tech/v1/responses \
  -H "Authorization: Bearer $GROK_API_DEV_KEY" \
  -H "Content-Type: application/json" \
  -H "Accept: text/event-stream" \
  -d '{
    "model": "grok-4.5",
    "input": "Объясни бинарный поиск в двух предложениях",
    "stream": true
  }'

Типы событий

Во время прослушивания стрима клиент получает события разных типов. Ниже приведены актуальные события для обычного текстового ответа и назначение каждого из них.

СобытиеЧто делать
response.createdСохранить response ID и отметить запрос как выполняющийся.
response.output_item.addedДобавлен типизированный output: текст, вызов инструмента или другой элемент.
response.content_part.addedВнутри сообщения началась новая часть content.
response.output_text.deltaДобавить delta к видимому ответу строго по порядку.
response.output_text.doneТекущая текстовая часть завершена.
response.completedЗафиксировать полный ответ, статус и usage.
errorОстановить вывод и показать безопасное состояние ошибки.

Пример потока событий

1
response.created
{"response":{"id":"resp_01abc","status":"in_progress"}}
2
response.output_text.delta
{"delta":"Binary"}
3
response.output_text.delta
{"delta":" search"}
4
response.completed
{"response":{"status":"completed","usage":{"input_tokens":18,"output_tokens":42,"total_tokens":60}}}

Пример кода для работы со стримингом

import json
import os
import httpx

with httpx.stream(
    "POST",
    "https://api.llm-gate.tech/v1/responses",
    headers={
        "Authorization": f"Bearer {os.environ['GROK_API_DEV_KEY']}",
        "Accept": "text/event-stream",
    },
    json={
        "model": "grok-4.5",
        "input": "Объясни бинарный поиск в двух предложениях",
        "stream": True,
    },
    timeout=3600.0,
) as response:
    response.raise_for_status()
    for line in response.iter_lines():
        if not line.startswith("data:"):
            continue

        data = line.removeprefix("data:").strip()
        if not data or data == "[DONE]":
            continue

        event = json.loads(data)
        if event.get("type") == "response.output_text.delta":
            print(event["delta"], end="", flush=True)

Обработка ошибок

HTTP-ошибка до SSE

Если статус ответа не 2xx, разберите обычное JSON-тело ошибки.

Событие error после подключения

Поток может оборваться после получения события типа error. Оставьте частичный результат, отметьте его незавершённым и сохраните response ID.

Соединение закрылось раньше

Если соединение закрылось до response.completed, считайте ответ незавершённым. Повторный запрос сформирует новый ответ, поэтому не объединяйте его автоматически с уже полученным текстом.

429 или временная 5xx

Используйте ограниченный exponential backoff с jitter, только если полезный результат ещё не зафиксирован.