六、向量存储

6.1 什么是向量数据库

上一章我们已经把文本转化为了向量。现在面临一个现实问题：这些向量存在哪里？

能不能直接存到MySQL之类的传统数据库里？技术上可以——把向量当作一个数组字段存储。但问题出在检索上：

简单来说，向量数据库的核心价值在于解决了传统数据库面对高维数据时的两个致命短板：

1. 查询逻辑的转变（相似度检索）：突破了传统关系型数据库只能找“绝对相等”或“标量范围”的限制，向量数据库支持在高维度空间中寻找“最相似”的特征。
2. 海量数据的检索速度（高维索引）：相比于传统数据库在计算相似度时无奈的“全表扫描”，向量数据库通过构建 ANN（近似最近邻）、HNSW 等专用的高维向量索引算法，巧妙地避免了全局遍历，在亿级数据中依然能实现毫秒级的快速返回。

6.2 主流选型

向量数据库	描述
FAISS	用于高效相似性搜索和密集向量聚类的库
Chroma	开源轻量级向量数据库，有极简API
Milvus	开源云原生向量数据库，性能强悍，覆盖轻量级原型到十亿级向量的大规模系统
Pgvector	PostgreSQL扩展，为PostgreSQL增加向量搜索功能
Redis	开源内存数据结构存储，已原生支持向量相似性搜索
Elasticsearch	开源分布式搜索引擎，统一管理结构化、非结构化和向量数据

选型众多，本课程选择Milvus作为主要向量存储。为什么？因为Milvus在生产环境中使用最为普遍，支持数百亿级别的向量存储和检索，并且原生支持稠密向量和稀疏向量的混合检索——这正好对应上一章我们用BGE-M3生成的两种向量。

6.3 Milvus是什么

Milvus是一个开源的云原生向量数据库，专门为向量搜索而设计。可以把它类比为"向量世界的MySQL"——MySQL擅长结构化数据的精确查询，Milvus擅长高维向量的相似度检索。

6.3.1 Milvus架构

结合官方架构图，Milvus 的设计核心可以总结为四个字："存算分离"。它采用了云原生和微服务架构，将整个系统解耦为四个主要层次，从而实现了极高的弹性和扩展性。

核心架构分层

1. 接入层（Access Layer）—— Proxy

角色：这是整个 Milvus 集群的"大门"（API 网关）。Client SDK 所有的请求都会先打到 Proxy。

功能：Proxy 负责请求的路由转发。它会将不同类型的请求分发给后端的不同组件：

• DCL/DDL（数据控制/定义语言，如建表、建集合）：转发给 Coordinator
• DML（数据操作语言，如插入向量）：转发给 Streaming Nodes
• DQL（数据查询语言，如向量相似度检索）：主要转发给 Query Nodes 执行核心检索；同时也会与 Streaming Nodes 交互，比如直接读取极热的流式数据。

2. 协调层（Coordinator Service）—— Coordinator

角色：集群的"大脑"。

功能：负责集群拓扑管理、负载均衡、数据分配和元数据管理。它会将系统的元数据（如集合的 Schema、节点的状态）注册并持久化到右侧的 Meta Storage (etcd) 中。同时，它向下的 Workers 层下发管理指令（Management）。

3. 计算/执行层（Worker Nodes）—— 真正干活的节点

这是图中虚线框内的部分，进一步细分为三种专注于特定任务的节点。它们可以独立进行横向扩容：

节点类型	职责
Streaming Nodes（流式节点）	负责处理所有的 DML（写入）请求。它将数据追加（Append）到 WAL 提供持久化保障，并负责将数据落盘（Flush）到对象存储中，形成初始的数据段（Segments）。
Data Nodes（数据节点）	专职后台优化。负责从对象存储中读取已落盘的数据，执行索引构建（Indexing）和碎片合并（Compaction），优化后写回对象存储，以提升检索效率。
Query Nodes（查询节点）	专门负责执行向量搜索和读取请求。它从对象存储加载已索引的历史数据到内存，同时订阅 Streaming Nodes 获取实时增量数据，合并两部分结果以保证查询的实时性和完整性。

