Milvus 向量数据库学习手册：从架构到实战

FLOAT_VECTOR      # 单精度浮点向量（最常用）
FLOAT16_VECTOR    # 半精度浮点向量（Milvus 2.4+）
BFLOAT16_VECTOR   # BFloat16 向量（Milvus 2.4+）
SPARSE_FLOAT_VECTOR  # 稀疏向量（Milvus 2.4+）
BINARY_VECTOR     # 二值向量

BOOL              # 布尔值
INT8              # 8位整数
INT16             # 16位整数
INT32             # 32位整数
INT64             # 64位整数（主键常用）
FLOAT             # 单精度浮点数
DOUBLE            # 双精度浮点数
VARCHAR           # 变长字符串
JSON              # JSON 类型
ARRAY             # 数组类型（Milvus 2.2+）

from pymilvus import FieldSchema, CollectionSchema, DataType

# 定义字段
fields = [
    # 主键字段（必须）
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
    # 向量字段（必须）
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=768),
    # 标量字段
    FieldSchema(name="title", dtype=DataType.VARCHAR, max_length=512),
    FieldSchema(name="content", dtype=DataType.VARCHAR, max_length=65535),
    FieldSchema(name="category", dtype=DataType.INT32),
    FieldSchema(name="create_time", dtype=DataType.INT64),
]

# 创建 Schema，开启动态字段支持
schema = CollectionSchema(
    fields=fields,
    description="文档知识库",
    enable_dynamic_field=True  # Milvus 2.3+，支持非预定义字段
)

wget https://github.com/milvus-io/milvus/releases/download/v2.3.15/milvus-standalone-docker-compose.yml -O docker-compose.yml

docker-compose up -d

docker-compose ps
# 应该看到 3 个容器：milvus-standalone, etcd, minio

version: '3.5'

services:
  etcd:
    container_name: milvus-etcd
    image: quay.io/coreos/etcd:v3.5.5
    environment:
      - ETCD_AUTO_COMPACTION_MODE=revision
      - ETCD_AUTO_COMPACTION_RETENTION=1000
      - ETCD_QUOTA_BACKEND_BYTES=4294967296
    volumes:
      - ${DOCKER_VOLUME_DIRECTORY:-.}/volumes/etcd:/etcd
    command: etcd -advertise-client-urls=http://127.0.0.1:2379 -listen-client-urls http://0.0.0.0:2379 --data-dir /etcd

  minio:
    container_name: milvus-minio
    image: minio/minio:RELEASE.2023-03-20T20-16-18Z
    environment:
      MINIO_ACCESS_KEY: minioadmin
      MINIO_SECRET_KEY: minioadmin
    volumes:
      - ${DOCKER_VOLUME_DIRECTORY:-.}/volumes/minio:/minio_data
    command: minio server /minio_data
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]
      interval: 30s
      timeout: 20s
      retries: 3

  standalone:
    container_name: milvus-standalone
    image: milvusdb/milvus:v2.3.15
    command: ["milvus", "run", "standalone"]
    environment:
      ETCD_ENDPOINTS: etcd:2379
      MINIO_ADDRESS: minio:9000
    volumes:
      - ${DOCKER_VOLUME_DIRECTORY:-.}/volumes/milvus:/var/lib/milvus
    ports:
      - "19530:19530"  # gRPC 端口
      - "9091:9091"    # RESTful API 端口
    depends_on:
      - "etcd"
      - "minio"

wget https://github.com/milvus-io/milvus/releases/download/v2.3.15/milvus-standalone-gpu-docker-compose.yml -O docker-compose.yml
docker-compose up -d

# 安装
pip install pymilvus

# 使用
from pymilvus import MilvusClient

# 数据存储在本地文件
client = MilvusClient(uri="./milvus_demo.db")

# 下载二进制包
wget https://github.com/milvus-io/milvus/releases/download/v2.3.15/milvus_2.3.15_Linux_x86_64.tar.gz
tar -xzf milvus_2.3.15_Linux_x86_64.tar.gz

# 配置文件
cp configs/milvus.yaml default.yaml

# 启动
./milvus run standalone

# 添加 Milvus Helm 仓库
helm repo add milvus https://zilliztech.github.io/milvus-helm/
helm repo update

# 安装单机版
helm install my-milvus milvus/milvus --set cluster.enabled=false

# 安装集群版
helm install my-milvus milvus/milvus --set cluster.enabled=true

# 查看 Pods
kubectl get pods

# values.yaml 关键配置
cluster:
  enabled: true
  replica:
    rootCoord: 1
    dataCoord: 1
    queryCoord: 1
    indexCoord: 1
    dataNode: 2
    queryNode: 2
    indexNode: 1
    proxy: 2

# 资源限制
resources:
  limits:
    cpu: "4"
    memory: "8Gi"
  requests:
    cpu: "2"
    memory: "4Gi"

# 存储配置
persistence:
  enabled: true
  storageClass: "standard"
  accessMode: ReadWriteOnce
  size: 50Gi

from pymilvus import connections

connections.connect(
    alias="default",
    uri="https://your-cluster-endpoint",
    token="api-key"
)

# 安装最新版本
pip install pymilvus

# 安装指定版本（推荐与服务端版本匹配）
pip install pymilvus==2.3.7

# 验证安装
python -c "from pymilvus import __version__; print(__version__)"

from pymilvus import connections, utility

# 连接本地 Milvus
connections.connect(
    alias="default",
    host="localhost",
    port="19530"
)

