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

Потоковые ответы через SSE

Шаг 1

В предыдущем разделе мы создали обычный HTTP-эндпоинт чата. Клиент отправляет сообщение, сервер обращается к модели, при необходимости выполняет инструменты и только после завершения всей операции возвращает ChatReplyResponse.

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

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

Что такое SSE?

Для передачи потока мы будем использовать Server-Sent Events, сокращённо SSE. Это текстовый формат, который позволяет серверу отправлять клиенту несколько последовательных событий в рамках одного HTTP-соединения.

В отличие от WebSocket, SSE рассчитан прежде всего на движение данных в одну сторону, а именно только от сервера к клиенту. И для нашего сценария этого достаточно, ведь пользователь один раз отправляет сообщение обычным HTTP-запросом, а затем сервер постепенно возвращает ответ модели в том же соединении.

Формат SSE определяет, как записывается каждое событие. В строке event указывается его название, а в строке data передаются данные. Пустая строка завершает текущее событие и отделяет его от следующего:

event: delta
data: {"text":"Ваш заказ"}

event: delta
data: {"text":" находится в пути."}

event: completed
data: {"session_id":"...","history_messages":2,...}

Названия событий стандарт SSE не устанавливает — их выбирает разработчик приложения. В нашем Chat API мы реализуем три вида событий:

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

Клиентский интерфейс будет последовательно добавлять текст каждого события delta к уже показанной части ответа.

Сейчас обычный эндпоинт FastAPI возвращает результат только после полного завершения функции. Например метод ChatService.reply() сначала ждёт ответ модели, выполняет необходимые проверки, сохраняет сообщения в истории и создаёт готовый ChatReply. Затем FastAPI преобразует результат в JSON и отправляет его клиенту целиком.

Но для потоковой выдачи такой порядок не подходит, ведь нам нужно отправлять первые части текста ещё до того, как модель закончит генерацию. Для этого в FastAPI используется StreamingResponse.

Вместо одного готового объекта StreamingResponse получает генератор. Генератор постепенно возвращает значения с помощью yield, а FastAPI сразу передаёт каждое из них клиенту. Серверу не требуется сначала собирать весь ответ в одну строку.

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

Чтобы клиент мог отличить часть ответа от сообщения об успешном завершении или ошибке, мы будем оформлять передаваемые строки по правилам SSE. Таким образом StreamingResponse обеспечит саму потоковую отправку, а SSE задаст понятный формат событий delta, completed и error.

Существующий эндпоинт удалять не будем:

POST /chat/sessions/{session_id}/messages

Он продолжит ждать завершения генерации и возвращать один готовый JSON. Такой вариант подходит клиентам, которым не нужно отображать текст постепенно. А рядом добавим отдельный потоковый эндпоинт:

POST /chat/sessions/{session_id}/messages/stream

Он будет передавать ответ последовательностью SSE-событий. При этом оба эндпоинта используют тот же ChatService, общие сессии, историю диалога и правила выполнения инструментов. Мы не создаём второй вариант чата, а меняется только способ доставки одного и того же результата клиенту.

Что уже умеет проект и что нужно изменить?

Во втором модуле мы уже научили LLMClient получать ответ модели постепенно. Метод stream() обращался к ProxyAPI в потоковом режиме и возвращал небольшие части текста по мере их появления.

Позднее мы добавили инструменты. Для управления ими появился BoundedToolLoop, который может несколько раз обратиться к модели, выполнить выбранную Python-функцию и продолжить диалог с полученным результатом. Однако этот цикл использует непотоковый метод generate_with_tools(). Он сначала дожидается полного ответа модели и только после этого возвращает результат.

Из-за этого сейчас цепочка работает следующим образом:

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

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

Нам нужно восстановить поток на каждом уровне приложения. LLMClient будет получать части ответа от ProxyAPI, BoundedToolLoop — обрабатывать их и при необходимости выполнять инструменты, ChatService — передавать текст выше, а FastAPI — преобразовывать внутренние события в SSE.

При этом возможность использовать инструменты должна сохраниться. Если модель запросит функцию, например get_order_status, клиент не должен увидеть служебные части её имени и JSON-аргументов. BoundedToolLoop сначала полностью соберёт вызов инструмента, проверит и выполнит его, а затем снова обратится к модели. Постепенно клиенту будет передаваться уже окончательный текст ответа.

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

У SSE есть ещё одна важная особенность. Пока передача данных не началась, FastAPI может вернуть подходящий HTTP-статус, например 404 Not Found или 409 Conflict. Но после отправки заголовков и первого события статус ответа уже зафиксирован. Если ошибка возникнет посреди генерации, заменить 200 OK на 502 Bad Gateway не получится. Поэтому ошибки до начала потока будем возвращать обычным HTTP-ответом, а ошибки после его начала — событием error.

В длительных SSE-соединениях иногда используют служебные события heartbeat. Сервер периодически отправляет такое пустое или небольшое событие, чтобы прокси-сервер не решил, что соединение зависло, и не закрыл его. В нашем чате модель обычно регулярно возвращает части текста, поэтому отдельно реализовывать heartbeat пока не будем.

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

Шаг 2

В прошлой лекции мы определили весь путь потокового ответа. Начнём его реализацию с LLMClient, потому что именно этот класс напрямую получает данные от ProxyAPI.

С обычным текстом всё достаточно просто. Модель создаёт ответ небольшими частями, а SDK возвращает каждую часть в отдельном объекте chunk. Внутри него поле choice.delta.content может содержать несколько новых символов или слов:

первый chunk  → "Ваш заказ"
второй chunk  → " находится"
третий chunk  → " в пути."

готовый текст → "Ваш заказ находится в пути."

Наш метод LLMClient.stream() уже умеет читать такие части и сразу возвращать для каждой из них LLMTextDelta.

Но так как наш проект умеет работать с реестром инструментов, появляется дополнительный сценарий. Ведь модель может не формировать ответ пользователю, а запросить выполнение Python-функции. Например для вопроса о заказе она выберет get_order_status и передаст ему order_id.