4. 存储层（Durable Storage）—— 数据基座

Milvus 自己不造存储轮子，而是依托成熟的第三方分布式存储组件：

存储组件	用途
Meta Storage (etcd)	存储集群的元数据（配置、节点信息等）
WAL (Write-Ahead Log - Pulsar/Kafka)	预写日志，由 Streaming Nodes 写入，提供底层数据的持久化保证。
Object Storage (MinIO/S3)	持久化存储最终的向量数据段（Segments）和构建好的索引文件

核心工作流

理解架构最好的方式是跑通读写两条链路。从图中我们可以清晰地看到数据的流动轨迹：

场景 A：数据插入流程 (Insert)

流程要点：

1. 接入与路由：

   Proxy 接收到客户端发起的 Insert 请求后，将其精准路由给后端的 Streaming Nodes。

2. Streaming Nodes 处理（日志追加缓冲 ---> 阈值触发落盘）：

   在这个节点，数据的处理分为明显的“即时”与“异步”两个阶段：

   • 即时日志追加与内存缓冲：数据到达后，立刻向 WAL（Kafka/Pulsar）追加日志，以保障数据绝对不丢失。此时，数据留在 Streaming Nodes 的内存中，成为 Growing Segment（生长中的数据段）。这部分内存中的“热数据”可以直接被 Query Nodes 订阅并进行“暴力扫描”检索。
   • 条件触发落盘 (Flush)：数据不会立刻写向底层存储，而是等待满足以下任一阈值时，Streaming Nodes 才会触发 Flush 动作将其写向 Object Storage：
     • 容量触发：积累到了配置的大小（默认是 512MB）。
     • 时间触发：达到了定时刷新的时间（防止数据写入极少时一直挂在内存不落盘）。
     • 手动触发：用户主动调用了 flush() 接口。
   • 形成初始段：落盘后的数据变成了 Sealed Segment（封存状态的初始小段），它安全地躺在 Object Storage 中，不再接收新数据。

3. Data Nodes 异步优化：

   Data Nodes 像后台的巡逻兵，异步从 Object Storage 读取这些刚刚落盘但未经优化的 Sealed Segments，默默在后台执行索引构建（Indexing，例如构建 HNSW 索引）以及 数据碎片合并（Compaction）。

4. 高可用检索就绪：

   优化完成后，Data Nodes 会将带有高效索引的、规整的大 Segments 重新写回 Object Storage。至此，数据彻底转化为持久化的“冷数据”，供 Query Nodes 加载并进行极速的向量相似度检索。

关键理解：Streaming Nodes 是写入链路的主角

关键注意：在 Milvus 中：

1. 它是写前日志（WAL），因为它承担着保证刚写入的数据不丢失、提供数据恢复基准的职责。
2. 它也是消息队列（Queue），因为它在物理上就是用 Kafka/Pulsar 部署的，并且被系统用来做组件之间的数据解耦和广播订阅。

---

场景 B：向量检索流程 (Search)

在 RAG 系统中，这是最关键、也是并发最高的环节。

流程要点：

1. Proxy 将 Search 请求路由给 Query Nodes
2. Query Nodes 同时从两个数据源检索：
   • 历史冷数据：从 Object Storage 加载已经被 Data Nodes 优化并构建好索引的 Segments 到内存，通过 ANN 索引（如 HNSW）进行高效检索
   • 实时热数据：订阅 Streaming Nodes，获取刚写入但尚未落盘的增量数据，对这部分数据进行暴力扫描
3. 将两部分检索结果合并、统一按相似度排序，取 TopK 后经 Proxy 返回客户端

关键理解：这种"冷热数据合并查询"的设计，保证了即使数据刚刚写入、还没来得及被 Data Nodes 优化，也能立刻被检索到，实现了写入即可查的实时性。

架构设计优势

总结来说，Milvus 的架构设计完美体现了"存算分离"的思想：

优势	说明
弹性扩展	各层组件可独立横向扩容，根据业务负载灵活调整
高可用性	组件解耦，单点故障不影响整体服务
低成本	计算节点可使用廉价存储，热数据缓存，冷数据落盘
高性能	Query Nodes 内存计算 + Streaming Nodes 订阅保证毫秒级实时检索
写入即可查	冷热数据合并查询机制，新写入数据无需等待索引构建即可被检索

