企业官网建设流程全解析

四，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_bge=model_name, model_weight=model_path) self.model.eval() def encode_query(self, image_path: str, text: str) -> list[float]: with torch.no_grad(): query_emb = self.model.encode(image=image_path, text=text) return query_emb.tolist()[0] def encode_image(self, image_path: str) -> list[float]: with torch.no_grad(): query_emb = self.model.encode(image=image_path) return query_emb.tolist()[0]

创建Collection

# 3. 初始化客户端 print("--> 正在初始化编码器和Milvus客户端...") encoder = Encoder(MODEL_NAME, MODEL_PATH) milvus_client = MilvusClient(uri=MILVUS_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_id=True) FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True), # 向量字段，维度与模型的输出向量维度一致 FieldSchema(name="vector", dtype=DataType.FLOAT_VECTOR, dim=dim), # 存储原图像路径的标量字段，最大长度限制 512 字符 FieldSchema(name="image_path", dtype=DataType.VARCHAR, max_length=512), ] # 创建集合 Schema schema = CollectionSchema(fields, description="多模态图文检索") print("Schema 结构:") print(schema) # 创建集合 milvus_client.create_collection(collection_name=COLLECTION_NAME, schema=schema) print(f"成功创建 Collection: '{COLLECTION_NAME}'") print("Collection 结构:") print(milvus_client.describe_collection(collection_name=COLLECTION_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_name=COLLECTION_NAME, data=data_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_name="vector", index_type="HNSW", metric_type="COSINE", params={"M": 16, "efConstruction": 256} ) milvus_client.create_index(collection_name=COLLECTION_NAME, index_params=index_params) print("成功为向量字段创建 HNSW 索引。") print("索引详情:") print(milvus_client.describe_index(collection_name=COLLECTION_NAME, index_name="vector")) milvus_client.load_collection(collection_name=COLLECTION_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_path=query_image_path, text=query_text) search_results = milvus_client.search( collection_name=COLLECTION_NAME, data=[query_vector], output_fields=["image_path"], limit=5, # ef: 搜索时的节点遍历深度，值越大召回率越高但耗时越长 search_params={"metric_type": "COSINE", "params": {"ef": 128}} )[0] retrieved_images = [] print("检索结果:") for i, hit in enumerate(search_results): print(f" Top {i+1}: 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 Retrieval）2。该技术巧妙地结合了两种方法的优点：它在检索时聚焦于高度精确的单个句子，在送入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_size=3, window_metadata_key="window", original_text_metadata_key="original_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"）。

1. 设置元数据排除项：这是一个非常关键的细节。在填充完元数据后，代码会执行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_size=512) 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_k=2, node_postprocessors=[ MetadataReplacementPostProcessor(target_metadata_key="window") ], ) base_query_engine = base_index.as_query_engine(similarity_top_k=2) # 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_name=sheet_name) # 为当前工作表创建一个 PandasQueryEngine query_engine = PandasQueryEngine(df=df, llm=Settings.llm, verbose=True) # 为当前工作表创建一个摘要节点（IndexNode） year = sheet_name.replace('年份_', '') summary = f"这个表格包含了年份为 {year} 的电影信息，可以用来回答关于这一年电影的具体问题。" node = IndexNode(text=summary, index_id=sheet_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_k=1) recursive_retriever = RecursiveRetriever( "vector", retriever_dict={"vector": vector_retriever}, query_engine_dict=df_query_engines, verbose=True, ) # 4. 创建查询引擎 query_engine = RetrieverQueryEngine.from_args(recursive_retriever) # 5. 执行查询 query = "1994年评分人数最多的电影是哪一部？" print(f"查询: {query}") response = query_engine.query(query) print(f"回答: {response}")

1. 创建 PandasQueryEngine：遍历 Excel 中的每个工作表，为每个工作表（即一个独立的 DataFrame）都实例化一个PandasQueryEngine。其强大之处在于，它能将关于表格的自然语言问题（如“评分人数最多的是哪个”）转换成实际的 Pandas 代码（如df.sort_values('评分人数').iloc[-1]）来执行。
2. 创建摘要节点 (IndexNode)：对每个工作表，都创建一个IndexNode，其内容是关于这个表格的一段摘要文本。这个节点将作为顶层检索的“指针”。
3. 构建顶层索引：使用所有创建的IndexNode构建一个VectorStoreIndex。这个索引不包含任何表格的详细数据，只包含指向各个表格的“指针”信息。
4. 创建RecursiveRetriever：这是实现递归检索的核心。将其配置为：
   • retriever_dict: 指定顶层的检索器，即在摘要节点中进行检索的vector_retriever。
   • query_engine_dict: 提供一个从节点 ID（即工作表名称）到其对应查询引擎的映射。当顶层检索器匹配到某个摘要节点后，递归检索器就知道该调用哪个PandasQueryEngine来处理后续查询。