一文詳解python如何實現(xiàn)流式輸出
一、創(chuàng)建fastapi的項目
1.創(chuàng)建虛擬環(huán)境(power shell)
python -m venv venv
2.運行虛擬環(huán)境（command prompt）
venv\Scripts\activate
3.虛擬環(huán)境下安裝fastapi
pip install fastapi
4.虛擬環(huán)境下安裝uvicorn
pip install uvicorn
5.虛擬環(huán)境同目錄創(chuàng)建app文件夾
6.app文件夾下創(chuàng)建main.py
7.配置main.py
from fastapi import FastAPI
app=FastAPI()
if __name__ == "__main__":
import uvicorn
uvicorn.run("app.main:app", host="0.0.0.0", port=8000, reload=True)8.測試
uvicorn app.main:app --reload
9.配置swagger
mian.py 添加如下代碼
from fastapi import applications
from fastapi.openapi.docs import get_swagger_ui_html
def swagger_monkey_patch(*args, **kwargs):
return get_swagger_ui_html(
*args, **kwargs,
swagger_js_url="https://cdn.bootcdn.net/ajax/libs/swagger-ui/5.1.0/swagger-ui-bundle.min.js",
swagger_css_url="https://cdn.bootcdn.net/ajax/libs/swagger-ui/5.1.0/swagger-ui.min.css"
)
applications.get_swagger_ui_html = swagger_monkey_patch 10.測試
訪問http://127.0.0.1:8000/docs
二、一個簡易循環(huán)發(fā)送
1、main.py代碼如下
app=FastAPI()
@app.get("/",response_class=StreamingResponse)
async def start():
return await send_sse()
app.include_router(sse.router, prefix="/v1")
if __name__ == "__main__":
uvicorn.run(app, host="127.0.0.1", port=8000)2、sse.py
async def event_generator():
for i in range(5):
# 標準 SSE 格式：data: 內(nèi)容\n\n
yield f"data: {{\"message\": \"Data chunk {i}\"}}\n\n".encode()
await asyncio.sleep(1) # 異步非阻塞延遲
# 返回體頭
headers = {
'Content-Type': 'text/event-stream',
'Cache-Control': 'no-cache',
'Connection': 'keep-alive'
}
router = APIRouter()
@router.post("/completions")
async def send_sse():
return StreamingResponse(event_generator(), headers=headers)
解釋：整個代碼很容易理解，調(diào)用send_see函數(shù)，其實工作中復雜的點在于，你是每一步都在發(fā)送。比如現(xiàn)在流行的fastgpt、dify這些基于工作流構(gòu)建應用，每次返回都是節(jié)點信息，將每個節(jié)點的結(jié)果通過stream返回。
三、問題
1、如何動態(tài)實現(xiàn)發(fā)送？
2、return后我還有其他操作怎么辦？
四、解決思路
題外話：解決問題的思路，才是你的成長，一味地依靠AI，永遠在原地踏步。AI只是輔助，思路才是你的價值
問題一
因為只要return之后，只能通過event_generator函數(shù)進行操作，所以要讓event_generator這個函數(shù)的循環(huán)是可控，因為外部無法直接調(diào)用event_generator。所以可以采用一個隊列實現(xiàn)。
1、為什么是隊列而不是集合？
因為要采用隊列的先進先出的思想，保證數(shù)據(jù)的先后順序。
2、是否隊列為空，整個循環(huán)就結(jié)束呢？
不是，因為每個節(jié)點執(zhí)行會有時間差，甚至說處理的比較慢從而導致，數(shù)據(jù)還沒進隊列，整個循環(huán)就已經(jīng)結(jié)束了。
3、循環(huán)結(jié)束的節(jié)點怎么做？
根據(jù)業(yè)務來看，因為發(fā)送事件信息是有個event鑒別數(shù)據(jù)的類型，可以通過這個確定最后一個事件是什么從而結(jié)束整個循環(huán)。如果無法確定，可以設置具體的超時時間比如10s。
下面是具體的代碼實現(xiàn)，以3s過期時間為例子。
async def event_generator(messags:deque):
timeout_seconds = 3 # 從props獲取超時時間，默認3秒
last_data_time = asyncio.get_running_loop().time()
start = True
while start:
# 標準 SSE 格式：data: 內(nèi)容\n\n
if messags:
message = messags.popleft()
yield f"data: {{\"message\": \"Data chunk {message}\"}}\n\n".encode()
last_data_time = asyncio.get_running_loop().time() # 重置計時器
else:
# 檢查是否超時
current_time = asyncio.get_running_loop().time()
if current_time - last_data_time > timeout_seconds:
logging.info("消息列表已空 {} 秒，結(jié)束任務",timeout_seconds)
break
# 無數(shù)據(jù)時發(fā)送心跳，避免客戶端斷開連接
await asyncio.sleep(0.5) # 降低 CPU 使用率
# 返回體頭
headers = {
'Content-Type': 'text/event-stream',
'Cache-Control': 'no-cache',
'Connection': 'keep-alive'
}
router = APIRouter()
messags = deque()
message=0
@router.post("/completions/send")
async def send_sse():
return StreamingResponse(event_generator(messags), headers=headers)
@router.get("/completions/addDeque")
async def send_sse(): # 定義發(fā)送信息，每次對message+1操作
global message
message=message+1
messags.append(message)
return "ok"效果圖：
問題二
流式的想法，是每次調(diào)用把結(jié)果給到前端。那么問題是我們寫這塊代碼是個很長的模塊，在中間會進行流式輸入，如果不return，所有的信息全部進了隊列，最后return其實是一個一次性返回，跟流式的理念相違背。那么如何去做，這里可以采用攜程去實現(xiàn)這個功能。
我們可以把自己的代碼塊邏輯丟到攜程讓攜程去做。整個思想邏輯，是用隊列的延展性實現(xiàn)流式的輸出。所以我們只需要保證，在發(fā)送的時候把數(shù)據(jù)給到隊列就行。代碼如下：
async def event_generator(messages:deque):
timeout_seconds = 3 # 從props獲取超時時間，默認3秒
last_data_time = asyncio.get_running_loop().time()
start = True
while start:
# 標準 SSE 格式：data: 內(nèi)容\n\n
if messages:
message = messages.popleft()
yield f"data: {{\"message\": \"Data chunk {message}\"}}\n\n".encode()
last_data_time = asyncio.get_running_loop().time() # 重置計時器
else:
# 檢查是否超時
current_time = asyncio.get_running_loop().time()
if current_time - last_data_time > timeout_seconds:
logging.info("消息列表已空 {} 秒，結(jié)束任務",timeout_seconds)
break
# 無數(shù)據(jù)時發(fā)送心跳，避免客戶端斷開連接
await asyncio.sleep(0.5) # 降低 CPU 使用率
# 返回體頭
headers = {
'Content-Type': 'text/event-stream',
'Cache-Control': 'no-cache',
'Connection': 'keep-alive'
}
router = APIRouter()
async def do_something(messages):
# 操作一
message1="節(jié)點1開始"
messages.append(message1)
#模擬每次操作耗時
await asyncio.sleep(0.5)
message2="節(jié)點1answer"
messages.append(message2)
#模擬每次操作耗時
await asyncio.sleep(0.5)
message2="節(jié)點1結(jié)束"
messages.append(message2)
#模擬每次操作耗時
await asyncio.sleep(0.5)
@router.post("/completions/send")
async def send_sse():
messages = deque()
asyncio.create_task(do_something(messages))
return StreamingResponse(event_generator(messages), headers=headers)注：messages是局部變量，這樣可以保證，線程安全。
結(jié)果如下：
以上就是一文詳解python如何實現(xiàn)流式輸出的詳細內(nèi)容，更多關于python流式輸出的資料請關注腳本之家其它相關文章！
相關文章
Pandas讀取excel合并單元格的正確方式(openpyxl合并單元格拆分并填充內(nèi)容)
Excel文件中可能包含合并單元格的數(shù)據(jù),下面這篇文章主要給大家介紹了關于Pandas讀取excel合并單元格的正確方式,主要介紹的openpyxl合并單元格拆分并填充內(nèi)容,需要的朋友可以參考下2023-06-06
這篇文章主要為大家詳細介紹了python多線程掃描端口，具有一定的參考價值，感興趣的小伙伴們可以參考一下2019-09-09
今天小編就為大家分享一篇在Python中輸入一個以空格為間隔的數(shù)組方法，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2018-11-11
詳解python第三方庫的安裝、PyInstaller庫、random庫
這篇文章主要介紹了python第三方庫的安裝、PyInstaller庫、random庫,本文給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2021-03-03
主流Python語音轉(zhuǎn)文字(STT)庫實戰(zhàn)指南
語音轉(zhuǎn)文字的核心是通過信號處理、特征提取和模式識別將聲波轉(zhuǎn)換為文本,下面這篇文章主要介紹了主流Python語音轉(zhuǎn)文字(STT)庫的相關資料,文中通過代碼介紹的非常詳細,需要的朋友可以參考下2026-01-01
這篇文章主要介紹了用Python做一個久坐提醒小助手的示例代碼，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2020-02-02
這篇文章主要介紹了Linux下Python獲取IP地址的代碼,需要的朋友可以參考下2014-11-11
這篇文章主要介紹了python selenium自動化測試框架搭建的方法步驟，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2020-03-03
這篇文章主要介紹了Python Socket編程詳細介紹，socket可以建立連接，傳遞數(shù)據(jù)，具有一定的參考價值，感興趣的小伙伴們可以參考一下。2017-03-03
近期在整理相冊的時候，發(fā)現(xiàn)相冊中有許多重復圖片，人工一張張篩查刪除太枯燥，便寫下這個python腳本，用于刪除文件夾下重復的圖片2021-05-05