资讯详情

深入了解 · 专业解读

8. 向量库存储

发布时间：2026/6/23 2:28:20
8. 向量库存储 # -*- coding: utf-8 -*- Created on 2026/6/16 17:00 creator er_nao File day_93.py Description 向量库存储 # 导入所有依赖库 importfaissimportnumpyasnpimportjsonimportosfromtypingimportList,Dict# 核心配置区替换成你自己的项目路径 # 1. 文本向量结果文件路径VECTOR_JSON_PATHrC:\\Users\\hp\\Desktop\\NLP学习数据\\Day91_文本转向量结果.json# 2. faiss索引文件保存路径后续检索直接加载这个文件INDEX_SAVE_PATHrF:\\RAG-Learning-Project\\vector-database\\rag_vector_index.faiss# 3. 元数据文件保存路径和索引一一对应存储原文内容等信息META_DATA_SAVE_PATHrF:\\RAG-Learning-Project\\vector-database\\rag_meta_data.json# 4. 向量维度和你之前生成的向量维度一致通义模型是1536BGE是384VECTOR_DIM1536# # 核心函数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,encodingutf-8)asf:vector_resultjson.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索引零基础首选IndexFlatL2100%匹配精度 :param vector_dim: 向量维度必须和你生成的向量维度完全一致 :return: 创建好的faiss索引对象 # 创建L2距离的索引最适合零基础、小项目场景indexfaiss.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要求必须是float32vectornp.array(item[embedding_vector],dtypenp.float32)# 校验向量维度是否和索引一致iflen(vector)!index.d:print(f 错误向量维度不匹配索引维度{index.d}当前向量维度{len(vector)})returnFalsevector_list.append(vector)# 把向量列表转换成二维numpy数组faiss要求的输入格式# 形状(向量数量, 向量维度)比如(100, 1536)vector_npnp.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_diros.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_diros.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,encodingutf-8)asf:json.dump(meta_data_list,f,ensure_asciiFalse,indent2)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_resultload_vector_result(VECTOR_JSON_PATH)ifnotvector_result:print( 全流程终止向量结果加载失败)return# 2. 创建faiss索引indexcreate_faiss_index(VECTOR_DIM)# 3. 批量向量入库add_successbatch_add_vector_to_index(index,vector_result)ifnotadd_success:print( 全流程终止向量入库失败)return# 4. 索引持久化存储到本地save_index_successsave_index_to_local(index,INDEX_SAVE_PATH)ifnotsave_index_success:print( 全流程终止索引存储失败)return# 5. 元数据结构化存储save_meta_successsave_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()

相关新闻