企业级 RAG 系统架构与实战教程
前言
RAG（Retrieval-Augmented Generation，检索增强生成）是当前大语言模型应用中最核心的技术范式之一。它通过将外部知识库与大模型相结合，有效解决了大模型"幻觉"、知识时效性差、无法访问私有数据等关键问题。
然而，搭建一个简单的 RAG Demo 只需要几十行代码，真正将其推向生产环境并支撑百万文档级的企业知识库，却面临大量工程挑战。本教程将系统性地讲解企业级 RAG 系统的完整架构设计与工程实践，帮助你从"能跑通"跨越到"能上线"。
第一章：从 Demo 到生产 — RAG 系统的企业级挑战
1.1 一个简单的 RAG Demo
大多数人在学习 RAG 时，会从类似下面的代码开始：
from langchain.document_loaders import TextLoader
from langchain.text_splitter import CharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS
from langchain.chains import RetrievalQA
from langchain.chat_models import ChatOpenAI
# 加载文档
loader = TextLoader("company_docs.txt")
documents = loader.load()
# 分块
splitter = CharacterTextSplitter(chunk_size=500, chunk_overlap=50)
chunks = splitter.split_documents(documents)
# 向量化并存储
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_documents(chunks, embeddings)
# 检索 + 生成
llm = ChatOpenAI(model="gpt-4")
qa_chain = RetrievalQA.from_chain_type(llm=llm, retriever=vectorstore.as_retriever())
answer = qa_chain.run("公司的年假政策是什么？")
print(answer)
这段代码能跑通，也能给出看起来不错的回答。但在企业生产环境中，它面临着以下核心问题：
1.2 企业级 RAG 的核心挑战
文档多样性挑战：企业文档格式复杂多样 — PDF（扫描版和数字版）、Word、Excel、PPT、Markdown、HTML 网页，甚至包含大量图表、表格、流程图。简单的 TextLoader 根本无法处理这些复杂格式。
质量挑战：Demo 级别的分块策略（如固定 500 字符切分）会破坏文档语义结构，导致检索到的内容碎片化，严重影响回答质量。
规模挑战：企业知识库动辄百万文档，FAISS 这种内存向量库无法支撑。需要分布式向量数据库、增量索引、增量更新机制。
延迟挑战：用户期望秒级响应，但完整的 RAG 管线包含文档解析、向量检索、重排序、LLM 推理等多个环节，每个环节都可能成为瓶颈。
安全挑战：企业数据有严格的权限控制要求，不同部门、不同角色能看到的文档不同。检索系统必须支持细粒度的访问控制。
可观测性挑战：生产系统需要完整的监控、日志、链路追踪，以及对 RAG 回答质量的持续评估。
1.3 企业级 RAG 架构总览
一个完整的企业级 RAG 系统通常包含以下核心模块：
┌─────────────────────────────────────────────────────────┐
│ 用户查询入口 │
│ ┌──────────┐ ┌──────────┐ ┌──────────────────────┐ │
│ │ 查询理解 │ │ 查询改写 │ │ 多轮对话管理 │ │
│ └──────────┘ └──────────┘ └──────────────────────┘ │
├─────────────────────────────────────────────────────────┤
│ 检索层 │
│ ┌──────────┐ ┌──────────┐ ┌──────────────────────┐ │
│ │ 向量检索 │ │ BM25检索 │ │ 知识图谱检索 │ │
│ └──────────┘ └──────────┘ └──────────────────────┘ │
├─────────────────────────────────────────────────────────┤
│ 后处理层 │
│ ┌──────────┐ ┌──────────┐ ┌──────────────────────┐ │
│ │ 重排序 │ │ 上下文压缩│ │ 去重与融合 │ │
│ └──────────┘ └──────────┘ └──────────────────────┘ │
├─────────────────────────────────────────────────────────┤
│ 生成层 │
│ ┌──────────┐ ┌──────────┐ ┌──────────────────────┐ │
│ │ LLM推理 │ │ 引用溯源 │ │ 质量校验 │ │
│ └──────────┘ └──────────┘ └──────────────────────┘ │
├─────────────────────────────────────────────────────────┤
│ 数据层 │
│ ┌──────────┐ ┌──────────┐ ┌──────────────────────┐ │
│ │ 文档处理 │ │ 智能分块 │ │ Embedding索引 │ │
│ └──────────┘ └──────────┘ └──────────────────────┘ │
└─────────────────────────────────────────────────────────┘
接下来，我们将逐一深入每个模块的设计与实现。
第二章：文档处理管线 — 从原始文件到可检索内容
2.1 文档处理管线架构
企业文档处理管线是 RAG 系统的"入口"，其质量直接决定了下游检索和生成的效果。一个典型的文档处理管线包括以下阶段：
原始文档 → 格式识别 → 内容提取 → 结构化解析 → 元数据抽取 → 清洗标准化
2.2 PDF 文档处理
PDF 是企业中最常见的文档格式，也是最难处理的格式之一。PDF 主要分为两种类型：
数字型 PDF：文本可以直接提取，使用 PyMuPDF 或 pdfplumber 等工具。
扫描型 PDF：本质上是图片，需要 OCR 识别文字。
import fitz # PyMuPDF
import pdfplumber
class PDFProcessor:
"""企业级 PDF 处理器"""
def __init__(self, ocr_engine="paddleocr"):
self.ocr_engine = ocr_engine
def process(self, pdf_path: str) -> list[dict]:
"""处理 PDF 文件，返回结构化内容块"""
results = []
# 尝试数字型提取
digital_text = self._extract_digital(pdf_path)
if len(digital_text.strip()) < 100:
# 文本过少，可能是扫描型 PDF，启用 OCR
ocr_text = self._extract_ocr(pdf_path)
results.append({
"type": "ocr_text",
"content": ocr_text,
"source": pdf_path
})
else:
# 数字型 PDF，保留结构
results = self._extract_structured(pdf_path)
return results
def _extract_digital(self, pdf_path: str) -> str:
"""数字型 PDF 文本提取"""
doc = fitz.open(pdf_path)
text_parts = []
for page in doc:
text_parts.append(page.get_text())
return "\n".join(text_parts)
def _extract_structured(self, pdf_path: str) -> list[dict]:
"""结构化提取，保留标题层级、表格等"""
results = []
with pdfplumber.open(pdf_path) as pdf:
for page_num, page in enumerate(pdf.pages):
# 提取文本
text = page.extract_text()
if text:
results.append({
"type": "text",
"content": text,
"page": page_num + 1,
"source": pdf_path
})
# 提取表格
tables = page.extract_tables()
for i, table in enumerate(tables):
# 表格转为 Markdown 格式
md_table = self._table_to_markdown(table)
results.append({
"type": "table",
"content": md_table,
"page": page_num + 1,
"table_index": i,
"source": pdf_path
})
return results
def _extract_ocr(self, pdf_path: str) -> str:
"""OCR 识别扫描型 PDF"""
if self.ocr_engine == "paddleocr":
from paddleocr import PaddleOCR
ocr = PaddleOCR(use_angle_cls=True, lang='ch')
doc = fitz.open(pdf_path)
all_text = []
for page in doc:
pix = page.get_pixmap(dpi=300)
img_path = f"/tmp/page_{page.number}.png"
pix.save(img_path)
result = ocr.ocr(img_path)
for line in result[0]:
all_text.append(line[1][0])
return "\n".join(all_text)
return ""
def _table_to_markdown(self, table: list) -> str:
"""将表格数据转为 Markdown 格式"""
if not table or not table[0]:
return ""
headers = [str(h) if h else "" for h in table[0]]
md_lines = ["| " + " | ".join(headers) + " |"]
md_lines.append("| " + " | ".join(["---"] * len(headers)) + " |")
for row in table[1:]:
cells = [str(c) if c else "" for c in row]
md_lines.append("| " + " | ".join(cells) + " |")
return "\n".join(md_lines)
2.3 Word 文档处理
Word 文档的处理相对简单，但需要注意保留标题层级结构：
from docx import Document
from docx.table import Table as DocxTable
class WordProcessor:
"""Word 文档处理器"""
def process(self, docx_path: str) -> list[dict]:
doc = Document(docx_path)
results = []
current_section = {"heading": "", "content": []}
for element in doc.element.body:
tag = element.tag.split('}')[-1]
if tag == 'p':
# 段落处理
para = None
for p in doc.paragraphs:
if p._element is element:
para = p
break
if para:
style_name = para.style.name if para.style else ""
if style_name.startswith("Heading"):
# 遇到标题，保存之前的 section
if current_section["content"]:
results.append({
"type": "section",
"heading": current_section["heading"],
"content": "\n".join(current_section["content"]),
"source": docx_path
})
level = style_name.replace("Heading ", "")
current_section = {
"heading": f"{'#' * int(level)} {para.text}",
"content": []
}
elif para.text.strip():
current_section["content"].append(para.text)
elif tag == 'tbl':
# 表格处理
for table in doc.tables:
md = self._table_to_markdown(table)
if md:
results.append({
"type": "table",
"content": md,
"source": docx_path
})
# 保存最后一个 section
if current_section["content"]:
results.append({
"type": "section",
"heading": current_section["heading"],
"content": "\n".join(current_section["content"]),
"source": docx_path
})
return results
def _table_to_markdown(self, table: DocxTable) -> str:
rows = []
for row in table.rows:
cells = [cell.text.strip() for cell in row.cells]
rows.append(cells)
if not rows:
return ""
headers = rows[0]
md = "| " + " | ".join(headers) + " |\n"
md += "| " + " | ".join(["---"] * len(headers)) + " |\n"
for row in rows[1:]:
md += "| " + " | ".join(row) + " |\n"
return md
2.4 Excel 表格处理
Excel 文件需要特殊处理，将表格数据转化为文本描述：
import pandas as pd
class ExcelProcessor:
"""Excel 表格处理器"""
def process(self, excel_path: str) -> list[dict]:
results = []
# 读取所有 sheet
sheets = pd.read_excel(excel_path, sheet_name=None)
for sheet_name, df in sheets.items():
# 方法1：转为 Markdown 表格（适合小表格）
if len(df) <= 50:
md_table = df.to_markdown(index=False)
results.append({
"type": "table",
"content": f"## {sheet_name}\n\n{md_table}",
"source": excel_path,
"sheet": sheet_name
})
# 方法2：逐行转为自然语言描述（适合大表格）
else:
for idx, row in df.iterrows():
desc = f"在 {sheet_name} 表中第 {idx + 1} 条记录："
for col in df.columns:
if pd.notna(row[col]):
desc += f" {col} 为 {row[col]}；"
results.append({
"type": "record",
"content": desc,
"row_index": idx,
"source": excel_path,
"sheet": sheet_name
})
return results
2.5 网页内容处理
企业内部文档往往也有网页形式（如 Confluence、Wiki）：
from bs4 import BeautifulSoup
import re
class WebPageProcessor:
"""网页内容处理器"""
def process(self, html_content: str, url: str) -> list[dict]:
soup = BeautifulSoup(html_content, 'html.parser')
# 移除无用标签
for tag in soup(['script', 'style', 'nav', 'footer', 'header']):
tag.decompose()
results = []
# 按标题层级提取内容
headings = soup.find_all(['h1', 'h2', 'h3', 'h4'])
for i, heading in enumerate(headings):
level = int(heading.name[1])
title = heading.get_text(strip=True)
# 收集到下一个同级或更高级标题之间的内容
content_parts = []
for sibling in heading.next_siblings:
if hasattr(sibling, 'name'):
if sibling.name in [f'h{l}' for l in range(1, level + 1)]:
break
text = sibling.get_text(strip=True)
if text:
content_parts.append(text)
if content_parts:
results.append({
"type": "web_section",
"heading": f"{'#' * level} {title}",
"content": "\n".join(content_parts),
"source_url": url
})
return results
第三章：智能分块策略 — 决定检索质量的关键
3.1 分块的核心原则
分块（Chunking）是 RAG 系统中最关键但最容易被忽视的环节。分块策略的好坏直接决定了检索的召回率和精确率。
核心原则：
- 语义完整性：每个 chunk 应该包含一个完整的语义单元
- 适当大小：太小丢失上下文，太大引入噪声
- 保留关联：相关联的内容应该在同一个 chunk 或相邻 chunk 中
3.2 递归分块策略
递归分块是目前最常用的通用分块方法：
from langchain.text_splitter import RecursiveCharacterTextSplitter
class SmartRecursiveSplitter:
"""智能递归分块器"""
def __init__(self, chunk_size=512, chunk_overlap=64):
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
# 中文文档的分隔符优先级
self.separators = [
"\n\n\n", # 大节分隔
"\n\n", # 段落分隔
"\n", # 行分隔
"。", # 句号（中文）
"；", # 分号
"，", # 逗号
" ", # 空格
"" # 字符级
]
def split(self, text: str, metadata: dict = None) -> list[dict]:
splitter = RecursiveCharacterTextSplitter(
chunk_size=self.chunk_size,
chunk_overlap=self.chunk_overlap,
separators=self.separators,
length_function=len,
)
chunks = splitter.split_text(text)
results = []
for i, chunk in enumerate(chunks):
chunk_meta = {
"content": chunk,
"chunk_index": i,
"total_chunks": len(chunks),
}
if metadata:
chunk_meta.update(metadata)
results.append(chunk_meta)
return results
3.3 语义分块策略
语义分块基于文本的语义相似度来确定分块边界，比固定规则更智能：
import numpy as np
from sentence_transformers import SentenceTransformer
import re
class SemanticChunker:
"""基于语义相似度的分块器"""
def __init__(self, model_name="BAAI/bge-large-zh-v1.5", threshold=0.5):
self.model = SentenceTransformer(model_name)
self.threshold = threshold
def split(self, text: str) -> list[str]:
# 第一步：按句子切分
sentences = self._split_sentences(text)
if len(sentences) <= 1:
return [text]
# 第二步：计算句子向量
embeddings = self.model.encode(sentences)
# 第三步：计算相邻句子的相似度
similarities = []
for i in range(len(embeddings) - 1):
sim = np.dot(embeddings[i], embeddings[i + 1]) / (
np.linalg.norm(embeddings[i]) * np.linalg.norm(embeddings[i + 1])
)
similarities.append(sim)
# 第四步：在相似度低于阈值处分割
chunks = []
current_chunk = [sentences[0]]
for i, sim in enumerate(similarities):
if sim < self.threshold:
# 语义断裂，开始新 chunk
chunks.append("".join(current_chunk))
current_chunk = [sentences[i + 1]]
else:
current_chunk.append(sentences[i + 1])
if current_chunk:
chunks.append("".join(current_chunk))
return chunks
def _split_sentences(self, text: str) -> list[str]:
"""中英文混合句子切分"""
# 中文句号、问号、叹号、分号 + 英文对应标点
pattern = r'([。！？；\.\!\?\;])'
parts = re.split(pattern, text)
sentences = []
current = ""
for part in parts:
current += part
if re.match(pattern, part):
sentences.append(current.strip())
current = ""
if current.strip():
sentences.append(current.strip())
return [s for s in sentences if s]
3.4 文档层级分块
对于有明确层级结构的文档（如技术手册、法规文档），保留层级关系非常重要：
class HierarchicalChunker:
"""文档层级分块器，保留标题层级上下文"""
def __init__(self, max_chunk_size=1024):
self.max_chunk_size = max_chunk_size
def split(self, structured_docs: list[dict]) -> list[dict]:
"""
输入：带标题层级的文档结构
输出：带有完整层级上下文的 chunks
"""
chunks = []
for doc in structured_docs:
heading = doc.get("heading", "")
content = doc.get("content", "")
# 构建层级上下文（面包屑路径）
breadcrumb = self._build_breadcrumb(heading)
if len(content) <= self.max_chunk_size:
# 内容在限制范围内，直接作为一个 chunk
chunks.append({
"content": f"{breadcrumb}\n\n{content}",
"heading": heading,
"breadcrumb": breadcrumb,
"source": doc.get("source", "")
})
else:
# 内容过长，需要进一步切分
sub_chunks = self._split_long_content(content)
for i, sub in enumerate(sub_chunks):
chunks.append({
"content": f"{breadcrumb}\n\n{sub}",
"heading": heading,
"breadcrumb": breadcrumb,
"chunk_index": i,
"source": doc.get("source", "")
})
return chunks
def _build_breadcrumb(self, heading: str) -> str:
"""构建面包屑导航路径"""
# 从 Markdown 标题解析层级
level = 0
for char in heading:
if char == '#':
level += 1
else:
break
return heading.strip()
def _split_long_content(self, content: str) -> list[str]:
"""切分过长的内容块"""
splitter = RecursiveCharacterTextSplitter(
chunk_size=self.max_chunk_size,
chunk_overlap=64,
separators=["\n\n", "\n", "。", "；", "，", " "]
)
return splitter.split_text(content)
第四章：Embedding 模型选型与微调
4.1 Embedding 模型选型
Embedding 模型将文本转化为高维向量，是语义检索的基础。选型时需要考虑以下维度：
| 模型 | 维度 | 中文能力 | 特点 | 适用场景 |
|---|---|---|---|---|
| BGE-Large-ZH | 1024 | 优秀 | 开源、可本地部署 | 中文企业场景首选 |
| Jina-Embeddings-v2 | 768 | 良好 | 支持8K长文本 | 长文档检索 |
| text-embedding-3-large | 3072 | 良好 | OpenAPI、性能强 | 快速原型、英文为主 |
| M3E-Large | 1024 | 优秀 | 开源、中文优化 | 中文场景性价比高 |
4.2 BGE 模型本地部署与使用
from sentence_transformers import SentenceTransformer
import numpy as np
class BGEEmbedding:
"""BGE 中文 Embedding 模型封装"""
def __init__(self, model_name="BAAI/bge-large-zh-v1.5"):
self.model = SentenceTransformer(model_name)
self.dimension = 1024
def embed_documents(self, texts: list[str]) -> np.ndarray:
"""批量文档向量化"""
# BGE 建议对文档添加前缀
prefixed_texts = [f"为这个句子生成表示以用于检索相关文章：{t}" for t in texts]
return self.model.encode(prefixed_texts, normalize_embeddings=True)
def embed_query(self, query: str) -> np.ndarray:
"""查询向量化"""
prefixed_query = f"为这个句子生成表示以用于检索相关文章：{query}"
return self.model.encode([prefixed_query], normalize_embeddings=True)[0]
4.3 Embedding 模型微调
当通用 Embedding 模型无法满足特定领域需求时，可以通过微调提升效果：
from sentence_transformers import SentenceTransformer, InputExample, losses
from torch.utils.data import DataLoader
class EmbeddingFineTuner:
"""Embedding 模型微调工具"""
def __init__(self, base_model="BAAI/bge-large-zh-v1.5"):
self.model = SentenceTransformer(base_model)
def prepare_training_data(self, query_doc_pairs: list[tuple]) -> list[InputExample]:
"""
准备训练数据
query_doc_pairs: [(query, positive_doc, negative_doc), ...]
"""
examples = []
for query, pos_doc, neg_doc in query_doc_pairs:
examples.append(InputExample(
texts=[query, pos_doc, neg_doc]
))
return examples
def fine_tune(self, training_examples: list[InputExample],
epochs=3, batch_size=16, output_path="./fine-tuned-embedding"):
"""执行微调"""
train_dataloader = DataLoader(training_examples, shuffle=True, batch_size=batch_size)
# 使用三元组损失
train_loss = losses.TripletLoss(model=self.model)
self.model.fit(
train_objectives=[(train_dataloader, train_loss)],
epochs=epochs,
output_path=output_path,
show_progress_bar=True
)
return output_path
第五章：混合检索架构 — 向量检索 + BM25 + 知识图谱
5.1 为什么需要混合检索
单一的向量检索存在明显局限：
- 关键词匹配弱：用户搜索"2024年Q3营收"时，向量检索可能匹配到其他季度的营收数据
- 精确查询差：搜索特定编号、型号、人名时，精确匹配更可靠
- 语义漂移：向量检索有时会返回语义相关但不是用户想要的内容
混合检索通过结合多种检索方式，取长补短：
import numpy as np
from rank_bm25 import BM25Okapi
from typing import Optional
class HybridRetriever:
"""混合检索器：向量检索 + BM25"""
def __init__(self, embedding_model, vector_store, alpha=0.7):
"""
alpha: 向量检索权重（0-1），1-alpha 为 BM25 权重
"""
self.embedding_model = embedding_model
self.vector_store = vector_store
self.alpha = alpha
self.bm25 = None
self.corpus = []
self.doc_ids = []
def build_index(self, documents: list[dict]):
"""构建混合索引"""
self.corpus = [doc["content"] for doc in documents]
self.doc_ids = [doc["id"] for doc in documents]
# 构建 BM25 索引
tokenized_corpus = [list(doc) for doc in self.corpus] # 中文按字符分词
self.bm25 = BM25Okapi(tokenized_corpus)
# 向量索引已在 vector_store 中构建
def retrieve(self, query: str, top_k: int = 10) -> list[dict]:
"""混合检索"""
# 向量检索
query_embedding = self.embedding_model.embed_query(query)
vector_results = self.vector_store.search(query_embedding, top_k=top_k * 2)
# BM25 检索
tokenized_query = list(query)
bm25_scores = self.bm25.get_scores(tokenized_query)
bm25_top_indices = np.argsort(bm25_scores)[::-1][:top_k * 2]
# 分数归一化
vector_scores = {r["id"]: r["score"] for r in vector_results}
max_bm25 = max(bm25_scores) if max(bm25_scores) > 0 else 1
# 融合分数
all_doc_ids = set()
for r in vector_results:
all_doc_ids.add(r["id"])
for idx in bm25_top_indices:
all_doc_ids.add(self.doc_ids[idx])
fused_results = []
for doc_id in all_doc_ids:
v_score = vector_scores.get(doc_id, 0)
bm25_idx = self.doc_ids.index(doc_id) if doc_id in self.doc_ids else -1
b_score = bm25_scores[bm25_idx] / max_bm25 if bm25_idx >= 0 else 0
# 加权融合
final_score = self.alpha * v_score + (1 - self.alpha) * b_score
fused_results.append({
"id": doc_id,
"score": final_score,
"vector_score": v_score,
"bm25_score": b_score
})
# 按融合分数排序
fused_results.sort(key=lambda x: x["score"], reverse=True)
return fused_results[:top_k]
5.2 RRF（Reciprocal Rank Fusion）融合算法
RRF 是一种更先进的融合方法，不需要归一化分数：
class RRFFuser:
"""RRF 融合算法"""
def __init__(self, k=60):
self.k = k # 平滑参数
def fuse(self, *ranked_lists: list[list[str]], weights: list[float] = None) -> list[tuple]:
"""
融合多个排序列表
ranked_lists: 每个元素是一个排序后的文档 ID 列表
weights: 每个列表的权重
"""
if weights is None:
weights = [1.0] * len(ranked_lists)
scores = {}
for ranked_list, weight in zip(ranked_lists, weights):
for rank, doc_id in enumerate(ranked_list):
if doc_id not in scores:
scores[doc_id] = 0
scores[doc_id] += weight * (1 / (self.k + rank + 1))
# 按 RRF 分数排序
sorted_docs = sorted(scores.items(), key=lambda x: x[1], reverse=True)
return sorted_docs
第六章：重排序与压缩 — 提升检索精度
6.1 Cross-Encoder 重排序
向量检索使用的是 Bi-Encoder（独立编码 query 和 doc），而 Cross-Encoder 可以同时编码 query 和 doc，捕捉更精细的交互关系：
from sentence_transformers import CrossEncoder
class CrossEncoderReranker:
"""Cross-Encoder 重排序器"""
def __init__(self, model_name="BAAI/bge-reranker-large"):
self.model = CrossEncoder(model_name)
def rerank(self, query: str, documents: list[dict], top_k: int = 5) -> list[dict]:
"""对检索结果重排序"""
# 构造 query-doc 对
pairs = [(query, doc["content"]) for doc in documents]
# 计算相关性分数
scores = self.model.predict(pairs)
# 按分数排序
for i, doc in enumerate(documents):
doc["rerank_score"] = float(scores[i])
reranked = sorted(documents, key=lambda x: x["rerank_score"], reverse=True)
return reranked[:top_k]
6.2 LLM-based Reranker
利用大模型本身的推理能力进行重排序，效果更好但成本更高：
class LLMReranker:
"""基于 LLM 的重排序器"""
def __init__(self, llm_client):
self.llm = llm_client
async def rerank(self, query: str, documents: list[dict], top_k: int = 5) -> list[dict]:
"""使用 LLM 进行重排序"""
# 构造 prompt
doc_list = "\n".join([
f"[{i+1}] {doc['content'][:200]}..."
for i, doc in enumerate(documents)
])
prompt = f"""你是一个文档排序专家。请根据查询与文档的相关性，对以下文档进行排序。
查询：{query}
候选文档：
{doc_list}
请返回排序后的文档编号列表（从最相关到最不相关），格式为 JSON 数组，例如：[3, 1, 5, 2, 4]
只返回 JSON 数组，不要其他内容。"""
response = await self.llm.generate(prompt)
# 解析排序结果
import json
try:
ranking = json.loads(response)
reranked = [documents[i - 1] for i in ranking if 1 <= i <= len(documents)]
return reranked[:top_k]
except:
return documents[:top_k]
6.3 上下文压缩
检索到的 chunk 可能包含与查询无关的内容，通过压缩可以减少噪声：
class ContextCompressor:
"""上下文压缩器"""
def __init__(self, llm_client):
self.llm = llm_client
async def compress(self, query: str, documents: list[dict]) -> list[dict]:
"""压缩文档，只保留与查询相关的部分"""
compressed = []
for doc in documents:
prompt = f"""请从以下文档中提取与查询最相关的内容，删除无关信息。
保留原文，不要改写。
查询：{query}
文档：
{doc['content']}
提取的相关内容："""
result = await self.llm.generate(prompt)
if result.strip():
compressed.append({
**doc,
"content": result.strip(),
"compressed": True
})
return compressed
第七章：查询理解与改写
7.1 HyDE（Hypothetical Document Embeddings）
HyDE 通过让 LLM 先生成一个假设性回答，再用这个回答去检索，可以显著提升检索效果：
class HyDEQueryRewriter:
"""HyDE 查询改写"""
def __init__(self, llm_client, embedding_model):
self.llm = llm_client
self.embedding_model = embedding_model
async def retrieve_with_hyde(self, query: str, vector_store, top_k: int = 10):
"""使用 HyDE 进行检索"""
# 第一步：让 LLM 生成假设性文档
hyde_prompt = f"""请根据以下问题，写一段可能包含答案的文档内容。
不需要准确，只需要看起来像是一个合理的技术文档片段。
问题：{query}
文档内容："""
hypothetical_doc = await self.llm.generate(hyde_prompt)
# 第二步：用假设性文档的 embedding 去检索
hyde_embedding = self.embedding_model.embed_query(hypothetical_doc)
results = vector_store.search(hyde_embedding, top_k=top_k)
return results
7.2 Multi-Query 改写
通过生成多个不同角度的查询，提升检索的召回率：
class MultiQueryRewriter:
"""Multi-Query 多角度查询改写"""
def __init__(self, llm_client):
self.llm = llm_client
async def generate_queries(self, original_query: str, num_queries: int = 3) -> list[str]:
"""生成多个不同角度的查询"""
prompt = f"""你是一个查询改写专家。请将以下查询改写为 {num_queries} 个不同的版本，
从不同角度表达相同的意思，以提升检索的召回率。
原始查询：{original_query}
请返回 {num_queries} 个改写后的查询，每行一个："""
response = await self.llm.generate(prompt)
queries = [q.strip() for q in response.strip().split("\n") if q.strip()]
# 加入原始查询
return [original_query] + queries[:num_queries]
async def retrieve_with_multi_query(self, query: str, retriever, top_k: int = 10):
"""使用 Multi-Query 进行检索"""
queries = await self.generate_queries(query)
all_results = []
for q in queries:
results = await retriever.retrieve(q, top_k=top_k)
all_results.extend(results)
# 去重并按分数排序
seen = set()
unique_results = []
for r in sorted(all_results, key=lambda x: x["score"], reverse=True):
if r["id"] not in seen:
seen.add(r["id"])
unique_results.append(r)
return unique_results[:top_k]
7.3 Step-Back Prompting
对于复杂查询，先将其"退一步"转化为更通用的问题：
class StepBackRewriter:
"""Step-Back 查询改写"""
def __init__(self, llm_client):
self.llm = llm_client
async def step_back(self, query: str) -> str:
"""将查询退一步，转化为更通用的问题"""
prompt = f"""你是一个查询改写专家。请将以下具体问题改写为一个更通用、更高层次的问题，
以便检索到更全面的背景知识。
原始问题：{query}
改写后的通用问题："""
response = await self.llm.generate(prompt)
return response.strip()
第八章：多模态 RAG
8.1 图片检索与理解
企业文档中包含大量图表、流程图、架构图，传统 RAG 系统无法处理这些视觉内容。
from PIL import Image
import base64
class MultimodalRAG:
"""多模态 RAG 系统"""
def __init__(self, vision_llm, text_embedding_model, clip_model=None):
self.vision_llm = vision_llm
self.text_embedding = text_embedding_model
self.clip_model = clip_model
def process_image(self, image_path: str) -> dict:
"""处理图片，生成描述和向量"""
# 使用 VLM 生成图片描述
with open(image_path, "rb") as f:
img_base64 = base64.b64encode(f.read()).decode()
description = self.vision_llm.describe_image(
image_base64=img_base64,
prompt="请详细描述这张图片的内容，包括文字、数据、图表类型、关键信息等。"
)
return {
"type": "image",
"description": description,
"image_path": image_path,
"content": description # 用描述作为检索内容
}
def process_table_image(self, image_path: str) -> dict:
"""处理表格图片，提取结构化数据"""
with open(image_path, "rb") as f:
img_base64 = base64.b64encode(f.read()).decode()
table_data = self.vision_llm.extract_table(
image_base64=img_base64,
prompt="请提取这张表格图片中的所有数据，以 Markdown 表格格式输出。"
)
return {
"type": "table_image",
"content": table_data,
"image_path": image_path
}
8.2 表格检索增强
表格数据的检索需要特殊处理，因为表格的语义不仅取决于单元格内容，还取决于行列关系：
class TableRetriever:
"""表格检索增强"""
def index_table(self, table_data: dict) -> list[dict]:
"""将表格转化为多个可检索的文本块"""
chunks = []
headers = table_data["headers"]
rows = table_data["rows"]
title = table_data.get("title", "")
# 策略1：整表描述
table_desc = f"表格标题：{title}\n列：{', '.join(headers)}\n共 {len(rows)} 行数据"
chunks.append({
"content": table_desc,
"type": "table_summary",
"table_id": table_data["id"]
})
# 策略2：逐行描述
for i, row in enumerate(rows):
row_desc = f"在 {title} 表中："
for header, value in zip(headers, row):
row_desc += f" {header} = {value}；"
chunks.append({
"content": row_desc,
"type": "table_row",
"table_id": table_data["id"],
"row_index": i
})
return chunks
第九章：评估体系
9.1 RAG 评估的核心指标
一个完善的 RAG 评估体系需要覆盖以下维度：
| 指标 | 说明 | 计算方式 |
|---|---|---|
| Context Recall | 检索内容对回答的覆盖度 | 检索到的相关文档 / 总相关文档 |
| Context Precision | 检索内容的精确度 | 相关文档 / 检索到的总文档 |
| Faithfulness | 回答的忠实度 | 回答中有依据的内容比例 |
| Answer Relevancy | 回答的相关性 | 回答与问题的语义相关度 |
9.2 使用 RAGAS 进行自动化评估
from ragas import evaluate
from ragas.metrics import (
context_recall,
context_precision,
faithfulness,
answer_relevancy
)
from datasets import Dataset
class RAGEvaluator:
"""RAG 系统评估器"""
def evaluate_with_ragas(self, eval_data: list[dict]) -> dict:
"""
使用 RAGAS 框架评估
eval_data 格式: [{"question": ..., "answer": ..., "contexts": [...], "ground_truth": ...}]
"""
dataset = Dataset.from_list(eval_data)
result = evaluate(
dataset=dataset,
metrics=[
context_recall,
context_precision,
faithfulness,
answer_relevancy
],
)
return {
"context_recall": result["context_recall"],
"context_precision": result["context_precision"],
"faithfulness": result["faithfulness"],
"answer_relevancy": result["answer_relevancy"],
"overall_score": (
result["context_recall"] +
result["context_precision"] +
result["faithfulness"] +
result["answer_relevancy"]
) / 4
}
def build_eval_dataset(self, questions: list[str],
rag_system, ground_truths: list[str]) -> list[dict]:
"""构建评估数据集"""
eval_data = []
for q, gt in zip(questions, ground_truths):
# 调用 RAG 系统获取回答和检索结果
result = rag_system.query(q)
eval_data.append({
"question": q,
"answer": result["answer"],
"contexts": [c["content"] for c in result["contexts"]],
"ground_truth": gt
})
return eval_data
9.3 A/B 测试评估
class ABTestEvaluator:
"""A/B 测试评估框架"""
def __init__(self):
self.results = {"A": [], "B": []}
def run_test(self, questions: list[str], system_a, system_b):
"""运行 A/B 测试"""
import random
for question in questions:
# 随机决定哪个系统先回答（避免顺序偏差）
if random.random() > 0.5:
result_a = system_a.query(question)
result_b = system_b.query(question)
order = "AB"
else:
result_b = system_b.query(question)
result_a = system_a.query(question)
order = "BA"
self.results["A"].append({
"question": question,
"answer": result_a["answer"],
"latency": result_a.get("latency", 0),
"order": order
})
self.results["B"].append({
"question": question,
"answer": result_b["answer"],
"latency": result_b.get("latency", 0),
"order": order
})
def get_comparison_report(self) -> dict:
"""生成对比报告"""
avg_latency_a = sum(r["latency"] for r in self.results["A"]) / len(self.results["A"])
avg_latency_b = sum(r["latency"] for r in self.results["B"]) / len(self.results["B"])
return {
"system_a_avg_latency": avg_latency_a,
"system_b_avg_latency": avg_latency_b,
"total_questions": len(self.results["A"]),
}
第十章：生产运维
10.1 缓存策略
import hashlib
import json
from functools import lru_cache
class RAGCache:
"""RAG 系统缓存层"""
def __init__(self, redis_client, ttl=3600):
self.redis = redis_client
self.ttl = ttl # 缓存过期时间（秒）
def _make_key(self, query: str, filters: dict = None) -> str:
"""生成缓存 key"""
content = json.dumps({"query": query, "filters": filters or {}}, sort_keys=True)
return f"rag:query:{hashlib.md5(content.encode()).hexdigest()}"
async def get_or_compute(self, query: str, compute_fn, filters: dict = None) -> dict:
"""获取缓存结果，不存在则计算并缓存"""
key = self._make_key(query, filters)
# 尝试从缓存获取
cached = self.redis.get(key)
if cached:
return json.loads(cached)
# 计算结果
result = await compute_fn(query)
# 存入缓存
self.redis.setex(key, self.ttl, json.dumps(result, ensure_ascii=False))
return result
def invalidate(self, pattern: str = "rag:query:*"):
"""批量失效缓存"""
keys = self.redis.keys(pattern)
if keys:
self.redis.delete(*keys)
10.2 限流与降级
from datetime import datetime, timedelta
import asyncio
class RateLimiter:
"""限流器"""
def __init__(self, redis_client, max_requests=100, window_seconds=60):
self.redis = redis_client
self.max_requests = max_requests
self.window = window_seconds
async def check(self, user_id: str) -> bool:
"""检查用户是否超过限流"""
key = f"ratelimit:{user_id}:{datetime.now().strftime('%Y%m%d%H%M')}"
current = self.redis.incr(key)
if current == 1:
self.redis.expire(key, self.window)
return current <= self.max_requests
class CircuitBreaker:
"""熔断器，当错误率过高时自动降级"""
def __init__(self, failure_threshold=5, recovery_timeout=30):
self.failure_count = 0
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.state = "closed" # closed, open, half-open
self.last_failure_time = None
async def call(self, func, fallback=None):
"""执行函数，带熔断保护"""
if self.state == "open":
if datetime.now() - self.last_failure_time > timedelta(seconds=self.recovery_timeout):
self.state = "half-open"
else:
if fallback:
return await fallback()
raise Exception("Circuit breaker is open")
try:
result = await func()
if self.state == "half-open":
self.state = "closed"
self.failure_count = 0
return result
except Exception as e:
self.failure_count += 1
self.last_failure_time = datetime.now()
if self.failure_count >= self.failure_threshold:
self.state = "open"
if fallback:
return await fallback()
raise e
10.3 可观测性
import time
from dataclasses import dataclass, field
from typing import Optional
import logging
@dataclass
class RAGTrace:
"""RAG 请求链路追踪"""
trace_id: str
query: str
start_time: float = field(default_factory=time.time)
# 各阶段耗时
query_rewrite_time: float = 0
retrieval_time: float = 0
rerank_time: float = 0
generation_time: float = 0
# 各阶段结果
rewritten_queries: list[str] = field(default_factory=list)
retrieved_docs: int = 0
reranked_docs: int = 0
answer_length: int = 0
# 质量指标
confidence_score: float = 0
def total_latency(self) -> float:
return time.time() - self.start_time
def to_dict(self) -> dict:
return {
"trace_id": self.trace_id,
"query": self.query,
"total_latency_ms": self.total_latency() * 1000,
"query_rewrite_time_ms": self.query_rewrite_time * 1000,
"retrieval_time_ms": self.retrieval_time * 1000,
"rerank_time_ms": self.rerank_time * 1000,
"generation_time_ms": self.generation_time * 1000,
"rewritten_queries": self.rewritten_queries,
"retrieved_docs": self.retrieved_docs,
"reranked_docs": self.reranked_docs,
"answer_length": self.answer_length,
"confidence_score": self.confidence_score
}
class RAGMonitor:
"""RAG 系统监控"""
def __init__(self, metrics_client=None):
self.metrics = metrics_client
self.logger = logging.getLogger("rag.monitor")
def record_trace(self, trace: RAGTrace):
"""记录追踪信息"""
trace_data = trace.to_dict()
self.logger.info(f"RAG Trace: {json.dumps(trace_data)}")
if self.metrics:
# 记录各阶段延迟
self.metrics.histogram("rag.latency.total", trace.total_latency() * 1000)
self.metrics.histogram("rag.latency.retrieval", trace.retrieval_time * 1000)
self.metrics.histogram("rag.latency.generation", trace.generation_time * 1000)
# 记录检索文档数
self.metrics.gauge("rag.docs.retrieved", trace.retrieved_docs)
self.metrics.gauge("rag.docs.reranked", trace.reranked_docs)
第十一章：实战项目 — 构建百万文档级企业知识库
11.1 系统架构设计
"""
企业知识库系统主入口
"""
from dataclasses import dataclass
from typing import Optional
import asyncio
@dataclass
class KnowledgeBaseConfig:
"""知识库配置"""
# Embedding 配置
embedding_model: str = "BAAI/bge-large-zh-v1.5"
embedding_dimension: int = 1024
# 向量数据库配置
vector_db_type: str = "milvus" # milvus / qdrant / pgvector
vector_db_url: str = "localhost:19530"
# 分块配置
chunk_size: int = 512
chunk_overlap: int = 64
# 检索配置
retrieval_top_k: int = 20
rerank_top_k: int = 5
hybrid_alpha: float = 0.7 # 向量检索权重
# 缓存配置
cache_enabled: bool = True
cache_ttl: int = 3600
class EnterpriseKnowledgeBase:
"""企业级知识库系统"""
def __init__(self, config: KnowledgeBaseConfig):
self.config = config
# 初始化各组件
self.embedding_model = BGEEmbedding(config.embedding_model)
self.vector_store = self._init_vector_store()
self.document_processor = DocumentProcessorPipeline()
self.chunker = SmartRecursiveSplitter(config.chunk_size, config.chunk_overlap)
self.retriever = HybridRetriever(self.embedding_model, self.vector_store, config.hybrid_alpha)
self.reranker = CrossEncoderReranker()
self.llm = ChatOpenAI(model="gpt-4")
self.cache = RAGCache(redis_client=None)
self.monitor = RAGMonitor()
def _init_vector_store(self):
"""初始化向量数据库"""
if self.config.vector_db_type == "milvus":
from pymilvus import connections, Collection
connections.connect(host=self.config.vector_db_url)
# 创建或获取 collection
return MilvusVectorStore(
collection_name="enterprise_kb",
dimension=self.config.embedding_dimension
)
elif self.config.vector_db_type == "qdrant":
from qdrant_client import QdrantClient
client = QdrantClient(host=self.config.vector_db_url)
return QdrantVectorStore(client, "enterprise_kb")
async def ingest_document(self, file_path: str, metadata: dict = None):
"""文档入库流程"""
# 第一步：文档处理
processed = self.document_processor.process(file_path)
# 第二步：分块
all_chunks = []
for doc in processed:
chunks = self.chunker.split(doc["content"], metadata={
"source": file_path,
"type": doc.get("type", "text"),
**(metadata or {})
})
all_chunks.extend(chunks)
# 第三步：向量化
texts = [c["content"] for c in all_chunks]
embeddings = self.embedding_model.embed_documents(texts)
# 第四步：存入向量数据库
self.vector_store.insert(texts, embeddings, [c for c in all_chunks])
# 第五步：更新 BM25 索引
self.retriever.build_index(all_chunks)
# 第六步：失效相关缓存
if self.config.cache_enabled:
self.cache.invalidate()
return {"status": "success", "chunks": len(all_chunks)}
async def query(self, question: str, user_id: str = None) -> dict:
"""查询入口"""
trace = RAGTrace(trace_id=str(uuid.uuid4()), query=question)
# 缓存检查
if self.config.cache_enabled:
cached = await self.cache.get_or_compute(
question,
lambda q: self._execute_query(q, trace)
)
return cached
return await self._execute_query(question, trace)
async def _execute_query(self, question: str, trace: RAGTrace) -> dict:
"""执行查询"""
# 第一步：查询改写
t0 = time.time()
rewriter = MultiQueryRewriter(self.llm)
queries = await rewriter.generate_queries(question)
trace.rewritten_queries = queries
trace.query_rewrite_time = time.time() - t0
# 第二步：混合检索
t0 = time.time()
all_results = []
for q in queries:
results = self.retriever.retrieve(q, top_k=self.config.retrieval_top_k)
all_results.extend(results)
# 去重
seen = set()
unique_results = []
for r in all_results:
if r["id"] not in seen:
seen.add(r["id"])
unique_results.append(r)
unique_results.sort(key=lambda x: x["score"], reverse=True)
unique_results = unique_results[:self.config.retrieval_top_k]
trace.retrieval_time = time.time() - t0
trace.retrieved_docs = len(unique_results)
# 第三步：重排序
t0 = time.time()
reranked = self.reranker.rerank(question, unique_results, top_k=self.config.rerank_top_k)
trace.rerank_time = time.time() - t0
trace.reranked_docs = len(reranked)
# 第四步：生成回答
t0 = time.time()
context = "\n\n---\n\n".join([doc["content"] for doc in reranked])
prompt = f"""你是一个企业知识库助手。请基于以下参考资料回答用户的问题。
如果参考资料中没有相关信息，请明确告知用户。
请在回答末尾标注引用来源。
参考资料：
{context}
用户问题：{question}
回答："""
answer = await self.llm.generate(prompt)
trace.generation_time = time.time() - t0
trace.answer_length = len(answer)
# 记录追踪
self.monitor.record_trace(trace)
return {
"answer": answer,
"sources": [{"content": d["content"][:200], "score": d["score"]} for d in reranked],
"trace_id": trace.trace_id
}
11.2 完整部署架构
生产环境的完整部署通常包括以下组件：
# docker-compose.yml
version: '3.8'
services:
# RAG API 服务
rag-api:
build: ./rag-api
ports:
- "8000:8000"
environment:
- MILVUS_HOST=milvus
- REDIS_HOST=redis
- EMBEDDING_MODEL=BAAI/bge-large-zh-v1.5
depends_on:
- milvus
- redis
# Milvus 向量数据库
milvus:
image: milvusdb/milvus:v2.3-latest
ports:
- "19530:19530"
volumes:
- milvus_data:/var/lib/milvus
# Redis 缓存
redis:
image: redis:7-alpine
ports:
- "6379:6379"
# MinIO 对象存储（存储原始文档）
minio:
image: minio/minio
ports:
- "9000:9000"
command: server /data
# 监控
prometheus:
image: prom/prometheus
ports:
- "9090:9090"
grafana:
image: grafana/grafana
ports:
- "3000:3000"
volumes:
milvus_data:
11.3 API 接口设计
from fastapi import FastAPI, UploadFile, File, HTTPException
from pydantic import BaseModel
app = FastAPI(title="企业知识库 API")
kb = EnterpriseKnowledgeBase(KnowledgeBaseConfig())
class QueryRequest(BaseModel):
question: str
user_id: str = None
top_k: int = 5
class QueryResponse(BaseModel):
answer: str
sources: list[dict]
trace_id: str
@app.post("/api/query", response_model=QueryResponse)
async def query_knowledge_base(req: QueryRequest):
"""查询知识库"""
try:
result = await kb.query(req.question, req.user_id)
return QueryResponse(**result)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/ingest")
async def ingest_document(file: UploadFile = File(...)):
"""上传并入库文档"""
# 保存文件
file_path = f"/tmp/{file.filename}"
with open(file_path, "wb") as f:
content = await file.read()
f.write(content)
# 入库
result = await kb.ingest_document(file_path)
return result
@app.get("/api/stats")
async def get_stats():
"""获取知识库统计信息"""
return {
"total_documents": kb.vector_store.count(),
"embedding_model": kb.config.embedding_model,
"vector_db": kb.config.vector_db_type
}
总结
本教程系统性地讲解了企业级 RAG 系统的完整技术栈，从文档处理到智能分块，从 Embedding 选型到混合检索，从重排序到查询改写，从评估体系到生产运维。
核心要点回顾：
- 文档处理是基础：支持多种格式、保留结构信息、处理表格和图片
- 分块策略是关键：语义分块优于固定分块，层级分块保留上下文
- 混合检索是标配：向量检索 + BM25 融合，RRF 算法简单有效
- 重排序提升精度：Cross-Encoder 性价比最高，LLM Reranker 效果最好
- 查询改写提升召回：HyDE、Multi-Query、Step-Back 各有适用场景
- 评估体系保障质量：RAGAS 自动化评估 + A/B 测试
- 生产运维不可忽视：缓存、限流、熔断、可观测性
掌握这些技术，你就能够构建一个真正可以上线的企业级 RAG 系统。
