当前位置：首页 > article >正文

python -【流式接口返回响应处理】

article 2024/10/24 19:44:32

一. 前言

在Python中实现流式处理通常涉及到处理大量的数据，这些数据可能来自于网络、磁盘或其他数据源。流式处理的一个关键点是数据不需要一次性加载到内存中，而是逐块处理。

二、设计原则

可读性：流式接口的设计应着重提高代码的可读性。这要求开发者在命名和构造代码时，要考虑到如何使代码更加直观和易于理解。
流畅性：通过链式调用，流式接口可以实现一系列操作的流畅组合，从而提高代码的编写效率和可读性。
灵活性：流式接口允许开发者根据需要动态地添加或修改操作序列，从而提高了代码的灵活性。

三、实现方式

返回对象本身：在流式接口的方法中，通常通过返回对象本身（例如使用this关键字）来维持链式调用的连续性。
泛型与类型安全：在需要处理不同类型数据时，可以使用泛型来保持类型安全，同时提供灵活的操作序列。
上下文管理：流式接口可能需要管理复杂的上下文信息，例如数据库连接、事务状态等。在设计时，应考虑到如何有效地管理这些上下文信息，以确保代码的正确性和稳定性。

四. 代码设计

1.python后端代码：

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": "*"}})  # 允许所有源访问，但请注意生产环境的安全性


@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 = "大模型通常指的是那些规模庞大、参数数量极多的人工智能模型，它们在训练时使用了海量的数据，能够处理复杂的任务并展现出强大的语言理解和生成能力。"

    def generate():
        for i in range(len(ret_str)):
            json_data["text"] = ret_str[i]
            time.sleep(0.1)  # 暂停一秒，模拟数据产生的时间间隔
            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():
    # 假设你有一个外部 URL，该 URL 提供流式数据
    external_url = 'http://127.0.0.1:7000/stream_json_data'

    # 使用 requests 库的流式响应来接收数据
    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:
        # 创建一个生成器来流式传输响应内容
        def generate():
            for chunk in response.iter_content(chunk_size=1024):  # 你可以设置合适的 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)

                    # 创建一个 Flask 响应对象，并使用 stream_with_context 来发送流式响应

        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": "储能公司在该项目攻关中取得了哪些关键性突破？"
            }
        ]
    }
    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'
    # 请求的数据
    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()

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 => {
            // 将接收到的数据添加到页面上
            var output = document.getElementById('single-output');
            output.textContent += data; // 注意这里使用了 = 而不是 +=，因为我们只想要一次性的内容
        })
        .catch(error => {
            console.error('There has been a problem with your fetch operation:', error);
        });
</script>

</body>
</html>