文本中提到的"解决了相似度检索和检索速度"这两个痛点，在这张架构图中得到了完美的工程体现：Query Nodes 负责在内存中极速计算相似度；而庞大的数据量和复杂的索引构建任务，则被优雅地甩给了后台的 Data Nodes 和廉价的 Object Storage，从而实现了性能和成本的最佳平衡。

6.3.2 部署方式

方式	说明	适用场景
Milvus Lite	通过pip安装，本地轻量化运行，仅支持FLAT索引，仅支持MacOS和Linux	本地开发调试
Milvus Standalone	单点部署，Docker一键启动	中小规模生产
Milvus Distributed	分布式部署，Kubernetes集群	大规模生产

本课程使用Milvus Standalone方式部署：

# 加载镜像
docker load -i milvus_image.tar

# 启动服务（Linux）
bash standalone_embed.sh start

# 启动服务（Windows Docker Desktop）
standalone.bat start

启动后可通过Milvus官方图形化客户端Attu查看数据，选择token方式直接连接即可。

6.4 Milvus核心概念

在动手写代码之前，先理解Milvus中几个核心概念以及它们之间的关系：

6.4.1 Collection与数据类型

Collection通过Schema定义有哪些字段及其类型。Milvus支持三大类字段：

Schema约束：一个Schema有一个主键、最多四个向量字段和若干标量字段。主键用于唯一标识实体，支持AutoId自动生成。标量字段通常用来存储元数据（如文档来源、页码等），在搜索时可通过标量条件进行过滤。

6.4.2 索引——加速搜索的关键

向量存进去之后，如果每次检索都暴力遍历所有向量，速度会很慢。索引是建立在数据之上的附加结构，可以大幅加快搜索速度。不同类型的向量需要不同的索引。

稠密向量索引 - HNSW

HNSW（分层导航小世界）是当下最常用的基于图的索引算法，具有出色的搜索精度和低延迟，但需要较高内存开销。它构建多层图（类似不同缩放级别的地图），底层包含所有数据点，上层由采样子集组成：

另一种稠密向量索引是FLAT，采用暴力搜索，每个查询直接与所有向量比较，能保证100%召回率，但速度最慢，仅适合小数据量或对精度要求极高的场景。

举例理解 HNSW——"在全国找餐厅"

假设你要在全中国几百万家餐厅中，找到和你口味最接近的那一家（口味 = 向量，口味的相似度 = 向量距离）。

暴力搜索（FLAT）：拿着全国餐厅名录，从第一家挨个比对到最后一家，几百万家全走一遍，速度极慢。

HNSW 的做法：事先准备三张不同比例的地图，搜索时从粗到细、逐层下降：

层级	地图比例	包含内容	搜索动作
Layer 2（顶层）	全国地图	只有几个省会城市（少量采样节点）	快速判断"目标大概在上海方向"，一步跳过几百公里
Layer 1（中间层）	城市地图	上海的主要街区：浦东、徐汇、静安…	缩小范围"应该在徐汇区附近"
Layer 0（底层）	街道地图	每一家餐厅都在这里（全部数据节点）	在徐汇区的街道上逐个比较邻居，找到最合口味的那家

对应到向量检索的关键点：

• 每个"餐厅"就是数据库里的一个向量，"口味的相似度"就是向量之间的距离
• 顶层节点少、连接跨度大，可以用很少的步数跨越大范围，快速缩小搜索空间
• 逐层下降，每层贪婪搜索：在当前层的邻居中找离目标最近的节点，然后跳到下一层继续
• 底层包含所有节点，保证最终能找到真正最近的向量
• 搜索时间复杂度从暴力遍历的 O(n) 降到 O(log n)——就像你不需要走遍全国每条街，只需要"全国→城市→街区→街道"几步就到

另一种稠密向量索引是FLAT，采用暴力搜索，每个查询直接与所有向量比较，能保证100%召回率，但速度最慢，仅适合小数据量或对精度要求极高的场景。

