当前博文未提及该问题，以下是基于C知道的知识内容为您回答。 构建向量数据库是 RAG（检索增强生成）系统的核心基础。在不依赖 LangChain 等高级框架的情况下，我们可以使用原生的 Python 结合向量数据库 SDK（如 Milvus、Faiss）来实现从文本向量化到存储检索的完整流程。 以下以目前业界广泛使用的 **Milvus** 向量数据库为例，展示如何使用原生 Python 代码构建向量数据库并进行基本的相似度检索。该方案能够提供更高的性能和更细粒度的控制，适合生产环境部署 [ref_2]。 ### 1. 环境准备与依赖安装 首先，需要安装 Milvus 的 Python SDK (`pymilvus`) 以及用于生成文本嵌入的库（这里使用 `sentence-transformers` 生成通用向量）。 ```bash pip install pymilvus sentence-transformers ``` ### 2. 完整构建代码示例 以下代码实现了以下逻辑： 1. **连接 Milvus 服务**：建立与向量数据库的连接。 2. **定义 Collection Schema**：设计数据表结构，包含向量字段和标量字段（如文本内容、主键）。 3. **生成 Embedding**：将原始文本转换为高维向量。 4. **数据插入**：将向量与原始文本写入数据库。 5. **索引构建**：创建索引以加速检索。 6. **相似度搜索**：输入查询文本，返回最相似的结果。 ```python from pymilvus import ( connections, utility, FieldSchema, CollectionSchema, DataType, Collection, ) from sentence_transformers import SentenceTransformer import time # --- 1. 初始化配置 --- # 假设 Milvus 服务已在本机默认端口 19530 启动 HOST = "127.0.0.1" PORT = "19530" COLLECTION_NAME = "demo_kb_collection" DIMENSION = 384 # 这里使用 all-MiniLM-L6-v2 模型的默认维度 # 连接到 Milvus 服务器 [ref_3] print(f"正在连接到 Milvus服务器 {HOST}:{PORT}...") connections.connect(host=HOST, port=PORT) # --- 2. 定义 Collection 结构 --- # 如果 Collection 已存在则删除，以便重复运行 Demo if utility.has_collection(COLLECTION_NAME): utility.drop_collection(COLLECTION_NAME) # 定义字段：ID (主键), 文本内容, 向量数据 fields = [ FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True), FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=65535), FieldSchema(name="vector", dtype=DataType.FLOAT_VECTOR, dim=DIMENSION) ] schema = CollectionSchema(fields, f"{COLLECTION_NAME} 的演示数据表") collection = Collection(name=COLLECTION_NAME, schema=schema) print(f"创建 Collection: {COLLECTION_NAME}") # --- 3. 准备数据与向量化 --- # 初始化 Embedding 模型 (all-MiniLM-L6-v2 轻量且高效) embedding_model = SentenceTransformer('sentence-transformers/all-MiniLM-L6-v2') # 模拟待存储的文本数据 raw_data = [ "LangChain 是一个用于开发由语言模型驱动的应用程序的框架。", "向量数据库是 AI 应用中用于存储和检索高维向量的关键组件。", "RAG 技术结合了信息检索的能力和生成模型的创造力。", "Python 是数据科学和机器学习领域最流行的编程语言。" ] # 生成向量数据 vectors = embedding_model.encode(raw_data).tolist() # 准备插入的数据实体 (这里只插入 text 和 vector，id 自动生成) data_entities = [raw_data, vectors] print(f"正在向量化并插入 {len(raw_data)} 条数据...") # --- 4. 插入数据 --- collection.insert(data_entities) # 确保数据已写入（Milvus 默认有缓冲，flush 保证可见性） collection.flush() # --- 5. 创建索引 --- # 为了加速搜索，必须为向量字段创建索引 # 这里使用 IVF_FLAT 索引类型，适合大多数场景 index_params = { "index_type": "IVF_FLAT", "metric_type": "IP", # 内积，适合计算余弦相似度（若向量已归一化） "params": {"nlist": 128} } print("正在创建向量索引...") collection.create_index(field_name="vector", index_params=index_params) # --- 6. 加载 Collection 并执行搜索 --- # 搜索前必须将数据加载到内存中 collection.load() # 定义搜索参数 search_params = { "metric_type": "IP", "params": {"nprobe": 10} } # 模拟用户查询 query_text = "什么是 LangChain？" query_vector = embedding_model.encode([query_text]).tolist() print(f"\n正在执行搜索查询: '{query_text}'...") # 执行搜索 (top_k=2 表示返回最相似的2条结果) results = collection.search( data=query_vector, anns_field="vector", param=search_params, limit=2, expr=None, # 可选的过滤表达式 output_fields=["text"] # 需要返回的原始字段 ) # --- 7. 解析并打印结果 --- for i, result in enumerate(results[0]): print(f"\n结果 #{i+1}:") print(f" 距离(得分): {result.distance:.4f}") print(f" 内容: {result.entity.get('text')}") # 断开连接 connections.disconnect("default") print("\n演示结束。") ``` ### 3. 关键步骤深度解析 #### 3.1 数据模型定义 在原生代码中，我们需要显式定义 `Schema`。这与 ORM 定义表结构类似，必须指定每个字段的数据类型。 * `DataType.FLOAT_VECTOR`：明确指定该字段用于存储高维浮点数组。 * `auto_id=True`：简化主键管理，让数据库自动生成唯一 ID [ref_2]。 #### 3.2 向量化处理 代码中使用了 `sentence-transformers` 库。这是脱离 LangChain 后必须自行处理的环节： 1. **模型选择**：`all-MiniLM-L6-v2` 是一个平衡了速度和效果的通用模型，输出维度为 384。 2. **数据转换**：模型输出的是 Numpy 数组，插入 Milvus 前必须使用 `.tolist()` 转换为 Python 原生 List。 #### 3.3 索引与搜索 * **IVF_FLAT 索引**：这是基于倒排文件的索引。`nlist` 参数决定了聚类中心的数量，通常设为 `sqrt(数据量)` 级别。如果不创建索引，Milvus 将执行暴力搜索，速度极慢 [ref_3]。 * **IP (Inner Product)**：内积距离。如果我们在生成 Embedding 时对向量进行了归一化，内积即等价于余弦相似度，数值越大表示越相似。 ### 4. 扩展：使用 Faiss 的内存版本 如果你的数据量较小且不需要持久化存储服务，可以使用 Facebook 的 Faiss 库，它完全基于内存，速度极快且无需安装服务端。 ```python import faiss import numpy as np from sentence_transformers import SentenceTransformer # 1. 初始化模型和模拟数据 model = SentenceTransformer('sentence-transformers/all-MiniLM-L6-v2') docs = ["LangChain 是一个应用框架。", "Faiss 是用于相似度搜索的库。"] vectors = model.encode(docs) # 2. 构建 Faiss 索引 (IndexFlatIP 表示内积，无压缩) index = faiss.IndexFlatIP(vectors.shape[1]) # 将向量添加到索引 index.add(vectors.astype('float32')) # 3. 搜索 query_vector = model.encode(["什么是 Faiss?"]) k = 1 distances, indices = index.search(query_vector.astype('float32'), k) print(f"最相似文档索引: {indices[0][0]}, 得分: {distances[0][0]:.4f}") print(f"内容: {docs[indices[0][0]]}") ``` 这段代码展示了构建向量数据库的最小化实现，通过对比可以看出，Milvus 适合构建服务级、可扩展的知识库，而 Faiss 适合本地、内存级的快速检索 [ref_4]。