Один SSE для четырёх LLM: стриминг OpenAI, Anthropic, DeepSeek и Kimi через один бэкенд

Мы делаем чат-агрегатор, где в одном окне доступны GPT, Claude, Kimi и DeepSeek. Фронтенду нужно отдавать ответ в реальном времени — токен за токеном, как в ChatGPT. Бэкенд при этом ходит к четырём разным API, и стриминг у них устроен по-разному. Расскажу, как мы свели это к единому SSE-потоку наружу, и про две грабли, на которые наступили: рваные UTF-8 символы и парсинг чужих SSE.

Статья будет полезна всем, кто проксирует LLM через свой сервер.

Зачем вообще свой прокси

Фронтенд не должен знать ключи провайдеров и не должен ходить к ним напрямую. Все запросы идут через наш Node.js-бэкенд: он подставляет ключ, дёргает нужный API с stream: true, парсит входящий поток и отдаёт фронту унифицированные события. Плюс на бэкенде живут лимиты, учёт токенов и подмена провайдера.

Задача: «получить поток от провайдера X → распарсить → отдать фронту в едином формате».

Два разных формата стриминга

Провайдеры делятся на два лагеря.

1. OpenAI-совместимые (GPT, DeepSeek, Kimi). SSE, где в каждом событии лежит delta:

   data: {“choices”:[{“delta”:{“content”:“При”}}]} data: {“choices”:[{“delta”:{“content”:“вет”}}]} data: [DONE]

2. Anthropic (Claude). Своя событийная модель с типами:

   data: {“type”:“message_start”,“message”:{“usage”:{“input_tokens”:10}}} data: {“type”:“content_block_delta”,“delta”:{“type”:“text_delta”,“text”:“При”}} data: {“type”:“message_delta”,“usage”:{“output_tokens”:5}}

Текст лежит в разных местах, события называются по-разному, токены usage приходят в разных местах потока. Нам нужно привести всё к одному виду.

Единый формат наружу

Договорились с фронтом о простом протоколе поверх SSE:

data: {“t”:“кусок текста”} // дельта data: {“done”:true,“full”:“весь текст”} data: [DONE]

Дальше — два обработчика, по одному на каждый лагерь.

Парсинг OpenAI-совместимого потока

Чанки из сокета не совпадают с границами SSE-событий, поэтому буферизуем и режем по разделителю \n\n:

let buf = “”; proxyRes.setEncoding(“utf-8”); proxyRes.on(“data”, (chunk) => { buf += chunk; const parts = buf.split(“\n\n”); buf = parts.pop() “”; // хвост — недособранное событие for (const part of parts) { for (const line of part.split(“\n”)) { const s = line.trim(); if (!s.startsWith(«data: «) s === “data: [DONE]”) continue; const evt = JSON.parse(s.slice(6)); const delta = evt.choices?.[0]?.delta?.content; if (delta) sseWrite(res, { t: delta }); } } });

Главное здесь — не парсить buf целиком на каждом чанке, а отрезать только завершённые события (до последнего \n\n), а недополученный хвост оставлять в буфере до следующего чанка.

Anthropic парсится так же, только вытаскиваем text из событий с типом content_block_delta, а usage собираем из message_start и message_delta.

Грабля №1: data += chunk ломает русские буквы

Сначала тело ответа мы собирали наивно:

let data = “”; proxyRes.on(“data”, chunk => data += chunk); // ❌

И в ответах периодически появлялись «битые символы» — кракозябры вместо части кириллицы или эмодзи. Причём не всегда, а будто случайно.

Причина: chunk — это Buffer, а не строка. Конкатенация data += chunk неявно вызывает chunk.toString() на каждом куске отдельно. Многобайтные UTF-8 символы (кириллица — 2 байта, эмодзи — 4) могут разорваться на границе сетевого пакета: первый байт символа уедет в конец одного чанка, второй — в начало следующего. toString() каждого чанка по отдельности декодирует половинку символа в U+FFFD — тот самый «ромбик с вопросом».

Чем выше нагрузка и длиннее ответ, тем чаще пакеты бьются не по символам — поэтому баг и казался плавающим.

Два рабочих решения:

1. Накапливать байты, декодировать один раз в конце:

   const chunks = []; proxyRes.on(“data”, c => chunks.push©); proxyRes.on(“end”, () => { const data = Buffer.concat(chunks).toString(“utf-8”); // ✅ });

2. Для стриминга, где декодировать нужно по ходу, — переложить склейку байтов на сам поток:

   proxyRes.setEncoding(“utf-8”); // ✅ теперь chunk — корректная строка, // поток сам держит «хвост» неполного символа

Второй вариант мы и используем в стриминговых обработчиках выше — обратите внимание на setEncoding(“utf-8”) перед on(“data”).

Вывод простой, но его легко забыть под нагрузкой: никогда не склеивайте сетевые чанки как строки. Либо Buffer.concat, либо setEncoding на потоке.

Грабля №2: usage приходит в последнем чанке

Количество токенов (для учёта и биллинга) у OpenAI прилетает в самом последнем событии перед [DONE], а у Anthropic — раздельно: input в message_start, output в message_delta. Если разбирать поток лениво и выходить по первому [DONE], можно потерять usage. Поэтому usage аккумулируем в переменные по ходу потока и фиксируем в on(“end”), там же отдаём фронту итоговое {done:true,full}.

Мелкие, но важные детали

— Таймаут на запрос к провайдеру (мы ставим 120с) + аккуратная отдача ошибки в том же SSE, а не обрыв соединения. — Если провайдер вернул не-200 — читаем тело ошибки целиком (через Buffer.concat, см. грабля №1) и отдаём фронту человеческое сообщение. — Фронт тоже буферизует по \n\n: частичное SSE-событие нельзя JSON.parse’ить.

Итог

Свести четыре разных стриминговых API к одному SSE-потоку — это в основном аккуратная работа с буферами и знание форматов каждого провайдера. Две главные ловушки — рваные UTF-8 на границах чанков и потерянный usage — стоили нам больше всего времени, хотя чинятся в одну строку.

Всё это крутится в нашем сервисе Nomi, но код и грабли универсальны для любого LLM-прокси. Если интересно, могу отдельно разобрать unified-формат сообщений и обработку отмены (abort) на стриме.

Пишите в комментариях, кто как решал UTF-8 на стриминге — встречали ли setEncoding-сюрпризы?