При обычном непотоковом запросе имя инструмента и все аргументы приходят сразу в одном ToolCall. В потоковом режиме они тоже могут быть разделены между несколькими чанками:

первый chunk:
name = "get_order_status"
arguments = "{\"order"

второй chunk:
name = None
arguments = "_id\":\"80"

третий chunk:
name = None
arguments = "41\"}"

после объединения:
name = "get_order_status"
arguments = "{\"order_id\":\"8041\"}"

Первый фрагмент нельзя сразу передать в ToolExecutor, ведь строка JSON ещё не закончена и не пройдёт проверку Pydantic. Поэтому текст будем передавать дальше сразу, а части вызова инструмента сначала сохраним. Готовый ToolCall появится только после завершения текущего ответа модели.

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

Откройте app/llm/client.py. Найдите класс LLMStreamCompleted и после response_id добавьте два поля:

    refusal: str | None = None
    tool_calls: tuple[ToolCall, ...] = ()

Теперь итоговое событие сможет сообщить не только статистику генерации, но и два особых результата. В refusal будет находиться отказ модели, если она не смогла выполнить запрос. В tool_calls попадут полностью собранные вызовы инструментов. В большинстве запросов отказа и инструментов нет, поэтому refusal останется равен None, а tool_calls будет пустым кортежем.

Далее замените публичный метод stream():

def stream(
        self,
        messages: list[Message],
        tools: list[ToolDefinition] | None = None,
        tool_choice: Literal["auto", "none"] = "auto",
) -> Generator[LLMStreamEvent, None, None]:
    try:
        yield from self._stream_once(
            messages=messages,
            tools=tools,
            tool_choice=tool_choice,
        )
    except openai.APIError as error:
        raise self._convert_error(error) from error

Раньше этот метод принимал только сообщения. Теперь в него также можно передать схемы инструментов. Если tools равен None, запрос останется обычным текстовым потоком. Если список передан, модель сможет выбрать одну из доступных функций. То есть tool_choice="auto" означает, что модель сама решает, отвечать текстом или запросить инструмент.

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

Полностью замените метод _stream_once():

def _stream_once(
        self,
        messages: list[Message],
        tools: list[ToolDefinition] | None,
        tool_choice: Literal["auto", "none"],
) -> Generator[LLMStreamEvent, None, None]:
    if tools is 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},
        )
    else:
        stream = self._client.chat.completions.create(
            model=self._settings.model,
            messages=messages,
            temperature=self._settings.temperature,
            max_completion_tokens=self._settings.max_output_tokens,
            tools=tools,
            tool_choice=tool_choice,
            parallel_tool_calls=False,
            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
    refusal_parts: list[str] = []
    tool_call_parts: dict[int, dict[str, str]] = {}

    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)

            if choice.delta.refusal:
                refusal_parts.append(choice.delta.refusal)

            for tool_call in choice.delta.tool_calls or []:
                parts = tool_call_parts.setdefault(
                    tool_call.index,
                    {"id": "", "name": "", "arguments": ""},
                )

                if tool_call.id:
                    parts["id"] += tool_call.id

                if tool_call.function is not None:
                    if tool_call.function.name:
                        parts["name"] += tool_call.function.name
                    if tool_call.function.arguments:
                        parts["arguments"] += (
                            tool_call.function.arguments
                        )
    finally:
        stream.close()

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

    tool_calls: list[ToolCall] = []
    for _, parts in sorted(tool_call_parts.items()):
        if not parts["id"] or not parts["name"]:
            raise LLMProviderError(
                "Провайдер вернул незавершённый вызов инструмента."
            )

        tool_calls.append(
            ToolCall(
                id=parts["id"],
                name=parts["name"],
                arguments=parts["arguments"],
            )
        )

    yield LLMStreamCompleted(
        model=model,
        finish_reason=finish_reason,
        prompt_tokens=prompt_tokens,
        completion_tokens=completion_tokens,
        total_tokens=total_tokens,
        response_id=response_id,
        refusal="".join(refusal_parts) or None,
        tool_calls=tuple(tool_calls),
    )

Разберём работу метода по порядку. Сначала он создаёт потоковый запрос. В обеих ветках используется stream=True. Разница только в том, передаём ли мы модели список инструментов.

Перед чтением ответа создаются переменные для итоговых метаданных. Их нельзя получить заранее, потому что ProxyAPI присылает информацию постепенно. А списки refusal_parts и tool_call_parts будут накапливать те данные, которые нельзя использовать до завершения потока.

Затем цикл for chunk in stream получает очередной фрагмент ответа. После получения каждого chunk мы вычисляем, сколько времени прошло с начала генерации. Если установленный лимит уже превышен, метод создаёт LLMTimeoutError и останавливает обработку потока. После этой проверки из полученного фрагмента сохраняются название модели и идентификатор ответа.

Важно понимать, что эта проверка выполняется только после получения очередного chunk. Она ограничивает общую продолжительность генерации, но не может самостоятельно прервать сетевое чтение, если ProxyAPI перестал передавать данные. Такое ожидание ограничивается отдельными сетевыми таймаутами SDK и HTTP-клиента.

Статистика токенов может прийти в самом последнем чанке. У него поле choices иногда остаётся пустым, поэтому сначала читается usage, а уже затем выполняется if not chunk.choices. Пустой список здесь не означает ошибку, ведь в таком фрагменте просто нет продолжения текста.

Если в choice.delta.content находится текст, метод немедленно возвращает LLMTextDelta. Если пришла часть отказа, она добавляется в refusal_parts.

Части вызова инструмента сохраняются в словаре tool_call_parts. SDK указывает для них одинаковый index, поэтому метод понимает, какие фрагменты относятся к одному вызову. Параллельные инструменты у нас отключены, и на практике индекс обычно будет равен нулю.

Имя, идентификатор и аргументы дополняются через +=. Так строка аргументов постепенно превращается из нескольких обрывков в законченный JSON. Пока это только сборка текста, а проверку аргументов через Pydantic позднее выполнит ToolExecutor.

