Flask服务中parse_json_stream可打印但无法yield流式LLM响应问题

问题概述 #

后台为搭载RAG pipeline的Flask应用，尝试流式获取LLM响应时，parse_json_stream函数内的print(result["text"])能正常输出内容，但通过yield生成的内容无法传递到客户端，客户端仅收到200状态码，无实际响应内容。

代码问题分析 #

1. 客户端未读取流式响应内容 #

当前客户端代码仅打印响应对象，未迭代读取流式返回的内容，导致即使服务端已yield内容，客户端也无法获取。

2. 请求参数名不匹配 #

服务端从request.form中获取question参数，但客户端请求时传递的是prompt参数，导致服务端拿到的问题为空，可能影响LLM返回结果（若用户测试时print正常，可能为代码笔误）。

3. 可能存在SSE格式前缀未处理 #

若LLM返回的是SSE（Server-Sent Events）格式响应，每行内容会带有data: 前缀，直接解析会导致JSON解码异常（但用户print正常，可能此情况不存在，但需排查）。

4. 生成器判断冗余 #

generate函数中if parsed_text:判断无意义，因为生成器对象永远为真值，不会为None，该判断不影响功能但属于冗余代码。

修复方案 #

方案1：修正客户端代码，读取流式内容 #

修改客户端请求代码，迭代读取响应的流式内容：

import requests

url = "http://ec2-11-11-1-1.eu-central-1.compute.amazonaws.com:5000/upload"  # 修正请求路径为/upload

files = {
    'file': ('file.pdf', open('file.pdf', 'rb'))
}
data = {
    'question': 'what is this file about?',  # 修正参数名为服务端期望的question
    'stream': 'True', 
    'min_tokens': 256,
    'max_tokens': 1024
}

response = requests.post(url, files=files, data=data, stream=True)
if response.status_code == 200:
    # 迭代读取流式内容
    for chunk in response.iter_content(chunk_size=None, decode_unicode=True):
        if chunk:
            print(chunk, end='')

方案2：修正服务端参数获取逻辑 #

若需保留客户端prompt参数名，修改服务端代码中的参数获取：

question = request.form.get('prompt', '')

方案3：处理SSE格式前缀（若适用） #

如果LLM返回的是SSE格式响应，修改parse_json_stream函数处理前缀：

def parse_json_stream(line):
    decoded_line = line.decode('utf-8')
    # 移除SSE的data前缀
    if decoded_line.startswith('data: '):
        decoded_line = decoded_line[6:]
    # 处理结束标记
    if decoded_line.strip() == '[DONE]':
        return
    decoder = json.JSONDecoder()
    pos = 0
    while pos < len(decoded_line):
        try:
            result, json_end = decoder.raw_decode(decoded_line[pos:])
            if "text" in result:
                print(result["text"])
                yield result["text"]
            pos += json_end
        except json.JSONDecodeError:
            pos += 1

方案4：优化Flask响应配置 #

确保流式响应正确传递，可添加direct_passthrough=True：

return Response(stream_with_context(generate()), content_type='text/plain', direct_passthrough=True)

方案5：移除冗余判断 #

删除generate函数中多余的if parsed_text:判断：

def generate():
    try:
        prompt_len, response = process_and_respond(file_path, question)
        print("********* Generate Funct **********")
        for line in response.iter_lines():
            if line:
                parsed_text = parse_json_stream(line)
                for text in parsed_text:
                    yield text
    finally:
        print(f"Deleting File {file_path}")
        os.remove(file_path)

验证步骤 #

1. 启动修改后的Flask服务
2. 运行修正后的客户端代码
3. 观察客户端是否能实时打印LLM的流式响应内容

内容的提问来源于stack exchange，提问作者Cihan Yalçın