本文详解如何基于 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 规范，同时赋予前端完全控制权——可实时渲染答案流、延迟展示来源、甚至支持「点击跳转原文」等高级交互。
