Перейти к содержанию

Потоковая генерация

Шаг 1

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

Но есть еще один аспект, который влияет на удобство работы с чатом, — то, как приложение получает ответ от модели.

Сейчас LLMClient.generate() ожидает, пока модель полностью закончит генерацию, собирает готовый LLMResult и только после этого возвращает его в ChatService. И для коротких ответов это практически незаметно. Но если модели требуется несколько секунд, пользователь все это время не видит никакого результата и не понимает, выполняется ли запрос.

Решить эту проблему позволяет потоковая генерация (streaming). В этом режиме приложение не ждет завершения генерации целиком, а начинает получать небольшие фрагменты текста по мере их появления и сразу может передавать их интерфейсу.

Streaming не заставляет модель генерировать быстрее и не уменьшает количество токенов. Он сокращает время до появления первого фрагмента ответа. Пользователь начинает читать текст раньше, хотя полное время запроса и его стоимость обычно остаются примерно такими же.

Мы продолжим использовать Chat Completions API через ProxyAPI. Согласно документации ProxyAPI, форматы запросов и ответов совпадают с оригинальным OpenAI API, поэтому менять base_url, API-ключ или SDK не нужно, а в запрос добавится параметр stream=True, а вместо одного объекта ответа SDK вернет поток объектов ChatCompletionChunk.

Из чего состоит поток?

Во время потоковой генерации SDK получает не один готовый ответ, а последовательность объектов ChatCompletionChunk. Каждый такой объект представляет собой одно событие потока (или один фрагмент потокового ответа) и содержит новую информацию, полученную от модели.

Внутри ChatCompletionChunk могут находиться:

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

Текстовый фрагмент находится в choices[0].delta.content.

Слово delta означает изменение, которое нужно добавить к уже полученному результату. Отдельная delta не обязана совпадать с одним словом или одним токеном. Она может содержать часть слова, несколько слов, пробел, знак препинания или даже пустое значение. Поэтому такие фрагменты нельзя обрабатывать как законченные предложения. Их нужно сразу передавать пользователю и одновременно накапливать в правильном порядке.

delta 1: "Ваш"
delta 2: " заказ"
delta 3: " принят"

результат: "Ваш заказ принят"

Передавать наружу объекты SDK напрямую не будем. Как и обычный LLMResult, поток должен иметь собственный контракт нашего приложения. Создадим два вида событий: текстовый фрагмент и успешное завершение с метаданными.

Давайте перейдем к практике, первым делом откройте app/config.py и после request_timeout_seconds добавьте общий предел времени потоковой генерации:

stream_timeout_seconds: float = Field(
    default=60.0,
    gt=0.0,
    validation_alias="STREAM_TIMEOUT_SECONDS",
)

REQUEST_TIMEOUT_SECONDS и STREAM_TIMEOUT_SECONDS решают разные задачи. Настройка SDK ограничивает ожидание сетевых операций, например подключение или слишком долгую паузу между данными. Новая настройка ограничивает полную продолжительность потока на стороне нашего приложения. Даже если сервер регулярно присылает небольшие фрагменты и сетевой таймаут не срабатывает, приложение не будет держать один поток бесконечно.

Добавьте переменную в .env:

STREAM_TIMEOUT_SECONDS=60

Теперь откройте app/llm/client.py. В начало файла добавьте импорты генератора и монотонных часов:

from collections.abc import Generator
from time import monotonic

Из app.llm.errors дополнительно импортируйте базовую ошибку LLMClientError, а остальные классы ошибок, созданные в первом модуле, остаются на месте:

from app.llm.errors import (
    LLMAccessError,
    LLMBalanceError,
    LLMClientError,
    LLMConnectionError,
    LLMProviderError,
    LLMRateLimitError,
    LLMRequestError,
    LLMTimeoutError,
)

После LLMResult добавьте события потока:

@dataclass(frozen=True)
class LLMTextDelta:
    text: str

@dataclass(frozen=True)
class LLMStreamCompleted:
    model: str
    finish_reason: str | None
    prompt_tokens: int | None
    completion_tokens: int | None
    total_tokens: int | None
    response_id: str

LLMStreamEvent = LLMTextDelta | LLMStreamCompleted

LLMTextDelta содержит только новую часть текста, а LLMStreamCompleted создается один раз после корректного чтения потока и переносит те же технические метаданные, которые раньше находились рядом с текстом в LLMResult.

Объединение LLMStreamEvent показывает, что генератор возвращает события только двух известных типов. Это делает контракт потока явным и удобным для проверки типов. Вместо словарей с разным набором ключей мы работаем с объектами, структура которых заранее известна. Поэтому обработчик может определить тип события через isinstance() и безопасно обращаться только к тем полям, которые гарантированно существуют у этого типа.

В обычном generate() уже есть преобразование ошибок OpenAI SDK в ошибки нашего приложения. Поток может завершиться такой же ошибкой не только при создании запроса, но и позже, во время чтения очередного фрагмента. Чтобы не копировать весь блок except, вынесем преобразование в отдельный метод.

Замените внешний метод generate() и рядом добавьте новый метод stream():

