Стриминг ответа LLM в Python: чек-лист перед продом

Стриминг ответа модели выглядит просто: открыл соединение с параметром stream=True, крутишь цикл по чанкам, отдаёшь токены клиенту. На демо работает с первого раза. В проде начинается другое: клиент закрыл вкладку, а генерация продолжает жечь токены; провайдер оборвал поток на середине, и пользователь получил обрезанный ответ без ошибки; вы решили стримить структурированный JSON и половину времени парсер падает на неполном объекте. Это чек-лист того, что нужно проверить до релиза, а не после первого инцидента.

Тема — именно надёжность стриминга под реальными клиентами. Выбор между SSE и WebSocket, устройство самих моделей и промпт-инжиниринг здесь не рассматриваются. Примеры на Python 3.11+, SDK openai 1.x и anthropic 0.39+; синтаксис у двух провайдеров разный, принципы одни.

Что вообще ломается при стриминге

Обычный запрос атомарен: либо получили полный ответ, либо ошибку. Стриминг растягивает ответ во времени, и каждая секунда этого времени — окно для сбоя. Основные точки отказа:

• Обрыв со стороны провайдера. Соединение закрылось до финального чанка. HTTP-статус при этом уже был 200 — ошибку вы получите в виде исключения посреди итерации или тихо пустого хвоста.
• Уход клиента. Пользователь закрыл вкладку, но ваш сервер продолжает читать поток от модели и платить за токены.
• Незакрытые соединения. Забыли закрыть стрим при исключении — соединение висит в пуле, под нагрузкой они кончаются.
• Частичный вывод. Клиент уже отрендерил половину ответа, а генерация упала. Откатить нельзя — токены на экране.

Проверка причины остановки

Самое коварное — тихий обрыв. Модель может остановиться не потому, что закончила мысль, а потому что упёрлась в лимит токенов или сработал фильтр. У OpenAI это поле finish_reason в последнем чанке, у Anthropic — событие message_delta с полем stop_reason.

# openai 1.x
from openai import OpenAI

client = OpenAI()
finish_reason = None

stream = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Объясни стриминг"}],
    stream=True,
)

for chunk in stream:
    choice = chunk.choices[0]
    if choice.delta.content:
        yield choice.delta.content
    # finish_reason приходит в последнем чанке, до него он None
    if choice.finish_reason is not None:
        finish_reason = choice.finish_reason

# stop — модель закончила сама; length — упёрлась в max_tokens
if finish_reason == "length":
    # ответ обрезан, это не полноценный результат
    log.warning("Ответ обрезан по лимиту токенов")

Значение stop — единственное, при котором ответ считается завершённым штатно. Если пришло length, content_filter или итерация закончилась без finish_reason вообще — это оборванный ответ, и обрабатывать его как полноценный нельзя.

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

Таймаут между чанками, а не на весь запрос

Обычный таймаут HTTP-запроса для стрима бесполезен: сам запрос длится столько, сколько идёт генерация, — это законно минуты. Опасна другая ситуация — поток завис между чанками. SDK openai и anthropic используют httpx, и таймаут задаётся на клиенте. Важна составляющая read — сколько ждать очередной порции данных.

import httpx
from openai import OpenAI

# connect — на установку соединения, read — между порциями данных
timeout = httpx.Timeout(connect=5.0, read=30.0, write=10.0, pool=5.0)
client = OpenAI(timeout=timeout)

Если между двумя чанками прошло больше read-секунд, httpx поднимет ReadTimeout — вы узнаете о зависании, а не будете ждать вечно.

Отмена, когда клиент ушёл

При стриминге через FastAPI или Starlette клиент может отключиться в любой момент. Если это не обработать, ваш сервер продолжит читать поток от модели до конца, оплачивая токены, которые никто не увидит. В async-стеке нужно закрывать стрим при отмене корутины.

# FastAPI + openai async, версия 1.x
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import AsyncOpenAI

app = FastAPI()
client = AsyncOpenAI()

@app.post("/chat")
async def chat(prompt: str):
    async def event_stream():
        stream = await client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": prompt}],
            stream=True,
        )
        try:
            async for chunk in stream:
                delta = chunk.choices[0].delta.content
                if delta:
                    yield delta
        finally:
            # закрываем поток при любом выходе, включая отмену клиентом
            await stream.close()

    return StreamingResponse(event_stream(), media_type="text/event-stream")

Ключевой момент — блок finally с явным закрытием. Когда клиент отключается, Starlette отменяет корутину-генератор, и finally отрабатывает, освобождая соединение к провайдеру. Без этого соединения утекают, и пул httpx под нагрузкой исчерпается.

Стриминг структурированного вывода

Если вы стримите JSON, помните: пока генерация не закончена, у вас на руках синтаксически невалидный кусок. Парсить каждый промежуточный чанк обычным json.loads нельзя — он упадёт на первой же неполной строке.

Есть два честных пути. Первый — накапливать весь ответ и парсить один раз в конце, потеряв смысл стриминга для клиента. Второй — использовать инкрементальный парсер, который умеет достраивать частичный JSON. Здесь помогает провайдерский режим структурированного вывода: у OpenAI это response_format с JSON-схемой, который гарантирует валидность финального объекта. Но именно финального — промежуточные чанки всё равно неполные.

• Нужен прогресс на экране — стримьте текст, а не JSON, и структурируйте отдельным быстрым запросом в конце.
• Нужен именно структурированный объект — не стримьте его по-настоящему, соберите буфер и распарсите один раз.
• Нужно и то и другое — смотрите в сторону библиотек инкрементального парсинга JSON; проверьте на сайте PyPI, какие поддерживаются и в каких версиях, вместо того чтобы полагаться на имя из памяти.

Сравнение подходов к сборке ответа

Типичные ошибки

• Не проверяют finish_reason. Обрезанный по лимиту ответ уходит клиенту как полноценный.
• Ставят общий таймаут на запрос. Либо режут долгие легитимные генерации, либо не ловят зависания между чанками.
• Не закрывают стрим в finally. Утечка соединений всплывает только под нагрузкой, на тестах всё зелёное.
• Парсят промежуточный JSON. Падение на неполном объекте, которое воспроизводится через раз.
• Не считают токены оборванного ответа. Провайдер тарифицирует сгенерированное, даже если клиент ушёл, — расход растёт незаметно.

Что настроить перед релизом

Соберите минимальный набор наблюдаемости: логируйте finish_reason каждого ответа, считайте долю обрывов и обрезаний, меряйте время до первого токена (TTFB) и время между чанками. Именно TTFB чувствуют пользователи, а рост межчанкового интервала — первый признак деградации на стороне провайдера. Проверьте поведение под read-таймаутом на стенде: искусственно замедлите ответ и убедитесь, что соединения закрываются, а не копятся в пуле.