Блок finally закрывает поток ProxyAPI после успешного завершения, ошибки, превышения проверяемого лимита времени или досрочной остановки генератора. Если SDK в этот момент уже ожидает данные по сети, выполнение finally начнётся только после завершения этого ожидания или срабатывания сетевого таймаута.

После окончания потока части каждого инструмента превращаются в обычный ToolCall. Если провайдер не прислал идентификатор или имя, метод завершится ошибкой вместо запуска неполной операции. В самом конце создаётся единственное событие LLMStreamCompleted с причиной завершения, токенами, возможным отказом и готовыми вызовами инструментов.

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

Шаг 3

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

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

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

Перейдем к практике, откройте app/services/tool_loop.py. В начало файла добавьте импорт Generator:

from collections.abc import Generator

Импорт из app.llm.client замените следующим:

from app.llm.client import (
    LLMResult,
    LLMStreamCompleted,
    LLMStreamEvent,
    LLMTextDelta,
    Message,
    ToolCall,
    ToolDefinition,
)

В ToolCallingClient после уже существующего generate_with_tools() добавьте описание потокового метода:

    def stream(
            self,
            messages: list[Message],
            tools: list[ToolDefinition] | None = None,
            tool_choice: Literal["auto", "none"] = "auto",
    ) -> Generator[LLMStreamEvent, None, None]:
        ...

Напомню, что ToolCallingClient является протоколом и он не выполняет запрос и не содержит настоящей реализации. С помощью этого класса мы только указываем, что объект внутри BoundedToolLoop должен поддерживать метод stream().

После ToolLoopResult добавьте два новых события:

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

@dataclass(frozen=True)
class ToolLoopStreamCompleted:
    result: ToolLoopResult

ToolLoopStreamEvent = ToolLoopTextDelta | ToolLoopStreamCompleted

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

Теперь перейдём к классу BoundedToolLoop. Полностью удалите прежний метод run() и вставьте на его место два метода:

def run(self, messages: list[Message]) -> ToolLoopResult:
    completed: ToolLoopStreamCompleted | None = None

    for event in self.stream(messages):
        if isinstance(event, ToolLoopStreamCompleted):
            completed = event

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

    return completed.result

def stream(
        self,
        messages: list[Message],
) -> Generator[ToolLoopStreamEvent, None, None]:
    conversation = list(messages)
    llm_results: list[LLMResult] = []
    tool_names: list[str] = []
    audit_records: list[ToolAuditRecord] = []

    for step in range(1, self._max_steps + 1):
        text_parts: list[str] = []
        completed: LLMStreamCompleted | None = None

        try:
            events = self._client.stream(
                messages=conversation,
                tools=self._registry.api_schemas(),
                tool_choice="auto",
            )

            try:
                for event in events:
                    if isinstance(event, LLMTextDelta):
                        text_parts.append(event.text)
                        yield ToolLoopTextDelta(text=event.text)
                    else:
                        completed = event
            finally:
                events.close()
        except LLMClientError as error:
            raise ToolLoopError(
                f"Не удалось обратиться к модели: {error}"
            ) from error

        if completed is None:
            raise ToolLoopProtocolError(
                "Поток модели завершился без итогового события."
            )

        result = LLMResult(
            text="".join(text_parts) or None,
            refusal=completed.refusal,
            tool_calls=completed.tool_calls,
            model=completed.model,
            finish_reason=completed.finish_reason,
            prompt_tokens=completed.prompt_tokens,
            completion_tokens=completed.completion_tokens,
            total_tokens=completed.total_tokens,
            response_id=completed.response_id,
        )
        llm_results.append(result)

        if result.refusal is not None:
            raise ToolLoopProtocolError(
                f"Модель отказалась отвечать: {result.refusal}"
            )

        if result.finish_reason == "stop":
            if result.tool_calls:
                raise ToolLoopProtocolError(
                    "Текстовый ответ содержит tool calls."
                )

            text = self._require_text(result)
            yield ToolLoopStreamCompleted(
                result=self._build_result(
                    text=text,
                    llm_results=llm_results,
                    tool_names=tool_names,
                    audit_records=audit_records,
                    confirmation=None,
                )
            )
            return

        if result.finish_reason != "tool_calls":
            raise ToolLoopProtocolError(
                "Неожиданная причина завершения: "
                f"{result.finish_reason}."
            )

        if len(result.tool_calls) != 1:
            raise ToolLoopProtocolError(
                "Ожидался ровно один tool call на шаг."
            )

        if text_parts:
            raise ToolLoopProtocolError(
                "Вызов инструмента неожиданно содержит текст."
            )

        if step == self._max_steps:
            raise ToolLoopLimitError(
                "Модель запросила инструмент на последнем "
                "разрешённом шаге."
            )

        tool_call = result.tool_calls[0]
        conversation.append(
            self._build_tool_call_message(result, tool_call)
        )

        try:
            execution = self._executor.execute(tool_call)
        except ToolConfirmationRequiredError as error:
            audit_records.append(error.audit)
            confirmation_text = self._build_confirmation_text(
                error.request
            )
            yield ToolLoopTextDelta(text=confirmation_text)
            yield ToolLoopStreamCompleted(
                result=self._build_result(
                    text=confirmation_text,
                    llm_results=llm_results,
                    tool_names=tool_names,
                    audit_records=audit_records,
                    confirmation=error.request,
                )
            )
            return
        except ToolExecutionError as error:
            raise ToolLoopExecutionError(
                str(error),
                error.audit,
            ) from error

        tool_names.append(tool_call.name)
        audit_records.append(execution.audit)
        conversation.append(
            {
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": execution.content,
            }
        )

    raise ToolLoopLimitError(
        "Цикл завершился без финального ответа."
    )

Сначала разберём короткий run(). Этот метод нужен для старого непотокового способа работы и он запускает новый stream(), пропускает промежуточные части текста и сохраняет только ToolLoopStreamCompleted. Поэтому остальной код проекта по-прежнему может запросить один готовый ToolLoopResult.

