Стриминг ответа LLM в Python: три подхода к обработке

Запрос к модели с длинным ответом висит пять-восемь секунд, пользователь смотрит на пустой экран, а вы в это время ждёте финальную строку в переменной. Стриминг решает ровно это: сервер отдаёт ответ по частям (chunks), и вы можете показывать текст по мере поступления. Но между «поставить stream=True» и «сделать так, чтобы это работало под нагрузкой и не текло памятью» — три разных инженерных решения. Разберём их на клиенте OpenAI Python SDK 1.x, но логика переносится на любой провайдер с потоковым API.

Задача одна: получить от модели поток частей и что-то с ним сделать — вывести в консоль, отдать во фронтенд через SSE, собрать в буфер для последующего разбора. Очевидное решение — синхронный цикл по генератору — упирается в проблему ровно тогда, когда запросов становится больше одного одновременно.

Что вообще приходит в потоке

При stream=True клиент возвращает не готовый объект ответа, а итератор по объектам-частям. Каждая часть в Chat Completions API несёт choices[0].delta.content — фрагмент текста, чаще всего в несколько символов. Часть может нести пустой delta (например, самый первый чанк с ролью или последний с finish_reason), поэтому наивное chunk.choices[0].delta.content без проверки на None уронит вам код на первом же ответе.

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

Ниже три варианта обработки. Первый — синхронный, второй — асинхронный, третий — генераторная обёртка для отдачи наружу. Каждый решает свою задачу и платит своей ценой.

Подход 1. Синхронный цикл

Простейший вариант: блокирующий вызов, итерация по частям, печать по мере поступления. Подходит для CLI-утилит, скриптов, ноутбуков — одна задача, один поток исполнения.

# 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": "Объясни, что такое SSE, в трёх абзацах"}],
    stream=True,
)

buffer = []
for chunk in stream:
    delta = chunk.choices[0].delta.content
    if delta is not None:
        print(delta, end="", flush=True)  # flush обязателен, иначе вывод буферизуется
        buffer.append(delta)

full_text = "".join(buffer)

Два места, где новички спотыкаются. Первое — забытый flush=True: без него print копит вывод во внутреннем буфере и вы всё равно видите текст рывком в конце, теряя весь смысл стриминга. Второе — сборка результата через += в цикле: на длинных ответах это квадратичная сложность по строкам. Собирайте в список и делайте join один раз.

Цена: код блокирует поток исполнения на всё время генерации. Пока модель отвечает, процесс ничего другого не делает. Для одного запроса это норма. Для веб-сервера, обслуживающего десятки одновременных соединений, — тупик.

Подход 2. Асинхронный цикл

Тот же поток, но через AsyncOpenAI и async for. Пока одна корутина ждёт следующую часть от сети, event loop обслуживает другие запросы. Это то, что нужно под FastAPI, aiohttp или любой асинхронный веб-фреймворк.

# openai==1.x, Python 3.11+
import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI(api_key="YOUR_API_KEY")

async def collect_response(prompt: str) -> str:
    stream = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        stream=True,
    )
    parts = []
    async for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta is not None:
            parts.append(delta)
    return "".join(parts)

async def main():
    # три запроса выполняются конкурентно, а не последовательно
    results = await asyncio.gather(
        collect_response("Что такое CAP-теорема"),
        collect_response("Что такое идемпотентность"),
        collect_response("Что такое обратное давление"),
    )
    for r in results:
        print(r[:80])

asyncio.run(main())

Разница с синхронным вариантом только в await и async for, но именно она позволяет держать сотни открытых потоков на одном процессе. asyncio.gather запускает три генерации конкурентно — общее время близко к времени самого долгого ответа, а не к сумме трёх.

Цена: весь стек вызовов должен быть асинхронным. Один синхронный блокирующий вызов внутри корутины (тяжёлый парсинг, запрос к синхронной БД, time.sleep) заморозит event loop и убьёт всю конкурентность. Смешивать sync и async без run_in_executor нельзя.

Подход 3. Генератор для отдачи наружу

Когда поток надо не собрать, а прокинуть дальше — во фронтенд через Server-Sent Events или в WebSocket, — оборачивайте обработку в асинхронный генератор. FastAPI умеет отдавать такой генератор через StreamingResponse.

# openai==1.x, fastapi, Python 3.11+
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import AsyncOpenAI

app = FastAPI()
client = AsyncOpenAI(api_key="YOUR_API_KEY")

async def sse_generator(prompt: str):
    stream = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        stream=True,
    )
    async for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta is not None:
            # формат SSE: каждое событие начинается с data: и заканчивается пустой строкой
            yield f"data: {delta}\n\n"
    yield "data: [DONE]\n\n"

@app.get("/chat")
async def chat(q: str):
    return StreamingResponse(
        sse_generator(q),
        media_type="text/event-stream",
    )

Три подводных камня этого варианта. Первый: если в delta попадёт символ перевода строки, а вы отдаёте его как есть, разметка SSE поедет — для сложного контента фрагменты стоит кодировать (например, оборачивать в JSON). Второй: клиент может закрыть соединение до конца генерации; тогда async for прервётся, и если у вас после цикла есть закрытие ресурсов, оборачивайте в try/finally. Третий: прокси-серверы вроде Nginx по умолчанию буферизуют ответ — на проде понадобится proxy_buffering off; для этого location, иначе клиент опять получит текст одним куском.

Сравнение

Когда что выбирать

• Один запрос за раз, результат нужен вам, а не пользователю в реальном времени — синхронный цикл. Не тащите async туда, где нет конкурентности.
• Бэкенд обслуживает много запросов, но клиенту отдаёте готовый ответ целиком — асинхронный цикл со сборкой в буфер. Выигрыш в пропускной способности без возни с потоковой отдачей.
• Пользователь должен видеть текст по мере генерации — генератор с StreamingResponse. Это единственный вариант, где стриминг доходит до экрана, а не остаётся у вас на сервере.

Что ломается на практике

Обрыв соединения в середине потока. Сеть моргнула, модель отдала половину ответа — SDK бросит исключение прямо из цикла итерации. Оборачивайте цикл в try/except и решайте: показать частичный результат, повторить запрос или отдать ошибку. Молча проглотить нельзя — у вас окажется обрезанный текст, который выглядит как валидный.

Подсчёт токенов. При stream=True классический ответ с полем usage по умолчанию не приходит — считать токены придётся либо самостоятельно на своей стороне, либо через дополнительный параметр запроса. Точное имя параметра и его доступность зависят от версии API, сверьте по актуальной документации провайдера, а не подставляйте на память.

Тайм-аут. У потокового запроса тайм-аут работает не как у обычного: важна не общая длительность, а пауза между частями. Настраивайте тайм-аут клиента с учётом того, что длинный ответ — это долго, но паузы между чанками короткие.

Что настроить дальше

Если ушли в SSE — проверьте на проде поведение прокси и балансировщика: отключите буферизацию для потокового эндпоинта и убедитесь, что промежуточные звенья не режут долгие соединения по тайм-ауту. Измеряйте time to first token — задержку до первого чанка: именно её чувствует пользователь, и именно она отличает «быстро» от «медленно» при одинаковом полном времени ответа. Инвалидация и повтор оборванных генераций — отдельная тема, здесь мы разобрали только способы обработки потока.