基于Flask与LangChain实现ChatGPT流式响应的问题及解决

问题场景 #

用Flask和LangChain开发基于ChatGPT的问答应用，目标是实现实时流式输出（像ChatGPT那样逐字显示回答），但当前代码只能等待完整响应生成后才一次性渲染内容。现有核心代码如下：

chain = VectorDBQA.from_chain_type(llm=ChatOpenAI(model_name="gpt-3.5-turbo", streaming=True), chain_type="stuff", vectorstore=docsearch)
...
result = chain({"query": query})
output = result['result']

前端通过Jinja2的{{ output }}渲染内容，尝试过stream_template但未达到预期效果。

解决步骤 #

1. 后端改造：实现流式接口 #

放弃同步调用chain({"query": query})的方式，改用LangChain的流式输出能力，结合Flask的Response对象返回逐块内容：

from flask import Flask, Response, request
from langchain.chat_models import ChatOpenAI
from langchain.chains import RetrievalQA
from langchain.vectorstores import Chroma  # 示例向量库，根据实际情况替换

app = Flask(__name__)

# 假设已初始化好vectorstore（docsearch）
docsearch = Chroma(persist_directory="./db", embedding_function=embeddings)

@app.route('/stream-answer', methods=['POST'])
def stream_answer():
    query = request.json.get('query')
    # 初始化带流式配置的LLM
    llm = ChatOpenAI(model_name="gpt-3.5-turbo", streaming=True, temperature=0)
    # 改用RetrievalQA（VectorDBQA已被标记为过时）
    qa_chain = RetrievalQA.from_chain_type(
        llm=llm,
        chain_type="stuff",
        retriever=docsearch.as_retriever()
    )
    
    def generate_stream():
        # 调用链的stream方法获取流式输出
        for chunk in qa_chain.stream({"query": query}):
            yield chunk['result']
    
    # 返回流式响应，设置content-type为text/plain或text/event-stream
    return Response(generate_stream(), content_type='text/plain')

注意：VectorDBQA已被LangChain标记为过时，建议改用RetrievalQA，确保链支持stream方法实现逐块输出。

2. 前端配合：实时接收并渲染 #

使用JavaScript的fetch API配合ReadableStream接收后端的流式数据，逐块更新页面内容：

<!DOCTYPE html>
<html>
<head>
    <title>流式问答</title>
</head>
<body>
    <input type="text" id="query-input" placeholder="输入问题">
    <button onclick="sendQuery()">提问</button>
    <div id="response-container"></div>

    <script>
        async function sendQuery() {
            const query = document.getElementById('query-input').value;
            const responseContainer = document.getElementById('response-container');
            responseContainer.textContent = ''; // 清空之前的回答

            const response = await fetch('/stream-answer', {
                method: 'POST',
                headers: {
                    'Content-Type': 'application/json',
                },
                body: JSON.stringify({ query: query })
            });

            // 读取流式响应
            const reader = response.body.getReader();
            const decoder = new TextDecoder('utf-8');

            while (true) {
                const { done, value } = await reader.read();
                if (done) break;
                // 逐块追加内容到页面
                responseContainer.textContent += decoder.decode(value);
            }
        }
    </script>
</body>
</html>

3. 正确使用stream_template的方法 #

如果想继续使用Flask的stream_template，需要确保后端传递的是生成器对象，模板中逐块渲染：

@app.route('/qa-page')
def qa_page():
    query = request.args.get('query')
    def content_generator():
        for chunk in qa_chain.stream({"query": query}):
            yield chunk['result']
    return stream_template('qa.html', content=content_generator())

对应的Jinja2模板qa.html：

<div id="response">
    {% for chunk in content %}
        {{ chunk }}
    {% endfor %}
</div>

内容的提问来源于stack exchange，提问作者Dante Noguez