SSE Turns Python Generators into Streaming LLM Chat Replies

Using SSE for Streaming Large Model Output in Python

In chat pages, answers often appear piece by piece, without waiting for the entire text to be generated. SSE can achieve this effect. Starting from an existing agent.execute_stream(), this guide explains how the backend receives chunks, wraps and returns them, and provides a simple web page.

1. Why use yield?

yield can hand over the current segment and next time continue execution from the paused position. Combined with StreamingResponse, it can return while generating. Python docs

Start with a simple example that returns only two text segments:

def generate_reply():
    yield "你好"  # Hand over this text, function pauses here
    yield "，欢迎学习 SSE。"  # Continue execution when the next value is fetched


for part in generate_reply():
    print(part)  # Process each segment as it is fetched

The loop will print "你好" and "，欢迎学习 SSE。" in turn. yield can continue execution after handing over content, which is suitable for providing content piece by piece. The following interface also uses this idea, only replacing printed text with sending SSE messages through StreamingResponse.

2. SSE format requirements

The common way to write it is f"event: {event_type}\ndata: {data}\n\n". Remember these requirements:

Requirement	Description
Response type and encoding	text/event-stream, use UTF-8.
event:	Optional event name; defaults to message when omitted.
data:	Message content, can be text or JSON.
Newlines and blank lines	Each field occupies one line; this example uses \n\n to mark the end of a message, and the trailing blank line cannot be omitted — it does not close the connection.

SSE official spec: message format and parsing rules

3. What does agent.execute_stream(messages) return?

In this example, with thinking mode enabled and returning thinking content, the existing agent.execute_stream(messages) returns two kinds of dicts piece by piece:

Returned data	Frontend handling
{"type": "reasoning", "content": "思考片段"}	Append content to the thinking area.
{"type": "answer", "content": "你好"}	Append content to the answer area.

The frontend handles the two types separately; both kinds may be returned many times in a row.

4. The role of StreamingResponse

yield provides chunks, StreamingResponse continuously reads and sends them. The configuration in the interface is as follows:

return StreamingResponse(
    generate_reply(),
    media_type="text/event-stream",
    headers={
        "Cache-Control": "no-cache",
        "X-Accel-Buffering": "no",
    },
)

Parameter or config	Effect
generate_reply()	Passed as the first parameter content, provides a generator that can be read piece by piece.
media_type	Sets the response type, here text/event-stream.
headers	Adds response headers, this example includes the two below.
headers["Cache-Control"]	no-cache: confirm with the server before reusing cache.
headers["X-Accel-Buffering"]	no: asks Nginx to disable response buffering and forward chunks promptly; requires the proxy config to allow it.

Parameter docs

5. Interface and page example

Effect preview:

The following code focuses on SSE sending and receiving. agent uses an existing instance; definition and wrapping are omitted here.

Backend: main.py

import json
import logging
from pathlib import Path

from fastapi import FastAPI
from fastapi.responses import FileResponse, StreamingResponse
from pydantic import BaseModel

# Create the app; agent reuses existing import or initialization code, omitted here.
app = FastAPI()

class ChatRequest(BaseModel):
    message: str

# Receive a question, reuse the existing Agent's streaming method.
@app.post("/chat", response_class=StreamingResponse)
async def chat(request: ChatRequest) -> StreamingResponse:
    # Compose the message list the Agent receives.
    messages = [{"role": "user", "content": request.message}]

    # Async generator: read a chunk, send an SSE message.
    async def generate_reply():
        reply_parts = []
        try:
            async for event in agent.execute_stream(messages):
                # Send both thinking and body; collect only body for later use.
                if event["type"] == "answer":
                    reply_parts.append(event["content"])

                # JSON escapes newlines in the body to avoid breaking SSE frame boundaries.
                event_data = json.dumps(event, ensure_ascii=False)
                yield f"event: {event['type']}\ndata: {event_data}\n\n"

            reply = "".join(reply_parts)
            if not reply:
                raise RuntimeError("模型没有返回回答正文")

            # If persisting the conversation, save successfully before notifying the frontend.
            event_data = json.dumps({"type": "done"}, ensure_ascii=False)
            yield f"event: done\ndata: {event_data}\n\n"
        except Exception:
            # Log server-side exceptions and notify the frontend of failure via SSE.
            logger.exception("聊天流式请求失败")
            event_data = json.dumps(
                {"type": "error", "message": "本次请求失败，请稍后重试。"},
                ensure_ascii=False,
            )
            yield f"event: error\ndata: {event_data}\n\n"

    # StreamingResponse continuously reads the generator and sends each message to the frontend.
    return StreamingResponse(
        generate_reply(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",
        },
    )

