网站建设 客户同程国内新闻最新消息今天简短
介绍
SSE（Server-Sent Events）是一种 服务器主动向客户端“推送”数据的技术，特别适合用来实现像 GPT 这种「流式输出」的效果。
SSE 是一种基于 HTTP 的单向通信协议：
• 客户端（浏览器）发起请求后
• 服务器可以不断「推送消息」给客户端
• 使用 MIME 类型：text/event-stream
• 天然支持断线重连（Last-Event-ID）
用途
• 🌊 流式响应（比如 GPT、AI 对话）
• 📢 实时通知 / 消息广播
• 📈 实时数据更新（股票、仪表盘）
• 🧠 AI 推理过程展示
服务端：
用两个换行分割
data: 这是第一条消息data: 这是第二条消息data: {"event": "update", "msg": "更新成功"}
实现
使用fastapi 非常容易实现
定义一个接口，在这个接口定义一个async 异步方法，这个方法使用yield 生产对应的信息。
使用StreamingResponse 来对这个方法进行生产捕获
from fastapi import FastAPI,Request
from fastapi.responses import StreamingResponse
from services.async_openai_out import async_openai_out
import asyncio
app = FastAPI()@app.post('/sse/v1')
async def root(request: Request):async def event_stream():for i in range(10):yield f"data: {i}\n\n"await asyncio.sleep(0.1)# 用了 Python 的参数顺序机制，你看到的 content 参数确实就是你传入的 event_stream()。r = StreamingResponse(event_stream(), media_type="text/event-stream")return r
这个StreamingResponse 是一个异步响应类，支持将内容「一块一块」地异步地发送给客户端，而不是像普通 Response 一次性构造整个响应体。
• 发送响应时，Starlette 会通过 __call__ 调用 stream_response 方法，将响应体通过 await send() 一块块地推送给客户端。
• 你传入的 AsyncGenerator 会成为 body_iterator。
ContentStream = typing.Union[AsyncContentStream, SyncContentStream]
class StreamingResponse(Response):body_iterator: AsyncContentStreamdef __init__(self,content: ContentStream,status_code: int = 200,headers: typing.Mapping[str, str] | None = None,media_type: str | None = None,background: BackgroundTask | None = None,) -> None:• AsyncContentStream: 异步生成器、异步迭代器（如 async def event_stream(): yield ...）• SyncContentStream: 同步的可迭代对象（如 def gen(): yield ...）
这个SSE的curl 和普通的post一样：
curl --location --request POST 'http://localhost:8000/sse/v1' \
--header 'User-Agent: Apifox/1.0.0 (https://apifox.com)' \
--header 'Accept: */*' \
--header 'Host: localhost:8000' \
--header 'Connection: keep-alive'
不过返回的响应头里面：
date Sat, 22 Mar 2025 07:29:57 GMT
server uvicorn
content-type text/event-stream; charset=utf-8
Transfer-Encoding chunked
FastAPI 会根据 返回值的类型自动推断并构造响应对象，这是因为 FastAPI 的底层用了 Pydantic + Starlette 自动序列化机制。
普通的post 响应请求：
@app.post('/sse/no_sse')
async def root(request: Request):r = {"message": "Hello World"} # 也可以直接返回return rreturn Response(content=json.dumps(r), media_type="application/json")from fastapi.responses import JSONResponse@app.post("/sse/no_sse")
async def root():return JSONResponse(content={"message": "Hello World"})
封装为openai使用的sse:
注意这个 yield f’data: {chunk}\n\n’ 少不了
@app.post("/sse/async_openai_out")
async def root(request: Request):user_message = "你好"async def gpt_stream():async for chunk in async_openai_out.gpt_stream(user_message=user_message,system_prompt="You are a helpful assistant."):print(chunk)yield f'data: {chunk}\n\n'r = StreamingResponse(gpt_stream(), media_type="text/event-stream")return r
一些思考
-
“StreamingResponse 内部是如何 send() 的？”
FastAPI 的 StreamingResponse 继承自 Starlette 的 Response，其关键机制是异步迭代器（AsyncIterable）。核心在 call() 中调用了：
async def stream_response(self, send):async for chunk in self.body_iterator:await send({"type": "http.response.body","body": chunk,"more_body": True})
-
为什么 text/event-stream 是长连接 + 推送？
HTTP 长连接：
• SSE 使用的是 HTTP/1.1 长连接，不会在发送完一条消息后断开。
• 响应头 Content-Type: text/event-stream 告诉浏览器或客户端：“我会不断推送数据”。
- 服务端推送格式：
• 每条消息格式是：data: xxxx\n\n