# 连接带认证的 Milvus
connections.connect(
    alias="default",
    host="localhost",
    port="19530",
    user="root",
    password="Milvus"
)

# 检查连接状态
print(utility.get_server_version())

# 断开连接
connections.disconnect("default")

from pymilvus import MilvusClient

# 本地文件模式（Milvus Lite）
client = MilvusClient(uri="./milvus_demo.db")

# 连接远程服务
client = MilvusClient(
    uri="http://localhost:19530",
    token="root:Milvus"
)

# Zilliz Cloud
client = MilvusClient(
    uri="https://your-cluster.zillizcloud.com:19530",
    token="your-api-key"
)

collection_name = "quick_start"

# 删除已存在的集合
if client.has_collection(collection_name):
    client.drop_collection(collection_name)

# 快速创建（自动生成 Schema）
client.create_collection(
    collection_name=collection_name,
    dimension=768,  # 向量维度
    metric_type="COSINE",  # 相似度度量
    auto_id=True,  # 自动生成主键
    enable_dynamic_field=True  # 支持动态字段
)

print(f"集合 {collection_name} 创建成功")

from pymilvus import Collection, CollectionSchema, FieldSchema, DataType

# 定义字段
fields = [
    FieldSchema(
        name="book_id",
        dtype=DataType.INT64,
        is_primary=True,
        auto_id=False,
        description="图书ID"
    ),
    FieldSchema(
        name="word_count",
        dtype=DataType.INT64,
        description="字数"
    ),
    FieldSchema(
        name="book_intro",
        dtype=DataType.FLOAT_VECTOR,
        dim=768,
        description="图书简介向量"
    ),
    FieldSchema(
        name="title",
        dtype=DataType.VARCHAR,
        max_length=256,
        description="书名"
    )
]

# 创建 Schema
schema = CollectionSchema(
    fields=fields,
    description="图书知识库",
    enable_dynamic_field=True
)

# 创建集合
collection = Collection(
    name="book_collection",
    schema=schema,
    using="default"  # 使用的连接别名
)

import numpy as np

# 生成模拟数据
data = []
for i in range(1000):
    data.append({
        "book_id": i,
        "word_count": i * 100 + 5000,
        "book_intro": np.random.rand(768).tolist(),
        "title": f"Book Title #{i}",
        "author": f"Author {i % 10}",  # 动态字段
        "publish_year": 2020 + (i % 5)  # 动态字段
    })

# 批量插入
res = client.insert(
    collection_name="quick_start",
    data=data
)

print(f"插入成功: {res['insert_count']} 条")
print(f"插入耗时: {res['cost']} ms")

import random

# 按列组织数据（Milvus 列存格式）
entities = [
    [i for i in range(1000)],  # book_id
    [random.randint(1000, 10000) for _ in range(1000)],  # word_count
    [[random.random() for _ in range(768)] for _ in range(1000)],  # book_intro
    [f"Book {i}" for i in range(1000)]  # title
]

# 插入数据
insert_result = collection.insert(entities)
print(f"插入了 {insert_result.insert_count} 条数据")

# 手动 flush（确保数据可查询）
collection.flush()

# 定义索引参数
index_params = {
    "index_type": "HNSW",
    "metric_type": "COSINE",
    "params": {
        "M": 16,  # 每层最大邻居数
        "efConstruction": 256  # 构建时搜索邻居的范围
    }
}

# 创建索引
client.create_index(
    collection_name="quick_start",
    index_params=index_params
)

# 查看索引信息
index_info = client.describe_index("quick_start")
print(index_info)

# 准备查询向量
query_vector = np.random.rand(768).tolist()

# 加载集合到内存（必须）
client.load_collection("quick_start")

# 执行搜索
search_result = client.search(
    collection_name="quick_start",
    data=[query_vector],  # 查询向量列表
    limit=10,  # 返回 Top 10
    output_fields=["book_id", "title", "word_count"],  # 返回的字段
    search_params={
        "metric_type": "COSINE",
        "params": {"ef": 128}  # HNSW 搜索参数
    }
)

# 解析结果
print("搜索结果:")
for hits in search_result:
    for hit in hits:
        print(f"ID: {hit['id']}, Score: {hit['distance']:.4f}, "
              f"Title: {hit['entity']['title']}")

# 只搜索 2022 年以后出版，且字数大于 5000 的图书
search_result = client.search(
    collection_name="quick_start",
    data=[query_vector],
    limit=10,
    filter="publish_year >= 2022 and word_count > 5000",
    output_fields=["book_id", "title", "word_count", "publish_year"]
)

print("过滤后的搜索结果:")
for hits in search_result:
    for hit in hits:
        print(f"ID: {hit['id']}, Score: {hit['distance']:.4f}, "
              f"Year: {hit['entity']['publish_year']}, "
              f"Word Count: {hit['entity']['word_count']}")

# 查询特定条件的实体
query_result = client.query(
    collection_name="quick_start",
    filter="word_count > 8000 and publish_year == 2023",
    output_fields=["book_id", "title", "word_count"],
    limit=100
)

print(f"查询到 {len(query_result)} 条结果")
for item in query_result:
    print(f"ID: {item['book_id']}, Title: {item['title']}, "
          f"Word Count: {item['word_count']}")

