为什么需要 RAG
大语言模型虽然知识丰富，但存在知识截止和幻觉问题。RAG（Retrieval-Augmented Generation）通过将外部知识库与大模型结合，让模型在回答时参考真实文档，显著提升回答的准确性和时效性。
在企业级应用场景中，RAG 系统可以接入内部文档库、知识库和数据库，使大模型能够基于企业私有数据生成精准回答。这对于客服系统、智能问答、文档检索等场景尤为重要。相比纯微调方案，RAG 具有数据更新快、成本低、可追溯等优势。
// SECTION 02技术架构选型
本次实战我们选择以下技术栈，每个组件都经过严格评估，确保在生产环境中具备高可用性和可扩展性：
- 框架: LangChain 0.1.x — 提供完整的 RAG 工具链和抽象层
- 向量数据库: Milvus 2.3 — 支持分布式部署的高性能向量检索引擎
- Embedding 模型: BGE-large-zh-v1.5 — 中文语义理解表现优异
- 大模型: GPT-4 / Qwen-72B — 根据预算和延迟需求灵活选择
- 部署方案: Docker Compose + Kubernetes — 从开发到生产无缝过渡
文档切分策略
文档切分直接影响检索质量。过大的切分块会引入噪声，过小则可能丢失上下文。我们采用递归字符切分器，经过多轮实验将 chunk_size 设为 512，overlap 设为 64，在中文文档场景下取得了最佳检索效果：
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.document_loaders import PyPDFLoader, UnstructuredMarkdownLoader
import os
def load_and_split_documents(doc_dir: str):
"""加载并切分多种格式的文档"""
documents = []
for filename in os.listdir(doc_dir):
filepath = os.path.join(doc_dir, filename)
if filename.endswith('.pdf'):
loader = PyPDFLoader(filepath)
elif filename.endswith('.md'):
loader = UnstructuredMarkdownLoader(filepath)
else:
continue
documents.extend(loader.load())
# 递归字符切分器：按段落 > 换行 > 句号 > 逗号 > 空格 优先级切分
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=512,
chunk_overlap=64,
separators=["\n\n", "\n", "。", "，", " "],
length_function=len,
)
chunks = text_splitter.split_documents(documents)
print(f"共加载 {len(documents)} 个文档，切分为 {len(chunks)} 个文档块")
return chunks
# 使用示例
chunks = load_and_split_documents("./docs")向量检索优化
Milvus 支持多种索引类型。对于百万级文档，推荐使用 IVF_FLAT 索引，它在检索精度和速度之间取得了良好平衡。以下代码展示了如何创建 Collection 并配置索引参数：
from pymilvus import (
connections, Collection, FieldSchema,
CollectionSchema, DataType, utility
)
def create_rag_collection(collection_name: str = "rag_docs"):
"""创建 Milvus Collection 用于 RAG 检索"""
connections.connect(host="localhost", port="19530")
# 定义字段
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=65535),
FieldSchema(name="source", dtype=DataType.VARCHAR, max_length=512),
FieldSchema(name="chunk_index", dtype=DataType.INT64),
]
schema = CollectionSchema(fields=fields, description="RAG 文档向量集合")
collection = Collection(name=collection_name, schema=schema)
# 创建 IVF_FLAT 索引
index_params = {
"metric_type": "COSINE",
"index_type": "IVF_FLAT",
"params": {"nlist": 1024}
}
collection.create_index(
field_name="embedding",
index_params=index_params
)
print(f"Collection '{collection_name}' 创建完成，索引已构建")
return collection
# 检索函数
def search_similar(collection, query_embedding, top_k=5):
"""执行相似度检索"""
collection.load()
results = collection.search(
data=[query_embedding],
anns_field="embedding",
param={"metric_type": "COSINE", "params": {"nprobe": 16}},
limit=top_k,
output_fields=["text", "source", "chunk_index"]
)
return results[0]// SECTION 03提示：生产环境中建议为每个文档块添加来源元数据（source、page_number、chunk_index），方便溯源和去重。同时可以使用 Metadata Filter 在检索时过滤特定来源的文档。
部署与监控
使用 Docker Compose 部署 Milvus，配合 Prometheus + Grafana 实现检索延迟和吞吐量的实时监控。建议关注 P99 延迟指标，确保用户体验。以下是完整的部署架构说明和关键配置建议。
在部署阶段，需要特别注意 Milvus 的 etcd 和 MinIO 依赖配置。etcd 用于元数据管理，MinIO 用于对象存储。建议在生产环境中使用独立的 etcd 集群，并配置持久化存储卷。
建议将 Milvus 部署为分布式模式（Cluster），至少配置 3 个 Query Node 和 2 个 Data Node。使用 Kubernetes 编排时，可通过 Horizontal Pod Autoscaler 根据检索 QPS 自动扩缩容。同时建议开启 Milvus 的内置监控面板，并与 Prometheus 联邦对接。
当文档量超过百万级时，需要注意以下几点：1) 合理设置 IVF_FLAT 索引的 nlist 参数（推荐值为 sqrt(N)，N 为向量总数）；2) 检索时 nprobe 参数越大精度越高但延迟越大，推荐设为 nlist 的 1%-5%；3) 定期执行 compaction 操作清理已删除的向量数据，避免存储空间膨胀。