def generate(self, messages: list[Message]) -> LLMResult:
    try:
        return self._generate_once(messages)
    except openai.APIError as error:
        raise self._convert_error(error) from error

def stream(
    self,
    messages: list[Message],
) -> Generator[LLMStreamEvent, None, None]:
    try:
        yield from self._stream_once(messages)
    except openai.APIError as error:
        raise self._convert_error(error) from error

generate() по-прежнему возвращает один результат и продолжит работать в SupportService и ConversationSummaryService. Метод stream() является генератором. Выражение yield from последовательно передает вызывающему коду все события, которые создает _stream_once().

Важное свойство генератора, это ленивое выполнение. Сам вызов client.stream(messages) еще не отправляет сетевой запрос. Код начнет выполняться, когда интерфейс запросит первое событие, например, войдет в цикл for. Благодаря этому ошибки, возникающие во время чтения потока, также проходят через блок try/except.

Добавьте в LLMClient преобразование ошибок:

@staticmethod
def _convert_error(error: openai.APIError) -> LLMClientError:
    request_id = getattr(error, "request_id", None)

    if isinstance(error, openai.AuthenticationError):
        return LLMAccessError(
            "Провайдер отклонил API-ключ.",
            request_id=request_id,
        )
    if isinstance(error, openai.PermissionDeniedError):
        return LLMAccessError(
            "Недостаточно прав для обращения к модели.",
            request_id=request_id,
        )
    if isinstance(
        error,
        (
            openai.BadRequestError,
            openai.NotFoundError,
            openai.UnprocessableEntityError,
        ),
    ):
        return LLMRequestError(
            "Провайдер отклонил модель или параметры запроса.",
            request_id=request_id,
        )
    if isinstance(error, openai.RateLimitError):
        return LLMRateLimitError(
            "Достигнуто ограничение частоты или квоты.",
            request_id=request_id,
        )
    if isinstance(error, openai.APITimeoutError):
        return LLMTimeoutError(
            "Провайдер не ответил за отведенное время."
        )
    if isinstance(error, openai.APIConnectionError):
        return LLMConnectionError(
            "Не удалось соединиться с провайдером."
        )
    if isinstance(error, openai.APIResponseValidationError):
        return LLMProviderError(
            "Провайдер вернул ответ неожиданного формата."
        )
    if isinstance(error, openai.APIStatusError):
        if error.status_code == 402:
            return LLMBalanceError(
                "Недостаточно средств на балансе ProxyAPI.",
                request_id=request_id,
            )
        return LLMProviderError(
            f"Провайдер вернул HTTP {error.status_code}.",
            request_id=request_id,
        )
    return LLMProviderError(
        "SDK сообщил о неизвестной ошибке провайдера."
    )

Логика ошибок не изменилась. Раньше каждый except сразу выбрасывал прикладное исключение, а теперь метод возвращает подходящий объект ошибки. И generate(), и stream() используют одно и то же преобразование, поэтому авторизация, недостаточный баланс, сетевой сбой и timeout имеют одинаковый смысл в обоих режимах.

Теперь добавьте основной метод чтения потока:

def _stream_once(
    self,
    messages: list[Message],
) -> Generator[LLMStreamEvent, None, None]:
    stream = self._client.chat.completions.create(
        model=self._settings.model,
        messages=messages,
        temperature=self._settings.temperature,
        max_completion_tokens=self._settings.max_output_tokens,
        stream=True,
        stream_options={"include_usage": True},
    )

    started_at = monotonic()
    model: str | None = None
    finish_reason: str | None = None
    response_id: str | None = None
    prompt_tokens: int | None = None
    completion_tokens: int | None = None
    total_tokens: int | None = None

    try:
        for chunk in stream:
            if monotonic() - started_at > self._settings.stream_timeout_seconds:
                raise LLMTimeoutError(
                    "Поток не завершился за отведенное время."
                )

            model = chunk.model
            response_id = chunk.id

            if chunk.usage is not None:
                prompt_tokens = chunk.usage.prompt_tokens
                completion_tokens = chunk.usage.completion_tokens
                total_tokens = chunk.usage.total_tokens

            if not chunk.choices:
                continue

            choice = chunk.choices[0]
            if choice.finish_reason is not None:
                finish_reason = choice.finish_reason

            text = choice.delta.content
            if text:
                yield LLMTextDelta(text=text)
    finally:
        stream.close()

    if model is None or response_id is None:
        raise LLMProviderError(
            "Провайдер завершил поток без метаданных."
        )

    yield LLMStreamCompleted(
        model=model,
        finish_reason=finish_reason,
        prompt_tokens=prompt_tokens,
        completion_tokens=completion_tokens,
        total_tokens=total_tokens,
        response_id=response_id,
    )

И вот тут остановимся и последовательно разберем, что делает _stream_once(). В начале метод отправляет почти такой же запрос, как знакомый _generate_once(). Модель, список сообщений, температура и ограничение ответа остаются прежними. Главное отличие это параметр stream=True:

stream=True

Без него SDK дождался бы полного ответа и вернул один объект. С этим параметром результатом становится поток, который можно читать в цикле for. По мере генерации ProxyAPI передает небольшие объекты chunk, а наше приложение обрабатывает их по одному.