# 准备 upsert 数据
upsert_data = [
    {
        "book_id": 999,  # 已存在的 ID
        "word_count": 99999,  # 更新字段
        "book_intro": np.random.rand(768).tolist(),
        "title": "Updated Book Title"
    },
    {
        "book_id": 1000,  # 新 ID
        "word_count": 10000,
        "book_intro": np.random.rand(768).tolist(),
        "title": "New Book"
    }
]

# 执行 Upsert
res = client.upsert(
    collection_name="quick_start",
    data=upsert_data
)

print(f"Upsert 完成，影响 {res['upsert_count']} 条数据")

# 按表达式删除
client.delete(
    collection_name="quick_start",
    filter="book_id < 100"
)

# 按主键删除
client.delete(
    collection_name="quick_start",
    pks=[100, 101, 102]
)

# 按主键获取实体
entities = client.get(
    collection_name="quick_start",
    ids=[500, 501, 502],
    output_fields=["book_id", "title", "word_count"]
)

print(f"获取到 {len(entities)} 条实体")
for entity in entities:
    print(entity)

# 列出所有集合
collections = client.list_collections()
print("所有集合:", collections)

# 检查集合是否存在
exists = client.has_collection("quick_start")
print(f"集合是否存在: {exists}")

# 获取集合详情
desc = client.describe_collection("quick_start")
print(f"集合详情: {desc}")

# 获取集合统计信息
stats = client.get_collection_stats("quick_start")
print(f"集合统计: {stats}")

# 加载到内存（查询前必须）
client.load_collection("quick_start")

# 异步加载
client.load_collection("quick_start", _async=True)

# 加载指定分区
client.load_partitions("quick_start", ["partition_1"])

# 从内存释放（节省内存）
client.release_collection("quick_start")

# 重命名集合（Milvus 2.3+）
client.rename_collection("old_name", "new_name")

# 删除集合
client.drop_collection("collection_to_delete")

index_params = {
    "index_type": "HNSW",
    "metric_type": "COSINE",
    "params": {
        "M": 16,           # 每层邻居数，8-64，越大精度越高但内存越大
        "efConstruction": 256  # 构建时搜索范围，100-500，越大构建越慢精度越高
    }
}

index_params = {
    "index_type": "IVF_FLAT",
    "metric_type": "L2",
    "params": {
        "nlist": 1024  # 聚类中心数量，sqrt(N) ~ 4*sqrt(N)
    }
}

# 搜索时参数
search_params = {
    "metric_type": "L2",
    "params": {
        "nprobe": 32  # 搜索的聚类数量，1-nlist，越大越准越慢
    }
}

index_params = {
    "index_type": "IVF_PQ",
    "metric_type": "L2",
    "params": {
        "nlist": 1024,
        "m": 16,  # 子向量数量，必须能整除维度
        "nbits": 8  # 每个子向量的比特数，通常 8
    }
}

# 比较运算符
"field > value"
"field >= value"
"field < value"
"field <= value"
"field == value"
"field != value"

# 逻辑运算符
"cond1 and cond2"
"cond1 or cond2"
"not cond"

# 集合运算符
"field in [value1, value2, value3]"
"field not in [value1, value2]"

# 字符串匹配（Milvus 2.4+）
"field like 'prefix%'"
"field like '%suffix'"
"field like '%contains%'"

# 复杂过滤示例
filter_expr = """
    (category == 1 or category == 2) and 
    view_count > 1000 and 
    create_time >= 1700000000 and 
    title like 'Tech%'
"""

results = client.search(
    collection_name="articles",
    data=[query_vector],
    limit=20,
    filter=filter_expr,
    output_fields=["title", "view_count", "category"]
)

# 动态字段中的 JSON 过滤（Milvus 2.4+）
filter_expr = "meta['tags'] contains 'python' and meta['rating'] > 4.5"

results = client.search(
    collection_name="articles",
    data=[query_vector],
    limit=10,
    filter=filter_expr
)

# 第一页
page1 = client.query(
    collection_name="quick_start",
    filter="word_count > 5000",
    output_fields=["book_id", "title"],
    limit=10,
    offset=0
)

# 第二页
page2 = client.query(
    collection_name="quick_start",
    filter="word_count > 5000",
    output_fields=["book_id", "title"],
    limit=10,
    offset=10
)

# 使用游标分页
all_results = []
cursor = None

while True:
    batch = client.query(
        collection_name="quick_start",
        filter="word_count > 5000",
        output_fields=["book_id", "title"],
        limit=100,
        cursor=cursor
    )

    if not batch:
        break

    all_results.extend(batch)
    # 获取下一页游标
    cursor = batch[-1]['book_id']

print(f"总共获取 {len(all_results)} 条记录")

# 建议单次批量大小：1000-5000 条
BATCH_SIZE = 2000

# 模拟大量数据
total_data = []
for i in range(100000):
    total_data.append({
        "id": i,
        "vector": np.random.rand(768).tolist(),
        "text": f"Document {i}"
    })

# 分批插入
for i in range(0, len(total_data), BATCH_SIZE):
    batch = total_data[i:i + BATCH_SIZE]
    res = client.insert("large_collection", batch)
    print(f"已插入 {i + len(batch)}/{len(total_data)} 条")

# 最后手动 flush
client.flush("large_collection")

# 使用 Milvus 官方 bulk load 工具
milvus_cli tools import \
    --path ./data.json \
    --collection large_collection \
    --batch-size 10000