Frontend: index.html

<!doctype html>
<html lang="zh-CN">
<head>
  <meta charset="UTF-8" />
  <title>SSE 流式聊天</title>
</head>
<body>
  <textarea id="question">什么是 SSE？</textarea>
  <button id="send">发送</button>
  <button id="stop">停止</button>
  <pre id="thinking"></pre>
  <pre id="answer"></pre>

  <script>
    const question = document.querySelector("#question");
    const thinking = document.querySelector("#thinking");
    const answer = document.querySelector("#answer");
    const sendButton = document.querySelector("#send");
    let controller = null;

    function parseEvent(frame) {
      // Matches the format where each message has one line of data; extract the JSON content.
      const line = frame.split(/\r?\n/).find((s) => s.startsWith("data:"));
      return line ? JSON.parse(line.slice(5)) : null;
    }

    async function chat() {
      sendButton.disabled = true; // Avoid duplicate sends while receiving
      controller = new AbortController();
      thinking.textContent = "";
      answer.textContent = "";
      let reader;
      let buffer = "";
      let finished = false;
      const decoder = new TextDecoder();

      try {
        const response = await fetch("/chat", {
          method: "POST",
          headers: {
            "Content-Type": "application/json",
            "Accept": "text/event-stream",
          },
          body: JSON.stringify({ message: question.value }),
          signal: controller.signal,
        });
        if (!response.ok || !response.body) {
          throw new Error("接口请求失败");
        }

        reader = response.body.getReader();
        while (!finished) {
          const { value, done } = await reader.read();
          // Stream-decode preserves incomplete Chinese characters; complete on end.
          buffer += decoder.decode(value, { stream: !done });

          // Split out complete messages by blank lines, keep the last partial one for next time.
          const frames = buffer.split(/\r?\n\r?\n/);
          buffer = frames.pop() || "";
          for (const frame of frames) {
            const event = parseEvent(frame);
            if (!event) continue;
            // The type here comes from JSON, not the outer event line.
            if (event.type === "reasoning") {
              thinking.textContent += event.content;
            } else if (event.type === "answer") {
              answer.textContent += event.content; // Append answer piece by piece
            } else if (event.type === "done") {
              finished = true;
              break;
            } else if (event.type === "error") {
              throw new Error(event.message);
            }
          }
          if (done) break;
        }
        if (!finished) throw new Error("连接提前结束");
      } catch (error) {
        answer.textContent += error.name === "AbortError"
          ? "\n[已停止]" : "\n[失败] " + error.message;
      } finally {
        // After completion, failure, or stop, end reading and restore the send button.
        if (reader) {
          await reader.cancel().catch(() => {});
          reader.releaseLock();
        }
        controller = null;
        sendButton.disabled = false;
      }
    }

    sendButton.onclick = chat;
    document.querySelector("#stop").onclick = () => controller?.abort();
  </script>
</body>
</html>

Summary

To implement SSE, remember this sequence: backend generates content piece by piece → wraps into SSE messages → frontend reads and appends display. This example uses async for to read the agent's chunks, passes them via yield to StreamingResponse for continuous sending, and the frontend displays thinking content and answer text separately according to type.

When using, pay attention to the trailing blank line at the end of a message, and keep incomplete messages in the frontend buffer. Receiving done confirms completion; receiving error shows a prompt. This allows step-by-step display of the answer while handling end and failure cases.

Feel free to comment in the discussion area. If any content is inaccurate or unclear, point it out and we can improve it together.