稀疏向量索引 - SPARSE_INVERTED_INDEX

利用倒排索引的原理，为稀疏数据创建高效的搜索结构。查询时先通过倒排索引找到包含query中token的文档，然后计算相似度分数：

6.4.3 相似度度量

索引解决了"怎么快速找"的问题，而metric_type解决了"怎么判断相似"的问题。建索引时必须指定度量方式，检索时也要保持一致。Milvus支持三种主要度量方式：

6.4.3.1 三者的数学关系

要想真正理解如何选择度量方式，必须先弄清三者在数学上的关系。假设有两个向量 A 和 B：

度量方式	公式	衡量的是
IP（内积）	A · B	方向 + 长度的综合相似性
COSINE（余弦相似度）	(A · B) / (‖A‖ × ‖B‖)	纯方向相似性（消除长度影响）
L2（欧氏距离）	‖A - B‖² = ‖A‖² + ‖B‖² - 2(A · B)	空间中的绝对距离

关键推导：当向量做了 L2 归一化（‖A‖ = 1 且 ‖B‖ = 1）时：

1. COSINE = IP：余弦公式的分母变成 1×1 = 1，因此 cos(θ) = A · B = IP，两者数值完全相等
2. L2² = 2 - 2 × IP：因为 ‖A‖² + ‖B‖² = 1 + 1 = 2，所以 L2² = 2 - 2(A · B) = 2 - 2 × IP

结论：归一化后，三者严格等价——IP 越大 ⇔ COSINE 越大 ⇔ L2 越小，排序结果完全一致。

6.4.3.2 如何选择？

场景	推荐	原因
稠密向量 + 文本语义检索（已归一化）	IP	BGE 等主流文本嵌入模型输出已归一化，此时 IP = COSINE，但 IP 计算最快（仅乘加运算，无需算模长和除法）
稠密向量 + 文本语义检索（未归一化/不确定）	COSINE	COSINE 自带归一化，对向量长度不敏感，兼容性最好
稠密向量 + 图像/通用检索	L2	图像特征向量通常未归一化，需要考虑绝对距离
稀疏向量检索	IP	稀疏向量本质是"关键词权重"，内积直接反映匹配程度

工程最佳实践：很多文本嵌入模型（如 BGE 系列）输出的稠密向量已经做了 L2 归一化。此时 COSINE、IP、L2 三者的排序结果完全一致（数学原理见上方推导）。但从计算性能的角度，应当优先选 IP——因为 IP 只需乘加运算，而 COSINE 额外需要算模长和除法，L2 额外需要减法和平方运算。在 Milvus、FAISS 等向量数据库中，常见做法是：在外部对向量做 L2 归一化，建索引时 metric_type 设为 IP，既保证等价于 COSINE 的语义召回效果，又获得最优的检索性能。如果不确定模型是否归一化，优先选 COSINE，它对向量长度不敏感。

6.5 基本操作流程

理解了核心概念后，Milvus的使用可以概括为以下流程：

下面沿着这个流程逐步演示代码。

6.5.1 定义Schema

def build_schema():
    from pymilvus import MilvusClient, DataType
    return (
        MilvusClient.create_schema(auto_id=True)
        # 主键：自动生成ID
        .add_field(field_name="id", datatype=DataType.INT64, is_primary=True)
        # 稠密向量：用于语义检索，维度与嵌入模型一致
        .add_field(field_name="vector", datatype=DataType.FLOAT_VECTOR, dim=1024)
        # 稀疏向量：用于关键词检索
        .add_field(field_name="sparse_vector", datatype=DataType.SPARSE_FLOAT_VECTOR)
        # 原始文本：存储Chunk内容，检索后返回给用户
        .add_field(field_name="text", datatype=DataType.VARCHAR, max_length=1500)
        # 元数据：存储来源、页码等信息，支持过滤
        .add_field(field_name="metadata", datatype=DataType.JSON)
    )

6.5.2 配置索引