Вся основная логика теперь находится в stream(). В начале создаются уже знакомые коллекции: conversation хранит сообщения текущей операции, llm_results — результаты всех обращений к модели, а tool_names и audit_records — сведения о выполненных функциях.

На каждом шаге цикл вызывает self._client.stream() и передаёт модели схемы из реестра. Когда приходит LLMTextDelta, текст выполняет сразу две задачи, а именно добавляется в text_parts для сборки полного ответа и немедленно возвращается выше как ToolLoopTextDelta.

LLMTextDelta("Ваш ")
→ сохранить в text_parts
→ вернуть ToolLoopTextDelta("Ваш ")

LLMTextDelta("заказ в пути.")
→ сохранить в text_parts
→ вернуть ToolLoopTextDelta("заказ в пути.")

Итоговое LLMStreamCompleted сохраняется в переменной completed. После окончания потока из накопленного текста и его метаданных создаётся обычный LLMResult. Благодаря этому ниже можно оставить знакомые проверки причин завершения.

Если finish_reason равен stop, модель закончила обычный ответ. Цикл проверяет текст, создаёт итоговый ToolLoopResult и завершает поток событием ToolLoopStreamCompleted.

Если причина равна tool_calls, цикл получает полностью собранный ToolCall. Затем он проверяет ограничение количества шагов и передаёт вызов в ToolExecutor. Результат функции добавляется в conversation, после чего начинается следующий шаг и модель формирует ответ с учётом новых данных.

Изменяющий данные инструмент по-прежнему не запускается сразу. ToolExecutor создаёт ToolConfirmationRequiredError, а цикл возвращает пользователю описание подготовленной операции и итоговый результат с заполненным confirmation.

Обратите внимание и на events.close() внутри finally. Если внешний код прекратит читать ответ, внутренний генератор LLMClient тоже будет закрыт. Это понадобится, когда пользователь остановит HTTP-запрос.

Теперь BoundedToolLoop умеет передавать финальный текст постепенно, не теряя вызовы инструментов, ограничения шагов и подтверждение опасных действий. В следующем шаге мы поднимем эти события ещё на один уровень — в ChatService, а затем опишем формат будущих SSE-событий.

Шаг 4

После предыдущей лекции BoundedToolLoop уже возвращает текст постепенно. Но ChatService пока вызывает его старый метод run(), получает готовый результат и только затем создаёт одно большое событие ChatTextDelta.

Теперь нужно передать события цикла дальше без ожидания полного текста:

Зачем создавать похожие события на двух уровнях? События ToolLoop... относятся к работе цикла инструментов, а события Chat... описывают работу чата. Благодаря этому ChatService не передаёт наружу внутренние детали BoundedToolLoop. Если реализация инструментов позднее изменится, внешний интерфейс чата можно будет сохранить.

Откройте app/services/chat_service.py и замените импорт из app.services.tool_loop:

from app.services.tool_loop import (
    BoundedToolLoop,
    ToolLoopError,
    ToolLoopStreamCompleted,
    ToolLoopTextDelta,
)

Затем найдите в stream_reply() следующий старый фрагмент:

            try:
                loop_result = self._tool_loop.run(prepared.messages)
            except ToolLoopError as error:
                raise ChatServiceError(
                    f"Не удалось получить ответ: {error}"
                ) from error

            elapsed_seconds = perf_counter() - started_at
            reply_text = self._validate_reply(loop_result.text)
            if loop_result.confirmation is not None:
                session.pending_confirmation_id = (
                    loop_result.confirmation.confirmation_id
                )
            yield ChatTextDelta(text=reply_text)

Полностью замените его новым кодом:

            try:
                loop_completed: ToolLoopStreamCompleted | None = None

                for event in self._tool_loop.stream(prepared.messages):
                    if isinstance(event, ToolLoopTextDelta):
                        yield ChatTextDelta(text=event.text)
                    else:
                        loop_completed = event
            except ToolLoopError as error:
                raise ChatServiceError(
                    f"Не удалось получить ответ: {error}"
                ) from error

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

            loop_result = loop_completed.result
            elapsed_seconds = perf_counter() - started_at
            reply_text = self._validate_reply(loop_result.text)

            if loop_result.confirmation is not None:
                session.pending_confirmation_id = (
                    loop_result.confirmation.confirmation_id
                )

Остальную часть stream_reply() оставьте без изменений. Ниже этого блока метод по-прежнему создаёт сообщение ассистента, обновляет историю, собирает ChatReply и возвращает ChatStreamCompleted.

Новый цикл получает события из self._tool_loop.stream(). Каждый ToolLoopTextDelta сразу превращается в ChatTextDelta и передаётся вызывающему коду. При этом части текста здесь не объединяются вручную: полный ответ уже находится в итоговом ToolLoopResult.

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

Здесь сохраняется важное правило, а именно что отдельные ChatTextDelta ещё не изменяют историю. Сохранение происходит только после успешного завершения цикла и проверки полного ответа. Если модель остановится из-за ошибки или внешний код закроет генератор раньше, выполнение не дойдёт до добавления сообщений.

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

Теперь внутренний поток доходит до границы HTTP-слоя. Перед созданием нового эндпоинта опишем формат данных, которые увидит клиент.

Откройте app/api/schemas/chat.py и после ChatReplyRequest добавьте модель одного текстового фрагмента:

class ChatStreamDeltaResponse(BaseModel):
    text: str

Для события delta достаточно одного поля: очередной части текста. Итоговые сведения в этот момент ещё могут быть неизвестны. Модель, количество токенов, состояние истории и признаки инструментов появятся позднее в событии completed.

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

Теперь создайте файл app/api/sse.py. Он будет одинаково преобразовывать любую Pydantic-модель в строку SSE:

from pydantic import BaseModel

def encode_sse(event: str, payload: BaseModel) -> str:
    data = payload.model_dump_json(exclude_none=True)
    return f"event: {event}\ndata: {data}\n\n"

После этого функция добавляет строку с названием события, строку с его данными и пустую строку в конце. Именно пустая строка показывает клиенту, что текущее SSE-событие закончилось:

event: delta
data: {"text":"Ваш заказ"}

event: delta
data: {"text":" находится в пути."}

Событие completed будет содержать полный ChatReplyResponse. Клиент не должен повторно добавлять его поле text к уже показанным частям. Это событие сообщает, что генерация успешно завершилась, а ответ сохранён в истории.

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

Шаг 5

Теперь ChatService возвращает части ответа постепенно, а encode_sse() умеет оформлять их как SSE. Осталось создать эндпоинт, через который этот поток получит HTTP-клиент.

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

Откройте app/api/routes/chat.py. В начало файла добавьте импорт генератора:

from collections.abc import AsyncGenerator
from starlette.concurrency import iterate_in_threadpool

В существующий импорт FastAPI добавьте Request, а рядом импортируйте StreamingResponse:

from fastapi import APIRouter, Depends, Request, Response, status
from fastapi.responses import StreamingResponse

Request понадобится для получения Request ID при формировании ошибки, а StreamingResponse принимает генератор и отправляет каждое возвращаемое им значение клиенту, не собирая весь ответ заранее.

Далее добавьте новые элементы в существующие группы импортов:

from app.api.sse import encode_sse
from app.api.schemas.chat import (
    ChatHistoryMessageResponse,
    ChatReplyRequest,
    ChatReplyResponse,
    ChatSessionCreatedResponse,
    ChatSessionResponse,
    ChatStreamDeltaResponse,
    ToolActionResponse,
)
from app.api.schemas.errors import (
    ErrorContentResponse,
    ErrorResponse,
)
from app.services.chat_service import (
    ChatServiceError,
    ChatSessionConflictError,
    ChatSessionNotFoundError,
    ChatStreamCompleted,
    ChatTextDelta,
)

После функции reply_to_chat() добавьте новый эндпоинт:

@router.post(
    "/chat/sessions/{session_id}/messages/stream",
    responses={
        404: {
            "model": ErrorResponse,
            "description": "Чат-сессия не найдена.",
        },
        409: {
            "model": ErrorResponse,
            "description": (
                "Состояние сессии не позволяет "
                "отправить сообщение."
            ),
        },
        422: {
            "model": ErrorResponse,
            "description": "Запрос не прошёл проверку.",
        },
    },
)
def stream_chat_reply(
        session_id: str,
        payload: ChatReplyRequest,
        request: Request,
        container: Annotated[
            ApplicationContainer,
            Depends(get_container),
        ],
) -> StreamingResponse:
    try:
        session = container.chat_service.get_session(session_id)
    except ChatSessionNotFoundError as error:
        raise ApiError(
            status_code=status.HTTP_404_NOT_FOUND,
            code="chat_session_not_found",
            message="Чат-сессия не найдена.",
        ) from error

    if session.pending_confirmation:
        raise ApiError(
            status_code=status.HTTP_409_CONFLICT,
            code="chat_session_conflict",
            message=(
                "Сначала подтвердите или отмените "
                "подготовленное действие."
            ),
        )

    events = _stream_chat_events(
        request=request,
        session_id=session_id,
        message=payload.message,
        container=container,
    )
    return StreamingResponse(
        events,
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",
        },
    )

Эндпоинт принимает ту же модель ChatReplyRequest, что и обычная отправка сообщения. Поэтому тело запроса не меняется. Клиент выбирает только способ получения результата, а именно один готовый JSON через /messages или последовательность событий через /messages/stream.

Сначала вызывается get_session(). Эта проверка выполняется до создания потокового ответа. Если сессия не существует, сервер сразу вернёт обычный 404 Not Found.

Затем проверяется pending_confirmation. Пока пользователь не подтвердил или не отменил подготовленное действие, новое сообщение отправлять нельзя. В такой ситуации сервер вернёт 409 Conflict.

Если проверки пройдены, создаётся генератор events. Обращение к модели в этот момент ещё не начинается. Код генератора начнёт выполняться, когда StreamingResponse приступит к чтению первого события.

media_type="text/event-stream" сообщает клиенту, что ответ записан в формате SSE. Заголовок Cache-Control запрещает кэшировать поток. X-Accel-Buffering: no просит некоторые прокси-серверы не накапливать маленькие части ответа, а передавать их сразу.

Теперь напишем сам генератор. Сразу после нового эндпоинта добавьте:

async def _stream_chat_events(
        request: Request,
        session_id: str,
        message: str,
        container: ApplicationContainer,
) -> AsyncGenerator[str, None]:
    stream = container.chat_service.stream_reply(
        session_id=session_id,
        user_text=message,
    )

    try:
        async for event in iterate_in_threadpool(stream):
            if await request.is_disconnected():
                break
            if isinstance(event, ChatTextDelta):
                yield encode_sse(
                    "delta",
                    ChatStreamDeltaResponse(text=event.text),
                )
                continue

            if isinstance(event, ChatStreamCompleted):
                result = event.result
                yield encode_sse(
                    "completed",
                    ChatReplyResponse(
                        text=result.text,
                        session_id=result.session_id,
                        history_messages=result.history_messages,
                        model=result.model,
                        prompt_id=result.prompt_id,
                        prompt_version=result.prompt_version,
                        elapsed_seconds=result.elapsed_seconds,
                        usage=TokenUsageResponse(
                            prompt_tokens=result.prompt_tokens,
                            completion_tokens=(
                                result.completion_tokens
                            ),
                            total_tokens=result.total_tokens,
                        ),
                        tool_calls_count=result.tool_calls_count,
                        tool_names=list(result.tool_names),
                        requires_confirmation=(
                            result.requires_confirmation
                        ),
                        summary_updated=result.summary_updated,
                        has_summary=result.has_summary,
                    ),
                )

    except ChatSessionNotFoundError:
        yield _stream_error(
            request=request,
            code="chat_session_not_found",
            message="Чат-сессия не найдена.",
        )
    except ChatSessionConflictError as error:
        yield _stream_error(
            request=request,
            code="chat_session_conflict",
            message=str(error),
        )
    except ChatServiceError:
        yield _stream_error(
            request=request,
            code="chat_stream_failed",
            message="Не удалось завершить потоковый ответ.",
        )
    finally:
        stream.close()