# 实时统计集合中的实体数量（无需 flush）
count = client.query(
    collection_name="quick_start",
    filter="",
    output_fields=["count(*)"]
)
print(f"集合总数据量: {count[0]['count(*)']}")

# 带条件统计
count_filtered = client.query(
    collection_name="quick_start",
    filter="word_count > 5000",
    output_fields=["count(*)"]
)
print(f"满足条件的数据量: {count_filtered[0]['count(*)']}")

精确最近邻（NN）：100% 准确率，但搜索速度 O(N)
         ↓
近似最近邻（ANN）：90%-99% 准确率，搜索速度 O(log N)

Layer 2 (顶层)：  A─────B           稀疏连接，用于快速导航
                   │     │
Layer 1：         A─────B─────C     中等连接密度
                   │     │     │
Layer 0 (底层)：  A──D──B──E──C    稠密连接，存储所有点

1. 从顶层的一个入口点开始
2. 在当前层进行贪心搜索，找到最近的节点
3. 下降到下一层，以上一层的最近节点为起点
4. 重复步骤 2-3，直到底层
5. 在底层扩展搜索，返回 Top K 结果

召回率要求 90%: ef ≈ 100
召回率要求 95%: ef ≈ 200
召回率要求 99%: ef ≈ 500

1. 使用 K-Means 将所有向量聚类为 nlist 个中心
   ┌─────────────────────────────────────────┐
   │    •     •     •     •     •     •      │ 向量空间
   │      •    ○    •    ○    •    ○        │ ○ = 聚类中心
   │    •     •     •     •     •     •      │
   └─────────────────────────────────────────┘

2. 为每个聚类中心维护一个倒排列表
   Cluster 0: [vec1, vec5, vec9, ...]
   Cluster 1: [vec2, vec3, vec7, ...]
   ...
   Cluster N: [vec4, vec6, vec8, ...]

1. 计算查询向量与所有 nlist 个聚类中心的距离
2. 选择距离最近的 nprobe 个聚类
3. 只在这 nprobe 个聚类内进行精确搜索

原始向量 (128维 × 4字节) = 512字节
     ↓ 切分为 m 个子向量
子向量0 (16维) → 聚类 → 码本索引 (1字节)
子向量1 (16维) → 聚类 → 码本索引 (1字节)
...
子向量m-1 (16维) → 聚类 → 码本索引 (1字节)
     ↓
压缩后大小 = m 字节 (m=16 时，压缩 32 倍！)

预计算：查询子向量 vs 码本中所有中心的距离
查表：通过码本索引快速获取距离值
求和：所有子向量的距离之和即为近似距离

import time
import numpy as np

def benchmark_search(client, collection_name, nq=1000, limit=10):
    """性能测试工具"""

    # 生成查询向量
    query_vectors = np.random.rand(nq, 768).tolist()

    # 预热
    client.search(collection_name, [query_vectors[0]], limit=limit)

    # 开始测试
    start_time = time.time()

    for vec in query_vectors:
        client.search(
            collection_name,
            [vec],
            limit=limit,
            search_params={"params": {"ef": 128}}
        )

    total_time = time.time() - start_time
    qps = nq / total_time
    avg_latency = total_time * 1000 / nq

    print(f"QPS: {qps:.2f}")
    print(f"Average Latency: {avg_latency:.2f} ms")
    print(f"Total Time: {total_time:.2f} s")

    return qps, avg_latency

索引内存 ≈ 向量数 × (维度 × 4字节 + 索引开销)

HNSW 开销: ~ 8-10 × 向量大小
IVF_FLAT 开销: ~ 1.2 × 向量大小
IVF_PQ 开销: 极低，取决于压缩率

示例：100万 × 768维 × HNSW
内存 ≈ 1,000,000 × (768 × 4) × 10 ≈ 30 GB

数据规模？
├─ < 100万 → 选择 FLAT（100%召回）或 HNSW（高性能）
├─ 100万 - 1亿 → 选择 HNSW（首选）或 IVF_FLAT
└─ > 1亿 → 
    内存充足？
    ├─ 是 → HNSW
    └─ 否 → 
        能接受精度损失？
        ├─ 是 → IVF_PQ / IVF_SQ8
        └─ 否 → DISKANN（基于磁盘）

# 批量查询比单条查询快 2-5 倍
results = client.search(
    collection_name="my_collection",
    data=[vec1, vec2, vec3, ..., vec100],  # 一次查 100 个向量
    limit=10
)

# 对一致性要求不高的场景，使用最终一致可提升 30% 性能
search_result = client.search(
    collection_name="my_collection",
    data=[query_vector],
    limit=10,
    consistency_level="Eventually"  # 最快
)

# 将数据按时间或类别分区，查询时只扫描相关分区
results = client.search(
    collection_name="logs",
    data=[query_vector],
    limit=10,
    partition_names=["2024_05"]  # 只查 5 月份的数据
)

# 创建分区
client.create_partition(
    collection_name="article_collection",
    partition_name="2024_01"
)

# 列出所有分区
partitions = client.list_partitions("article_collection")
print(partitions)

# 插入到指定分区
client.insert(
    collection_name="article_collection",
    data=data_batch,
    partition_name="2024_01"
)

# 只查询指定分区（大幅提升性能）
results = client.search(
    collection_name="article_collection",
    data=[query_vector],
    limit=10,
    partition_names=["2024_01", "2024_02"]  # 只查这两个月
)

# 创建集合时设置 TTL
client.create_collection(
    collection_name="temp_logs",
    dimension=512,
    ttl=3600  # 1 小时后自动过期（秒）
)

