快捷導(dǎo)航

python 實現(xiàn)流式接口返回響應(yīng)處理示例

更新時間：2025年12月05日 10:49:55 作者：宥鈞

流式接口設(shè)計是一種強大的編程范式,它能夠提高代碼的可讀性、靈活性和可維護性,特別是在現(xiàn)在大模型的應(yīng)用中很常見,下面就來詳解一下python 實現(xiàn)流式接口返回響應(yīng)處理示例,感興趣的可以了解一下

一. 前言

在Python中實現(xiàn)流式處理通常涉及到處理大量的數(shù)據(jù)，這些數(shù)據(jù)可能來自于網(wǎng)絡(luò)、磁盤或其他數(shù)據(jù)源。流式處理的一個關(guān)鍵點是數(shù)據(jù)不需要一次性加載到內(nèi)存中，而是逐塊處理。

二、設(shè)計原則

可讀性：流式接口的設(shè)計應(yīng)著重提高代碼的可讀性。這要求開發(fā)者在命名和構(gòu)造代碼時，要考慮到如何使代碼更加直觀和易于理解。
流暢性：通過鏈?zhǔn)秸{(diào)用，流式接口可以實現(xiàn)一系列操作的流暢組合，從而提高代碼的編寫效率和可讀性。
靈活性：流式接口允許開發(fā)者根據(jù)需要動態(tài)地添加或修改操作序列，從而提高了代碼的靈活性。

三、實現(xiàn)方式

返回對象本身：在流式接口的方法中，通常通過返回對象本身（例如使用this關(guān)鍵字）來維持鏈?zhǔn)秸{(diào)用的連續(xù)性。
泛型與類型安全：在需要處理不同類型數(shù)據(jù)時，可以使用泛型來保持類型安全，同時提供靈活的操作序列。
上下文管理：流式接口可能需要管理復(fù)雜的上下文信息，例如數(shù)據(jù)庫連接、事務(wù)狀態(tài)等。在設(shè)計時，應(yīng)考慮到如何有效地管理這些上下文信息，以確保代碼的正確性和穩(wěn)定性。

四. 代碼設(shè)計

1.python后端代碼：

1.1 Flask服務(wù)

import json

import requests
from flask import Flask, Response, stream_with_context, request
import time

from flask_cors import CORS
from gevent.pywsgi import WSGIServer

app = Flask(__name__)

CORS(app, resources={r"/*": {"origins": "*"}})  # 允許所有源訪問，但請注意生產(chǎn)環(huán)境的安全性


@app.route('/ret_json_data', methods=['GET'])
def ret_json_data():
    print('start request!')
    return {'hello': 'hello flask'}


@app.route('/stream_json_data',  methods=["GET", "POST"])
def stream_json_data():
    json_data = {
        'state': 0,
        'data': [],
        'text': '',
        'message': ''
    }
    ret_str = "大模型通常指的是那些規(guī)模龐大、參數(shù)數(shù)量極多的人工智能模型，它們在訓(xùn)練時使用了海量的數(shù)據(jù)，能夠處理復(fù)雜的任務(wù)并展現(xiàn)出強大的語言理解和生成能力。"

    def generate():
        for i in range(len(ret_str)):
            json_data["text"] = ret_str[i]
            time.sleep(0.1)  # 暫停一秒，模擬數(shù)據(jù)產(chǎn)生的時間間隔
            print(f"Data: {json_data}\n")
            yield json.dumps(json_data, ensure_ascii=False)

    print('stream-data')
    return Response(stream_with_context(generate()), content_type='application/json; charset=utf-8')
    # return stream_data(generate())


@app.route('/stream_data_to_frontend')
def stream_data_to_frontend():
    # 假設(shè)你有一個外部 URL，該 URL 提供流式數(shù)據(jù)
    external_url = 'http://127.0.0.1:7000/stream_json_data'

    # 使用 requests 庫的流式響應(yīng)來接收數(shù)據(jù)
    headers = {"Content-Type": "application/json"}
    response = requests.get(external_url, stream=True, headers=headers)
    # with requests.get(external_url, stream=True, headers=headers) as r:
    #     # print('r========', r.json())
    #     # 檢查外部請求是否成功
    if response.status_code == 200:
        # 創(chuàng)建一個生成器來流式傳輸響應(yīng)內(nèi)容
        def generate():
            for chunk in response.iter_content(chunk_size=1024):  # 你可以設(shè)置合適的 chunk_size
                if chunk:
                    print('chunk: ', chunk)
                    # yield chunk
                    # yield chunk.decode("utf-8", "ignore")
                    chunk = json.loads(chunk)
                    chunk['tt'] = 'ttttt'
                    yield json.dumps(chunk, ensure_ascii=False)

                    # 創(chuàng)建一個 Flask 響應(yīng)對象，并使用 stream_with_context 來發(fā)送流式響應(yīng)

        return Response(stream_with_context(generate()), content_type='application/json; charset=utf-8')
    else:
        # 如果外部請求失敗，返回錯誤信息
        return 'Failed to retrieve data from external source', 500


@app.route('/stream_data_to_frontend_test')
def stream_data_to_frontend_test():
    url = "http://ip:port/llm"
    params = {
        "messages": [
            {
                "role": "user",
                "content": "儲能公司在該項目攻關(guān)中取得了哪些關(guān)鍵性突破？"
            }
        ]
    }
    headers = {"Content-Type": "application/json"}
    response = requests.post(url=url, json=params, stream=True, headers=headers)

    def generate():
        for chunk in response.iter_content(1024):
            print(chunk.decode("utf-8", "ignore"))
            if chunk:
                yield chunk.decode("utf-8", "ignore")

    return Response(stream_with_context(generate()), content_type='application/json; charset=utf-8')


@app.route('/stream_faq_data_to_frontend_test')
def stream_faq_data_to_frontend_test():
    url = 'http://ip:port/url'
    # 請求的數(shù)據(jù)
    data = {
        "query": "測試"
    }
    headers = {"Content-Type": "application/json"}
    response = requests.post(url=url, json=data, stream=True, headers=headers)

    def generate():
        for chunk in response.iter_content(1024):
            print('chunk', chunk.decode("utf-8", "ignore"))
            if chunk:
                # chunk = json.loads(chunk)
                # chunk = json.dumps(chunk)
                yield chunk
                # yield chunk.decode("utf-8", "ignore")

    return stream_data(generate())


def stream_data(generate):
    """
    :param generate: generator object
    :return:
    """
    return Response(stream_with_context(generate), content_type='application/json; charset=utf-8')


if __name__ == '__main__':
    app.run(host='0.0.0.0', port=7000)
    # server = WSGIServer(('0.0.0.0', 7000), app)
    # server.serve_forever()

1.2 FastAPI服務(wù)

基礎(chǔ)流式文本輸出

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio

app = FastAPI()

async def data_generator():
    for i in range(5):
        # 模擬耗時操作（如 AI 推理、實時數(shù)據(jù)采集）
        await asyncio.sleep(1)
        yield f"數(shù)據(jù)塊 {i}\n"

@app.get("/stream")
async def stream_data():
    return StreamingResponse(
        data_generator(),
        media_type="text/plain",  # 或 "text/event-stream"（SSE）
    )

流式傳輸大文件

@app.get("/stream-file")
def stream_large_file():
    def file_iterator():
        with open("large_file.zip", "rb") as f:
            while chunk := f.read(4096):  # 每次讀取 4KB
                yield chunk

    return StreamingResponse(
        file_iterator(),
        media_type="application/octet-stream",
        headers={"Content-Disposition": "attachment; filename=large_file.zip"}
    )

Server-Sent Events (SSE)

@app.get("/sse")
async def sse():
    async def event_generator():
        for i in range(5):
            await asyncio.sleep(1)
            # SSE 格式要求
            yield f"data: 事件 {i}\n\n"

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream"
    )

2.前端代碼

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Single Request Demo</title>
</head>
<body>

<h1>Single Request Demo</h1>
<pre id="single-output"></pre>

<script>
    fetch('http://127.0.0.1:7000/stream_json_data')
        .then(response => {
            if (!response.ok) {
                throw new Error('Network response was not ok');
            }
            return response.text(); // 或者使用 response.json() 如果后端返回的是JSON
        })
        .then(data => {
            // 將接收到的數(shù)據(jù)添加到頁面上
            var output = document.getElementById('single-output');
            output.textContent += data; // 注意這里使用了 = 而不是 +=，因為我們只想要一次性的內(nèi)容
        })
        .catch(error => {
            console.error('There has been a problem with your fetch operation:', error);
        });
</script>

</body>
</html>