部署在Google Cloud的Flask应用无法实现API响应流式输出

我用OpenAI API开发测试网站，目标是实现类似ChatGPT的流式输出功能。本地Flask服务器运行时功能正常，但部署到Google Cloud后，响应变为一次性返回，而非流式输出。已经按要求禁用了缓冲，但问题依旧，怀疑是Google Cloud的应用配置存在问题。

以下是本地可正常运行的代码及部署信息：

main.py #

@app.route('/stream_response', methods=['POST'])
def stream_response():
    prompt_text = request.form['prompt']

    def generate():
        for chunk in gpt_model.get_response(prompt_text, stream=True):
            for choice in chunk['choices']:
                dictionary: dict = choice['delta']
                if 'content' in dictionary:
                    yield dictionary['content']
    
    response = Response(generate(), content_type='text/html')
    response.headers['X-Accel-Buffering'] = 'no'
    return response

prompt.html #

<script>
    function streamResponse() {
        var promptText = document.getElementById("prompt").value;
        var xhr = new XMLHttpRequest();
        xhr.open("POST", "/stream_response", true);
        xhr.setRequestHeader("Content-Type", "application/x-www-form-urlencoded");
        xhr.onprogress = function () {
            document.getElementById("response-container").style.display = "block";
            document.getElementById("response").innerHTML = xhr.responseText;
            console.log(xhr.responseText)
        };
        xhr.send("prompt=" + encodeURIComponent(promptText));
    }
</script>

Google Cloud app.yaml #

runtime: python310

handlers:
- url: /.*
  script: auto

部署流程 #

gcloud app deploy

解决方法 #

1. 切换到App Engine灵活环境
   标准环境会强制缓冲响应，导致流式输出失效。修改app.yaml切换到灵活环境，并配置基础资源：

runtime: python310
env: flex
resources:
  cpu: 1
  memory_gb: 0.5
  disk_size_gb: 10

2. 优化后端响应流
   使用stream_with_context包装生成器，并手动刷新缓冲区，确保每个chunk及时输出：

from flask import Response, stream_with_context
import sys

@app.route('/stream_response', methods=['POST'])
def stream_response():
    prompt_text = request.form['prompt']

    def generate():
        for chunk in gpt_model.get_response(prompt_text, stream=True):
            for choice in chunk['choices']:
                dictionary: dict = choice['delta']
                if 'content' in dictionary:
                    yield dictionary['content']
                    sys.stdout.flush()
    
    response = Response(stream_with_context(generate()), content_type='text/plain')
    response.headers['X-Accel-Buffering'] = 'no'
    response.headers['Cache-Control'] = 'no-cache'
    response.headers['Transfer-Encoding'] = 'chunked'
    return response

3. 修复前端增量更新逻辑
   当前前端每次用xhr.responseText覆盖内容，会导致看起来像一次性返回。修改为增量追加：

<script>
    function streamResponse() {
        var promptText = document.getElementById("prompt").value;
        var xhr = new XMLHttpRequest();
        xhr.open("POST", "/stream_response", true);
        xhr.setRequestHeader("Content-Type", "application/x-www-form-urlencoded");
        xhr.responseType = 'text';
        let accumulatedContent = '';
        xhr.onprogress = function () {
            document.getElementById("response-container").style.display = "block";
            const newContent = xhr.responseText.slice(accumulatedContent.length);
            accumulatedContent = xhr.responseText;
            document.getElementById("response").innerHTML += newContent;
            console.log(newContent);
        };
        xhr.send("prompt=" + encodeURIComponent(promptText));
    }
</script>

内容的提问来源于stack exchange，提问作者Milan Dierick