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

Вы включили stream=True, получили генератор и печатаете токены по мере поступления. В демо всё работает. В проде под сотней параллельных запросов пользователи жалуются: ответ обрывается на середине, иногда висит минуту без единого символа, а в логах — Connection reset by peer и растущее число открытых сокетов. Ниже — разбор четырёх мест, где потоковая обработка ломается, и как их закрыть. Примеры на Python 3.11 и SDK openai 1.x, но проблемы общие для любого провайдера, отдающего Server-Sent Events.

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

Модель отдаёт ответ не целиком, а порциями через протокол SSE^*. Каждая порция — это отдельное сообщение вида data: {...}, а конец потока помечается служебной строкой. SDK разбирает эти сообщения за вас и отдаёт объекты-дельты, где полезный текст лежит в choices[0].delta.content. Ключевое: дельта может быть пустой. Первый чанк часто несёт только роль (role: "assistant") без текста, а последние — причину остановки без контента.

Стриминг — это не «тот же ответ, только по буквам». Это поток событий, где текстовые куски перемешаны со служебными, и любой из них может оказаться пустым.

^* SSE (Server-Sent Events) — однонаправленный протокол поверх HTTP: сервер держит соединение открытым и шлёт клиенту сообщения по мере готовности. В отличие от WebSocket, канал только один — от сервера к клиенту.

Ошибка 1. Падение на пустой дельте

Самый частый способ прострелить себе ногу — обратиться к .content напрямую, считая, что там всегда строка:

# openai 1.x — так делать не надо
from openai import OpenAI

client = OpenAI(api_key="YOUR_API_KEY")

stream = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Привет"}],
    stream=True,
)

for chunk in stream:
    # На первом чанке content == None -> конкатенация упадёт
    print(chunk.choices[0].delta.content, end="")

В консоли вы увидите None в начале вывода, а если складываете куски в строку через +=, получите TypeError: can only concatenate str (not "NoneType") to str. Дельта без контента — норма, а не сбой. Проверяйте наличие текста явно:

parts = []
for chunk in stream:
    delta = chunk.choices[0].delta
    if delta.content:              # None и пустая строка отсекаются разом
        parts.append(delta.content)
        print(delta.content, end="", flush=True)

full_text = "".join(parts)

Два нюанса. Собирайте текст в список и склеивайте один раз через join — конкатенация строк в цикле на длинных ответах даёт квадратичную сложность. И не забудьте flush=True в print: без него буфер stdout копит вывод, и «потоковость» превращается в один залп в конце.

Ошибка 2. Битый JSON при потоковом парсинге аргументов

Отдельная боль — стриминг вызова инструмента (function calling). Аргументы функции приходят кусками, и каждый кусок — это фрагмент строки JSON, а не валидный JSON сам по себе. Попытка распарсить каждую дельту через json.loads обречена: вы получите json.decoder.JSONDecodeError: Expecting value на первом же чанке вроде {"loc.

Аргументы копятся в delta.tool_calls[i].function.arguments и парсятся только после закрытия потока:

import json
from collections import defaultdict

# Копим аргументы по индексу вызова: их может быть несколько
args_buffers = defaultdict(str)
tool_names = {}

for chunk in stream:
    for call in (chunk.choices[0].delta.tool_calls or []):
        idx = call.index
        if call.function.name:
            tool_names[idx] = call.function.name
        if call.function.arguments:
            args_buffers[idx] += call.function.arguments

# Парсим ТОЛЬКО когда поток закончился и строка собрана целиком
for idx, raw in args_buffers.items():
    try:
        parsed = json.loads(raw)
    except json.JSONDecodeError:
        # Поток мог оборваться — строка неполная, не считаем это валидным вызовом
        parsed = None
    print(tool_names.get(idx), parsed)

Здесь важен call.index: модель может запросить несколько инструментов сразу, их дельты приходят вперемешку, и без разбивки по индексу вы склеите аргументы разных вызовов в одну кашу.

Ошибка 3. Утечка соединений при обрыве на стороне клиента

SSE держит TCP-соединение открытым всё время генерации. Если пользователь закрыл вкладку, а вы просто вышли из цикла по break или поймали исключение и пошли дальше — соединение и HTTP-ответ остаются висеть. Под нагрузкой пул соединений исчерпывается, и новые запросы встают в очередь или падают по таймауту.

Стрим нужно закрывать явно. SDK openai 1.x возвращает объект, поддерживающий менеджер контекста и метод .close():

with client.chat.completions.create(
    model="gpt-4o-mini",
    messages=messages,
    stream=True,
) as stream:
    for chunk in stream:
        content = chunk.choices[0].delta.content
        if content:
            yield content        # если клиент отвалится, __exit__ закроет соединение

В веб-приложении, где вы пробрасываете токены дальше через свой SSE-эндпоинт, оборачивайте генератор так, чтобы finally отрабатывал при отмене запроса клиентом. В FastAPI/Starlette отменённый запрос приходит как asyncio.CancelledError внутри асинхронного генератора — не глушите его молча, дайте finally закрыть ресурс.

Ошибка 4. Один таймаут на весь поток

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

Правильная модель — таймаут между чанками, а не на весь ответ. Пока токены идут, соединение живёт; замолчало дольше порога — рвём. В SDK openai таймаут задаётся при создании клиента или на конкретный запрос через with_options, но межчанковую логику проще держать у себя:

import time

INTER_CHUNK_TIMEOUT = 20  # секунд тишины между токенами

last = time.monotonic()
for chunk in stream:
    now = time.monotonic()
    if now - last > INTER_CHUNK_TIMEOUT:
        raise TimeoutError("Модель молчит дольше 20 с — обрываем")
    last = now
    content = chunk.choices[0].delta.content
    if content:
        handle(content)

Для асинхронного кода эту проверку удобнее делать через asyncio.wait_for вокруг получения следующего элемента, а не сравнением времени вручную. Точные имена параметров таймаута в вашей версии SDK сверьте в документации — они менялись между релизами библиотеки httpx, на которой openai 1.x строит транспорт.

Сводка ошибок и лекарств

Что проверить в проде

• Метрика TTFT — время до первого токена. Оно важнее общего времени ответа: пользователь видит, что «пошло». Логируйте отдельно.
• Долю оборванных потоков — сколько ответов завершилось без finish_reason. Рост говорит о таймаутах или проблемах на стороне провайдера.
• Число активных соединений в пуле — если оно ползёт вверх и не спадает, у вас течёт стрим на обрывах клиента.

Дальше стоит посмотреть в сторону обработки finish_reason (модель могла упереться в лимит токенов и обрезать ответ на полуслове — это не то же самое, что нормальное завершение) и в асинхронный вариант через AsyncOpenAI, если вы обслуживаете много параллельных стримов в одном процессе. Ретраи на стриминге — отдельная тема: повторять весь запрос при обрыве в середине дорого и ломает уже показанный пользователю текст, здесь мы её не разбираем.