Flask中使用LangChain的astream_events流式传输数据至前端失败

核心问题原因 #

• Flask流式响应要求返回可迭代对象（比如生成器），如果代码里直接返回了函数本身，或者对astream_events的返回值处理不当，就会触发TypeError: 'function' object is not iterable错误。
• Colab是交互式环境，对生成器的处理逻辑和Flask的WSGI运行环境完全不同，所以能在Colab正常跑的代码放到Flask里会出问题。
• 调试中间件提示的“响应头已发送时捕获到异常”，是因为流式响应开始发送数据后，生成器内部抛出了异常，但此时HTTP响应头已经返回，无法再修改响应状态或头信息，导致中间件捕获到冲突。

修复步骤 #

1. 正确返回流式响应对象 #

Flask需要你用Response对象包装生成器，并设置合适的响应头（比如用text/event-stream做SSE，或者application/json做JSON分块传输）。

修正后的示例代码：

from flask import Flask, Response
from langchain.chat_models import ChatOpenAI
from langchain.schema import HumanMessage

app = Flask(__name__)

def generate_stream():
    llm = ChatOpenAI(streaming=True)
    messages = [HumanMessage(content="给我讲个100字左右的小故事")]
    # 遍历astream_events的结果，逐个输出内容
    for event in llm.astream_events(messages, version="v1"):
        if event["event"] == "on_chat_model_stream":
            chunk_content = event["data"]["chunk"].content
            # 用SSE格式返回，前端可直接用EventSource接收
            yield f"data: {chunk_content}\n\n"

@app.route('/stream-story')
def stream_story():
    # 用Response包装生成器，指定正确的MIME类型
    return Response(generate_stream(), mimetype='text/event-stream')

if __name__ == '__main__':
    app.run()

2. 避免返回函数本身 #

检查你的代码，别犯这种低级错误：把生成器函数直接返回（没加括号调用），这会导致Flask拿到的是函数对象而非可迭代的生成器。

错误示例：

@app.route('/stream')
def stream():
    return generate_stream  # 错：返回了函数，不是生成器实例

正确示例：

@app.route('/stream')
def stream():
    return Response(generate_stream(), mimetype='text/event-stream')  # 对：返回调用后的生成器

3. 捕获生成器内部异常 #

为了避免响应头发送后抛出异常触发中间件报错，在生成器内部提前捕获异常，优雅收尾：

def generate_stream():
    try:
        llm = ChatOpenAI(streaming=True)
        messages = [HumanMessage(content="给我讲个100字左右的小故事")]
        for event in llm.astream_events(messages, version="v1"):
            if event["event"] == "on_chat_model_stream":
                chunk_content = event["data"]["chunk"].content
                yield f"data: {chunk_content}\n\n"
    except Exception as e:
        # 异常发生时返回错误信息，通知前端
        yield f"data: 出错了：{str(e)}\n\n"
        # 可选：发送关闭信号，让前端断开连接
        yield "event: close\n\n"

4. 调整Flask调试配置（可选） #

如果调试中间件的报错干扰排查，可以暂时关闭调试模式，或者修改配置让异常正常传播：

app.config['PROPAGATE_EXCEPTIONS'] = True

关键注意点 #

• 确保使用的WSGI服务器（比如Werkzeug、Gunicorn）支持流式传输，部分轻量服务器可能不兼容。
• 确认LangChain版本和astream_events的version参数匹配，不同版本的参数要求可能有差异。
• 前端接收时要对应后端的格式：用SSE就用EventSource，用JSON分块就用fetch处理ReadableStream。

内容的提问来源于stack exchange，提问作者Abdul Haadi