8. 向量库存储

# -*- coding: utf-8 -*-""" @Created on ： 2026/6/16 17:00 @creator ： er_nao @File ：day_93.py @Description ：向量库存储 """# ====================== 导入所有依赖库 ======================importfaissimportnumpyasnpimportjsonimportosfromtypingimportList,Dict# ====================== 核心配置区：替换成你自己的项目路径 ======================# 1. 文本向量结果文件路径VECTOR_JSON_PATH=r"C:\\Users\\hp\\Desktop\\NLP学习数据\\Day91_文本转向量结果.json"# 2. faiss索引文件保存路径（后续检索直接加载这个文件）INDEX_SAVE_PATH=r"F:\\RAG-Learning-Project\\vector-database\\rag_vector_index.faiss"# 3. 元数据文件保存路径（和索引一一对应，存储原文内容等信息）META_DATA_SAVE_PATH=r"F:\\RAG-Learning-Project\\vector-database\\rag_meta_data.json"# 4. 向量维度（和你之前生成的向量维度一致，通义模型是1536，BGE是384）VECTOR_DIM=1536# ==================================================================================# ====================== 核心函数1：加载文本向量结果 ======================defload_vector_result(json_path:str)->List[Dict]:""" 加载之前生成的文本向量结果JSON文件 :param json_path: 向量结果文件路径 :return: 结构化的向量结果列表 """# 校验文件是否存在ifnotos.path.exists(json_path):print(f"错误：向量结果文件不存在，路径：{json_path}")return[]try:withopen(json_path,"r",encoding="utf-8")asf:vector_result=json.load(f)print(f" 向量结果加载成功，共{len(vector_result)}个文本块向量")returnvector_resultexceptExceptionase:print(f"加载失败错误信息：{str(e)}")return[]# ====================== 核心函数2：创建faiss索引 ======================defcreate_faiss_index(vector_dim:int)->faiss.IndexFlatL2:""" 创建faiss索引，零基础首选IndexFlatL2，100%匹配精度 :param vector_dim: 向量维度，必须和你生成的向量维度完全一致 :return: 创建好的faiss索引对象 """# 创建L2距离的索引，最适合零基础、小项目场景index=faiss.IndexFlatL2(vector_dim)print(f" faiss索引创建成功，向量维度：{index.d}")returnindex# ====================== 核心函数3：批量向量入库 ======================defbatch_add_vector_to_index(index:faiss.IndexFlatL2,vector_result:List[Dict])->bool:ifnotindexornotvector_result:print("错误：索引或向量结果为空，无法入库")returnFalse# 提取所有向量，转换成faiss要求的numpy数组格式vector_list=[]foriteminvector_result:# 提取向量列表，转成float32类型（faiss要求必须是float32）vector=np.array(item["embedding_vector"],dtype=np.float32)# 校验向量维度是否和索引一致iflen(vector)!=index.d:print(f" 错误：向量维度不匹配，索引维度{index.d}，当前向量维度{len(vector)}")returnFalsevector_list.append(vector)# 把向量列表转换成二维numpy数组，faiss要求的输入格式# 形状：(向量数量, 向量维度)，比如(100, 1536)vector_np=np.array(vector_list)print(f" 待入库向量形状：{vector_np.shape}，共{vector_np.shape[0]}个向量")# 批量添加向量到索引中try:index.add(vector_np)print(f" 批量向量入库完成，索引中当前总向量数：{index.ntotal}")returnTrueexceptExceptionase:print(f" 向量入库失败，错误信息：{str(e)}")returnFalse# ====================== 核心函数4：索引持久化存储到本地 ======================defsave_index_to_local(index:faiss.IndexFlatL2,save_path:str)->bool:""" 把faiss索引存储到本地硬盘文件，实现持久化，后续可直接加载使用 :param index: 完成入库的faiss索引对象 :param save_path: 索引文件保存路径 :return: 保存成功返回True，失败返回False """# 自动创建保存路径的文件夹save_dir=os.path.dirname(save_path)ifsave_dirandnotos.path.exists(save_dir):os.makedirs(save_dir)print(f"自动创建索引保存文件夹：{save_dir}")# 把索引写入本地文件try:faiss.write_index(index,save_path)print(f"索引持久化存储完成，文件路径：{save_path}")print(f"存储的索引总向量数：{index.ntotal},维度：{index.d}")returnTrueexceptExceptionase:print(f"索引存储失败，错误信息：{str(e)}")returnFalse# ====================== 核心函数5：元数据结构化存储 ======================defsave_meta_data(vector_result:List[Dict],save_path:str)->bool:""" 存储元数据，和索引一一对应，包含原文内容、ID等信息 faiss索引只存向量数字，必须单独存元数据，检索后才能找到原文内容 :param vector_result: 加载好的向量结果列表 :param save_path: 元数据文件保存路径 :return: 保存成功返回True，失败返回False """# 自动创建保存路径的文件夹save_dir=os.path.dirname(save_path)ifsave_dirandnotos.path.exists(save_dir):os.makedirs(save_dir)# 结构化元数据，只保留需要的内容，和索引的顺序完全一致meta_data_list=[]foridx,iteminenumerate(vector_result):meta_data_list.append({"index_id":idx,"chunk_id":item["chunk_id"],"paragraph_id":item.get("paragraph_id",0),"chunk_content":item["chunk_content"],"chunk_length":item.get("chunk_length",0),"create_time":item.get("create_time","")})# 保存元数据到JSON文件try:withopen(save_path,"w",encoding="utf-8")asf:json.dump(meta_data_list,f,ensure_ascii=False,indent=2)print(f" 元数据存储完成，文件路径：{save_path}")print(f" 共存储{len(meta_data_list)}条元数据，和索引一一对应")returnTrueexceptExceptionase:print(f" 元数据存储失败，错误信息：{str(e)}")returnFalse# ====================== 主函数：一键执行向量库存储全流程 ======================defmain_vector_storage_pipeline():print("="*80)print(" Day93 向量库存储全流程开始执行")print("="*80)# 1. 加载文本向量结果vector_result=load_vector_result(VECTOR_JSON_PATH)ifnotvector_result:print(" 全流程终止：向量结果加载失败")return# 2. 创建faiss索引index=create_faiss_index(VECTOR_DIM)# 3. 批量向量入库add_success=batch_add_vector_to_index(index,vector_result)ifnotadd_success:print(" 全流程终止：向量入库失败")return# 4. 索引持久化存储到本地save_index_success=save_index_to_local(index,INDEX_SAVE_PATH)ifnotsave_index_success:print(" 全流程终止：索引存储失败")return# 5. 元数据结构化存储save_meta_success=save_meta_data(vector_result,META_DATA_SAVE_PATH)ifnotsave_meta_success:print(" 全流程终止：元数据存储失败")returnprint("\n"+"="*80)print(" Day93 向量库存储全流程执行完成！")print("="*80)print(f" 最终成果：")print(f" - 成功入库向量总数：{index.ntotal}个")print(f" - 向量维度：{index.d}维")print(f" - 索引文件路径：{INDEX_SAVE_PATH}")print(f" - 元数据文件路径：{META_DATA_SAVE_PATH}")print(f" - 输出的向量库可直接用于后续的相似度检索环节")# ====================== 程序入口：一键运行全流程 ======================if__name__=="__main__":main_vector_storage_pipeline()