AI图鉴

AI图鉴首页 › AI 资讯 › Python 中使用 SSE 实现大模型流式输出

Python 中使用 SSE 实现大模型流式输出

来源:掘金 · 发布时间:11-02 · 由 AI图鉴每小时自动聚合

聊天页面里的回答常常是一点点出现的,不用等整段文字生成完。SSE 就能实现这样的效果。下面从已有的 agent.execute_stream() 出发,讲清后端如何接收片段、包装并返回内容,再配上一个

聊天页面里的回答常常是一点点出现的,不用等整段文字生成完。SSE 就能实现这样的效果。下面从已有的 agent.execute_stream() 出发,讲清后端如何接收片段、包装并返回内容,再配上一个简单网页。

一、为什么使用 yield?

yield 可以先交出当前这一段,下次再从暂停的位置继续执行。 配合 StreamingResponse,就能边生成边返回。Python 说明

先看一个只返回两段文字的简单例子:

def generate_reply():

yield "你好" # 交出这段文字,函数在这里暂停

yield ",欢迎学习 SSE。" # 下次取值时继续执行

for part in generate_reply():

print(part) # 每取到一段,就处理一段

循环会依次打印“你好”和“,欢迎学习 SSE。”。yield 交出内容后还能继续执行,适合逐段提供内容。 后面的接口也是这个思路,只是把打印文字换成通过 StreamingResponse 发送 SSE 消息。

二、SSE 的格式要求

常用写法是 f"event: {event_type}\ndata: {data}\n\n",记住以下要求:

三、agent.execute_stream(messages) 返回什么?

本例模型开启思考模式并返回思考内容时,已有的 agent.execute_stream(messages) 会逐段返回两类字典:

前端根据 type 分别处理,两类片段都可能连续返回多次。

四、StreamingResponse 的作用

yield 提供片段,StreamingResponse 持续读取并发送。 接口中的配置如下:

return StreamingResponse(

generate_reply(),

media_type="text/event-stream",

headers={

"Cache-Control": "no-cache",

"X-Accel-Buffering": "no",

},

)

五、接口与页面示例

效果展示:

以下代码聚焦 SSE 的发送与接收,agent 使用已有实例,定义和封装在此省略。

后端:main.py

import json

import logging

from pathlib import Path

from fastapi import FastAPI

from fastapi.responses import FileResponse, StreamingResponse

from pydantic import BaseModel

# 创建应用;agent 沿用已有的导入或初始化代码,此处省略。

app = FastAPI()

class ChatRequest(BaseModel):

message: str

# 接收问题,沿用已有 Agent 的流式方法。

@app.post("/chat", response_class=StreamingResponse)

async def chat(request: ChatRequest) -> StreamingResponse:

# 将本次问题组成 Agent 接收的消息列表。

messages = [{"role": "user", "content": request.message}]

# 异步生成器:读取一段,就发出一条 SSE 消息。

async def generate_reply():

reply_parts = []

try:

async for event in agent.execute_stream(messages):

# 思考和正文都发送,只收集正文供后续使用。

if event["type"] == "answer":

reply_parts.append(event["content"])

# JSON 会转义正文中的换行,避免打乱 SSE 消息边界。

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("模型没有返回回答正文")

# 如需保存会话,应保存成功后再通知前端完成。

event_data = json.dumps({"type": "done"}, ensure_ascii=False)

yield f"event: done\ndata: {event_data}\n\n"

except Exception:

# 记录服务端异常,并通过 SSE 通知前端失败。

logger.exception("聊天流式请求失败")

event_data = json.dumps(

{"type": "error", "message": "本次请求失败,请稍后重试。"},

ensure_ascii=False,

)

yield f"event: error\ndata: {event_data}\n\n"

# StreamingResponse 持续读取生成器,将每条消息发送给前端。

return StreamingResponse(

generate_reply(),

media_type="text/event-stream",

headers={

"Cache-Control": "no-cache",

"X-Accel-Buffering": "no",

},

)

前端: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) {

// 对应本例每条消息只有一行 data 的格式,取出 JSON 内容。

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; // 接收期间避免重复发送

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();

// 流式解码保留未收齐的中文字符;读取结束时补完解码。

buffer += decoder.decode(value, { stream: !done });

// 按空行取出完整消息,把最后的半条留到下次。

const frames = buffer.split(/\r?\n\r?\n/);

buffer = frames.pop() || "";

for (const frame of frames) {

const event = parseEvent(frame);

if (!event) continue;

// 这里的 type 来自 JSON,而不是外层的 event 行。

if (event.type === "reasoning") {

thinking.textContent += event.content;

} else if (event.type === "answer") {

answer.textContent += event.content; // 逐段追加答案

} 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 {

// 完成、失败或停止后,都结束读取并恢复发送按钮。

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>

总结

实现 SSE,可以记住这条顺序:后端逐段生成内容 → 包装成 SSE 消息 → 前端读取并追加显示。 本例用 async for 读取 Agent 的片段,通过 yield 交给 StreamingResponse 持续发送,前端根据 type 分别展示思考内容和回答正文。

使用时留意消息末尾的空行,并在前端保留尚未收完整的消息。收到 done 确认完成,收到 error 显示提示,这样既能逐段展示回答,也能处理结束和失败的情况。

欢迎在评论区留言交流。如果文中有不准确或不清楚的地方,也欢迎指出,一起完善。

本文由 AI图鉴自动抓取并翻译,版权归原作者所有;如需查看原文请点下方链接。

相关 AI 工具

相关免费羊毛

同源资讯

阅读原文(juejin.cn)

标题、摘要与正文均来自该信源的公开页面,版权归原作者所有;本站自动聚合与翻译,仅作出处跳转。

← 返回 AI 资讯列表AI工具图鉴首页

返回 AI图鉴首页 · 提交工具 · 商务合作 · 关于本站 · 607 款 AI 工具分类导航 · 每日 AI 热门项目与动态