# 修改现有集合的 TTL
client.alter_collection(
    collection_name="temp_logs",
    ttl=86400  # 改为 1 天
)

class MultiTenantMilvus:
    def __init__(self, client, base_collection):
        self.client = client
        self.base_collection = base_collection

    def _get_partition_name(self, tenant_id):
        return f"tenant_{tenant_id}"

    def insert_for_tenant(self, tenant_id, data):
        partition = self._get_partition_name(tenant_id)
        if not self.client.has_partition(self.base_collection, partition):
            self.client.create_partition(self.base_collection, partition)

        return self.client.insert(
            self.base_collection,
            data,
            partition_name=partition
        )

    def search_for_tenant(self, tenant_id, query_vector, limit=10):
        partition = self._get_partition_name(tenant_id)
        return self.client.search(
            self.base_collection,
            [query_vector],
            limit=limit,
            partition_names=[partition]
        )

# 创建集合时开启动态字段
client.create_collection(
    collection_name="flexible_data",
    dimension=768,
    enable_dynamic_field=True  # 关键参数
)

# 插入包含任意字段的数据
data = [
    {
        "id": 1,
        "vector": [...],
        "name": "Alice",  # 动态字段
        "age": 30,        # 动态字段
        "tags": ["a", "b"],  # 动态字段
        "extra": {"k": "v"}  # 嵌套 JSON
    }
]

client.insert("flexible_data", data)

# 动态字段也可以用于过滤
results = client.search(
    collection_name="flexible_data",
    data=[query_vector],
    limit=10,
    filter="age > 25 and tags contains 'a'",  # 使用动态字段过滤
    output_fields=["name", "age", "tags"]  # 返回动态字段
)

# 安装备份工具
pip install milvus-backup

# 创建备份
milvus-backup create \
    --name backup_20240519 \
    --collections collection1,collection2 \
    --path ./backups

# 恢复备份
milvus-backup restore \
    --name backup_20240519 \
    --path ./backups

# 列出备份
milvus-backup list

# 查询并导出所有数据
def export_collection(client, collection_name, batch_size=1000):
    """导出集合数据"""
    all_data = []
    offset = 0

    while True:
        batch = client.query(
            collection_name=collection_name,
            filter="",
            output_fields=["*"],
            limit=batch_size,
            offset=offset
        )

        if not batch:
            break

        all_data.extend(batch)
        offset += len(batch)

    return all_data

# 导出到 JSON
import json
data = export_collection(client, "my_collection")
with open("backup.json", "w") as f:
    json.dump(data, f)

# Query 相关指标
milvus_query_latency_ms         # 查询延迟分布
milvus_query_qps                # 查询 QPS
milvus_query_recall_rate        # 召回率（需自行计算）

# 写入相关指标
milvus_insert_latency_ms        # 插入延迟
milvus_insert_throughput        # 写入吞吐量（条/秒）

# 资源相关
milvus_memory_usage_bytes       # 内存使用
milvus_cpu_usage_percent        # CPU 使用率
milvus_segment_count            # Segment 数量
milvus_vector_count             # 向量总数

# Prometheus Alert Rules
groups:
  - name: milvus_alerts
    rules:
      - alert: MilvusHighLatency
        expr: milvus_query_latency_ms{quantile="0.99"} > 1000
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Milvus 查询 P99 延迟 > 1s"

      - alert: MilvusHighMemoryUsage
        expr: milvus_memory_usage_bytes / 1024^3 > 0.9 * total_memory_gb
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "Milvus 内存使用率 > 90%"

      - alert: MilvusSegmentGrowing
        expr: milvus_segment_count{type="growing"} > 1000
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Growing Segment 数量过多，可能存在 Flush 问题"

写入数据流
    ↓
Growing Segment（增量索引）
    ↓ 达到大小阈值（默认 512MB）
Flush → Sealed Segment
    ↓
后台构建索引
    ↓
Sealed Segment（历史索引）

用户问题
    ↓
Embedding 模型 → 查询向量
    ↓
Milvus 向量检索 → Top K 相关文档
    ↓
Prompt 组装：问题 + 上下文
    ↓
大语言模型 → 生成答案

from pymilvus import MilvusClient
from sentence_transformers import SentenceTransformer
import openai
import numpy as np