Параметр stream_options={"include_usage": True} нужен для уже знакомых метрик токенов. В обычном запросе usage находится внутри готового LLMResult. В потоковом режиме заранее готового результата нет, поэтому мы просим провайдера прислать статистику отдельным событием ближе к завершению потока.

Перед циклом создаются переменные model, finish_reason, response_id и три счетчика токенов. Сначала в них находится None, потому что ни одного события еще не прочитано. Во время цикла мы постепенно заполним эти значения, а после завершения соберем из них LLMStreamCompleted.

Цикл for chunk in stream получает очередное событие. Сразу после получения проверяется общий таймаут. Если поток работает дольше значения STREAM_TIMEOUT_SECONDS, метод останавливается с LLMTimeoutError.

Затем из каждого chunk сохраняются модель и ID ответа:

model = chunk.model
response_id = chunk.id

Эти значения обычно повторяются во всех событиях одного потока. Мы сохраняем последние полученные значения, чтобы после цикла передать их в итоговые метаданные.

Теперь у нас повляется проверка chunk.usage:

if chunk.usage is not None:
    prompt_tokens = chunk.usage.prompt_tokens
    completion_tokens = chunk.usage.completion_tokens
    total_tokens = chunk.usage.total_tokens

Большинство событий несут текст, поэтому их поле usage будет равно None. После завершения генерации провайдер может прислать отдельный chunk со статистикой всего запроса. В таком событии есть usage, но список choices может оказаться пустым, потому что нового текста уже нет.

Именно поэтому код сначала читает usage и только потом выполняет:

if not chunk.choices:
    continue

Если поменять проверки местами, событие без choices будет пропущено раньше, чем мы сохраним статистику токенов. Тогда успешный поток вернет текст, но консоль сообщит, что провайдер не передал метрики.

Когда choices присутствует, мы берем первый вариант ответа и проверяем finish_reason. В промежуточных событиях это поле обычно равно None. Значение stop, length или другая причина появляется ближе к завершению, поэтому сохраняем ее только тогда, когда она действительно передана.

После этого читаем choice.delta.content. Если там находится непустая строка, создаем событие нашего приложения:

text = choice.delta.content
if text:
    yield LLMTextDelta(text=text)

LLMClient не соединяет части текста. Он сразу передает очередную delta вызывающему коду. В следующем шаге ChatService будет одновременно показывать эти части пользователю и складывать их в список, чтобы после завершения получить полный ответ.

Блок finally закрывает сетевой поток при любом исходе: после успешного завершения, при ошибке, таймауте или отмене генератора. Это особенно важно для долгих соединений. Если пользователь больше не ждет результат, приложение должно освободить HTTP-соединение.

Когда цикл закончился, проверяем, удалось ли получить модель и ID ответа. Их отсутствие означает, что провайдер завершил поток неожиданным образом. В нормальном случае создается LLMStreamCompleted с причиной завершения и собранными метаданными.

Сам LLMClient пока не решает, является ли finish_reason успешным. Его задача только прочитать поток без потери текста и технических данных, а решение о сохранении ответа в историю примет ChatService, потому что только он знает правила диалога.

На этом уровне поток уже умеет отдавать части текста и финальные метаданные, но чат пока продолжает вызывать generate(). В следующем шаге мы подключим события к ChatService, накопим полный ответ и сделаем так, чтобы в историю попадала только успешно завершенная генерация.

Шаг 2

В прошлом шаге LLMClient научился возвращать два вида событий: LLMTextDelta с очередным фрагментом текста и LLMStreamCompleted с причиной завершения и метаданными. Но пользователь пока не увидит разницы, потому что ChatService.reply() по-прежнему вызывает обычный generate() и получает ответ целиком.

Недостаточно просто вывести delta.content в консоль непосредственно из LLMClient. Клиент интеграции не знает, к какой сессии относится ответ, какую историю нужно обновить и можно ли считать генерацию успешной. Эти решения остаются в ChatService.

Потоковый сценарий должен выполнить ту же бизнес-логику, что и обычный ответ:

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

События уровня чата

LLMClient сообщает, что происходит при получении ответа от модели, то есть передает очередной фрагмент текста или итоговые метаданные. Но консоль работает не напрямую с LLMClient, а с ChatService. Поэтому ChatService преобразует технические события клиента в собственные события чата, понятные интерфейсу, часть ответа для немедленного вывода и завершенный ChatReply со всеми метриками.

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

Откройте app/chat_service.py и добавьте импорт Generator:

from collections.abc import Generator

Из app.llm.client импортируйте потоковые события:

from app.llm.client import (
    LLMClient,
    LLMStreamCompleted,
    LLMTextDelta,
    Message,
)

После ChatReply добавьте классы событий:

@dataclass(frozen=True)
class ChatTextDelta:
    text: str

@dataclass(frozen=True)
class ChatStreamCompleted:
    result: ChatReply

ChatStreamEvent = ChatTextDelta | ChatStreamCompleted

ChatTextDelta можно сразу показать пользователю, а ChatStreamCompleted появляется только после проверки и сохранения полного ответа. Внутри него находится уже знакомый ChatReply, поэтому нам не придется создавать второй набор полей для Session ID, версии промпта, токенов и времени.

