四milvus一个向量数据库实践导入# 1. 初始化设置 MODEL_NAME BAAI/bge-base-en-v1.5 MODEL_PATH ../../models/bge/Visualized_base_en_v1.5.pth DATA_DIR ../../data/C3 COLLECTION_NAME multimodal_demo MILVUS_URI http://localhost:19530 # 2. 定义工具 (编码器和可视化函数) class Encoder: 编码器类用于将图像和文本编码为向量。 def __init__(self, model_name: str, model_path: str): self.model Visualized_BGE(model_name_bgemodel_name, model_weightmodel_path) self.model.eval() def encode_query(self, image_path: str, text: str) - list[float]: with torch.no_grad(): query_emb self.model.encode(imageimage_path, texttext) return query_emb.tolist()[0] def encode_image(self, image_path: str) - list[float]: with torch.no_grad(): query_emb self.model.encode(imageimage_path) return query_emb.tolist()[0]创建Collection# 3. 初始化客户端 print(-- 正在初始化编码器和Milvus客户端...) encoder Encoder(MODEL_NAME, MODEL_PATH) milvus_client MilvusClient(uriMILVUS_URI) # 4. 创建 Milvus Collection print(f\n-- 正在创建 Collection {COLLECTION_NAME}) if milvus_client.has_collection(COLLECTION_NAME): milvus_client.drop_collection(COLLECTION_NAME) print(f已删除已存在的 Collection: {COLLECTION_NAME}) image_list glob(os.path.join(DATA_DIR, dragon, *.png)) if not image_list: raise FileNotFoundError(f在 {DATA_DIR}/dragon/ 中未找到任何 .png 图像。) dim len(encoder.encode_image(image_list[0])) # 定义Schema,建表语句 fields [ # 主键字段设置自增 (auto_idTrue) FieldSchema(nameid, dtypeDataType.INT64, is_primaryTrue, auto_idTrue), # 向量字段维度与模型的输出向量维度一致 FieldSchema(namevector, dtypeDataType.FLOAT_VECTOR, dimdim), # 存储原图像路径的标量字段最大长度限制 512 字符 FieldSchema(nameimage_path, dtypeDataType.VARCHAR, max_length512), ] # 创建集合 Schema schema CollectionSchema(fields, description多模态图文检索) print(Schema 结构:) print(schema) # 创建集合 milvus_client.create_collection(collection_nameCOLLECTION_NAME, schemaschema) print(f成功创建 Collection: {COLLECTION_NAME}) print(Collection 结构:) print(milvus_client.describe_collection(collection_nameCOLLECTION_NAME))准备并插入数据# 5. 准备并插入数据 print(f\n-- 正在向 {COLLECTION_NAME} 插入数据) data_to_insert [] for image_path in tqdm(image_list, desc生成图像嵌入): vector encoder.encode_image(image_path) data_to_insert.append({vector: vector, image_path: image_path}) if data_to_insert: result milvus_client.insert(collection_nameCOLLECTION_NAME, datadata_to_insert) print(f成功插入 {result[insert_count]} 条数据。)创建索引-HNSW# 6. 创建索引 print(f\n-- 正在为 {COLLECTION_NAME} 创建索引) index_params milvus_client.prepare_index_params() index_params.add_index( field_namevector, index_typeHNSW, metric_typeCOSINE, params{M: 16, efConstruction: 256} ) milvus_client.create_index(collection_nameCOLLECTION_NAME, index_paramsindex_params) print(成功为向量字段创建 HNSW 索引。) print(索引详情:) print(milvus_client.describe_index(collection_nameCOLLECTION_NAME, index_namevector)) milvus_client.load_collection(collection_nameCOLLECTION_NAME) print(已加载 Collection 到内存中。)执行检索--把问题也向量化# 7. 执行多模态检索 print(f\n-- 正在 {COLLECTION_NAME} 中执行检索) query_image_path os.path.join(DATA_DIR, dragon, query.png) query_text 一条龙 query_vector encoder.encode_query(image_pathquery_image_path, textquery_text) search_results milvus_client.search( collection_nameCOLLECTION_NAME, data[query_vector], output_fields[image_path], limit5, # ef: 搜索时的节点遍历深度值越大召回率越高但耗时越长 search_params{metric_type: COSINE, params: {ef: 128}} )[0] retrieved_images [] print(检索结果:) for i, hit in enumerate(search_results): print(f Top {i1}: ID{hit[id]}, 距离{hit[distance]:.4f}, 路径{hit[entity][image_path]}) retrieved_images.append(hit[entity][image_path])五索引优化本节将基于LlamaIndex的高性能生产级RAG构建方案1对索引优化进行更深入的探讨。一上下文扩展--Sentence Window Retrieval在RAG系统中常常面临一个权衡问题使用小块文本进行检索可以获得更高的精确度但小块文本缺乏足够的上下文可能导致大语言模型LLM无法生成高质量的答案而使用大块文本虽然上下文丰富却容易引入噪音降低检索的相关性。为了解决这一矛盾LlamaIndex 提出了一种实用的索引策略——句子窗口检索Sentence Window Retrieval2。该技术巧妙地结合了两种方法的优点它在检索时聚焦于高度精确的单个句子在送入LLM生成答案前又智能地将上下文扩展回一个更宽的“窗口”从而同时保证检索的准确性和生成的质量。句子窗口检索的思想可以概括为为检索精确性而索引小块为上下文丰富性而检索大块。其工作流程如下1索引阶段在构建索引时文档被分割成单个句子。每个句子都作为一个独立的“节点Node”存入向量数据库。同时每个句子节点都会在元数据metadata中存储其上下文窗口即该句子原文中的前N个和后N个句子。这个窗口内的文本不会被索引仅仅是作为元数据存储。2检索阶段当用户发起查询时系统会在所有单一句子节点上执行相似度搜索。因为句子是表达完整语义的最小单位所以这种方式可以非常精确地定位到与用户问题最相关的核心信息。3后处理阶段在检索到最相关的句子节点后系统会使用一个名为MetadataReplacementPostProcessor的后处理模块。该模块会读取到检索到的句子节点的元数据并用元数据中存储的完整上下文窗口来替换节点中原来的单一句子内容。4生成阶段最后这些被替换了内容的、包含丰富上下文的节点被传递给LLM用于生成最终的答案。# 假设 Settings.llm 和 Settings.embed_model 已经预先配置好 # 1. 加载文档 documents SimpleDirectoryReader( input_files[../../data/C3/pdf/IPCC_AR6_WGII_Chapter03.pdf] ).load_data() # 2. 创建节点与构建索引 # 2.1 句子窗口索引 node_parser SentenceWindowNodeParser.from_defaults( window_size3, window_metadata_keywindow, original_text_metadata_keyoriginal_text, ) sentence_nodes node_parser.get_nodes_from_documents(documents) sentence_index VectorStoreIndex(sentence_nodes)根据 LlamaIndex 的底层源码SentenceWindowNodeParser的核心逻辑位于build_window_nodes_from_documents方法中。其实现过程可以分解为以下几个关键步骤1句子切分 (sentence_splitter)解析器首先接收一个文档Document然后调用self.sentence_splitter(doc.text)方法。这个sentence_splitter是一个可配置的函数默认为split_by_sentence_tokenizer它负责将文档的全部文本精确地切分成一个句子列表text_splits。2创建基础节点 (build_nodes_from_splits)切分出的text_splits列表被传递给build_nodes_from_splits工具函数。这个函数会为列表中的每一个句子都创建一个独立的TextNode。此时每个TextNode的text属性就是这个句子的内容。3构建窗口并填充元数据 (主要循环)接下来解析器会遍历所有新创建的TextNode。对于位于第i个位置的节点它会执行以下操作定位窗口通过列表切片nodes[max(0, i - self.window_size) : min(i self.window_size 1, len(nodes))]来获取一个包含中心句子及其前后window_size默认为3个邻近节点的列表window_nodes。这个切片操作很巧妙地处理了文档开头和结尾的边界情况。组合窗口文本将window_nodes列表中所有节点的text即所有在窗口内的句子用空格拼接成一个长字符串。填充元数据将上一步生成的长字符串完整的上下文窗口存入当前节点第i个节点的元数据中键为self.window_metadata_key默认为window。同时也会将节点自身的文本原始句子存入元数据键为self.original_text_metadata_key默认为original_text。设置元数据排除项这是一个非常关键的细节。在填充完元数据后代码会执行node.excluded_embed_metadata_keys.extend(...)和node.excluded_llm_metadata_keys.extend(...)。这行代码的作用是告诉后续的嵌入模型和LLM在处理这个节点时应当忽略window和original_text这两个元数据字段。这确保了只有单个句子的纯净文本被用于生成向量嵌入从而保证了检索的高精度。而window字段仅供后续的MetadataReplacementPostProcessor使用。通过以上步骤SentenceWindowNodeParser最终返回一个TextNode列表。列表中的每个节点都代表一个独立的句子其text属性用于精确检索而其metadata中则“隐藏”了用于生成答案的丰富上下文窗口。# 2.2 常规分块索引 (基准) base_parser SentenceSplitter(chunk_size512) base_nodes base_parser.get_nodes_from_documents(documents) base_index VectorStoreIndex(base_nodes) # 3. 构建查询引擎 sentence_query_engine sentence_index.as_query_engine( similarity_top_k2, node_postprocessors[ MetadataReplacementPostProcessor(target_metadata_keywindow) ], ) base_query_engine base_index.as_query_engine(similarity_top_k2) # 4. 执行查询并对比结果 query What are the concerns surrounding the AMOC? print(f查询: {query}\n) print(--- 句子窗口检索结果 ---) window_response sentence_query_engine.query(query) print(f回答: {window_response}\n) print(--- 常规检索结果 ---) base_response base_query_engine.query(query) print(f回答: {base_response}\n)1构建句子窗口索引这一步利用了SentenceWindowNodeParser。它将文档解析为以单个句子为单位的Node同时将包含上下文的“窗口”文本默认为前后各3个句子存储在每个Node的元数据中。这一步是实现“为检索精确性而索引小块”思想的关键。2构建查询引擎与后处理查询引擎的构建是实现“为生成质量而扩展上下文”的关键。在创建sentence_query_engine时配置中加入了一个重要的后处理器MetadataReplacementPostProcessor。它的作用是当检索器根据用户查询找到最相关的节点也就是单个句子后这个后处理器会立即介入。它会从该节点的元数据中读取出预先存储的完整“窗口”文本并用它替换掉节点中原来的单个句子内容。这样最终传递给大语言模型的就不再是孤立的句子而是包含丰富上下文的完整文本段落从而确保了生成答案的质量和连贯性。二结构化索引随着知识库的规模不断扩大例如包含数百个PDF文件传统的RAG方法即对所有文本块进行top-k相似度搜索会遇到瓶颈。当一个查询可能只与其中一两个文档相关时在整个文档库中进行无差别的向量搜索不仅效率低下还容易被不相关的文本块干扰导致检索结果不精确。为了解决这个问题一个有效的方法是利用结构化索引。其原理是在索引文本块的同时为其附加结构化的元数据Metadata。这些元数据可以是任何有助于筛选和定位信息的标签例如文件名文档创建日期章节标题作者任何自定义的分类标签实际上在第二章“文本分块”中介绍的基于文档结构的分块方法就是实现结构化索引的一种前置步骤。例如在使用MarkdownHeaderTextSplitter时分块器会自动将Markdown文档的各级标题如Header 1,Header 2等提取并存入每个文本块的元数据中。这些标题信息就是非常有价值的结构化数据可以直接用于后续的元数据过滤。# 1. 为每个工作表创建查询引擎和摘要节点 excel_file ../../data/C3/excel/movie.xlsx xls pd.ExcelFile(excel_file) df_query_engines {} all_nodes [] for sheet_name in xls.sheet_names: df pd.read_excel(xls, sheet_namesheet_name) # 为当前工作表创建一个 PandasQueryEngine query_engine PandasQueryEngine(dfdf, llmSettings.llm, verboseTrue) # 为当前工作表创建一个摘要节点IndexNode year sheet_name.replace(年份_, ) summary f这个表格包含了年份为 {year} 的电影信息可以用来回答关于这一年电影的具体问题。 node IndexNode(textsummary, index_idsheet_name) all_nodes.append(node) # 存储工作表名称到其查询引擎的映射 df_query_engines[sheet_name] query_engine # 2. 创建顶层索引只包含摘要节点 vector_index VectorStoreIndex(all_nodes) # 3. 创建递归检索器 vector_retriever vector_index.as_retriever(similarity_top_k1) recursive_retriever RecursiveRetriever( vector, retriever_dict{vector: vector_retriever}, query_engine_dictdf_query_engines, verboseTrue, ) # 4. 创建查询引擎 query_engine RetrieverQueryEngine.from_args(recursive_retriever) # 5. 执行查询 query 1994年评分人数最多的电影是哪一部 print(f查询: {query}) response query_engine.query(query) print(f回答: {response})创建 PandasQueryEngine遍历 Excel 中的每个工作表为每个工作表即一个独立的 DataFrame都实例化一个PandasQueryEngine。其强大之处在于它能将关于表格的自然语言问题如“评分人数最多的是哪个”转换成实际的 Pandas 代码如df.sort_values(评分人数).iloc[-1]来执行。创建摘要节点 (IndexNode)对每个工作表都创建一个IndexNode其内容是关于这个表格的一段摘要文本。这个节点将作为顶层检索的“指针”。构建顶层索引使用所有创建的IndexNode构建一个VectorStoreIndex。这个索引不包含任何表格的详细数据只包含指向各个表格的“指针”信息。创建RecursiveRetriever这是实现递归检索的核心。将其配置为retriever_dict: 指定顶层的检索器即在摘要节点中进行检索的vector_retriever。query_engine_dict: 提供一个从节点 ID即工作表名称到其对应查询引擎的映射。当顶层检索器匹配到某个摘要节点后递归检索器就知道该调用哪个PandasQueryEngine来处理后续查询。