class RAGSystem:
    def __init__(
        self,
        milvus_uri="./milvus_rag.db",
        collection_name="knowledge_base",
        embedding_model="BAAI/bge-small-zh-v1.5",
        openai_api_key="your-api-key"
    ):
        # 初始化 Milvus
        self.client = MilvusClient(milvus_uri)
        self.collection_name = collection_name

        # 初始化 Embedding 模型
        self.embedding_model = SentenceTransformer(embedding_model)
        self.dim = self.embedding_model.get_sentence_embedding_dimension()

        # 初始化 OpenAI
        openai.api_key = openai_api_key

        # 初始化集合
        self._init_collection()

    def _init_collection(self):
        if not self.client.has_collection(self.collection_name):
            self.client.create_collection(
                collection_name=self.collection_name,
                dimension=self.dim,
                metric_type="COSINE",
                enable_dynamic_field=True
            )

            # 创建索引
            self.client.create_index(
                self.collection_name,
                {
                    "index_type": "HNSW",
                    "metric_type": "COSINE",
                    "params": {"M": 16, "efConstruction": 256}
                }
            )

        self.client.load_collection(self.collection_name)

    def add_documents(self, documents, batch_size=32):
        """添加文档到知识库"""
        data = []

        for i in range(0, len(documents), batch_size):
            batch = documents[i:i + batch_size]

            # 批量生成 Embedding
            embeddings = self.embedding_model.encode(
                [doc["content"] for doc in batch],
                normalize_embeddings=True
            )

            for j, doc in enumerate(batch):
                data.append({
                    "id": i + j,
                    "vector": embeddings[j].tolist(),
                    "title": doc["title"],
                    "content": doc["content"],
                    "source": doc.get("source", ""),
                    "category": doc.get("category", "general")
                })

        # 批量插入
        res = self.client.insert(self.collection_name, data)
        print(f"成功添加 {res['insert_count']} 条文档")
        return res

    def retrieve(self, query, top_k=5, filter_expr=None):
        """检索相关文档"""
        # 生成查询向量
        query_vec = self.embedding_model.encode(
            query,
            normalize_embeddings=True
        ).tolist()

        # 检索
        results = self.client.search(
            collection_name=self.collection_name,
            data=[query_vec],
            limit=top_k,
            filter=filter_expr,
            output_fields=["title", "content", "source", "category"],
            search_params={"params": {"ef": 128}}
        )

        return results[0]

    def generate_answer(self, query, retrieved_docs):
        """基于检索结果生成答案"""
        # 构建上下文
        context = "\n\n".join([
            f"文档 {i+1} (相似度: {hit['distance']:.3f}):\n"
            f"标题: {hit['entity']['title']}\n"
            f"内容: {hit['entity']['content']}"
            for i, hit in enumerate(retrieved_docs)
        ])

        # 构建 Prompt
        prompt = f"""你是一个专业的问答助手。请基于以下文档回答用户的问题。
如果文档中没有相关信息，请回答"根据现有资料无法回答"。

参考文档：
{context}

用户问题：{query}

回答："""

        # 调用 LLM
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.7,
            max_tokens=1024
        )

        return response.choices[0].message.content

    def query(self, question, top_k=5, filter_expr=None):
        """完整的 RAG 流程"""
        # 1. 检索
        retrieved = self.retrieve(question, top_k, filter_expr)

        # 2. 生成答案
        answer = self.generate_answer(question, retrieved)

        return {
            "answer": answer,
            "sources": [
                {
                    "title": hit["entity"]["title"],
                    "source": hit["entity"]["source"],
                    "similarity": hit["distance"]
                }
                for hit in retrieved
            ]
        }


# 使用示例
if __name__ == "__main__":
    # 初始化 RAG 系统
    rag = RAGSystem(
        milvus_uri="./milvus_rag.db",
        openai_api_key="your-api-key"
    )

    # 添加示例文档
    sample_docs = [
        {
            "title": "Milvus 简介",
            "content": "Milvus 是一个开源的向量数据库，专为 AI 应用设计，支持十亿级向量的毫秒级检索。",
            "source": "官网",
            "category": "技术"
        },
        {
            "title": "RAG 技术原理",
            "content": "检索增强生成（RAG）结合了向量检索和大语言模型，通过检索相关文档来增强生成答案的准确性。",
            "source": "技术博客",
            "category": "AI"
        }
    ]

    rag.add_documents(sample_docs)

    # 提问
    result = rag.query("什么是 Milvus 和 RAG？")
    print("回答:", result["answer"])
    print("\n参考来源:")
    for src in result["sources"]:
        print(f"- {src['title']} (相似度: {src['similarity']:.3f})")

from pymilvus import AnnSearchRequest, WeightedRanker

# 构建两个检索请求
dense_req = AnnSearchRequest(
    data=[dense_vector],
    ann_field="dense_vector",
    param={"metric_type": "COSINE", "params": {"ef": 128}},
    limit=100
)

sparse_req = AnnSearchRequest(
    data=[sparse_vector],
    ann_field="sparse_vector",
    param={"metric_type": "IP", "params": {"drop_ratio_search": 0.2}},
    limit=100
)

# 加权融合重排序
rerank = WeightedRanker(0.7, 0.3)  # 70% 语义 + 30% 关键词

# 执行混合检索
results = client.hybrid_search(
    collection_name="hybrid_rag",
    reqs=[dense_req, sparse_req],
    ranker=rerank,
    limit=10,
    output_fields=["title", "content"]
)

class RecommendationSystem:
    def __init__(self, client):
        self.client = client
        self.item_collection = "item_vectors"
        self.user_collection = "user_vectors"

    def get_user_recommendations(self, user_id, top_n=20):
        """基于用户向量获取推荐"""
        # 获取用户向量
        user = self.client.get(
            self.user_collection,
            ids=[user_id],
            output_fields=["user_vector"]
        )

        if not user:
            return []

        user_vector = user[0]["user_vector"]

        # 检索相似物品
        results = self.client.search(
            collection_name=self.item_collection,
            data=[user_vector],
            limit=top_n,
            filter="is_active == true",  # 只推荐上架的商品
            output_fields=["item_id", "name", "category", "price"]
        )

        return results[0]

    def get_similar_items(self, item_id, top_n=10):
        """相似物品推荐"""
        item = self.client.get(
            self.item_collection,
            ids=[item_id],
            output_fields=["item_vector", "category"]
        )

        if not item:
            return []

        # 同类目优先推荐
        category_filter = f"category == '{item[0]['category']}'"

        results = self.client.search(
            collection_name=self.item_collection,
            data=[item[0]["item_vector"]],
            limit=top_n + 1,  # +1 排除自己
            filter=category_filter,
            output_fields=["item_id", "name", "price"]
        )

        # 排除查询的物品自身
        return [hit for hit in results[0] if hit["id"] != item_id][:top_n]

