LangChain+Flask流式输出异常：控制台正常网页无法显示

1. 核心问题定位 #

你遇到的网页端显示queryresultsource_documents，本质是后端直接返回了RetrievalQA的完整结果字典（包含query/result/source_documents字段），前端未正确解析就直接拼接显示了键名；同时控制台能正常输出，说明LLM的流式回调在本地输出层面是正常的，但未将流式token传递给前端响应流。

2. 后端回调与响应调整 #

需要替换仅输出控制台的回调处理器，改用支持HTTP流式响应的自定义回调，并通过StreamingResponse返回实时token。

错误示例（导致返回完整结果） #

from langchain.chains import RetrievalQA
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

# 仅输出到控制台的回调，无法传递给前端
@app.post("/query")
def qa_query(query: str):
    qa = RetrievalQA.from_chain_type(llm=..., retriever=...)
    result = qa({"query": query}, callbacks=[StreamingStdOutCallbackHandler()])
    return result  # 返回完整字典，前端解析错误会显示键名拼接

正确实现代码 #

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from langchain.chains import RetrievalQA
from langchain.callbacks.base import BaseCallbackHandler
from typing import Iterator

app = FastAPI()

class StreamingResponseCallback(BaseCallbackHandler):
    def __init__(self):
        self.token_queue = []
    
    def on_llm_new_token(self, token: str, **kwargs) -> None:
        # 收集每个生成的token
        self.token_queue.append(token)

@app.post("/stream-query")
def stream_qa(query: str):
    qa = RetrievalQA.from_chain_type(
        llm=...,  # 你的LLM实例
        retriever=...,  # 你的检索器
        chain_type="stuff",
        verbose=True
    )
    callback = StreamingResponseCallback()

    def generate_stream() -> Iterator[str]:
        # 同步执行QA，同时通过回调获取token
        qa.run(query, callbacks=[callback])
        for token in callback.token_queue:
            yield token

    # 返回流式响应，媒体类型设为纯文本
    return StreamingResponse(generate_stream(), media_type="text/plain")

3. 前端流式数据处理 #

前端需要监听ReadableStream，逐段接收并拼接token，而非直接解析完整响应体。

错误示例（直接获取完整响应） #

async function submitQuery() {
    const res = await fetch("/query", {
        method: "POST",
        headers: {"Content-Type": "application/json"},
        body: JSON.stringify({query: "你的问题"})
    });
    const data = await res.json();
    document.getElementById("result").innerText = data;
    // 会直接显示字典的字符串形式，出现键名拼接
}

正确实现代码 #

async function streamQuery() {
    const queryInput = document.getElementById("query-input").value;
    const resultDiv = document.getElementById("result");
    resultDiv.innerText = "";

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

    if (!res.ok) {
        resultDiv.innerText = "请求失败";
        return;
    }

    // 处理流式响应
    const reader = res.body.getReader();
    const decoder = new TextDecoder("utf-8");

    while (true) {
        const { done, value } = await reader.read();
        if (done) break;
        const token = decoder.decode(value, { stream: true });
        resultDiv.innerText += token; // 逐个拼接显示
    }
}

4. 回调处理器选择说明 #

• 不要用StreamingStdOutCallbackHandler：它仅实现控制台输出，无法将token传递给前端响应流。
• 自定义BaseCallbackHandler是正确方向：必须实现on_llm_new_token方法收集token，再通过迭代器输出到HTTP流中。

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