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: ～
（筆記尚未整理完畢）