from PIL import Image
import torch
from torchvision import models, transforms

class ImageSearchEngine:
    def __init__(self, milvus_client, collection_name="images"):
        self.client = milvus_client
        self.collection_name = collection_name

        # 加载预训练的 ResNet 模型
        self.model = models.resnet50(pretrained=True)
        self.model.eval()

        # 图像预处理
        self.transform = transforms.Compose([
            transforms.Resize(256),
            transforms.CenterCrop(224),
            transforms.ToTensor(),
            transforms.Normalize(
                mean=[0.485, 0.456, 0.406],
                std=[0.229, 0.224, 0.225]
            )
        ])

        self.dim = 2048  # ResNet50 特征维度
        self._init_collection()

    def _init_collection(self):
        if not self.client.has_collection(self.collection_name):
            fields = [
                FieldSchema("image_id", DataType.INT64, is_primary=True),
                FieldSchema("vector", DataType.FLOAT_VECTOR, dim=self.dim),
                FieldSchema("path", DataType.VARCHAR, max_length=512),
                FieldSchema("category", DataType.VARCHAR, max_length=128)
            ]
            schema = CollectionSchema(fields)
            self.client.create_collection(self.collection_name, schema=schema)

            # 创建索引
            self.client.create_index(
                self.collection_name,
                {"index_type": "HNSW", "metric_type": "L2", "params": {"M": 32}}
            )

        self.client.load_collection(self.collection_name)

    def extract_feature(self, image_path):
        """提取图像特征向量"""
        img = Image.open(image_path).convert('RGB')
        img_tensor = self.transform(img).unsqueeze(0)

        with torch.no_grad():
            features = self.model(img_tensor)
            features = torch.nn.functional.normalize(features, p=2, dim=1)

        return features.squeeze().numpy().tolist()

    def add_image(self, image_id, image_path, category="general"):
        """添加图像到索引"""
        vector = self.extract_feature(image_path)

        self.client.insert(
            self.collection_name,
            [{
                "image_id": image_id,
                "vector": vector,
                "path": image_path,
                "category": category
            }]
        )

    def search_similar(self, query_image_path, top_k=10, category=None):
        """搜索相似图像"""
        query_vector = self.extract_feature(query_image_path)

        filter_expr = f"category == '{category}'" if category else None

        results = self.client.search(
            collection_name=self.collection_name,
            data=[query_vector],
            limit=top_k,
            filter=filter_expr,
            output_fields=["path", "category"],
            search_params={"params": {"ef": 128}}
        )

        return results[0]

class SemanticSearch:
    def __init__(self, client, embedding_model_name):
        self.client = client
        self.collection_name = "semantic_docs"
        self.embedding_model = SentenceTransformer(embedding_model_name)

    def search(self, query, top_k=20, rerank=True):
        """语义搜索 + 重排序"""
        # 1. 向量召回
        query_vector = self.embedding_model.encode(query).tolist()

        candidates = self.client.search(
            collection_name=self.collection_name,
            data=[query_vector],
            limit=100,  # 召回更多候选
            output_fields=["title", "content"]
        )[0]

        if not rerank:
            return candidates[:top_k]

        # 2. Cross-Encoder 精排（可选）
        from sentence_transformers import CrossEncoder
        reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

        # 构建查询-文档对
        pairs = [(query, hit['entity']['content']) for hit in candidates]
        scores = reranker.predict(pairs)

        # 重新排序
        reranked = sorted(
            zip(candidates, scores),
            key=lambda x: x[1],
            reverse=True
        )

        return [item[0] for item in reranked[:top_k]]

# 1. 检查容器状态
docker-compose ps

# 2. 查看日志
docker-compose logs milvus-standalone

# 3. 测试端口连通性
telnet localhost 19530

# 4. 检查 etcd 状态
docker exec milvus-etcd etcdctl endpoint health

# 重启服务
docker-compose down
docker-compose up -d

# 等待服务就绪（约 30-60 秒）
sleep 30

# 确认用户名密码正确
connections.connect(
    host="localhost",
    port="19530",
    user="root",
    password="Milvus"  # 默认密码
)

# 重置 root 密码（需要进入容器）
docker exec -it milvus-standalone /milvus/bin/milvus-cli reset-root-password

# 1. 手动 flush（不建议频繁调用）
collection.flush()

# 2. 等待一小段时间
import time
time.sleep(2)  # 等待 2 秒

# 3. 调整一致性级别
results = client.search(
    ...,
    consistency_level="Strong"  # 强一致，确保能查到最新数据
)

# 1. 检查集合是否加载
load_state = client.get_load_state("my_collection")
print(f"加载状态: {load_state}")

# 2. 检查索引状态
index_info = client.describe_index("my_collection")
print(f"索引状态: {index_info}")

# 3. 先查询看是否有数据
count = client.query("my_collection", "", ["count(*)"])
print(f"数据量: {count}")

# 4. 验证向量维度
desc = client.describe_collection("my_collection")
print(f"向量维度: {desc['schema']['fields'][1]['params']['dim']}")

