RAG架构原理与核心组件
RAG（Retrieval-Augmented Generation）检索增强生成是解决大语言模型知识时效性和幻觉问题的主流方案。传统大模型依赖训练数据中的参数化知识，无法访问私有文档或实时信息。RAG通过在生成前从外部知识库检索相关文档片段，将检索结果作为上下文注入提示词，使模型基于真实文档生成回答。
RAG系统的核心组件包括文档处理模块、向量嵌入模型、向量数据库、检索器和生成模型。文档处理模块负责解析PDF、Word、HTML等格式文件并切分为语义块。向量嵌入模型将文本块映射为高维向量。向量数据库存储向量并支持相似度检索。检索器根据用户查询召回Top-K相关文档。生成模型将检索结果与用户问题拼接后生成最终回答。
向量嵌入模型选型与文本向量化
向量嵌入模型直接影响检索质量。常用的嵌入模型包括OpenAI text-embedding-3、BGE-large-zh、E5-mistral等。选择时需关注向量维度、最大输入长度、语言支持和推理速度。中文场景推荐BGE-large-zh-v1.5，维度1024，在C-MTEB基准上表现优异。
使用HuggingFace Transformers加载嵌入模型的实现：
from sentence_transformers import SentenceTransformer
import numpy as np
model = SentenceTransformer('BAAI/bge-large-zh-v1.5')
def embed_texts(texts, batch_size=32):
'''批量文本向量化，返回归一化向量'''
embeddings = model.encode(
texts,
batch_size=batch_size,
normalize_embeddings=True,
show_progress_bar=True
)
return embeddings
# 示例：将文档块向量化
chunks = [
"RAG通过检索外部知识库增强大模型生成能力",
"向量数据库支持高效的近似最近邻搜索",
"文档分块策略影响检索粒度和召回率"
]
vectors = embed_texts(chunks)
print(f"向量形状: {vectors.shape}") # (3, 1024)向量归一化（L2标准化）后可使用内积计算相似度，避免向量长度差异对排序的影响。批处理时控制batch_size防止显存溢出，BERT类模型建议batch_size不超过64。
文档分块策略与语义切分
文档分块是RAG效果的决定性因素。固定长度切分简单粗暴，可能截断语义。推荐使用滑动窗口加重叠区域的策略，重叠部分保持上下文连续性。更高级的做法是按段落、标题层级进行结构化切分。
import re
from typing import List
def semantic_chunk(text: str, chunk_size=512, overlap=64) -> List[str]:
'''按句子边界滑动窗口切分文档'''
# 按句子分割
sentences = re.split(r'(?<=[。！？\n])', text)
sentences = [s.strip() for s in sentences if s.strip()]
chunks = []
current_chunk = ""
for sent in sentences:
# 添加句子后若超出chunk_size，保存当前块并保留overlap
if len(current_chunk) + len(sent) > chunk_size and current_chunk:
chunks.append(current_chunk)
# 保留重叠部分
overlap_text = current_chunk[-overlap:] if len(current_chunk) > overlap else current_chunk
current_chunk = overlap_text + sent
else:
current_chunk += sent
if current_chunk:
chunks.append(current_chunk)
return chunks
# 按Markdown标题层级切分
def chunk_by_heading(text: str) -> List[str]:
'''按##标题切分文档，保持章节完整性'''
sections = re.split(r'(?=^#{1,3}\s)', text, flags=re.MULTILINE)
return [s.strip() for s in sections if s.strip()]chunk_size通常设置256-1024字符。较小的块提高检索精度但可能丢失上下文，较大的块保留完整语义但降低检索粒度。建议根据文档类型调整：技术文档按标题切分，长文按512字符加64重叠切分，问答对每个QA作为独立块。
向量数据库对比与Milvus部署
主流向量数据库包括Milvus、Chroma、Qdrant、Weaviate和FAISS。Milvus支持十亿级向量，提供丰富的索引类型（IVF_FLAT、IVF_SQ8、HNSW），适合生产环境。Chroma轻量级，适合原型开发。Qdrant用Rust编写，性能优秀。FAISS是Meta开源的库，适合本地部署。
Milvus Docker部署与集合创建：
# docker-compose.yml 部署Milvus standalone
# docker compose up -d
from pymilvus import connections, FieldSchema, CollectionSchema, DataType, Collection
# 连接Milvus
connections.connect(host="localhost", port="19530")
# 定义集合Schema
fields = [
FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=1024),
FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=4096),
FieldSchema(name="source", dtype=DataType.VARCHAR, max_length=256),
]
schema = CollectionSchema(fields, description="RAG知识库向量集合")
collection = Collection("rag_knowledge_base", schema)
# 创建HNSW索引
index_params = {
"index_type": "HNSW",
"metric_type": "IP", # 内积（向量已归一化时等价于余弦相似度）
"params": {"M": 16, "efConstruction": 256}
}
collection.create_index("embedding", index_params)
# 插入向量数据
entities = [
{"embedding": vectors[i].tolist(), "text": chunks[i], "source": "doc1.pdf"}
for i in range(len(chunks))
]
collection.insert(entities)
collection.load()检索增强生成管线实现
完整的RAG检索管线包括查询向量化、向量检索、结果重排序和上下文组装四个阶段。重排序（Reranking）使用交叉编码器对检索结果精排，显著提升相关性。
from pymilvus import Collection
def rag_retrieve(query: str, top_k=5) -> list:
'''RAG检索管线'''
# 1. 查询向量化
query_vec = embed_texts([query])[0]
# 2. 向量检索
collection = Collection("rag_knowledge_base")
collection.load()
results = collection.search(
data=[query_vec.tolist()],
anns_field="embedding",
param={"metric_type": "IP", "params": {"ef": 64}},
limit=top_k * 3, # 多召回用于重排序
output_fields=["text", "source"]
)
# 3. 交叉编码器重排序
candidates = [(hit.entity.get("text"), hit.score, hit.entity.get("source"))
for hit in results[0]]
reranked = cross_encoder_rerank(query, candidates, top_k)
return reranked
def cross_encoder_rerank(query, candidates, top_k):
'''使用BGE-reranker重排序'''
from sentence_transformers import CrossEncoder
reranker = CrossEncoder('BAAI/bge-reranker-large')
pairs = [(query, c[0]) for c in candidates]
scores = reranker.predict(pairs)
ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
return [{"text": r[0][0], "score": float(r[1]), "source": r[0][2]}
for r in ranked[:top_k]]
def generate_answer(query, retrieved_docs):
'''组装上下文调用大模型生成'''
context = "\n\n".join([d["text"] for d in retrieved_docs])
prompt = f'''基于以下检索到的文档内容回答问题。如果文档中没有相关信息，请说明无法回答。
文档内容：
{context}
问题：{query}
回答：'''
# 调用LLM API生成回答
return call_llm(prompt)RAG系统评估指标与调优
RAG系统评估从检索质量和生成质量两个维度进行。检索评估指标包括召回率（Recall@K）、精确率（Precision@K）和MRR（Mean Reciprocal Rank）。生成评估关注答案相关性、忠实度（Faithfulness，答案是否基于检索内容）和正确性。
常见调优方向：分块策略调整、嵌入模型升级、索引参数调优（HNSW的M和ef参数）、查询改写（Query Rewriting）和混合检索（向量加BM25关键词检索融合）。混合检索通过倒数排名融合（RRF）合并两路检索结果，兼顾语义匹配和关键词匹配：
def hybrid_search(query, top_k=5):
'''向量检索 + BM25关键词检索融合'''
# 向量检索结果
vec_results = vector_search(query, top_k * 2)
# BM25检索结果
bm25_results = bm25_search(query, top_k * 2)
# RRF倒数排名融合
rrf_k = 60
scores = {}
for rank, doc in enumerate(vec_results):
scores[doc["text"]] = scores.get(doc["text"], 0) + 1 / (rank + rrf_k)
for rank, doc in enumerate(bm25_results):
scores[doc["text"]] = scores.get(doc["text"], 0) + 1 / (rank + rrf_k)
return sorted(scores.items(), key=lambda x: x[1], reverse=True)[:top_k]RAG系统的效果调优是迭代过程，建议建立标准评估数据集，每次参数变更后量化评估，避免凭直觉调整导致的回归。生产环境中持续收集用户反馈（点赞/点踩），构建评估闭环。
原创文章，作者：小编，如若转载，请注明出处：https://www.yunthe.com/rag-jian-suo-zeng-qiang-sheng-cheng-xi-tong-da-jian-shi/
