大模型RAG检索增强生成架构与向量数据库Milvus语义检索实战配置

RAG检索增强生成架构核心原理

RAG（Retrieval-Augmented Generation）检索增强生成将大语言模型的生成能力与外部知识库的检索能力结合，解决大模型幻觉、知识时效性和领域专有信息缺失的问题。RAG架构分为离线索引和在线检索两个阶段：离线阶段将文档切分为文本块，通过Embedding模型生成向量表示，存入向量数据库；在线阶段将用户查询向量化，在向量数据库中做相似度检索，将检索结果作为上下文拼接到Prompt中送入大模型生成回答。

与直接微调大模型相比，RAG的优势在于知识更新成本低——只需更新向量数据库中的文档，无需重新训练模型；同时保留了模型本身的推理能力，通过检索结果约束生成范围，减少幻觉。在企业知识问答、法律文书检索、医疗辅助诊断等场景中，RAG已成为主流的技术方案。

向量数据库是RAG架构的核心组件，负责存储文档向量并提供高效的相似度检索。Milvus作为开源向量数据库，支持十亿级向量规模的近似最近邻搜索（ANN），提供IVF_FLAT、IVF_SQ8、HNSW等多种索引类型，适用于大规模RAG生产环境部署。

Milvus向量数据库部署与Collection创建

Milvus支持单机模式和集群模式部署。开发环境使用Docker快速启动单机实例：

# docker-compose.yml
version: '3.5'
services:
  etcd:
    image: quay.io/coreos/etcd:v3.5.5
    environment:
      - ETCD_AUTO_COMPACTION_MODE=revision
      - ETCD_AUTO_COMPACTION_RETENTION=1000
    volumes:
      - etcd_data:/etcd
    command: etcd -advertise-client-urls=http://127.0.0.1:2379 -listen-client-urls http://0.0.0.0:2379 --data-dir /etcd

  milvus:
    image: milvusdb/milvus:v2.4.0
    command: ["milvus", "run", "standalone"]
    environment:
      ETCD_ENDPOINTS: etcd:2379
    ports:
      - "19530:19530"
      - "9091:9091"
    depends_on:
      - etcd

volumes:
  etcd_data:

启动后使用pymilvus客户端创建Collection并定义Schema：

from pymilvus import MilvusClient

client = MilvusClient(uri="http://localhost:19530")

# 创建Collection，定义字段结构
client.create_collection(
    collection_name="knowledge_base",
    dimension=768,  # Embedding向量维度，与模型输出一致
    metric_type="COSINE",  # 余弦相似度
    index_type="HNSW",  # 混合导航小世界图索引
    index_params={
        "M": 16,           # 每个节点的最大连接数
        "efConstruction": 200  # 构建索引时的搜索宽度
    }
)

# 插入向量数据
data = [
    {"id": i, "vector": embedding_vec, "text": chunk_text, "source": doc_name}
    for i, (embedding_vec, chunk_text, doc_name) in enumerate(documents)
]
client.insert(collection_name="knowledge_base", data=data)

HNSW索引在检索精度和速度之间取得了较好的平衡，M值设置为16、efConstruction设置为200适用于大多数RAG场景。对于超大规模数据集（亿级以上），可考虑IVF_SQ8索引配合量化压缩降低内存占用。

文档切分策略与Embedding模型选择

文档切分质量直接影响RAG系统的检索准确率。常见的切分策略包括固定长度切分、按段落切分和语义切分。固定长度切分实现简单但可能截断语义完整的段落；按段落切分保留了文档结构，但段落长度不均匀；语义切分通过模型识别语义边界，效果最好但计算成本高。

实际工程中推荐使用滑动窗口切分，设置chunk_size=512、overlap=50，在保持上下文连贯性的同时控制每个文本块的长度：

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=50,
    separators=["\n\n", "\n", "。", "！", "？", "；", " ", ""]
)

chunks = splitter.split_text(document_text)
# 每个chunk通过Embedding模型生成向量
embeddings = embedding_model.encode(chunks)

Embedding模型的选择需要平衡精度、速度和维度。BGE-large-zh-v1.5在中文语义检索任务上表现优异，输出1024维向量；text2vec-base-chinese轻量高效，输出768维向量，适合资源受限的部署环境。使用商API时OpenAI text-embedding-3-small输出1536维向量，通用性强但存在数据出境合规问题。

检索增强生成Pipeline完整实现

完整的RAG Pipeline包含查询向量化、向量检索、上下文构造和生成四个步骤。以下代码实现端到端的RAG问答流程：

import torch
from transformers import AutoTokenizer, AutoModelForCausalLM
from pymilvus import MilvusClient

class RAGPipeline:
    def __init__(self, collection_name, embedding_model, llm_model_path):
        self.milvus = MilvusClient(uri="http://localhost:19530")
        self.collection_name = collection_name
        self.embedding_model = embedding_model
        self.tokenizer = AutoTokenizer.from_pretrained(llm_model_path)
        self.llm = AutoModelForCausalLM.from_pretrained(
            llm_model_path,
            torch_dtype=torch.float16,
            device_map="auto"
        )

    def retrieve(self, query, top_k=5):
        # 查询向量化
        query_vec = self.embedding_model.encode([query])[0]
        # 向量检索
        results = self.milvus.search(
            collection_name=self.collection_name,
            data=[query_vec],
            limit=top_k,
            output_fields=["text", "source"]
        )
        return [hit["entity"]["text"] for hit in results[0]]

    def generate(self, query, context_chunks):
        context = "\n\n".join(context_chunks)
        prompt = f"请根据以下参考资料回答问题。\n\n参考资料：\n{context}\n\n问题：{query}\n\n回答："
        inputs = self.tokenizer(prompt, return_tensors="pt").to(self.llm.device)
        output = self.llm.generate(**inputs, max_new_tokens=512, temperature=0.3)
        return self.tokenizer.decode(output[0], skip_special_tokens=True)

    def answer(self, query):
        chunks = self.retrieve(query)
        return self.generate(query, chunks)

# 使用示例
rag = RAGPipeline("knowledge_base", bge_model, "Qwen/Qwen2.5-7B-Instruct")
response = rag.answer("MySQL主从复制延迟如何排查？")
print(response)

RAG系统优化与调优方向

RAG系统上线后的优化集中在三个方向。检索质量优化方面，引入混合检索（Hybrid Search）结合BM25关键词检索和向量语义检索，通过RRF（Reciprocal Rank Fusion）算法融合排序结果，提升召回率。引入重排序模型（Reranker）对检索结果二次排序，将最相关的文档排在前面。常用Reranker包括bge-reranker-large和Cohere Rerank API。

生成质量优化方面，调整Prompt模板的结构化程度，加入引用标记要求模型标注信息来源。设置temperature=0.3降低生成随机性，max_new_tokens控制输出长度。对生成结果做事实性校验，检测生成内容与检索上下文的一致性。

性能优化方面，使用连接池管理Milvus客户端连接，批量插入替代单条插入提升索引构建速度。对高频查询结果做缓存，减少重复检索和生成开销。在多用户并发场景下，使用异步IO处理检索请求，配合vLLM等推理框架实现高吞吐生成。