В начале функции вызывается ChatService.stream_reply(). Он возвращает внутренний генератор чата, а цикл async for получает события синхронного генератора через iterate_in_threadpool(stream).

Если пришёл ChatTextDelta, мы берём очередную часть текста, создаём ChatStreamDeltaResponse и превращаем его в событие delta. Оператор yield передаёт готовую SSE-строку в StreamingResponse, после чего она отправляется клиенту.

Если пришёл ChatStreamCompleted, значит генерация полностью завершилась. Внутри события находится готовый ChatReply. Из него создаётся знакомый ChatReplyResponse со всем текстом и итоговыми метаданными, а затем отправляется событие completed.

Блок finally отвечает за завершение внутреннего генератора. Он выполняется после успешного окончания потока, при ошибке или при досрочном прекращении чтения ответа. Внутри него вызывается stream.close(), поэтому после завершения текущего чтения запускается очистка всей цепочки генераторов и освобождается блокировка сессии.

ChatService.stream_reply() возвращает синхронный генератор, поскольку внутри проекта используется синхронный SDK. Мы читаем его через iterate_in_threadpool(), чтобы ожидание следующего фрагмента не останавливало асинхронную работу FastAPI.

После получения каждого события вызывается request.is_disconnected(). Этот метод проверяет, продолжает ли HTTP-клиент ждать ответ. Если клиент уже закрыл соединение, цикл завершается и управление переходит в блок finally, где вызывается stream.close().

Однако проверка выполняется только между событиями. Если в этот момент синхронный SDK уже ожидает следующий фрагмент от ProxyAPI, Python не может мгновенно остановить выполняющееся сетевое чтение. Сначала провайдер должен передать очередной фрагмент либо должен сработать сетевой таймаут. Только после этого выполнение дойдёт до finally, генератор закроется, а session.lock будет освобождён.

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

Ошибки внутри генератора обрабатываются иначе, чем предварительные проверки. Если проблема с session_id обнаружилась до создания StreamingResponse, сервер может вернуть обычный статус 404 или 409. Но после отправки первого события HTTP-ответ уже начался со статусом 200 OK, и заменить его на 502 Bad Gateway невозможно.

Поэтому исключения, возникшие внутри потока, превращаются в последнее SSE-событие error. Для его создания после _stream_chat_events() добавьте:

def _stream_error(
        request: Request,
        code: str,
        message: str,
) -> str:
    payload = ErrorResponse(
        error=ErrorContentResponse(
            code=code,
            message=message,
            request_id=getattr(
                request.state,
                "request_id",
                "unknown",
            ),
        )
    )
    return encode_sse("error", payload)

Функция создаёт уже знакомый ErrorResponse. В него попадают программный код ошибки, безопасное сообщение и Request ID текущего HTTP-запроса. Затем encode_sse() превращает модель в событие error.

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

Шаг 6

Мы закончили реализацию потокового ответа на всех уровнях приложения. Теперь проверим не отдельные методы, а всю цепочку целиком, где FastAPI примет сообщение, ChatService подготовит историю, BoundedToolLoop при необходимости выполнит инструмент, а клиент начнёт получать текст ещё до окончания генерации.

Запустите сервер из корня проекта:

fastapi dev app/api/main.py

Откройте http://127.0.0.1:8000/docs, выполните POST /chat/sessions и скопируйте полученный session_id.

Сам поток лучше проверять не в Swagger UI. Интерфейс документации может дождаться окончания запроса и показать всё тело сразу, поэтому постепенная выдача будет незаметна. Для наглядной проверки воспользуемся curl во втором окне терминала.

Сначала отправьте обычный вопрос. Вместо <session_id> подставьте полученный идентификатор без угловых скобок:

curl -N -X POST \
  -H "Content-Type: application/json" \
  -d '{"message":"Объясните простыми словами, как оформить возврат товара."}' \
  http://127.0.0.1:8000/chat/sessions/<session_id>/messages/stream

Параметр -N отключает буферизацию вывода в curl. Благодаря этому терминал показывает события сразу после получения, а не накапливает их до завершения ответа.

Во время генерации будут появляться события delta. В конце должно прийти одно событие completed:

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

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

Теперь вызовите в Swagger:

GET /chat/sessions/{session_id}

В истории должны находиться две записи, только вопрос пользователя и один полный ответ ассистента.

Отдельные delta не становятся отдельными сообщениями. Они нужны только для постепенного отображения одного ответа.

Далее проверим, что потоковая выдача не отключила инструменты. Отправьте в ту же сессию:

curl -N -X POST \
  -H "Content-Type: application/json" \
  -d '{"message":"Где сейчас заказ 8041?"}' \
  http://127.0.0.1:8000/chat/sessions/<session_id>/messages/stream

И в результате мы получим:

Внутри приложения произойдёт несколько действий:

Служебные фрагменты имени функции и её JSON-аргументов в терминале появляться не должны. Их собирает и обрабатывает BoundedToolLoop. Клиент видит только окончательный текст модели.

Посмотрите на событие completed. Поле tool_calls_count должно быть равно 1, а в tool_names должно находиться имя get_order_status. Так мы видим, что инструмент действительно использовался, даже несмотря на то что его внутренние сообщения не передавались пользователю.

При желании можно проверить и действие, требующее подтверждения:

curl -N -X POST \
  -H "Content-Type: application/json" \
  -d '{"message":"Создайте заявку на возврат заказа 7718. Товар пришёл повреждённым."}' \
  http://127.0.0.1:8000/chat/sessions/<session_id>/messages/stream

Если модель выберет create_return_request, приложение передаст описание подготовленной операции, а в completed поле requires_confirmation станет равно true.

Сама заявка ещё не создаётся. Для выполнения или отмены по-прежнему используются отдельные эндпоинты /confirm и /cancel.

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