Старый код внутри reply() мы удалим, но имя метода, его аргументы и возвращаемый ChatReply сохраним. Поэтому существующий вызов service.reply(session_id, user_text) продолжит работать. Просто теперь reply() будет запускать stream_reply(), полностью читать его события и возвращать только итоговый результат, не показывая промежуточные фрагменты.

Замените текущий reply() коротким адаптером:

def reply(self, session_id: str, user_text: str) -> ChatReply:
    result: ChatReply | None = None

    for event in self.stream_reply(session_id, user_text):
        if isinstance(event, ChatStreamCompleted):
            result = event.result

    if result is None:
        raise ChatServiceError(
            "Поток завершился без итогового результата."
        )

    return result

Таким образом, в приложении остается один основной сценарий генерации, только потоковый. Метод reply() не дублирует подготовку истории и сохранение сессии, а только потребляет все события, не показывая промежуточный текст.

Теперь после reply() добавьте новый метод stream_reply(). Большая часть кода уже знакома по разделу 2.4, и мы по-прежнему готовим историю, обновляем summary при переполнении и работаем под блокировкой конкретной сессии. Изменится участок обращения к LLMClient и момент сохранения ответа.

def stream_reply(
    self,
    session_id: str,
    user_text: str,
) -> Generator[ChatStreamEvent, None, None]:
    session_key = self._normalize_session_id(session_id)

    text = user_text.strip()
    if not text:
        raise ChatServiceError("Сообщение не должно быть пустым.")

    session = self._get_or_create_session(session_key)

    with session.lock:
        developer_message = self._build_developer_message()
        user_message = self._build_user_message(text)
        summary_message = self._build_summary_message(session.summary)
        context_messages = (
            [summary_message] if summary_message is not None else []
        )

        try:
            prepared = self._history_policy.prepare(
                developer_message=developer_message,
                history=session.messages,
                user_message=user_message,
                context_messages=context_messages,
            )
        except ChatHistoryError as error:
            raise ChatServiceError(
                f"Не удалось подготовить историю: {error}"
            ) from error

        started_at = perf_counter()
        new_summary = session.summary
        summary_updated = False

        if prepared.removed_history_messages > 0:
            try:
                new_summary = self._summary_service.summarize(
                    previous_summary=session.summary,
                    messages=session.messages,
                )
            except ConversationSummaryError as error:
                raise ChatServiceError(
                    f"Не удалось сократить историю: {error}"
                ) from error

            summary_updated = True
            summary_message = self._build_summary_message(new_summary)
            if summary_message is None:
                raise ChatServiceError(
                    "Не удалось подготовить новое summary."
                )

            try:
                prepared = self._history_policy.prepare(
                    developer_message=developer_message,
                    history=[],
                    user_message=user_message,
                    context_messages=[summary_message],
                )
            except ChatHistoryError as error:
                raise ChatServiceError(
                    "Новое summary не помещается "
                    f"в бюджет: {error}"
                ) from error

        reply_parts: list[str] = []
        llm_completed: LLMStreamCompleted | None = None

        try:
            for event in self._llm_client.stream(prepared.messages):
                if isinstance(event, LLMTextDelta):
                    reply_parts.append(event.text)
                    yield ChatTextDelta(text=event.text)
                elif isinstance(event, LLMStreamCompleted):
                    llm_completed = event
        except LLMClientError as error:
            raise ChatServiceError(
                f"Не удалось получить ответ модели: {error}"
            ) from error

        elapsed_seconds = perf_counter() - started_at

        if llm_completed is None:
            raise ChatServiceError(
                "Поток завершился без итоговых метаданных."
            )
        if llm_completed.finish_reason == "length":
            raise ChatServiceError(
                "Ответ остановлен из-за ограничения длины."
            )
        if llm_completed.finish_reason != "stop":
            raise ChatServiceError(
                "Модель не вернула готовый ответ. "
                f"Причина завершения: {llm_completed.finish_reason}."
            )

        reply_text = self._validate_reply("".join(reply_parts))
        assistant_message: Message = {
            "role": "assistant",
            "content": reply_text,
        }

        if summary_updated:
            session.summary = new_summary
            session.messages = [user_message, assistant_message]
        else:
            session.messages.extend([user_message, assistant_message])

        history_messages = len(session.messages)

        result = ChatReply(
            text=reply_text,
            session_id=session_key,
            history_messages=history_messages,
            context_history_messages=prepared.context_history_messages,
            removed_history_messages=prepared.removed_history_messages,
            estimated_prompt_tokens=prepared.estimated_tokens,
            model=llm_completed.model,
            prompt_id=SUPPORT_CHAT_PROMPT.prompt_id,
            prompt_version=SUPPORT_CHAT_PROMPT.version,
            elapsed_seconds=elapsed_seconds,
            prompt_tokens=llm_completed.prompt_tokens,
            completion_tokens=llm_completed.completion_tokens,
            total_tokens=llm_completed.total_tokens,
            response_id=llm_completed.response_id,
            summary_updated=summary_updated,
            has_summary=session.summary is not None,
        )

    yield ChatStreamCompleted(result=result)

