Skip to content

[Python] FastAPI 使用 Server-Sent Events (SSE) 進行串流回覆

Last Updated on 2024-10-31 by Clay

最近建立了許多 Chatbot 的後台 API Server,一開始我是接收到使用者的訊息後回傳,將 LLM 的生成回覆一口氣顯示在前端界面,但這樣使用者體驗並不好;之後改成了 HTTP 串流,每生成一個 Token 就回傳前端界面,但後來發現在部份使用者的裝置上會發生黏包,所以最後改成了使用 WebSocket。

不過最近做前端的同事跟我討論到,或許使用 Server-Sent Event(SSE)是一種更好的選擇,所以我開始摸索使用 FastAPI 來建立 SSE API。

簡單看了一下,SSE 可以同樣是基於 HTTP,可以視為是 WebSocket 的輕量級替代方案。

import asyncio

from fastapi import FastAPI
from sse_starlette.sse import EventSourceResponse

app = FastAPI()

async def event_generator(sent: str):
    for char in sent:
        yield {"event": "message", "data": char}
        await asyncio.sleep(0.2)

@app.get("/api/chatbot/stream")
async def sse_endpoint(sent: str):
    return EventSourceResponse(event_generator(sent))


這是一個經典的重複句子回應機器人,以此先來測試取代 LLM 的效果。可以使用以下指令啟動:

uvicorn app:app --port 8080 --reload



接著一旦我們使用了以下請求:

import httpx

url = "http://127.0.0.1:8080/api/chatbot/stream"
params = {"sent": "你好,今天天氣不錯~"}

with httpx.stream("GET", url, params=params) as response:
    for line in response.iter_text():
        if line:
            print(line)


Output:

event: message
data: 你

event: message
data: 好

event: message
data: ,

event: message
data: 今

event: message
data: 天

event: message
data: 天

event: message
data: 氣

event: message
data: 不

event: message
data: 錯

event: message
data: ~

(筆記尚未整理完畢)


References


Read More

Tags:

Leave a Reply