企业级RAG系统架构与实战完全教程
教程简介
检索增强生成（Retrieval-Augmented Generation, RAG）已成为企业级AI应用的核心架构模式。本教程从RAG的基础原理出发，系统讲解企业级RAG系统的架构设计、核心组件、实战部署与运维优化，帮助开发者构建生产级别的智能检索系统。
第一章：RAG架构演进与设计原则
1.1 从朴素RAG到高级RAG
RAG的核心思想是将外部知识库与大语言模型（LLM）结合，通过检索相关文档来增强模型的生成能力。
朴素RAG（Naive RAG） 的基本流程：
用户查询 → 文档检索 → 拼接上下文 → LLM生成 → 返回结果
高级RAG（Advanced RAG） 在此基础上增加了查询优化、多路检索、重排序、幻觉检测等环节：
用户查询 → 查询理解/改写 → 多路检索 → 重排序 → 上下文压缩 → LLM生成 → 事实性验证 → 返回结果
模块化RAG（Modular RAG） 将系统拆分为可组合的独立模块：
class ModularRAG:
"""模块化RAG架构"""
def __init__(self):
self.query_processor = QueryProcessor() # 查询处理
self.retriever = HybridRetriever() # 混合检索
self.reranker = CrossEncoderReranker() # 重排序
self.context_manager = ContextManager() # 上下文管理
self.generator = LLMGenerator() # 生成
self.validator = HallucinationValidator() # 幻觉验证
self.memory = ConversationMemory() # 对话记忆
1.2 企业级RAG的特殊挑战
企业场景与Demo级RAG有本质区别：
| 维度 | Demo级 | 企业级 |
|---|---|---|
| 文档量 | 几十篇 | 百万级 |
| 文档类型 | 纯文本 | PDF/表格/图片/音视频 |
| 查询复杂度 | 简单问答 | 多轮对话/复杂推理 |
| 准确性要求 | 大致正确 | 必须可溯源 |
| 延迟要求 | 几秒可接受 | P99 < 3秒 |
| 安全性 | 无要求 | 权限控制/审计 |
第二章：多模态文档解析
2.1 PDF文档解析
企业文档大量以PDF格式存在，解析质量直接影响RAG效果。
# 使用 PyMuPDF4LLM 进行高质量PDF解析
import pymupdf4llm
def parse_pdf_with_pymupdf(pdf_path: str) -> str:
"""将PDF转换为Markdown格式，保留结构信息"""
md_text = pymupdf4llm.to_markdown(
pdf_path,
page_chunks=True, # 按页分块
write_images=False, # 不提取图片
image_path="images/",
dpi=150
)
return md_text
# 使用 Unstructured 进行复杂PDF解析
from unstructured.partition.pdf import partition_pdf
def parse_complex_pdf(pdf_path: str) -> list:
"""解析包含表格、图片的复杂PDF"""
elements = partition_pdf(
filename=pdf_path,
strategy="hi_res", # 高精度模式
model_name="yolox", # 版面分析模型
extract_images_in_pdf=True,
infer_table_structure=True,
chunking_strategy="by_title",
max_characters=1000,
new_after_n_chars=800,
overlap=200
)
return elements
2.2 表格数据处理
表格是企业文档中最常见也最难处理的结构。
import pandas as pd
from io import StringIO
class TableProcessor:
"""表格处理器：将表格转换为LLM可理解的文本"""
def table_to_natural_language(self, table: pd.DataFrame, context: str = "") -> str:
"""将表格转换为自然语言描述"""
descriptions = []
# 表头信息
headers = ", ".join(table.columns.tolist())
descriptions.append(f"该表格包含以下列：{headers}")
# 逐行描述
for idx, row in table.iterrows():
row_desc = []
for col in table.columns:
row_desc.append(f"{col}为{row[col]}")
descriptions.append(f"第{idx+1}行：{'，'.join(row_desc)}")
return "\n".join(descriptions)
def table_to_markdown(self, table: pd.DataFrame) -> str:
"""将DataFrame转为Markdown表格"""
return table.to_markdown(index=False)
def smart_chunk_table(self, table: pd.DataFrame, max_rows: int = 20) -> list:
"""大表格智能分块"""
chunks = []
for i in range(0, len(table), max_rows):
chunk = table.iloc[i:i+max_rows]
chunks.append(self.table_to_markdown(chunk))
return chunks
2.3 图片与多模态内容
import base64
class ImageProcessor:
"""图片处理器：提取图片中的文字和语义信息"""
def __init__(self, vlm_client):
self.vlm = vlm_client # 视觉语言模型
async def extract_from_image(self, image_path: str) -> dict:
"""使用VLM提取图片中的文字和语义"""
with open(image_path, "rb") as f:
img_base64 = base64.b64encode(f.read()).decode()
prompt = """请分析这张图片，提取以下信息：
1. 图片中的所有文字内容（OCR）
2. 图片的类型（图表/流程图/截图/照片等）
3. 图片所表达的核心信息
4. 如果是图表，请描述数据趋势和关键数值
请以JSON格式返回结果。"""
result = await self.vlm.chat(
messages=[{
"role": "user",
"content": [
{"type": "text", "text": prompt},
{"type": "image_url", "image_url": {"url": f"data:image/png;base64,{img_base64}"}}
]
}]
)
return result
第三章：Chunking策略与优化
3.1 常见分块策略
Chunking是RAG系统中影响检索质量的关键环节。
from typing import List
class TextChunker:
"""文本分块器"""
def fixed_size_chunk(self, text: str, chunk_size: int = 512, overlap: int = 50) -> List[str]:
"""固定大小分块"""
chunks = []
start = 0
while start < len(text):
end = start + chunk_size
chunks.append(text[start:end])
start = end - overlap
return chunks
def recursive_split(self, text: str, chunk_size: int = 1000,
separators: List[str] = None) -> List[str]:
"""递归字符分割：按语义边界分块"""
if separators is None:
separators = ["\n\n", "\n", "。", "！", "？", ".", "!", "?", " "]
chunks = []
current_chunks = [text]
for sep in separators:
new_chunks = []
for chunk in current_chunks:
if len(chunk) <= chunk_size:
new_chunks.append(chunk)
else:
splits = chunk.split(sep)
current = ""
for split in splits:
if len(current) + len(split) + len(sep) <= chunk_size:
current = current + sep + split if current else split
else:
if current:
new_chunks.append(current)
current = split
if current:
new_chunks.append(current)
current_chunks = new_chunks
return [c.strip() for c in current_chunks if c.strip()]
def semantic_chunk(self, text: str, embedding_model, threshold: float = 0.5) -> List[str]:
"""语义分块：基于嵌入相似度的智能分块"""
import numpy as np
# 按句子分割
sentences = self._split_sentences(text)
# 计算每个句子的嵌入
embeddings = embedding_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 < threshold:
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
3.2 高级分块：Parent-Child策略
class ParentChildChunker:
"""父子分块策略：检索小块，返回大块"""
def __init__(self, child_size=256, parent_size=1024, overlap=64):
self.child_size = child_size
self.parent_size = parent_size
self.overlap = overlap
def chunk(self, text: str) -> dict:
"""生成父子块"""
# 生成大块（父块）
parent_chunks = self._split(text, self.parent_size, self.overlap)
result = {"parents": [], "children": []}
for parent_idx, parent in enumerate(parent_chunks):
result["parents"].append({
"id": f"parent_{parent_idx}",
"text": parent,
"metadata": {"parent_id": f"parent_{parent_idx}"}
})
# 将大块拆分为小块（子块）
child_chunks = self._split(parent, self.child_size, self.overlap // 2)
for child_idx, child in enumerate(child_chunks):
result["children"].append({
"id": f"child_{parent_idx}_{child_idx}",
"text": child,
"parent_id": f"parent_{parent_idx}",
"metadata": {"parent_id": f"parent_{parent_idx}"}
})
return result
def _split(self, text, size, overlap):
"""按句子边界分块"""
sentences = text.replace("。", "。\n").replace("！", "！\n").replace("？", "？\n").split("\n")
chunks = []
current = ""
for sent in sentences:
if len(current) + len(sent) <= size:
current += sent
else:
if current:
chunks.append(current.strip())
current = sent
if current:
chunks.append(current.strip())
return chunks
第四章：向量数据库选型与实战
4.1 主流向量数据库对比
| 特性 | Milvus | Qdrant | Weaviate | Chroma |
|---|---|---|---|---|
| 语言 | Go/C++ | Rust | Go | Python |
| 部署复杂度 | 高 | 中 | 中 | 低 |
| 性能 | 极高 | 高 | 高 | 中 |
| 标量过滤 | ✅ | ✅ | ✅ | ✅ |
| 多租户 | ✅ | ✅ | ✅ | ❌ |
| 适用规模 | 十亿级 | 千万级 | 千万级 | 百万级 |
| 云服务 | Zilliz Cloud | Qdrant Cloud | WCS | ❌ |
4.2 Milvus实战
from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType, utility
class MilvusVectorStore:
"""Milvus向量数据库操作封装"""
def __init__(self, host="localhost", port="19530"):
connections.connect(host=host, port=port)
self.collection = None
def create_collection(self, name: str, dim: int = 1536):
"""创建集合"""
fields = [
FieldSchema(name="id", dtype=DataType.VARCHAR, is_primary=True, max_length=64),
FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=65535),
FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=dim),
FieldSchema(name="metadata", dtype=DataType.JSON),
]
schema = CollectionSchema(fields, description="RAG document chunks")
self.collection = Collection(name, schema)
# 创建向量索引
index_params = {
"metric_type": "COSINE",
"index_type": "HNSW",
"params": {"M": 16, "efConstruction": 256}
}
self.collection.create_index("embedding", index_params)
return self.collection
def insert(self, ids: list, texts: list, embeddings: list, metadata: list):
"""批量插入数据"""
data = [ids, texts, embeddings, metadata]
self.collection.insert(data)
self.collection.flush()
def search(self, query_embedding: list, top_k: int = 10,
filter_expr: str = None) -> list:
"""向量检索"""
self.collection.load()
search_params = {"metric_type": "COSINE", "params": {"ef": 128}}
results = self.collection.search(
data=[query_embedding],
anns_field="embedding",
param=search_params,
limit=top_k,
expr=filter_expr,
output_fields=["text", "metadata"]
)
return [
{
"id": hit.id,
"text": hit.entity.get("text"),
"score": hit.score,
"metadata": hit.entity.get("metadata")
}
for hit in results[0]
]
4.3 Qdrant实战
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct, Filter, FieldCondition
class QdrantVectorStore:
"""Qdrant向量数据库操作封装"""
def __init__(self, host="localhost", port=6333):
self.client = QdrantClient(host=host, port=port)
def create_collection(self, name: str, dim: int = 1536):
"""创建集合"""
self.client.create_collection(
collection_name=name,
vectors_config=VectorParams(size=dim, distance=Distance.COSINE)
)
def upsert(self, collection: str, points: list):
"""插入或更新数据"""
self.client.upsert(
collection_name=collection,
points=[
PointStruct(
id=p["id"],
vector=p["embedding"],
payload={"text": p["text"], "metadata": p.get("metadata", {})}
)
for p in points
]
)
def search(self, collection: str, query_vector: list,
top_k: int = 10, filter_conditions: dict = None) -> list:
"""向量检索"""
query_filter = None
if filter_conditions:
query_filter = Filter(
must=[
FieldCondition(key=k, match={"value": v})
for k, v in filter_conditions.items()
]
)
results = self.client.search(
collection_name=collection,
query_vector=query_vector,
limit=top_k,
query_filter=query_filter
)
return [
{
"id": r.id,
"text": r.payload.get("text"),
"score": r.score,
"metadata": r.payload.get("metadata", {})
}
for r in results
]
第五章：混合检索策略
5.1 稠密检索 + 稀疏检索
单一检索方式难以覆盖所有场景，混合检索能显著提升召回率。
from sentence_transformers import SentenceTransformer
from rank_bm25 import BM25Okapi
import numpy as np
class HybridRetriever:
"""混合检索器：稠密向量 + BM25稀疏检索"""
def __init__(self, embedding_model_name: str = "BAAI/bge-large-zh-v1.5"):
self.dense_model = SentenceTransformer(embedding_model_name)
self.bm25 = None
self.documents = []
self.dense_index = None
def build_index(self, documents: list):
"""构建索引"""
self.documents = documents
texts = [doc["text"] for doc in documents]
# BM25索引
tokenized = [list(text) for text in texts] # 中文按字符分词
self.bm25 = BM25Okapi(tokenized)
# 稠密向量索引
self.dense_index = self.dense_model.encode(texts, show_progress_bar=True)
def search(self, query: str, top_k: int = 10,
dense_weight: float = 0.7, sparse_weight: float = 0.3) -> list:
"""混合检索"""
# 稠密检索
query_embedding = self.dense_model.encode([query])[0]
dense_scores = np.dot(self.dense_index, query_embedding)
# 稀疏检索
query_tokens = list(query)
sparse_scores = self.bm25.get_scores(query_tokens)
# 归一化
dense_scores = (dense_scores - dense_scores.min()) / (dense_scores.max() - dense_scores.min() + 1e-8)
sparse_scores = (sparse_scores - sparse_scores.min()) / (sparse_scores.max() - sparse_scores.min() + 1e-8)
# 加权融合
combined_scores = dense_weight * dense_scores + sparse_weight * sparse_scores
# 排序
top_indices = np.argsort(combined_scores)[::-1][:top_k]
results = []
for idx in top_indices:
results.append({
"text": self.documents[idx]["text"],
"score": float(combined_scores[idx]),
"dense_score": float(dense_scores[idx]),
"sparse_score": float(sparse_scores[idx]),
"metadata": self.documents[idx].get("metadata", {})
})
return results
5.2 ColBERT晚期交互检索
# 使用 ColBERT 进行晚期交互检索
# pip install ragatouille
from ragatouille import RAGPretrainedModel
class ColBERTRetriever:
"""ColBERT晚期交互检索器"""
def __init__(self, model_path: str = "colbert-ir/colbertv2.0"):
self.model = RAGPretrainedModel.from_pretrained(model_path)
def index(self, documents: list, collection_name: str = "my_index"):
"""构建ColBERT索引"""
self.model.index(
collection=[doc["text"] for doc in documents],
index_name=collection_name,
max_document_length=256,
split_documents=True
)
def search(self, query: str, top_k: int = 10) -> list:
"""ColBERT检索"""
results = self.model.search(query, k=top_k)
return results
第六章：重排序（Reranking）
6.1 Cross-Encoder重排序
检索返回的候选文档需要经过精排来提升相关性。
from sentence_transformers import CrossEncoder
class Reranker:
"""Cross-Encoder重排序器"""
def __init__(self, model_name: str = "BAAI/bge-reranker-v2-m3"):
self.model = CrossEncoder(model_name, max_length=512)
def rerank(self, query: str, documents: list, top_k: int = 5) -> list:
"""对检索结果重排序"""
pairs = [[query, doc["text"]] 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]
class LLMReranker:
"""使用LLM进行重排序（适合高质量场景）"""
def __init__(self, llm_client):
self.llm = llm_client
async def rerank(self, query: str, documents: list, top_k: int = 5) -> list:
"""LLM重排序"""
doc_list = "\n".join([
f"[{i+1}] {doc['text'][:200]}..."
for i, doc in enumerate(documents)
])
prompt = f"""请根据与查询的相关性，对以下文档进行排序。
查询：{query}
文档列表：
{doc_list}
请返回排序后的文档编号（从最相关到最不相关），用逗号分隔。
例如：3,1,5,2,4
排序结果："""
result = await self.llm.chat(messages=[{"role": "user", "content": prompt}])
# 解析排序结果
try:
order = [int(x.strip()) - 1 for x in result.strip().split(",")]
reranked = [documents[i] for i in order if i < len(documents)]
return reranked[:top_k]
except:
return documents[:top_k]
第七章：查询理解与改写
7.1 查询意图识别
class QueryProcessor:
"""查询处理器"""
def __init__(self, llm_client):
self.llm = llm_client
async def classify_intent(self, query: str) -> str:
"""查询意图分类"""
prompt = f"""请对以下查询进行意图分类：
查询：{query}
可选分类：
- factual：事实性问答
- comparison：对比分析
- how-to：操作指导
- summary：总结归纳
- analysis：深入分析
请只返回分类名称。"""
result = await self.llm.chat(messages=[{"role": "user", "content": prompt}])
return result.strip().lower()
async def rewrite_query(self, query: str, history: list = None) -> str:
"""查询改写：优化检索效果"""
history_text = ""
if history:
history_text = "\n对话历史：\n" + "\n".join([
f"用户：{h['user']}\n助手：{h['assistant']}"
for h in history[-3:]
])
prompt = f"""请将以下查询改写为更适合检索的形式，保持核心意图不变。
{history_text}
原始查询：{query}
改写后的查询（只返回改写结果）："""
result = await self.llm.chat(messages=[{"role": "user", "content": prompt}])
return result.strip()
async def generate_sub_queries(self, query: str) -> list:
"""查询分解：将复杂查询拆分为子查询"""
prompt = f"""请将以下复杂查询分解为2-4个更具体的子查询：
原始查询：{query}
请以JSON数组格式返回子查询列表。
示例：["子查询1", "子查询2", "子查询3"]"""
result = await self.llm.chat(messages=[{"role": "user", "content": prompt}])
import json
try:
return json.loads(result.strip())
except:
return [query]
第八章：幻觉检测与事实性验证
8.1 基于检索的验证
class HallucinationValidator:
"""幻觉检测器"""
def __init__(self, llm_client, embedding_model):
self.llm = llm_client
self.embedding_model = embedding_model
async def validate(self, query: str, answer: str,
source_documents: list) -> dict:
"""验证生成内容的事实性"""
# 提取答案中的关键声明
claims = await self._extract_claims(answer)
# 对每个声明进行验证
validations = []
for claim in claims:
# 在源文档中查找支持证据
evidence = await self._find_evidence(claim, source_documents)
# 判断是否有充分支持
is_supported = await self._check_support(claim, evidence)
validations.append({
"claim": claim,
"is_supported": is_supported,
"evidence": evidence
})
# 计算事实性得分
supported_count = sum(1 for v in validations if v["is_supported"])
factuality_score = supported_count / len(validations) if validations else 1.0
return {
"factuality_score": factuality_score,
"validations": validations,
"is_faithful": factuality_score >= 0.8
}
async def _extract_claims(self, text: str) -> list:
"""提取文本中的关键声明"""
prompt = f"""请从以下文本中提取所有事实性声明（可验证的陈述）：
{text}
请以JSON数组格式返回，每个元素为一个声明。"""
result = await self.llm.chat(messages=[{"role": "user", "content": prompt}])
import json
try:
return json.loads(result.strip())
except:
return [text]
async def _find_evidence(self, claim: str, documents: list) -> list:
"""在源文档中查找支持证据"""
claim_embedding = self.embedding_model.encode([claim])[0]
evidence = []
for doc in documents:
doc_embedding = self.embedding_model.encode([doc["text"]])[0]
similarity = np.dot(claim_embedding, doc_embedding) / (
np.linalg.norm(claim_embedding) * np.linalg.norm(doc_embedding)
)
if similarity > 0.5:
evidence.append({"text": doc["text"], "score": float(similarity)})
return sorted(evidence, key=lambda x: x["score"], reverse=True)[:3]
async def _check_support(self, claim: str, evidence: list) -> bool:
"""判断声明是否有充分证据支持"""
if not evidence:
return False
evidence_text = "\n".join([e["text"][:300] for e in evidence])
prompt = f"""根据以下证据，判断声明是否得到支持。
声明：{claim}
证据：
{evidence_text}
请回答：SUPPORTED（有支持）或 NOT_SUPPORTED（无支持）
只返回判断结果。"""
result = await self.llm.chat(messages=[{"role": "user", "content": prompt}])
return "SUPPORTED" in result.upper()
第九章：RAG评估框架
9.1 RAGAS评估
# pip install ragas
from ragas import evaluate
from ragas.metrics import (
faithfulness, # 忠实度
answer_relevancy, # 答案相关性
context_precision, # 上下文精确度
context_recall # 上下文召回率
)
from datasets import Dataset
class RAGEvaluator:
"""RAG系统评估器"""
def evaluate_with_ragas(self, eval_data: list) -> dict:
"""使用RAGAS框架评估"""
dataset = Dataset.from_list(eval_data)
result = evaluate(
dataset=dataset,
metrics=[
faithfulness,
answer_relevancy,
context_precision,
context_recall
]
)
return {
"faithfulness": result["faithfulness"],
"answer_relevancy": result["answer_relevancy"],
"context_precision": result["context_precision"],
"context_recall": result["context_recall"],
"overall": result["faithfulness"] * 0.3 + result["answer_relevancy"] * 0.3 +
result["context_precision"] * 0.2 + result["context_recall"] * 0.2
}
def evaluate_retrieval(self, queries: list, retrieved_docs: list,
ground_truth: list) -> dict:
"""评估检索质量"""
hits = 0
mrr_sum = 0
for i, (retrieved, relevant) in enumerate(zip(retrieved_docs, ground_truth)):
retrieved_ids = [d["id"] for d in retrieved]
# Hit@K
if any(doc_id in relevant for doc_id in retrieved_ids[:5]):
hits += 1
# MRR
for rank, doc_id in enumerate(retrieved_ids[:10]):
if doc_id in relevant:
mrr_sum += 1 / (rank + 1)
break
n = len(queries)
return {
"hit_at_5": hits / n,
"mrr_at_10": mrr_sum / n
}
第十章：完整RAG Pipeline实战
10.1 端到端RAG系统
class EnterpriseRAG:
"""企业级RAG系统完整实现"""
def __init__(self, config: dict):
self.config = config
# 初始化各组件
self.document_parser = DocumentParser()
self.chunker = ParentChildChunker(
child_size=config.get("child_size", 256),
parent_size=config.get("parent_size", 1024)
)
self.embedding_model = SentenceTransformer(config.get("embedding_model", "BAAI/bge-large-zh-v1.5"))
self.vector_store = MilvusVectorStore()
self.reranker = Reranker()
self.query_processor = QueryProcessor(llm_client=None)
self.validator = HallucinationValidator(llm_client=None, embedding_model=self.embedding_model)
self.memory = ConversationMemory(max_turns=5)
async def ingest(self, file_path: str):
"""文档摄入流程"""
# 1. 解析文档
content = self.document_parser.parse(file_path)
# 2. 分块
chunks = self.chunker.chunk(content)
# 3. 生成嵌入
child_texts = [c["text"] for c in chunks["children"]]
embeddings = self.embedding_model.encode(child_texts)
# 4. 存入向量数据库
self.vector_store.insert(
ids=[c["id"] for c in chunks["children"]],
texts=child_texts,
embeddings=embeddings.tolist(),
metadata=[c["metadata"] for c in chunks["children"]]
)
# 5. 存储父块映射
self._store_parent_mapping(chunks["parents"])
return len(chunks["children"])
async def query(self, question: str, session_id: str = None) -> dict:
"""查询流程"""
# 1. 获取对话历史
history = self.memory.get_history(session_id)
# 2. 查询处理
rewritten = await self.query_processor.rewrite_query(question, history)
# 3. 混合检索
query_embedding = self.embedding_model.encode([rewritten])[0]
raw_results = self.vector_store.search(
query_embedding.tolist(), top_k=20
)
# 4. 重排序
reranked = self.reranker.rerank(rewritten, raw_results, top_k=5)
# 5. 获取父块上下文
parent_ids = list(set(r["metadata"].get("parent_id") for r in reranked))
context = self._get_parent_context(parent_ids)
# 6. 生成回答
answer = await self._generate(question, context, history)
# 7. 幻觉验证
validation = await self.validator.validate(question, answer, reranked)
# 8. 更新对话记忆
self.memory.add(session_id, question, answer)
return {
"answer": answer,
"sources": [{"text": r["text"][:200], "score": r.get("rerank_score", 0)} for r in reranked],
"factuality_score": validation["factuality_score"],
"is_faithful": validation["is_faithful"]
}
async def _generate(self, question: str, context: str, history: list) -> str:
"""生成回答"""
history_text = ""
if history:
history_text = "\n".join([
f"用户：{h['user']}\n助手：{h['assistant']}"
for h in history[-3:]
])
prompt = f"""基于以下上下文信息回答用户的问题。如果上下文中没有相关信息，请明确说明无法回答。
上下文：
{context}
对话历史：
{history_text}
用户问题：{question}
请给出准确、详细的回答，并在回答中标注信息来源。"""
return await self.llm.chat(messages=[{"role": "user", "content": prompt}])
第十一章：大规模RAG系统运维
11.1 索引更新策略
class IndexManager:
"""索引管理器"""
def __init__(self, vector_store, embedding_model):
self.vector_store = vector_store
self.embedding_model = embedding_model
async def incremental_update(self, new_documents: list):
"""增量更新索引"""
for doc in new_documents:
# 检查文档是否已存在
existing = await self._check_exists(doc["id"])
if existing:
# 文档已更新，删除旧数据后重新索引
await self._delete_document(doc["id"])
# 索引新文档
await self._index_document(doc)
async def rebuild_index(self, all_documents: list):
"""全量重建索引"""
# 创建临时集合
temp_name = f"{self.collection_name}_temp"
self.vector_store.create_collection(temp_name)
# 批量索引
batch_size = 1000
for i in range(0, len(all_documents), batch_size):
batch = all_documents[i:i+batch_size]
embeddings = self.embedding_model.encode([d["text"] for d in batch])
self.vector_store.insert(
ids=[d["id"] for d in batch],
texts=[d["text"] for d in batch],
embeddings=embeddings.tolist(),
metadata=[d.get("metadata", {}) for d in batch]
)
# 原子切换
self._swap_collection(self.collection_name, temp_name)
11.2 性能优化建议
- 缓存策略：对高频查询结果进行缓存
- 异步处理：全文档摄入采用异步流水线
- 批量嵌入：批量调用embedding模型减少API调用次数
- 预加载：向量数据库索引预加载到内存
- 连接池：数据库和API连接复用
最佳实践总结
- 文档预处理是关键：投入足够精力优化文档解析和分块策略
- 混合检索优于单一检索：结合稠密和稀疏检索提升召回率
- 重排序必不可少：Cross-Encoder重排序能显著提升精确度
- 查询改写提升体验：LLM驱动的查询改写能理解用户真实意图
- 幻觉检测保障质量：生产环境必须有事实性验证机制
- 持续评估迭代：建立评估体系，定期评估和优化系统
- 权限控制不可忽视：企业场景必须考虑文档级权限控制
总结
本教程系统讲解了企业级RAG系统的完整技术栈，从文档解析、分块策略、向量数据库、混合检索、重排序、查询改写到幻觉检测和评估框架。RAG系统的核心在于将正确的上下文以正确的方式传递给LLM，每个环节都直接影响最终效果。
随着技术的发展，Agentic RAG、GraphRAG、多模态RAG等新范式不断涌现，RAG正在从简单的检索增强向智能化、自适应的方向演进。掌握本教程的核心技术，将为构建生产级AI应用打下坚实基础。
