Стриминг ответов LLM по SSE и стоимость в последнем чанке

Стриминг делает интерфейс чата мгновенным на вид, но меняет и то, как работает биллинг: usage неизвестен, пока модель не закончит генерацию, поэтому он передаётся внутри самого потока, а не в заголовке ответа. В этом посте — разбор SSE-потока genmux вручную (полезно, если вы не используете OpenAI SDK или хотите понимать, что он делает под капотом) и что происходит при обрыве соединения посреди генерации.

Формат потока

POST /v1/chat/completions с "stream": true возвращает text/event-stream: последовательность строк data: {...}\n\n, каждая — JSON-объект chat.completion.chunk, а в конце — буквально data: [DONE]\n\n. Каждый чанк, кроме последнего, несёт delta со следующим кусочком текста:

data: {"id":"gnx_9f2c...","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"Волга"},"finish_reason":null}]}

data: {"id":"gnx_9f2c...","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":" течёт"},"finish_reason":null}]}

data: {"id":"gnx_9f2c...","object":"chat.completion.chunk","choices":[{"index":0,"delta":{},"finish_reason":"stop"}],"usage":{"prompt_tokens":12,"completion_tokens":47,"total_tokens":59,"cost":0.071,"cost_source":"estimated"}}

data: [DONE]

Последний чанк перед [DONE] содержит пустую choices[].delta и заполненный usage на верхнем уровне — это единственная строка во всём потоке, которая сообщает стоимость вызова. cost_source для стриминговых вызовов всегда равен "estimated": апстрим-вендоры не отдают точный usage для потокового ответа, поэтому genmux вычисляет его по числу видимых токенов.

Разбор на Python

import json
import requests

def stream_chat(model: str, messages: list[dict], **kwargs):
    with requests.post(
        "https://api.genmux.tech/v1/chat/completions",
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={"model": model, "messages": messages, "stream": True, **kwargs},
        stream=True,
        timeout=60,
    ) as resp:
        resp.raise_for_status()
        usage = None
        for line in resp.iter_lines(decode_unicode=True):
            if not line or not line.startswith("data: "):
                continue
            payload = line[len("data: "):]
            if payload == "[DONE]":
                break
            chunk = json.loads(payload)
            delta = chunk["choices"][0]["delta"] if chunk["choices"] else {}
            if "content" in delta:
                print(delta["content"], end="", flush=True)
            if chunk.get("usage"):
                usage = chunk["usage"]
        print()
        return usage

usage = stream_chat("deepseek-v4-flash", [{"role": "user", "content": "Опиши Волгу одним предложением."}])
print(usage)  # {'prompt_tokens': 12, 'completion_tokens': 47, 'total_tokens': 59, 'cost': 0.071, 'cost_source': 'estimated'}

Два момента стоит держать в голове: проверяйте chunk["choices"] перед обращением по индексу (в последнем чанке массив может быть пустым) и проверяйте chunk.get("usage") в каждом чанке, а не предполагайте, что он есть только в последнем, — так код останется рабочим, даже если формат чанков немного изменится.

Разбор на Node

async function streamChat(model, messages, extra = {}) {
  const resp = await fetch("https://api.genmux.tech/v1/chat/completions", {
    method: "POST",
    headers: {
      Authorization: `Bearer ${process.env.GENMUX_API_KEY}`,
      "Content-Type": "application/json",
    },
    body: JSON.stringify({ model, messages, stream: true, ...extra }),
  });

  const reader = resp.body.getReader();
  const decoder = new TextDecoder();
  let buffer = "";
  let usage = null;

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    buffer += decoder.decode(value, { stream: true });
    const lines = buffer.split("\n\n");
    buffer = lines.pop(); // оставляем незавершённый хвост для следующего чтения

    for (const line of lines) {
      if (!line.startsWith("data: ")) continue;
      const payload = line.slice(6);
      if (payload === "[DONE]") return usage;
      const chunk = JSON.parse(payload);
      const delta = chunk.choices[0]?.delta ?? {};
      if (delta.content) process.stdout.write(delta.content);
      if (chunk.usage) usage = chunk.usage;
    }
  }
  return usage;
}

Если вы уже используете openai SDK для Node или Python, весь этот разбор — не ваша забота: передайте stream: true (в Python — stream=True) и читайте chunk.usage в том чанке, где оно есть; разбиение на строки и парсинг JSON SDK берёт на себя.

Обрыв посреди стрима

Если клиент закрывает соединение или пользователь уходит со страницы до [DONE], genmux списывает ровно за токены, сгенерированные к этому моменту — не больше, и без фиксированного «штрафа за отмену». Здесь есть важная асимметрия, вокруг которой стоит проектировать:

На практике это значит, что кнопку «отменить» в чат-интерфейсе можно подключать без специальной обработки на стороне биллинга: обрывайте соединение, когда захочет пользователь, — счёт будет отражать ровно то, что он увидел на экране.

Сверка постфактум

Если процесс упал посреди стрима и чанк с usage потерян, вызов не пропадает: каждый чат-вызов получает id задания genmux (поле id в каждом чанке), и GET /v1/generations/{id} вернёт полную запись с usage и cost после того, как задание рассчитается. Для периодической сверки GET /v1/generations?capability=text.chat выдаёт список всех текстовых вызовов аккаунта — это более надёжный источник истины, чем то, что успел захватить клиент вживую.

С чего начать

Стриминг работает одинаково для всех текстовых моделей платформы — DeepSeek, Qwen3, Doubao и другие модели по мере запуска — без особенностей формата потока по моделям. Полное описание endpoint'а — в документации; для reasoning-моделей стоит также прочитать как тарифицируется thinking-режим, поскольку токены рассуждений идут через то же поле delta.content перед финальным ответом.