Разберем новую часть метода подробнее. Список reply_parts накапливает фрагменты в том порядке, в котором они поступают. Когда приходит LLMTextDelta, сервис сначала добавляет текст в список, а затем сразу передает соответствующий ChatTextDelta интерфейсу.

reply_parts.append(event.text)
yield ChatTextDelta(text=event.text)

Операция yield временно приостанавливает stream_reply(). Интерфейс получает фрагмент, выводит его, а при запросе следующего события выполнение продолжается с той же строки и с теми же локальными переменными. Поэтому reply_parts, prepared, new_summary и текущая сессия не теряются между фрагментами.

Событие LLMStreamCompleted не выводится как текст. Мы сохраняем его в llm_completed, чтобы после завершения цикла проверить finish_reason и получить метаданные.

Полный текст создается выражением:

reply_text = self._validate_reply("".join(reply_parts))

В join() передается пустой разделитель. Мы не добавляем пробелы самостоятельно, потому что каждый фрагмент уже содержит ровно ту часть строки, которую вернула модель. Добавление " " могло бы разорвать слово или создать двойные пробелы перед знаками препинания.

Когда изменяется история?

И в конце этого шага, я бы хотел разобрать этот важный момент. Ведь пользователь начинает видеть ответ сразу после получения первых событий ChatTextDelta. Однако это еще не означает, что показанный текст сохранен в истории диалога. Пока поток не завершился успешно, поля session.messages и session.summary остаются без изменений.

Сообщения user и assistant добавляются в сессию только после того, как сервис убедится, что получен законченный и корректный ответ. Для этого должны выполниться все условия:

  • поток дошел до события LLMStreamCompleted
  • модель завершила генерацию с finish_reason="stop"
  • отдельные фрагменты были соединены в полный текст
  • полный текст прошел проверку _validate_reply()

Только после этого ChatService сохраняет новую пару сообщений и при необходимости обновляет резюме диалога — summary.

Если соединение оборвалось, пользователь отменил генерацию, провайдер вернул ошибку или модель остановилась из-за ограничения длины с finish_reason="length", история не изменяется. Пользователь мог успеть увидеть несколько первых фрагментов, но они остаются частью незавершенной генерации и не становятся новым сообщением assistant.

Благодаря этому каждый сохраненный ход истории всегда состоит из законченной пары:

user → assistant

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

Блокировка session.lock удерживается на протяжении всей потоковой генерации. Пока модель отвечает на одну реплику, обработка следующей реплики той же сессии не начнется. Это защищает порядок сообщений: второй вопрос не сможет попасть в историю раньше, чем закончится ответ на первый.

У каждой сессии собственная блокировка. Поэтому длительная генерация в одной сессии не блокирует другие сессии на уровне ChatService. Однако все запросы используют общий LLMClient и одного провайдера, поэтому на них по-прежнему могут влиять общие сетевые ограничения, rate limit и доступная конкурентность ProxyAPI.

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

Теперь ChatService умеет постепенно передавать текст пользователю и сохранять только полностью завершенные ответы. Однако консоль пока продолжает вызывать reply() и показывает результат целиком. В следующем шаге мы подключим к интерфейсу метод stream_reply(), выведем текст по мере генерации, добавим отмену через Ctrl+C и проверим разные сценарии завершения потока.

Шаг 3

Поток уже проходит через два уровня приложения. LLMClient читает события ProxyAPI, а ChatService накапливает полный ответ и обновляет историю только после успешного завершения. Осталось изменить консоль, ведь сейчас она вызывает reply(), поэтому сама полностью потребляет поток и по-прежнему возвращает интерфейсу только готовый ChatReply.

Для потокового интерфейса консоль должна выполнять две разные операции. Текстовые события нужно печатать сразу и без перевода строки, а финальное событие использовать для вывода Session ID, токенов и остальных метрик.

Откройте app/console.py и замените его содержимое:

from uuid import uuid4

from app.chat_service import (
    ChatReply,
    ChatService,
    ChatServiceError,
    ChatStreamCompleted,
    ChatTextDelta,
)

def print_usage(result: ChatReply) -> None:
    if result.total_tokens is None:
        print("Провайдер не вернул статистику токенов.")
        return

    print(f"Входные токены: {result.prompt_tokens}")
    print(f"Выходные токены: {result.completion_tokens}")
    print(f"Всего токенов: {result.total_tokens}")

def print_metadata(result: ChatReply) -> None:
    print(f"Session ID ответа: {result.session_id}")
    print(f"Сообщений в истории: {result.history_messages}")
    print(
        "Сообщений истории в запросе: "
        f"{result.context_history_messages}"
    )
    print(
        "Сообщений истории отброшено: "
        f"{result.removed_history_messages}"
    )
    print(
        "Оценка входных токенов до запроса: "
        f"{result.estimated_prompt_tokens}"
    )
    print(
        "Summary сохранено: "
        f"{'да' if result.has_summary else 'нет'}"
    )
    print(
        "Summary обновлено на этом ходу: "
        f"{'да' if result.summary_updated else 'нет'}"
    )
    print(f"Промпт: {result.prompt_id}@{result.prompt_version}")
    print(f"Время ответа: {result.elapsed_seconds:.2f} с")
    print_usage(result)
    print(f"ID ответа: {result.response_id}")