curl -N -X POST \
  -H "Content-Type: application/json" \
  -d '{"message":"Подробно перечислите шаги диагностики кофемашины, которая не включается."}' \
  http://127.0.0.1:8000/chat/sessions/<session_id>/messages/stream

После появления первых событий delta остановите curl сочетанием Control+C. Если команда запущена через IDE, можно нажать кнопку остановки процесса.

После остановки curl клиент закрывает соединение. FastAPI прекращает передачу ответа, а функция _stream_chat_events() переходит к завершению внутреннего генератора. Если синхронный SDK в этот момент ожидает очередной фрагмент от ProxyAPI, очистка может произойти не мгновенно, а после получения следующего фрагмента или срабатывания сетевого таймаута.

Когда генератор завершится, ChatService освободит блокировку сессии. Если модель ещё не успела сформировать и сохранить успешный результат, незавершённая пара сообщений не должна появиться в истории. Проверьте это через GET /chat/sessions/{session_id}. Если модель закончила ответ раньше, чем был остановлен curl, готовая пара уже может находиться в истории — это нормальное поведение.

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

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

Осталось посмотреть на ошибки, которые возникают до начала потока. Повторите запрос с несуществующим session_id. Сервер должен вернуть обычный 404 Not Found, потому что сессия проверяется до создания StreamingResponse.

Если отправить пустое значение message или вообще не передать это поле, Pydantic остановит запрос ещё раньше и FastAPI вернёт 422 Unprocessable Content.

На этом раздел завершён. Обычный эндпоинт чата продолжает возвращать один готовый ChatReplyResponse, а новый потоковый эндпоинт передаёт события delta и заканчивается событием completed или error. Оба варианта используют одни сессии, историю и инструменты.

В следующем разделе подключим к FastAPI возможности Document AI из четвёртого модуля. И наш сервер научится принимать загруженный файл, временно сохранять его, передавать существующему DocumentAIService и удалять временные данные после завершения обработки.

Шаг 7

Зачем пустая строка в конце SSE-записи нужна клиенту?

event: delta
data: {"text":"Привет"}
Правильный ответ
  • ✅ Она завершает текущее SSE-событие

Шаг 8

Какой media_type должен иметь потоковый ответ SSE?

Правильный ответ
  • ✅ text/event-stream

Шаг 9

Почему ошибка, возникшая после начала потоковой передачи, отправляется событием error, а не новым HTTP-статусом?

Правильный ответ
  • ✅ HTTP-заголовки и исходный статус уже отправлены клиенту

Шаг 10

Что следует делать при отключении HTTP-клиента от SSE-потока?

Выберите все правильные варианты.

Правильный ответ
  • ✅ Не сохранять незавершённый ответ как полноценную историю
  • ✅ Прекратить отправку новых SSE-событий после обнаружения отключения
  • ✅ Проверять request.is_disconnected() во время чтения событий

Шаг 11

Кодирование одного SSE-события

SSE-эндпоинт отправляет клиенту текстовые события, состоящие из строк event, data и пустой строки-разделителя. Вам нужно написать вспомогательную функцию, которая преобразует имя события и Pydantic-модель в одну готовую строку SSE.

Задача

Реализуйте вспомогательную функцию encode_sse(event, payload). Получите JSON из переданной Pydantic-модели и верните одну строку SSE с очищенным именем события и правильным разделителем в конце.

Требования

  • Не изменяйте имя функции encode_sse и её аргументы.
  • Удалите пробелы по краям event с помощью strip().
  • Если после очистки имя события оказалось пустым, выбросьте ValueError.
  • Для получения JSON вызовите payload.model_dump_json(exclude_none=True).
  • Верните одну строку в формате event: EVENT data: JSON , где EVENT — очищенное имя события.
  • После JSON должны находиться ровно два символа перевода строки: первый завершает строку data, второй создаёт пустую строку между событиями.
  • Не преобразовывайте JSON обратно в словарь и не добавляйте текст после завершающей пустой строки.

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

Функция используется для формирования тела потокового HTTP-ответа:

GET /demo-stream

200 OK
Content-Type: text/event-stream; charset=utf-8

event: delta
data: {"text":"Привет"}
Правильный ответ 
def encode_sse(event: str, payload) -> str:
    event = event.strip()

    if not event:
        raise ValueError

    data = payload.model_dump_json(exclude_none=True)

    return f"event: {event}\ndata: {data}\n\n"

Шаг 12

Публичные данные событий чата

Чат-сервис возвращает внутренние события ChatTextDelta и ChatStreamCompleted, которые нельзя отправлять клиенту напрямую. Вам нужно написать вспомогательную функцию, которая определит тип события и соберёт из него публичное имя и словарь данных.

Задача

Реализуйте вспомогательную функцию serialize_chat_event(event). Для текстового фрагмента верните данные события "delta", для завершения — данные события "completed", а неизвестный тип отклоните.

Требования

  • Не изменяйте имя функции и её аргумент.
  • Для объекта ChatTextDelta верните кортеж ("delta", {"text": event.text}).
  • Для объекта ChatStreamCompleted сначала получите итоговый объект из event.result.
  • Для завершения верните имя "completed" и словарь ровно с ключами "text", "session_id", "history_messages" и "requires_confirmation".
  • Значения этих четырёх полей возьмите из одноимённых атрибутов event.result.
  • Для объекта любого другого типа выбросьте TypeError.
  • Не изменяйте переданное событие и вложенный объект результата.

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

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

POST /chat/sessions/abc123/messages/stream
Content-Type: application/json

{"message":"Привет"}

200 OK
Content-Type: text/event-stream; charset=utf-8

event: delta
data: {"text":"Здрав"}

event: completed
data: {"text":"Здравствуйте!","session_id":"abc123","history_messages":2,"requires_confirmation":false}
Правильный ответ 
from task_environment import ChatStreamCompleted, ChatTextDelta

