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: ~
(筆記尚未整理完畢)