构建一个能基于自有文档回答问题的问答系统，是当前大模型落地企业场景中最常见的需求之一。直接把文档塞给大模型存在上下文长度限制和成本问题，而RAG（Retrieval-Augmented Generation，检索增强生成）通过先检索后生成的方式，把最相关的文档片段找出来交给模型组织答案，既解决了知识时效性问题，也显著减少了幻觉。这篇文章将从架构设计讲起，一步步实现文档解析、向量化存储、检索召回、大模型回答和前端流式交互的完整链路。
一、RAG系统的整体架构与文档处理流程
一个典型的RAG知识库问答系统由离线和在线两条链路组成。离线链路负责知识库的构建：把各类文档（PDF、Word、Markdown、网页等）读取出来，清洗后切成合适大小的文本块，调用Embedding模型将文本块转为向量，最后写入向量数据库。在线链路负责问答：用户提问后，先把问题同样转为向量，在向量库中做相似度检索，取回最相关的若干文本块，拼接成上下文交给大模型，由模型生成最终答案。
文档处理环节最容易踩坑的是分块策略。块切得太大，单块内容超出Embedding模型的有效长度，语义被稀释，检索精度下降；块切得太小，上下文信息不完整，召回的内容碎片化，模型难以据此给出准确回答。实践中常用的大小是300到800个字符，并且保留10%到20%的重叠区域，避免关键句子被硬生生切断在两个块的边界上。
下面用Python演示一段完整的文档加载与分块代码，使用langchain的工具库处理本地Markdown和文本文件：
from langchain_community.document_loaders import DirectoryLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
# 加载目录下所有txt文档
loader = DirectoryLoader("./docs", glob="**/*.txt", loader_cls=TextLoader)
docs = loader.load()
# 递归分块：优先按段落切，再按句子，最后按字符
splitter = RecursiveCharacterTextSplitter(
chunk_size=500, # 每块最大500字符
chunk_overlap=80, # 相邻块重叠80字符，防止语义被切断
separators=["\n\n", "\n", "。", "，", " "]
)
chunks = splitter.split_documents(docs)
print(f"共生成 {len(chunks)} 个文本块")
RecursiveCharacterTextSplitter的优势在于它会按照分隔符的优先级递归尝试，先尝试按空段落切分，切不下来再按单行、按中文句号切，尽量让每个块保持完整的语义单元。对于PDF这类排版复杂的文档，建议先用PyMuPDF或pdfplumber提取文字，经过清洗（去掉页眉页脚、乱码字符）后再进入分块流程。
二、Embedding模型选型与向量库的写入检索
Embedding模型决定了检索质量的上限。选型时需要关注三个指标：检索精度（MTEB榜单排名）、支持的语种（中文场景必须选择中文能力强的模型）、以及向量维度（维度越高存储和计算成本越大）。常用的选择包括开源的BGE系列（bge-large-zh-v1.5，中文效果好、可本地部署）、text2vec系列，以及各云厂商提供的Embedding API。如果数据涉及敏感信息不能出网，本地部署BGE是最稳妥的方案。
向量数据库方面，中小规模知识库（百万级以下向量）用FAISS或Chroma完全够用，部署简单且免费；数据量大、需要多用户并发和权限管理时，再考虑Milvus或Qdrant。下面演示用Chroma构建向量库并写入数据：
from langchain_community.embeddings import HuggingFaceBgeEmbeddings
from langchain_community.vectorstores import Chroma
# 本地加载BGE中文Embedding模型
embedding = HuggingFaceBgeEmbeddings(
model_name="BAAI/bge-large-zh-v1.5",
model_kwargs={"device": "cpu"},
encode_kwargs={"normalize_embeddings": True}
)
# 将分块结果写入Chroma并持久化到磁盘
vectordb = Chroma.from_documents(
documents=chunks,
embedding=embedding,
persist_directory="./chroma_db"
)
# 相似度检索：取回最相关的4个文本块
results = vectordb.similarity_search_with_score("如何申请年假？", k=4)
for doc, score in results:
print(round(score, 3), doc.page_content[:60])
这里有个细节值得注意：normalize_embeddings参数设为True后向量会做归一化，此时用余弦相似度和内积计算结果一致，能获得更好的检索性能。检索时除了普通向量检索，还建议启用MMR（最大边际相关性）算法，它会在保证相关性的同时尽量让返回的文本块内容多样化，避免召回四段几乎重复的话。
实际项目中还应该做一层检索后的过滤：对相似度分数设置阈值，把分数过低（说明和问题关系不大）的块直接丢弃，否则这些噪声内容进入提示词后反而会干扰模型判断。进一步可以用重排序模型（如bge-reranker）对初筛的20个候选块做精排，取前5个进入生成环节，准确率通常还能提升一到两成。
三、提示词拼接、大模型调用与流式输出
拿到检索结果后，需要把它们和用户问题组装成结构化的提示词。提示词设计的原则是：明确告诉模型只能依据提供的参考资料回答，资料中没有的内容要直接说不知道，并要求模型标注答案来自哪段资料，方便用户核实。一个经过验证的提示词模板如下：
PROMPT_TEMPLATE = """你是一个企业知识库助手，请严格根据下面的参考资料回答问题。
参考资料：
{context}
用户问题：{question}
要求：
1. 只使用参考资料中的信息作答，不得编造
2. 如果资料不足以回答，请回答：知识库中暂无相关信息
3. 回答末尾注明依据的资料编号
"""
def build_prompt(question, docs):
context = "\n\n".join(
f"【资料{i+1}】{d.page_content}" for i, d in enumerate(docs)
)
return PROMPT_TEMPLATE.format(context=context, question=question)
调用大模型时强烈建议开启流式输出。问答场景下用户对等待非常敏感，一次性等十几秒才看到完整答案体验很差，而流式模式下模型每生成几个字就推送到前端，用户第一时间就能看到内容在逐步展开，感知等待时间大幅缩短。以OpenAI兼容接口为例：
from openai import OpenAI
client = OpenAI(base_url="https://api.ippipp.com/v1", api_key="your_key")
def stream_answer(prompt):
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
stream=True # 开启流式返回
)
for chunk in response:
delta = chunk.choices[0].delta.content
if delta:
yield delta # 逐段产出文本
四、后端接口与前端交互的完整实现
后端用FastAPI搭建问答接口，通过SSE（Server-Sent Events）把模型的流式输出实时推送给浏览器。相比WebSocket，SSE是单向推送，实现更简单，对于问答这种只推送不双向通信的场景完全够用：
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
app = FastAPI()
class AskRequest(BaseModel):
question: str
@app.post("/ask")
def ask(req: AskRequest):
# 1. 检索相关知识块
docs = vectordb.similarity_search(req.question, k=4)
# 2. 组装提示词
prompt = build_prompt(req.question, docs)
# 3. 流式生成并返回SSE响应
def event_stream():
yield "data: 检索到 %d 条相关资料，开始生成回答...\n\n" % len(docs)
for text in stream_answer(prompt):
yield f"data: {text}\n\n"
yield "data: [DONE]\n\n"
return StreamingResponse(event_stream(), media_type="text/event-stream")
前端页面只需要一个输入框、一个发送按钮和一个答案展示区域，用浏览器原生的EventSource或fetch读取SSE流，把收到的文字片段逐步追加到页面。注意展示区域要开启自动滚动，让最新内容始终可见。核心的前端逻辑如下：
<div class="chat-box">
<div id="answer"></div>
<input id="question" placeholder="请输入你的问题" />
<button onclick="ask()">发送</button>
</div>
<script>
async function ask() {
const q = document.getElementById("question").value;
const box = document.getElementById("answer");
box.innerText = "";
const resp = await fetch("/ask", {
method: "POST",
headers: {"Content-Type": "application/json"},
body: JSON.stringify({question: q})
});
const reader = resp.body.getReader();
const decoder = new TextDecoder();
let buffer = "";
while (true) {
const {done, value} = await reader.read();
if (done) break;
buffer += decoder.decode(value, {stream: true});
// 按SSE格式解析每条data事件
const parts = buffer.split("\n\n");
buffer = parts.pop();
for (const p of parts) {
const data = p.replace(/^data: /, "");
if (data !== "[DONE]") box.innerText += data;
}
box.scrollTop = box.scrollHeight; // 自动滚到底部
}
}
</script>至此整套系统就跑通了。上线前还有几个优化方向值得投入：给文档块补充元数据（来源文件、章节标题、更新时间），检索后可以按来源过滤和展示引用出处；引入对话历史管理，让系统能处理追问式的多轮对话；定期对知识库做增量更新，只对新增或修改的文档重新向量化。这套架构在十万级文档规模内都能稳定运行，先跑通最小闭环，再按实际检索效果逐步迭代优化，是落地RAG系统最务实的路径。