Agent开发实战-持久化保存与调取

observer？

390人浏览 · 2026-04-09 22:46:55

observer？ · 2026-04-09 22:46:55 发布

RAG系统技术流程详解：向量化、持久化、双路查询与重排序

在构建高效的RAG（Retrieval-Augmented Generation）系统时，技术流程的设计至关重要。本文将详细介绍一个完整的RAG技术流程，包括向量化、持久化、双路查询和重排序四个核心环节。

一、向量化：文本到向量的转换

1. 什么是向量化？

向量化是将文本转换为高维向量表示的过程，使计算机能够理解文本的语义内容。在RAG系统中，向量化是实现语义检索的基础。

2. 模型选择

我们选择BAAI/bge-small-zh-v1.5作为Embedding模型，它是一个专为中文优化的轻量级模型，具有以下优势：

体积小（约120MB），加载速度快
性能优秀，在中文语义理解任务上表现良好
支持本地部署，适合资源有限的环境

3. 实现代码

# 模型本地存储路径
model_path = "./local_models/bge-small-zh-v1.5"

# 加载 Embedding 模型
print("正在加载 Embedding 模型...")
if os.path.exists(model_path):
    print("使用本地模型...")
    embeddings = HuggingFaceEmbeddings(
        model_name=model_path,
        model_kwargs={'device': 'cuda' if torch.cuda.is_available() else 'cpu'}
    )
else:
    print("首次运行，正在下载模型到本地...")
    os.makedirs(os.path.dirname(model_path), exist_ok=True)
    from sentence_transformers import SentenceTransformer
    model = SentenceTransformer("BAAI/bge-small-zh-v1.5")
    model.save(model_path)
    embeddings = HuggingFaceEmbeddings(
        model_name=model_path,
        model_kwargs={'device': 'cuda' if torch.cuda.is_available() else 'cpu'}
    )

4. 技术要点

本地模型存储：避免重复下载，提高加载速度
设备选择：根据硬件情况自动选择CPU或GPU
模型版本控制：确保使用固定版本的模型，保证结果一致性

二、持久化：向量数据的存储

1. 为什么需要持久化？

避免重复向量化：文本向量化是计算密集型操作，重复执行会浪费资源
数据持久性：确保系统重启后数据不丢失
快速检索：直接从本地加载索引，提高检索速度

2. 向量数据库选择

我们选择Chroma作为向量数据库，它具有以下特点：

轻量级，易于部署
支持本地持久化
与LangChain集成良好
适合中小规模数据集

3. 实现代码

# 保存结果到Chroma向量数据库
from langchain_community.vectorstores import Chroma

db_path = "./chroma_db_hybrid"
print(f"\n正在创建向量数据库...")

vector_db = Chroma.from_documents(
    documents=final_docs,
    embedding=embeddings,
    collection_name="ziwei_hybrid",
    persist_directory=db_path
)
vector_db.persist()
print(f"数据库已保存至: {db_path}")

4. 技术要点

存储路径：使用persist_directory参数指定本地存储路径
集合名称：使用collection_name区分不同类型的数据
持久化操作：调用persist()确保数据写入磁盘
文档结构：使用Document对象存储文本和元数据

三、双路查询：向量检索与BM25检索

1. 为什么需要双路查询？

向量检索：基于语义相似度，能理解查询的意图，但对特定名词、专业术语的检索精度可能不足
BM25检索：基于词频，对关键词匹配效果好，特别是对于人名、产品型号等具体词汇
双路结合：取两者之长，提高检索的全面性和准确性

2. 实现代码

def retrieve(self, query, top_k=10):
    """
    混合检索（向量检索 + BM25检索）
    
    Args:
        query: 查询语句
        top_k: 返回结果数量
        
    Returns:
        混合检索结果（去重后的文档列表）
    """
    # 1. 向量检索
    v_results = self.vector_db.similarity_search(query, k=top_k)
    
    # 2. BM25检索
    tokenized_query = list(jieba.cut(query))
    bm25_scores = self.bm25.get_scores(tokenized_query)
    
    # 归一化BM25分数
    if np.max(bm25_scores) != 0:
        bm25_scores = bm25_scores / np.max(bm25_scores)
    
    # 找到BM25得分最高的前k个索引
    top_n_idx = np.argsort(bm25_scores)[-top_k:][::-1]
    b_results = [self.corpus_docs[idx] for idx in top_n_idx if bm25_scores[idx] > 0]
    
    # 3. 合并去重
    # 创建一个字典来存储唯一的文档内容
    unique_docs = {}
    for doc in v_results + b_results:
        if doc.page_content not in unique_docs:
            unique_docs[doc.page_content] = doc
    
    # 转换回列表
    all_candidates = list(unique_docs.values())
    return all_candidates