def run_console(
    service: ChatService,
    app_env: str,
    configured_model: str,
) -> None:
    print(f"Окружение: {app_env}")
    print(f"Настроенная модель: {configured_model}")

    session_id = uuid4().hex
    print(f"Session ID: {session_id}")
    print("Команды: /new — новый диалог, /exit — выход.")
    print("Во время ответа нажмите Ctrl+C, чтобы отменить генерацию.")

    while True:
        user_text = input("\nВы: ")
        command = user_text.strip()

        if command == "/exit":
            return

        if command == "/new":
            service.reset_session(session_id)
            session_id = uuid4().hex
            print(f"Новый Session ID: {session_id}")
            continue

        stream = service.stream_reply(session_id, user_text)
        result: ChatReply | None = None
        received_text = False

        print("Assistant: ", end="", flush=True)

        try:
            for event in stream:
                if isinstance(event, ChatTextDelta):
                    received_text = True
                    print(event.text, end="", flush=True)
                elif isinstance(event, ChatStreamCompleted):
                    result = event.result
        except KeyboardInterrupt:
            stream.close()
            print("\nГенерация отменена. Ответ не сохранен в истории.")
            continue
        except ChatServiceError as error:
            stream.close()
            print()
            if received_text:
                print(f"Поток прерван: {error}")
            else:
                print(f"Ошибка чата: {error}")
            continue

        print()

        if result is None:
            print("Ошибка чата: поток завершился без результата.")
            continue

        print_metadata(result)

Функцию print_usage() мы сохранили, а остальные метрики вынесли в print_metadata(). Это позволяет не смешивать код чтения событий с длинной последовательностью обычных print().

Перед циклом потока консоль печатает начало строки:

print("Assistant: ", end="", flush=True)

end="" запрещает добавлять перевод строки после каждого вызова. Без этого отдельные фрагменты располагались бы столбиком. Параметр flush=True просит Python немедленно передать текст терминалу, а не ждать заполнения внутреннего буфера вывода. Именно благодаря этому пользователь видит фрагмент сразу после получения.

Мы не добавляем между событиями пробелы:

print(event.text, end="", flush=True)

Ведь поток сам может вернуть фрагменты "номер", " заказа" и " 7718". Если консоль будет добавлять разделитель, итоговая строка изменится. Поэтому интерфейс печатает delta без преобразований.

Так же обратите внимание на флаг received_text, он отвечает за обработку ошибки до начала и ошибка во время потока и нужен только для интерфейса. Если ошибка возникла до первого текстового события, пользователь увидит обычное сообщение Ошибка чата. Причиной может быть пустой ввод, неправильная конфигурация, отказ ProxyAPI или невозможность подготовить историю.

Если несколько фрагментов уже появились, стереть их из терминала уже нельзя. Консоль переводит строку и явно сообщает Поток прерван. Отображенный текст мы считаем неполным, а значит ChatService не добавит его в историю и не создаст финальный ChatReply.

При потоковом ответе интерфейс обязан различать «пользователь увидел часть текста» и «приложение успешно завершило операцию». Это особенно важно для веб-интерфейса, где частичный ответ можно визуально пометить как незавершенный, а кнопки дальнейших действий активировать только после события завершения.

Как работает отмена?

Во время чтения потока нажатие Ctrl+C создает в Python исключение KeyboardInterrupt. Мы перехватываем его только вокруг генерации, закрываем генератор методом close() и возвращаемся к следующему приглашению Вы:.

Закрытие проходит по всей цепочке. stream_reply() прекращает выполнение, а блок finally внутри LLMClient._stream_once() закрывает HTTP-поток. Код сохранения session.messages находится после полного чтения и уже не выполняется.

Ветка отмены показана на общей схеме выше, когда консоль закрывает генератор, LLMClient освобождает HTTP-поток, а код сохранения законченного хода не выполняется.

Закрытие соединения означает, что приложение перестало ждать и передавать фрагменты пользователю. Оно не гарантирует мгновенную остановку вычислений на стороне внешнего провайдера, поэтому часть токенов могла быть уже сгенерирована и учтена в балансе. Кроме того, при отмене можно не получить последнее событие usage. По этой причине консоль не выводит метрики отмененного запроса.

Сетевой таймаут и общий STREAM_TIMEOUT_SECONDS проходят тем же путем, но вместо KeyboardInterrupt возникает ChatServiceError.

Давайте запустим и проверим проект. А начнем мы с обычного обращения:

В заказе 7718 у монитора треснута подставка. Это первое обращение.

После строки Assistant: текст должен появляться постепенно. Скорость и размер отдельных фрагментов заранее не определены: иногда SDK вернет несколько коротких delta, а иногда один более крупный фрагмент.

Затем отправьте продолжение:

Фотографию повреждения уже сделал. Что уточнить еще?

Модель должна учитывать предыдущий ход. После финального события значение Сообщений в истории увеличится еще на два: сохранится новая пара user/assistant.