def build_index():
    from pymilvus import MilvusClient
    index_params = MilvusClient.prepare_index_params()

    # 稠密向量：使用HNSW索引 + L2度量
    index_params.add_index(
        field_name="vector",
        index_type="HNSW",
        metric_type="L2",
    )

    # 稀疏向量：使用倒排索引 + IP度量
    index_params.add_index(
        field_name="sparse_vector",
        index_type="SPARSE_INVERTED_INDEX",
        metric_type="IP",
    )

    return index_params

6.5.3 创建Collection

def get_milvus_client():
    from pymilvus import MilvusClient
    return MilvusClient(uri="http://localhost:19530", token="")

def create_collection(client):
    collection_name = "demo_collection"
    client.drop_collection(collection_name=collection_name)
    if not client.has_collection(collection_name=collection_name):
        client.create_collection(
            collection_name=collection_name,
            schema=build_schema(),
            index_params=build_index(),
        )

6.5.4 插入数据

将前面几章的成果串起来：加载文档 → 切分 → 嵌入 → 插入Milvus：

def insert_data(client: MilvusClient, collection_name: str):
    from langchain_community.document_loaders import UnstructuredWordDocumentLoader
    from langchain_text_splitters import RecursiveCharacterTextSplitter
    from FlagEmbedding import BGEM3FlagModel

    # 1、加载文件
    doc_list = UnstructuredWordDocumentLoader(
        "assets/sample.docx", mode="single"
    ).load()

    # 2、切分文件
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=500, chunk_overlap=50, separators=["\n\n", "\n", "。"]
    )
    splitted_doc_list = text_splitter.split_documents(doc_list)

    # 3、构建向量（同时生成稠密和稀疏向量）
    model = BGEM3FlagModel("assets/models/bge-m3")
    all_vectors = model.encode(
        [doc.page_content for doc in splitted_doc_list],
        return_dense=True, return_sparse=True
    )

    # 4、准备数据：组装成List[Dict]，每个Dict对应Schema中的字段
    insert_data_list = []
    for doc, dense_vector, sparse_vector in zip(
        splitted_doc_list,
        all_vectors["dense_vecs"],
        all_vectors['lexical_weights']
    ):
        insert_data_list.append({
            "vector": dense_vector,
            "sparse_vector": sparse_vector,
            "metadata": doc.metadata,
            "text": doc.page_content
        })

    # 5、插入数据
    res = client.insert(collection_name=collection_name, data=insert_data_list)
    print(res)

6.5.5 删除数据

可以通过ID列表或过滤条件删除实体：

def delete_demo(client):
    res = client.delete(
        collection_name="demo_collection",
        # 通过ID删除，也可通过其他字段过滤
        filter="id in [463480757150366907, 463480757150366908]",
    )
    print(res)  # {'delete_count': 2}

数据已经存好、索引已经建好，接下来的关键问题是：当用户提问时，如何从海量向量中找到最相关的那几个？

6.6.6 完整案例

import os
from pymilvus import MilvusClient, DataType
from langchain_community.document_loaders import UnstructuredWordDocumentLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from FlagEmbedding import BGEM3FlagModel

# ==========================================
# 1. 定义 Schema (表结构)
# ==========================================
def build_schema():
    print("-> 正在构建 Schema...")
    return (
        MilvusClient.create_schema(auto_id=True)
        # 主键：自动生成ID
        .add_field(field_name="id", datatype=DataType.INT64, is_primary=True)
        # 稠密向量：用于语义检索，维度需要与你的 BGE-M3 模型输出一致 (通常是 1024)
        .add_field(field_name="vector", datatype=DataType.FLOAT_VECTOR, dim=1024)
        # 稀疏向量：用于关键词检索
        .add_field(field_name="sparse_vector", datatype=DataType.SPARSE_FLOAT_VECTOR)
        # 原始文本：存储Chunk内容，检索后返回给用户
        .add_field(field_name="text", datatype=DataType.VARCHAR, max_length=1500)
        # 元数据：存储来源、页码等信息，支持过滤
        .add_field(field_name="metadata", datatype=DataType.JSON)
    )

