Почему нельзя парсить LLM-стрим по сетевым чанкам: SSE на Python и тест на каждый байт

Потоковый ответ может начаться нормально и закончиться полусловом. Разберу на воспроизводимом примере, где заканчивается сетевой чанк, начинается SSE-событие и почему ни одно из них ещё не означает «ответ готов». Ниже — парсер на стандартной библиотеке Python и тесты без платных запросов к модели.

Не проблема модели, а ошибка на границе слоёв

Представим небольшой backend: он получает поток от LLM-провайдера и передаёт текст в интерфейс. В тесте всё работает, а после прокси или изменения размера буфера внезапно появляются ошибки JSON и пропавшие фрагменты.

Подозрительное место — json.loads(chunk) внутри цикла чтения HTTP. У сетевого фрагмента нет обязанности совпадать с сообщением приложения: в нём может лежать половина строки, несколько событий или часть многобайтного символа. Для разбора возьму синтетический поток: ошибка воспроизводится без API-ключа и не зависит от выбранной модели.

Я разделяю четыре уровня: байты транспорта → строки UTF-8 → события SSE → состояние ответа конкретного API. Если смешать их в одном цикле, ошибка парсинга легко превращается в «провайдер нестабилен».

Минимальный контракт до кода

Для SSE событие отделяется пустой строкой. Несколько полей data объединяются через перевод строки; комментарии с двоеточия не являются текстом ответа. Переносы могут быть LF, CRLF или CR, включая разрыв CRLF между двумя чтениями.

Ключевая деталь: EOF не заменяет пустую строку. Незавершённое событие надо отбросить, даже если его data выглядит как корректный JSON. Эти правила сверены с HTML Standard, разделы 9.2.5–9.2.6.

Ниже — ограниченный парсер data/event для backend-клиента. Это не полная реализация браузерного EventSource: здесь нет id, retry, автоматического reconnect и Last-Event-ID. Невалидный UTF-8 намеренно вызывает ошибку, а не заменяется символом �. Это более строгая политика клиента, не попытка побайтно повторить браузер.

Парсер: сначала восстановить событие, потом читать JSON

Сохраните код как sse_parser.py. На вход поступает итератор bytes уже после HTTP-декодирования; на выходе — пары (тип события, данные). Если SDK уже выдаёт типизированные события, этот слой писать заново не нужно.

import codecs def parse_sse(chunks, *, max_line=65_536, max_event=262_144, max_total=8_388_608): """Yield (event_type, data). No reconnect, id or retry handling.""" def bounded_chunks(): total = 0 for chunk in chunks: total += len(chunk) if total > max_total: raise ValueError("stream byte limit") yield chunk line, data = [], [] kind, skip_lf, event_size = "", False, 0 # Strict UTF-8 is an intentional fail-closed policy for this API client. for text in codecs.iterdecode(bounded_chunks(), "utf-8-sig", errors="strict"): for char in text: if skip_lf: skip_lf = False if char == "\n": continue event_size += 1 if event_size > max_event: raise ValueError("event character limit") if char not in "\r\n": line.append(char) if len(line) > max_line: raise ValueError("line character limit") continue skip_lf = char == "\r" value = "".join(line) line.clear() if not value: if data: yield kind or "message", "\n".join(data) data, kind, event_size = [], "", 0 continue if value.startswith(":"): continue field, colon, value = value.partition(":") if colon and value.startswith(" "): value = value[1:] if field == "data": data.append(value) elif field == "event": kind = value # EOF is not a blank line: do NOT dispatch a pending event.

Важны не столько строки кода, сколько ограничения. max_line и max_event считаются в символах после декодирования; CRLF считается одним разделителем. max_total — в байтах входного потока. Значения в примере — учебные настройки, не универсальные лимиты для любого провайдера.

Бюджет события учитывает и комментарии до пустой строки. Общий лимит защищает от бесконечной последовательности коротких heartbeat. Но он не ограничивает время ожидания: read timeout, общий абсолютный deadline и закрытие ответа при отмене должен обеспечить HTTP-слой. Размер одного чтения там тоже нужно ограничить — парсер не отменит уже выделенную память.

Тест, который полезнее одного успешного запроса

Один и тот же корректный поток должен давать одинаковые события независимо от разбиения на чанки. Проверим все позиции разреза, а затем подадим поток по одному байту. В строке намеренно есть кириллица, иероглиф и emoji.

Запустите следующий код рядом с sse_parser.py:

from sse_parser import parse_sse wire = ( "\ufeff: ping\r\nevent: delta\r\n" "data: Привет, 猫 🐈\r\n" "data: второй ряд\r\n\r\n" ).encode("utf-8") expected = [("delta", "Привет, 猫 🐈\nвторой ряд")] for cut in range(len(wire) + 1): assert list(parse_sse([wire[:cut], wire[cut:]])) == expected assert list(parse_sse(bytes([b]) for b in wire)) == expected assert list(parse_sse([b"data: complete\n\n"])) == [ ("message", "complete") ] assert list(parse_sse([b"data: incomplete\n"])) == [] print("all byte boundaries + EOF: OK")

Для статьи я также прогнал расширенный набор из 9 тестов: BOM, три варианта переноса, пустое data, неизвестные поля, незавершённый хвост, повреждённый UTF-8, ограничения размеров и пустые чанки. В тестах корректных потоков проверяются все одиночные разрезы и случайные многократные разбиения с фиксированным seed. Все 9 прошли локально; это проверка примера, не сертификация интеграции с каждым API.

Событие прочитано — ответ ещё не завершён

Парсер не знает, что у провайдера означает успешное завершение. [DONE] — не специальное слово стандарта SSE. Названия terminal-событий, ошибки внутри потока, причины остановки и финальная статистика расходов относятся к контракту конкретного API.

Поэтому поверх парсера нужен отдельный адаптер. Сначала он разбирает payload и проверяет тип события. Затем накапливает допустимые дельты. Статус completed выставляет только после предусмотренного контрактом успешного финала и нужных проверок результата.

Если соединение закончилось без такого финала, я бы оставлял состояние incomplete, а уже показанный текст помечал частичным. Такой ответ нельзя незаметно положить в кэш как полный. Не надо и склеивать его с новой генерацией: повтор — отдельная попытка, а не гарантированное продолжение.

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

Когда свой парсер оправдан

Для обычного клиента я сначала выберу поддерживаемый SDK или проверенную SSE-библиотеку. Свой код имеет смысл, если вы строите gateway, исследуете несовместимость прокси или должны одинаково тестировать несколько транспортных адаптеров.

Цена собственного слоя — не только поддержка синтаксиса. Придётся тестировать отмену, backpressure, disconnect клиента, лимиты и обновления провайдерских событий. Синхронный пример выше проверяет framing; он не является готовым асинхронным gateway и не должен блокировать event loop.

Практический минимум перед релизом: тестировать произвольные границы байтов; отдельно учитывать stream_started и stream_completed; измерять incomplete_rate; не путать heartbeat с прогрессом генерации; не логировать целиком приватные ответы ради диагностики.

Мой вывод: поток нельзя считать успешным только потому, что он открылся и показал несколько слов. Надёжность здесь начинается с двух независимых проверок: «событие собрано правильно» и «задача действительно завершена».

Вы доверяете обработку стрима SDK или держите собственный gateway? Как проверяете обрыв после первого текста, но до финального события?

Техническая основа: HTML Standard, Server-sent events, §9.2.5–9.2.6 https://html.spec.whatwg.org/multipage/server-sent-events.html#parsing-an-event-stream

Python codecs.iterdecode https://docs.python.org/3/library/codecs.html#codecs.iterdecode