Теперь отправьте новое сообщение и нажмите Ctrl+C, как только начнет появляться ответ. Консоль сообщит, что генерация отменена. После этого задайте следующий вопрос. Отмененное сообщение и неполный ответ не должны участвовать в новом контексте.

Для проверки общего таймаута можно временно добавить в .env очень маленькое значение, например:

STREAM_TIMEOUT_SECONDS=0.1

После перезапуска поток, скорее всего, завершится ошибкой до полного ответа. Конкретный момент зависит от скорости сети и модели: иногда пользователь не увидит ни одного фрагмента, иногда успеет появиться начало текста. После проверки верните обычное значение или удалите переменную из .env.

Подведем итоги раздела

В этом разделе мы добавили потоковую генерацию в чат AI Support Desk. Теперь пользователь начинает видеть ответ сразу после появления первых текстовых фрагментов и не ждет завершения всей генерации.

Для этого мы изменили несколько уровней приложения. LLMClient научился читать поток от ProxyAPI и возвращать события с фрагментами текста и итоговыми метаданными. ChatService начал собирать из этих фрагментов полный ответ, проверять причину завершения и сохранять в историю только законченную пару user → assistant. Консоль, в свою очередь, выводит текст по мере получения и умеет обрабатывать ошибку, таймаут и отмену генерации.

При этом streaming не ускоряет работу модели и не уменьшает стоимость запроса. Он улучшает пользовательский интерфейс, ведь человек раньше видит начало ответа и понимает, что приложение продолжает работать.

Потоковая генерация особенно полезна для чатов и длинных текстовых ответов. Но использовать ее для каждого обращения к модели необязательно. Если приложение получает короткую классификацию, создает внутреннее summary, выполняет фоновую задачу или должно проверить весь результат до показа пользователю, проще дождаться готового LLMResult.

Поэтому ConversationSummaryService продолжает использовать обычный метод generate(). Пользователь не видит внутреннее резюме, а основной запрос чата нельзя правильно собрать, пока суммаризация не завершена. Из-за этого на ходу с переполнением истории перед первым фрагментом ответа может появиться дополнительная пауза, когда сначала приложение обновляет summary и только после этого открывает основной поток.

В некоторых сценариях ответ нельзя показывать частями, ведь сначала приложение должно получить и проверить весь текст. Для этого можно использовать метод reply(). Внутри он запускает stream_reply() и полностью читает поток, но не передает промежуточные ChatTextDelta интерфейсу. Вызывающий код получает только проверенный и завершенный ChatReply.

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

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

Шаг 4

Поток вернул три текстовых delta: «Ваш», « заказ» и « принят». Как получить итоговый ответ?

Правильный ответ
  • ✅ Сохранить части в порядке получения и объединить без добавления разделителей

Шаг 5

Какие правила корректны при разборе потока Chat Completions?

Правильный ответ
  • ✅ Исходный поток следует закрыть в finally
  • ✅ Пустой текст delta не нужно выдавать как отдельное текстовое событие
  • ✅ Chunk без choices может содержать usage и поэтому не является ошибкой сам по себе
  • ✅ Причину завершения и метрики нужно передать отдельным финальным событием

Шаг 6

Пользователь отменил генерацию после нескольких текстовых delta. Нужно ли сохранять частичный ответ в историю как сообщение assistant?

Правильный ответ
  • ✅ Нет, история изменяется только после успешного финального события и проверки полного ответа

Шаг 7

Какие действия нужны при отмене или ошибке потокового ответа?

Правильный ответ
  • ✅ Показать пользователю, что генерация была отменена или прервана
  • ✅ Закрыть генератор потока
  • ✅ Освободить сетевой поток и блокировку сессии
  • ✅ Не сохранять частичный assistant-ответ в историю

Шаг 8

Сборка текста из событий потока

Проверяющая система передаёт в переменной events текстовые и финальное события. Классы TextDelta и StreamCompleted уже доступны.

Задача

Допишите программу без создания отдельной функции. Соберите текстовые части и сохраните финальное событие.

Требования

  • Создайте пустой список parts и переменную completed = None.
  • Для каждого TextDelta добавляйте непустой event.text в parts.
  • При получении StreamCompleted сохраните событие в completed.
  • Не изменяйте текст delta и не добавляйте разделители самостоятельно.
  • После цикла объедините части через "".join(parts), примените strip() и сохраните строку в answer.

Пример результата

# events содержат TextDelta("  Ваш"), TextDelta(" заказ"),
# TextDelta(" принят.  ") и StreamCompleted(...)

answer == "Ваш заказ принят."
completed.finish_reason == "stop"

Проверяющая система заранее создаёт список events и классы TextDelta и StreamCompleted. Используйте эти объекты и не создавайте их повторно.

Правильный ответ 
# events, TextDelta и StreamCompleted уже созданы проверяющей системой.
parts = []
completed = None
answer = None

for event in events:
    if isinstance(event, TextDelta):
        if event.text:
            parts.append(event.text)

    elif isinstance(event, StreamCompleted):
        completed = event

answer = "".join(parts).strip()

Шаг 9

Преобразование chunk в события приложения

