如何在 FastAPI 中实现 RAG 流式响应并同步返回结构化源文档-Python教程-PHP中文网

本文详解如何基于 LangChain + FastAPI 构建支持答案流式输出与源文档结构化返回的 RAG 服务，避免重复检索、确保低延迟，并通过 SSE（Server-Sent Events）统一传输生成内容与元数据。

本文详解如何基于 langchain + fastapi 构建支持**答案流式输出**与**源文档结构化返回**的 rag 服务，避免重复检索、确保低延迟，并通过 sse（server-sent events）统一传输生成内容与元数据。

在构建生产级 RAG 应用时，仅流式返回模型生成文本是不够的——用户还需可信溯源：即明确知道答案依据哪些原始文档生成。但若将源文档混入流式响应（如逐条 yield），前端难以区分“答案流”与“元数据”，且破坏用户体验；若另起接口单独查询源文档，则触发二次检索，违背 RAG 的效率原则。

理想方案是：一次检索、一次调用、双路输出——前段持续接收 data: chunk（答案流），末尾以独立事件发送 data: {"source_documents": [...]}（结构化元数据）。LangChain 的 RunnableWithMessageHistory 或自定义 Runnable 配合 StreamingResponse 完全支持该模式，关键在于解耦流式生成与后处理逻辑，而非在 for resp in llm.stream(...) 中穿插 yield 元数据。

以下为优化后的完整实现（兼容 LangChain v0.1+ 和 FastAPI v0.110+）：

from fastapi import FastAPI, Query, HTTPException
from fastapi.responses import StreamingResponse
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_community.vectorstores import FAISS
from langchain_community.embeddings import HuggingFaceEmbeddings
import json
import time

# 初始化组件（保持原逻辑）
embeddings = HuggingFaceEmbeddings(
    model_name="intfloat/multilingual-e5-large-instruct", 
    model_kwargs={'device': "mps"}
)
db = FAISS.load_local(
    "streamlit_vectorstores/vectorstores/db_maxiw_testfreitag", 
    embeddings, 
    allow_dangerous_deserialization=True
)
retriever = db.as_retriever(
    search_kwargs={
        'k': cfg.STREAMLIT_VECTOR_COUNT, 
        'score_threshold': cfg.SCORE_THRESHOLD,
        'sorted': True
    }, 
    search_type="similarity_score_threshold"
)
llm = build_llm(cfg.MIXTRAL_PATH)  # 确保 streaming=True

# 构建可流式执行的 RAG Chain（推荐使用 LCEL）
prompt = ChatPromptTemplate.from_messages([
    ("system", "Du bist RagBot, ein hilfsbereiter Assistent. Antworte nur auf Deutsch."),
    ("human", "{context}\n\n{question}")
])

# 关键：链式组合，显式保留 context 供后续提取
rag_chain = (
    {
        "context": retriever | (lambda docs: "\n\n".join([d.page_content for d in docs])),
        "question": RunnablePassthrough()
    }
    | prompt
    | llm
    | StrOutputParser()
)

# 自定义流式生成器：分离 answer 流与 source_docs 结束事件
def rag_stream_generator(question: str):
    start_time = time.time()

    # 一次性检索（避免重复）
    retrieved_docs = retriever.get_relevant_documents(question)

    # 构建上下文字符串（用于 prompt）
    context_str = "\n\n".join([doc.page_content for doc in retrieved_docs])

    # 流式生成答案（不阻塞）
    try:
        stream = rag_chain.stream({"context": context_str, "question": question})
        first_token = True
        for chunk in stream:
            if first_token:
                yield f"data: {{\"event\": \"response_start\", \"elapsed_ms\": {int((time.time() - start_time) * 1000)}}}\n\n"
                first_token = False
            yield f"data: {json.dumps({'chunk': chunk}, ensure_ascii=False)}\n\n"
    except Exception as e:
        yield f"data: {{\"error\": \"Generation failed: {str(e)}\"}}\n\n"
        return

    # 流式结束后，发送 source_documents（单次 JSON 事件）
    source_metadata = [
        {
            "source": doc.metadata.get("source", "").split("/")[-1],
            "page": doc.metadata.get("page", 0) + 1,
            "score": getattr(doc, 'metadata', {}).get('score', None)
        }
        for doc in retrieved_docs
    ]

    yield f"data: {json.dumps({'source_documents': source_metadata}, ensure_ascii=False)}\n\n"

# FastAPI 路由（SSE 媒体类型）
app = FastAPI(
    title="RAG Streaming API",
    description="RAG service with streamed answers and structured source documents",
    version="1.0"
)

@app.get("/rag_model_response")
async def rag_endpoint(question: str = Query(..., min_length=1)):
    if not question.strip():
        raise HTTPException(400, "Question cannot be empty")
    return StreamingResponse(
        rag_stream_generator(question),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "Connection": "keep-alive"}
    )

✅ 前端消费示例（JavaScript）

const eventSource = new EventSource(`/rag_model_response?question=${encodeURIComponent(question)}`);
let answer = "";

eventSource.onmessage = (e) => {
    try {
        const data = JSON.parse(e.data);
        if ("chunk" in data) {
            answer += data.chunk;
            document.getElementById("answer").textContent = answer;
        } else if ("source_documents" in data) {
            renderSources(data.source_documents); // 自定义渲染函数
            eventSource.close();
        }
    } catch (err) {
        console.warn("Invalid SSE data:", e.data);
    }
};

eventSource.addEventListener("response_start", (e) => {
    console.log(`First token in ${JSON.parse(e.data).elapsed_ms}ms`);
});

⚠️ 注意事项

禁止在流中混杂非 JSON 数据：原代码中 yield "(Response Time: ...)" 会破坏 SSE 解析，必须统一为 data: {...} 格式；
源文档需预处理：retriever.get_relevant_documents() 必须在流开始前调用一次，其结果复用于 prompt 构建与最终元数据封装；
错误隔离：流式过程中异常应捕获并发送 error 事件，避免连接中断；
客户端健壮性：前端需忽略非 data: 行（如注释、空行），并校验 JSON 结构。

此方案彻底规避了双检索开销，严格遵循 SSE 规范，同时赋予前端完全控制权——可实时渲染答案流、延迟展示来源、甚至支持「点击跳转原文」等高级交互。