3. 技术要点

中文分词：使用jieba对查询和文档进行分词，提高BM25检索效果
结果去重：避免重复文档，提高检索效率
分数归一化：确保不同检索方法的分数在同一量级
参数调优：根据实际情况调整top_k参数

四、重排序：提升检索结果质量

1. 为什么需要重排序？

粗排与精排：双路检索是粗排，返回的结果可能包含相关性不高的文档
语义理解：重排序模型能更深入理解查询与文档的语义关系
结果优化：确保最相关的文档排在前面，提高后续生成质量

2. 重排序模型选择

我们选择BAAI/bge-reranker-base作为重排序模型，它具有以下优势：

专为RAG场景优化
理解"问题-文档"对的语义关系
支持中文文本
计算效率适中，适合在线应用

3. 实现代码

def post_process(self, query, candidates, top_n=3):
    """
    重排序检索结果
    
    Args:
        query: 查询语句
        candidates: 检索结果列表
        top_n: 返回前n个结果
        
    Returns:
        重排序后的结果列表
    """
    # 构建模型输入的[问题, 文档]对
    pairs = [[query, doc.page_content] for doc in candidates]
    
    # 计算相关性分数
    scores = self.reranker.predict(pairs)
    
    # 将分数与文档组合并排序
    scored_results = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
    
    # 返回前top_n个结果
    final_results = [item[0] for item in scored_results[:top_n]]
    
    # 打印结果
    print(f"--- Rerank 后的 Top {top_n} 结果 ---")
    for i, (doc, score) in enumerate(scored_results[:top_n]):
        print(f"[{i+1}] 得分: {score:.4f} | 内容: {doc.page_content[:100]}...")
    
    return final_results

4. 技术要点

输入构造：将查询与每个文档组合成[问题, 文档]对
分数计算：使用CrossEncoder模型计算相关性分数
结果排序：按分数降序排列，选择前N个结果
性能优化：只对粗排结果进行重排序，平衡精度与速度

五、全链路集成

1. 完整流程

def query(self, question):
    """
    全链路问答
    
    Args:
        question: 用户问题
        
    Returns:
        最终生成的Prompt和检索到的文档
    """
    # 1. 混合检索
    candidates = self.retrieve(question)
    
    # 2. 重排序
    refined_context = self.post_process(question, candidates)
    
    # 3. 构造Prompt
    context_str = "\n".join([f"资料{i+1}: {c.page_content}" for i, c in enumerate(refined_context)])
    prompt = f"""你是一个基于私有知识库的助手。请严格根据以下参考资料回答问题。
如果资料中没有相关信息，请直接回答"资料中未提及"，不要胡乱猜测。

参考资料：
{context_str}

问题：{question}
回答："""
    
    print("\n--- 最终生成的 Prompt ---")
    print(prompt)
    
    return prompt, refined_context

2. 技术要点

流程集成：将向量化、持久化、双路查询和重排序整合到一个完整的流程中
Prompt构造：将重排序后的结果组织成结构化的Prompt
防幻觉：在Prompt中明确要求模型只基于参考资料回答
结果返回：返回Prompt和检索到的文档，便于后续处理

六、调用示例

# 模型和数据库路径
vector_db_path = "./chroma_db_hybrid"
embedding_model_path = "./local_models/bge-small-zh-v1.5"
reranker_model_path = "./local_models/bge-reranker-base"

# 初始化RAG pipeline
rag_system = Advanced_RAG_Pipeline(
    vector_db_path=vector_db_path,
    embedding_model_path=embedding_model_path,
    reranker_model_path=reranker_model_path
)

# 测试查询
test_questions = [
    "贪狼星的特点是什么？",
    "紫微星在命宫有什么影响？",
    "天机星在事业宫代表什么？"
]

for question in test_questions:
    print(f"\n{'='*80}")
    print(f"测试问题: {question}")
    print(f"{'='*80}")
    prompt, context = rag_system.query(question)
    print(f"\n{'='*80}")