Поток SDK состоит из chunk. Текст, причина завершения, метаданные и usage могут приходить в разных объектах. Преобразуйте их в простой список событий.

Задача

Реализуйте parse_stream_chunks(chunks). Функция должна преобразовать последовательность технических chunk-объектов в список событий приложения.

Требования

  • Просмотрите chunk по порядку.
  • Из каждого chunk сохраняйте последние значения model и id.
  • Если chunk.usage is not None, сохраните три значения токенов.
  • Если chunk.choices пуст, продолжите обработку следующего chunk.
  • Используйте первый choice. Непустой choice.delta.content превращайте в TextDelta.
  • Если choice.finish_reason is not None, сохраните причину завершения.
  • После всех TextDelta добавьте ровно один StreamCompleted. Передайте в него последние сохранённые model, finish_reason, значения токенов и response_id.
  • Если поток не сообщил непустые model или id, выбросьте StreamParseError.

Пример результата

events = parse_stream_chunks(chunks)
# [
#   TextDelta("Ваш"),
#   TextDelta(" заказ"),
#   StreamCompleted(model="test-model", finish_reason="stop", ...),
# ]

Классы TextDelta, StreamCompleted и StreamParseError, а также необходимые импорты уже находятся в заготовке. Не изменяйте предоставленные классы.

Правильный ответ 
from dataclasses import dataclass
from typing import Any

class StreamParseError(ValueError):
    pass

@dataclass(frozen=True)
class TextDelta:
    text: str

@dataclass(frozen=True)
class StreamCompleted:
    model: str
    finish_reason: str | None
    prompt_tokens: int | None
    completion_tokens: int | None
    total_tokens: int | None
    response_id: str

def parse_stream_chunks(chunks: list[Any]) -> list[object]:
    events = []

    model = None
    response_id = None
    finish_reason = None

    prompt_tokens = 0
    completion_tokens = 0
    total_tokens = 0

    for chunk in chunks:
        if chunk.model:
            model = chunk.model

        if chunk.id:
            response_id = chunk.id

        if chunk.usage is not None:
            prompt_tokens = chunk.usage.prompt_tokens
            completion_tokens = chunk.usage.completion_tokens
            total_tokens = chunk.usage.total_tokens

        if not chunk.choices:
            continue

        choice = chunk.choices[0]

        if choice.delta.content:
            events.append(TextDelta(choice.delta.content))

        if choice.finish_reason is not None:
            finish_reason = choice.finish_reason

    if not model or not response_id:
        raise StreamParseError

    events.append(
        StreamCompleted(
            model=model,
            finish_reason=finish_reason,
            prompt_tokens=prompt_tokens,
            completion_tokens=completion_tokens,
            total_tokens=total_tokens,
            response_id=response_id,
        )
    )

    return events

Шаг 10

Сохранение только завершённого потокового ответа

История диалога не должна содержать частичный ответ. Функция получает уже собранные события и добавляет новый ход только при корректном завершении.

Задача

Реализуйте commit_streamed_reply(history, user_text, events). Функция должна проверить потоковый ответ и добавить в историю только полностью завершённый ход.

Требования

  • user_text должен быть строкой и не должен быть пустым после strip(). При нарушении этого требования выбросьте StreamReplyError.
  • Объедините текст всех TextDelta в порядке получения и примените strip().
  • В потоке должно находиться ровно одно событие StreamCompleted.
  • Принимайте ответ только при finish_reason == "stop".
  • Пустой итоговый текст недопустим.
  • При любой ошибке выбросьте StreamReplyError и не изменяйте history.
  • После успешной проверки добавьте сообщения user и assistant.
  • Верните очищенный текст assistant.

Пример результата

commit_streamed_reply(
    history,
    "Где мой заказ?",
    [TextDelta("Заказ "), TextDelta("в пути."), StreamCompleted("stop")],
)
# "Заказ в пути."

# При finish_reason="length" функция выбрасывает StreamReplyError,
# а history остаётся без изменений.

Классы TextDelta, StreamCompleted и StreamReplyError уже находятся в заготовке. Не удаляйте и не переименовывайте их.

Правильный ответ 
from dataclasses import dataclass

class StreamReplyError(ValueError):
    pass

@dataclass(frozen=True)
class TextDelta:
    text: str

@dataclass(frozen=True)
class StreamCompleted:
    finish_reason: str | None

def commit_streamed_reply(
    history: list[dict[str, str]],
    user_text: str,
    events: list[object],
) -> str:
    if not isinstance(user_text, str):
        raise StreamReplyError

    user_text = user_text.strip()

    if not user_text:
        raise StreamReplyError

    parts = []
    completed_events = []

    for event in events:
        if isinstance(event, TextDelta):
            parts.append(event.text)
        elif isinstance(event, StreamCompleted):
            completed_events.append(event)

    if len(completed_events) != 1:
        raise StreamReplyError

    completed = completed_events[0]

    if completed.finish_reason != "stop":
        raise StreamReplyError

    assistant_text = "".join(parts).strip()

    if not assistant_text:
        raise StreamReplyError

    history.append({
        "role": "user",
        "content": user_text,
    })

    history.append({
        "role": "assistant",
        "content": assistant_text,
    })

    return assistant_text