SSE(Server-Sent Events)是一种基于 HTTP 协议的服务器向客户端单向推送数据的技术，允许服务器主动向已建立连接的客户端持续发送事件流（如实时通知、更新数据等），无需客户端频繁轮询。

核心特点：基于 HTTP 长连接，单向通信（仅服务器→客户端），数据以 “事件” 格式传输（包含事件类型、数据体等结构化信息），天然支持断线重连。
典型场景：大语言模型（LLM）客户端、股票行情实时更新、新闻推送、系统通知等只需服务器主动下发数据的场景。

它与websocket的主要区别是：

  • 若场景仅需 “服务器推数据给客户端”（单向），优先选 SSE（实现简单、基于 HTTP 无额外协议成本）
  • 若需 “客户端与服务器实时双向通信”（如聊天、互动），必须选 WebSocket（全双工能力是核心优势）

下面我们使用大语言模型qwen3实现翻译功能。它可以自动识别源语言，就可以翻译为目标语种。

像千问这种大模型是基于多语言训练的，所以它支持中文、英文、法文、西班牙等多个语种的翻译。

这是实现的效果： 前端显示LLM(大语言模型)的流式输出内容

构建提示词

使用 langchain 的 SystemMessage、HumanMessage 构建提示词：

def build_prompt(language_dst:str,text:str):
    """构建提示词
    在不指定源语言的情况下，LLM也可以翻译。
    """
    prompt_sysytem = "你是一个专业的翻译助手。"
    prompt_user = "你是一个专业的翻译助手。请将以下文本翻译成客户指定语言。只输出翻译结果，不要包含任何其他解释、说明或额外文本。默认翻译成中英文互译，用户指定语言的话使用指定语言。"
    prompt_user = f"""你是一个专业的翻译助手。请将以下文本翻译成 {language_dst}。只输出翻译结果，不要包含任何其他解释、说明或额外文本。
        原文：
        {text}
        翻译："""
    return [
        SystemMessage(content =prompt_sysytem),
        HumanMessage(content=prompt_user)
    ]

大模型流式响应

我们让大模型一点一点吐出响应内容：

model = ChatOllama(model=llm_model_name,temperature=0.2,verbose=True)

def stream_generator(language_dst:str,text:str):
    """流式输出大模型的回答"""
    prompt = build_prompt(language_dst=language_dst,text=text)
    inside_think = False # 标记是否在<think>区间
    for chunk in model.stream(prompt):
        if isinstance(chunk, AIMessage):
            # 过滤掉<think>...</think>部分            
            if "<think>" in chunk.content:
                inside_think = True
            elif "</think>" in chunk.content:
                inside_think = False
                continue    # 跳过 </think>

            if not inside_think:
                yield {
                    "event": "message",
                    "data": chunk.content
                }
    yield {
        "event": "end",
        "data": "[[END]]"
    }

本地qwen3 8b的会自动使用思考模式，这里把思考过程去掉了。

定义接口

我们用Fast API实现接口，用 sse_starlette 实现 SSE(Server-Sent Events)

class TranslateRequest(BaseModel):
    """请求消息体"""
    text: str = Field(..., min_length=1, description="要翻译的文本")
    language_dst: str = Field(..., min_length=1, description="目标语言")

app = FastAPI(title="翻译接口")

@app.post("/translate_stream",tags=["接口"],summary="翻译接口，流式返回内容")
async def stream_translation(req: TranslateRequest):
    return EventSourceResponse(stream_generator(language_dst=req.language_dst,text=req.text))

实现 SSE 的代码量很小，很优雅：）

实现前端

为了能够完整展示 SSE 的实际效果，下面实现了一个get接口，它可以返回前端的html页面：

@app.get("/translate",tags=["测试客户端"],summary="返回翻译的前端界面")
async def get_translate_html():
    """返回翻译页面
    """

    file_path = os.path.join(os.path.dirname(__file__), "./33.SSE_client.html")
    with open(file_path, "r", encoding="utf-8") as f:
        html_content = f.read()
    return HTMLResponse(content=html_content)

下面是前端html页面的主要代码：

document.getElementById("translateForm").addEventListener("submit", async function (e) {
    e.preventDefault();

    const languageDst = document.getElementById("language_dst").value.trim();
    const text = document.getElementById("text").value.trim();

    if (!languageDst || !text) {
        alert("目标语言和翻译文本不能为空！");
        return;
    }

    const outputDiv = document.getElementById("output");
    outputDiv.textContent = "";

    // 关闭前一个连接
    if (window.currentEventSource) {
        window.currentEventSource.close();
    }

    // 发起 POST 请求初始化 SSE
    const response = await fetch("http://127.0.0.1:9000/translate_stream", {
        method: "POST",
        headers: {
            "Content-Type": "application/json"
        },
        body: JSON.stringify({
            text: text,
            language_dst: languageDst
        })
    });

    if (!response.ok) {
        outputDiv.textContent = "请求失败: " + response.statusText;
        return;
    }

    const reader = response.body.getReader();
    const decoder = new TextDecoder("utf-8");

    let buffer = "";
    while (true) {
        const { value, done } = await reader.read();
        if (done) break;
        buffer += decoder.decode(value, { stream: true });

        // 处理 SSE 消息（简单解析）
        const lines = buffer.split("\n");
        buffer = lines.pop(); // 可能是 incomplete 的最后一行
        for (const line of lines) {
            if (line.startsWith("data:")) {
                const data = line.slice(5).trim();
                if (data === "[[END]]") {
                    return;
                }
                outputDiv.textContent += data+" ";
            }
        }
    }
})

见证效果

通过 uvicorn 启动：

import uvicorn
if __name__ == '__main__':   
    uvicorn.run(app, host="127.0.0.1", port=9000)

现在我们可以启动后端接口，然后打开浏览器，输入地址：http://127.0.0.1:9000/translate_stream 即可。

总结

使用 qwen3 Fast API 和 sse_starlette 实现 SSE(Server-Sent Events) 只需要很少的代码量，可能正式因为具有 简单明了 的特点，才会有这么多人喜欢用python吧。

代码

本文涉及的所有代码以及相关资源都已经共享，参见：

为便于找到代码，程序文件名称最前面的编号与本系列文章的文档编号相同。

🪐感谢您观看，祝好运🪐