# ==========================================
# 2. 配置索引 (加速检索)
# ==========================================
def build_index():
    print("-> 正在配置索引参数...")
    index_params = MilvusClient.prepare_index_params()

    # 稠密向量：使用HNSW索引 + L2度量
    index_params.add_index(
        field_name="vector",
        index_type="HNSW",
        metric_type="COSINE",
    )

    # 稀疏向量：使用倒排索引 + IP度量
    index_params.add_index(
        field_name="sparse_vector",
        index_type="SPARSE_INVERTED_INDEX",
        metric_type="IP",
    )

    return index_params

# ==========================================
# 3. 创建客户端与 Collection (集合/表)
# ==========================================
def get_milvus_client():
    print("-> 正在连接 Milvus 客户端...")
    # 假设你的 Milvus 运行在本地默认端口
    return MilvusClient(uri="http://localhost:19530", token="")

def create_collection(client, collection_name):
    print(f"-> 准备创建 Collection: {collection_name}")

    # 为了演示方便，如果存在同名集合，先删除
    if client.has_collection(collection_name=collection_name):
        print(f"   发现已存在集合 {collection_name}，正在删除以重新创建...")
        client.drop_collection(collection_name=collection_name)

    # 创建集合，绑定前面定义的 Schema 和 索引参数
    print(f"   正在创建新集合...")
    client.create_collection(
        collection_name=collection_name,
        schema=build_schema(),
        index_params=build_index(),
    )
    print(f"   Collection {collection_name} 创建成功！")

# ==========================================
# 4. 数据处理与插入 (核心流程)
# ==========================================
def insert_data(client: MilvusClient, collection_name: str, doc_path: str, model_path: str):
    print(f"\n-> 开始处理数据并插入 Milvus...")

    # 检查文件是否存在
    if not os.path.exists(doc_path):
        print(f"   [错误] 找不到文档: {doc_path}。请确保文件存在。")
        return
    if not os.path.exists(model_path):
        print(f"   [警告] 找不到本地模型: {model_path}。如果是首次运行，它可能会自动下载。")

    print(f"   1. 加载文档: {doc_path}")
    doc_list = UnstructuredWordDocumentLoader(
        doc_path, mode="single"
    ).load()

    print(f"   2. 切分文档...")
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=500, chunk_overlap=50, separators=["\n\n", "\n", "。"]
    )
    splitted_doc_list = text_splitter.split_documents(doc_list)
    print(f"      共切分为 {len(splitted_doc_list)} 个片段 (Chunks)。")

    print(f"   3. 加载 BGE-M3 模型并构建向量 (这可能需要一些时间)...")
    model = BGEM3FlagModel(model_path)
    all_vectors = model.encode(
        [doc.page_content for doc in splitted_doc_list],
        return_dense=True,
        return_sparse=True
    )

    print(f"   4. 准备数据列表...")
    insert_data_list = []
    for doc, dense_vector, sparse_vector in zip(
        splitted_doc_list,
        all_vectors["dense_vecs"],
        all_vectors['lexical_weights']
    ):
        insert_data_list.append({
            "vector": dense_vector,
            "sparse_vector": sparse_vector,
            "metadata": doc.metadata,
            "text": doc.page_content
        })

    print(f"   5. 正在将数据插入 {collection_name}...")
    res = client.insert(collection_name=collection_name, data=insert_data_list)
    print(f"-> 插入完成！Milvus 返回结果: {res}")


# ==========================================
# 主运行入口
# ==========================================
if __name__ == "__main__":
    COLLECTION_NAME = "demo_collection"
    # 如果没有示例 word 文档，请随便建一个测试文档。
    DOC_PATH = "assets/sample.docx"
    # 如果你没有下载模型到本地，让它从 HuggingFace 自动下载
    MODEL_PATH = r"D:\ai_models\huggingface_cache\bge-m3"

    try:
        # 获取客户端连接
        milvus_client = get_milvus_client()

        # # 创建数据库集合
        create_collection(milvus_client, COLLECTION_NAME)

        # # 执行数据切分、向量化并插入
        insert_data(milvus_client, COLLECTION_NAME, DOC_PATH, MODEL_PATH)

        print("\n 全部流程执行完毕！可以打开 Attu 客户端查看插入的数据了。")

    except Exception as e:
        print(f"\n 运行过程中发生错误: {e}")