# 1. 检查索引类型（FLAT 比 HNSW 慢很多）
# 确保使用了 HNSW 或 IVF 索引

# 2. 调整搜索参数
search_params = {
    "metric_type": "COSINE",
    "params": {
        "ef": 64,  # 降低 ef 可提速，默认 128
        # 或
        "nprobe": 16  # IVF 索引降低 nprobe
    }
}

# 3. 使用更低的一致性级别
results = client.search(
    ...,
    consistency_level="Eventually"
)

# 4. 检查内存是否足够（OOM 会导致严重降速）
# free -h

# 1. 释放不常用的集合
client.release_collection("unused_collection")

# 2. 使用更省内存的索引
# HNSW → IVF_SQ8 → IVF_PQ
index_params = {
    "index_type": "IVF_PQ",  # PQ 压缩比最高
    "metric_type": "L2",
    "params": {"nlist": 1024, "m": 16}
}

# 3. 使用分区，只加载需要的分区
client.release_partitions("large_collection", ["old_partition"])

# 4. 开启 MMap（Milvus 2.3+）
# 修改配置文件：
# queryNode:
#   mmap:
#     enabled: true

def estimate_memory(num_vectors, dim, index_type="HNSW"):
    """估算索引内存占用（GB）"""
    vector_size = num_vectors * dim * 4  # 4 bytes per float32

    if index_type == "FLAT":
        overhead = 1.0
    elif index_type == "IVF_FLAT":
        overhead = 1.1
    elif index_type == "IVF_SQ8":
        overhead = 0.28  # 约 1/4
    elif index_type == "IVF_PQ":
        overhead = 0.125  # m=16, dim 倍数时约 1/8
    elif index_type == "HNSW":
        overhead = 10.0  # HNSW 内存开销大
    else:
        overhead = 1.0

    return vector_size * overhead / 1024**3

# 示例：100万 × 768维 HNSW 索引
print(f"预估内存: {estimate_memory(1000000, 768, 'HNSW'):.2f} GB")
# 输出: 预估内存: 28.61 GB

# 并发写入示例
from concurrent.futures import ThreadPoolExecutor

def insert_batch(batch):
    return client.insert("my_collection", batch)

# 使用 8 线程并发写入
with ThreadPoolExecutor(max_workers=8) as executor:
    futures = [executor.submit(insert_batch, batch) for batch in batches]
    results = [f.result() for f in futures]

# 更快的索引参数（牺牲一些精度）
index_params = {
    "index_type": "HNSW",
    "metric_type": "COSINE",
    "params": {
        "M": 8,              # 从 16 降到 8
        "efConstruction": 128  # 从 256 降到 128
    }
}

# 触发 Compaction（Milvus 2.3+）
client.compact_collection("my_collection")

# 等待 Compaction 完成
import time
time.sleep(60)  # 等待 1 分钟

# 查看 Segment 数量变化
stats = client.get_collection_stats("my_collection")
print(stats)

# 查看服务端版本
python -c "from pymilvus import utility; print(utility.get_server_version())"

# 安装匹配版本
pip install pymilvus==2.3.7  # 对应服务端 2.3.x

# Docker 部署
docker logs milvus-standalone

# 查看最近 100 行错误日志
docker logs milvus-standalone --tail 100 2>&1 | grep -i error

# 进入容器查看详细日志
docker exec -it milvus-standalone bash
cd /var/lib/milvus/logs/
ls -la

客户端 → Proxy → 消息队列（Pulsar）→ DataNode
                                          ↓
                                   写入 Growing Segment
                                          ↓
                                 达到阈值触发 Flush
                                          ↓
                              写入对象存储（MinIO）→ Sealed Segment
                                          ↓
                              IndexNode 异步构建索引

客户端 → Proxy → QueryCoord（分发）→ QueryNode
                                          ↓
                              Growing Segment（暴力搜）
                              Sealed Segment（ANN索引搜）
                                          ↓
                                    结果合并返回

数据量小 (< 100万) → FLAT（100%召回）
数据量大 →
    内存充足？
    ├─ 是 → HNSW（首选，高性能高精度）
    └─ 否 →
        能接受精度损失？
        ├─ 是 → IVF_PQ / IVF_SQ8
        └─ 否 → 加内存用 HNSW 或 DISKANN

from pymilvus import connections, utility, MilvusClient

# 传统方式
connections.connect(
    alias="default",
    host="localhost",
    port="19530",
    user="root",
    password="Milvus"
)

# MilvusClient 方式
client = MilvusClient(uri="http://localhost:19530", token="root:Milvus")

# 创建
client.create_collection(collection_name, dimension)

# 删除
client.drop_collection(collection_name)

# 检查存在
client.has_collection(collection_name)

# 列出所有
client.list_collections()

# 加载/释放
client.load_collection(collection_name)
client.release_collection(collection_name)

# 插入
client.insert(collection_name, data)

# 更新/插入
client.upsert(collection_name, data)

# 删除
client.delete(collection_name, filter="id > 100")

# 查询
client.query(collection_name, filter, output_fields, limit)

# 搜索
client.search(collection_name, data, limit, filter, output_fields)

# 获取
client.get(collection_name, ids)

# 创建
client.create_index(
    collection_name,
    index_params={
        "index_type": "HNSW",
        "metric_type": "COSINE",
        "params": {"M": 16, "efConstruction": 256}
    }
)

# 查看
client.describe_index(collection_name)

# 删除
client.drop_index(collection_name)