def serialize_chat_event(event) -> tuple[str, dict]:
    if isinstance(event, ChatTextDelta):
        return (
            "delta",
            {
                "text": event.text,
            },
        )

    if isinstance(event, ChatStreamCompleted):
        result = event.result

        return (
            "completed",
            {
                "text": result.text,
                "session_id": result.session_id,
                "history_messages": result.history_messages,
                "requires_confirmation": result.requires_confirmation,
            },
        )

    raise TypeError

Шаг 13

Потоковый ответ FastAPI

В функцию create_sse_response передаётся готовый генератор строк SSE. Вам нужно обернуть этот генератор в StreamingResponse, чтобы FastAPI передавал события клиенту постепенно.

Задача

Реализуйте вспомогательную функцию create_sse_response(stream). Оберните исходный поток в StreamingResponse с типом text/event-stream и верните ответ, не начиная чтение генератора внутри функции.

Требования

  • Не изменяйте имя функции create_sse_response и аргумент stream.
  • Создайте и верните объект StreamingResponse.
  • Передайте исходный stream первым аргументом.
  • Укажите media_type="text/event-stream".
  • Не преобразовывайте stream в список, строку или другой контейнер.
  • Не перебирайте поток внутри create_sse_response, чтение должен начать FastAPI во время отправки ответа.

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

Маршрут передаёт в функцию генератор с одним событием, после чего клиент получает:

GET /demo-stream

200 OK
Content-Type: text/event-stream; charset=utf-8

event: delta
data: {"text":"Привет"}
Правильный ответ 
from fastapi.responses import StreamingResponse

def create_sse_response(stream) -> StreamingResponse:
    return StreamingResponse(
        stream,
        media_type="text/event-stream",
    )

Шаг 14

Проверка сессии перед открытием потока

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

Задача

Реализуйте вспомогательную функцию validate_stream_session(service, session_id). Получите состояние сессии, верните его при корректном состоянии, а два ожидаемых случая ошибки преобразуйте в соответствующие ApiError.

Требования

  • Не изменяйте имя функции и её аргументы.
  • Вызовите service.get_session(session_id) ровно один раз.
  • Если сервис выбросил SessionNotFoundError, выбросьте ApiError(404, "chat_session_not_found", "Чат-сессия не найдена.").
  • Если у полученного снимка pending_confirmation is True, выбросьте ApiError(409, "chat_session_conflict", "Сначала подтвердите или отмените подготовленное действие.").
  • Если сессия существует и pending_confirmation is False, верните тот же объект снимка, который вернул сервис.
  • Не вызывайте service.stream_reply(), запуск потока выполняется только после успешной проверки.

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

Для существующей сессии без ожидающего действия эндпоинт может начать поток:

POST /chat/sessions/abc123/messages/stream
Content-Type: application/json

{"message":"Привет"}

200 OK
Content-Type: text/event-stream; charset=utf-8

event: completed
data: {}

Неизвестный session_id должен привести к статусу 404. Если сессия ожидает подтверждения или отмены действия, должен вернуться статус 409 ещё до начала SSE-потока.

Правильный ответ 
from task_environment import ApiError, SessionNotFoundError

def validate_stream_session(service, session_id: str):
    try:
        snapshot = service.get_session(session_id)
    except SessionNotFoundError:
        raise ApiError(
            404,
            "chat_session_not_found",
            "Чат-сессия не найдена.",
        )

    if snapshot.pending_confirmation is True:
        raise ApiError(
            409,
            "chat_session_conflict",
            "Сначала подтвердите или отмените подготовленное действие.",
        )

    return snapshot

Шаг 15

События и ошибки внутри SSE-потока

Чат-сервис выдаёт части ответа через генератор, а функция-эндпоинт передаёт их клиенту как SSE-события. Вам нужно написать промежуточный генератор, который преобразует обычные события в строки SSE, а ошибку сервиса — в последнее событие error.

Задача

Реализуйте генератор stream_chat_events(service, session_id, message, request_id). Получите поток событий от чат-сервиса, преобразуйте каждое событие в строку SSE, а при ошибке сервиса завершите поток событием error.

Требования

  • Не изменяйте имя функции и её аргументы.
  • Вызовите service.stream_reply(session_id=session_id, user_text=message) ровно один раз.
  • Последовательно перебирайте события полученного потока, не собирая их заранее в список.
  • Для каждого обычного события вызовите serialize_chat_event(event) и получите имя события вместе со словарём данных.
  • Передайте полученные имя и словарь в encode_sse_data(name, payload), а возвращённую строку сразу выдайте через yield.
  • Если ChatServiceError возник при создании или переборе потока, выдайте ровно одно событие "error" через encode_sse_data.
  • Данные события ошибки должны содержать ровно три ключа: "code": "chat_stream_failed", "message": "Не удалось завершить потоковый ответ." и "request_id" со значением аргумента request_id.
  • После события error завершите генератор и не пробрасывайте ChatServiceError наружу.
  • Не включайте исходный текст исключения в данные, отправляемые клиенту.

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

При успешной генерации клиент получает части текста и итоговое событие:

POST /chat/sessions/abc123/messages/stream
Content-Type: application/json

{"message":"Привет"}

200 OK
Content-Type: text/event-stream; charset=utf-8

event: delta
data: {"text":"Здрав"}

event: completed
data: {"text":"Здравствуйте!"}

Если ошибка возникла после первого delta, HTTP-статус остаётся 200, а последней записью становится:

event: error
data: {"code":"chat_stream_failed","message":"Не удалось завершить потоковый ответ.","request_id":"req-7"}
Правильный ответ 
from task_environment import ChatServiceError, encode_sse_data, serialize_chat_event

def stream_chat_events(service, session_id: str, message: str, request_id: str):
    try:
        stream = service.stream_reply(
            session_id=session_id,
            user_text=message,
        )

        for event in stream:
            name, payload = serialize_chat_event(event)
            yield encode_sse_data(name, payload)

    except ChatServiceError:
        yield encode_sse_data(
            "error",
            {
                "code": "chat_stream_failed",
                "message": "Не удалось завершить потоковый ответ.",
                "request_id": request_id,
            },
        )