> For the complete documentation index, see [llms.txt](https://docs.clore.ai/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.clore.ai/guides/guides_v2-zh/rag-yu-xiang-liang-shu-ju-ku/milvus.md).

# Milvus

> **面向 AI 应用、最具可扩展性的开源向量数据库——可扩展至数十亿个向量**

Milvus 是一款专为可扩展相似度搜索和 AI 应用而打造的开源向量数据库。它最初由 Zilliz 创建，后捐赠给 LF AI & Data Foundation，Milvus 为包括 NVIDIA、AT\&T、IBM 和 Salesforce 在内的公司提供生产级 AI 工作负载支持。当你需要扩展到数十亿个向量时，它是不二之选。

**GitHub：** [milvus-io/milvus](https://github.com/milvus-io/milvus) — 32K+ ⭐

***

## Milvus vs Qdrant——何时选择谁

| 标准       | Milvus         | Qdrant       |
| -------- | -------------- | ------------ |
| 规模       | 数十亿个向量         | 数亿个          |
| 架构       | 分布式（多个服务）      | 单一二进制文件      |
| 设置复杂度    | 更高             | 将            |
| GPU 索引支持 | ✅ 原生 GPU FAISS | 有限           |
| 多租户      | ✅ 分区 + 别名      | 基于集合         |
| 流式摄取     | ✅ Kafka/Pulsar | 有限           |
| 混合搜索     | ✅ 稠密 + 稀疏      | ✅            |
| 云托管选项    | Zilliz Cloud   | Qdrant Cloud |

{% hint style="success" %}
**当以下情况选择 Milvus：** 你需要扩展到数十亿个向量，需要 GPU 加速索引（IVF\_FLAT\_GPU），或者需要多租户、流式摄取和基于角色的访问控制等企业级功能。
{% endhint %}

***

## Milvus 架构

单机模式下的 Milvus（单服务器）包含：

* **milvus** — 主服务（代理、查询、数据、索引协调器）
* **etcd** — 元数据存储和服务发现
* **MinIO** — 用于段数据的对象存储

在分布式模式（集群）下，每个组件都可独立扩展。

***

## 前提条件

* 带 GPU 租赁的 Clore.ai 账户
* Docker Compose（通常已预装）
* 基础 Python 知识
* 16GB+ 内存（生产环境建议 32GB）

***

## 第 1 步——在 Clore.ai 上租用 GPU 服务器

1. 前往 [clore.ai](https://clore.ai) → **市场**
2. **推荐 GPU：** RTX 4090 或 A100，用于 GPU 加速索引
3. **CPU 替代方案：** 任何带有 32GB+ 内存的服务器，用于基于 CPU 的索引

**最低要求：**

* CPU：8 核
* 内存：16GB（建议 32GB）
* 磁盘：50GB SSD/NVMe
* GPU：可选（仅 GPU 索引类型需要）

{% hint style="info" %}
**Milvus 中的 GPU 索引类型** （IVF\_FLAT\_GPU、IVFSQ8\_GPU）需要支持 CUDA 的 GPU，并能显著加速大集合的索引构建。如果你计划频繁索引 1000 万+ 向量，GPU 索引很快就能回本。
{% endhint %}

***

## 第 2 步——部署 Milvus 单机版

**Docker 镜像：**

```
milvusdb/milvus:v2.4.0
```

Milvus 单机版需要 etcd 和 MinIO。使用 Docker Compose 是最简单的设置方式。

**端口：**

```
22
19530
```

* **端口 19530：** Milvus SDK/gRPC 端口（主端口）
* **端口 9091：** Milvus REST API 和健康检查（内部）

**环境变量：**

```
NVIDIA_VISIBLE_DEVICES=all
NVIDIA_DRIVER_CAPABILITIES=compute,utility
```

***

## 第 3 步——使用 Docker Compose 配置

通过 SSH 登录到你的 Clore.ai 服务器并创建 compose 文件：

```bash
ssh root@<server-ip> -p <ssh-port>

# 如果尚未安装，则安装 Docker Compose
which docker-compose || pip install docker-compose
# 或使用 Docker 插件：
docker compose version

# 创建项目目录
mkdir -p /opt/milvus && cd /opt/milvus

# 下载官方 Milvus 单机版 compose 文件
wget https://github.com/milvus-io/milvus/releases/download/v2.4.0/milvus-standalone-docker-compose.yml \
    -O docker-compose.yml

# 查看 compose 文件
cat docker-compose.yml
```

### 自定义 docker-compose.yml

```yaml
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
      - ETCD_SNAPSHOT_COUNT=50000
    volumes:
      - /opt/milvus/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
    healthcheck:
      test: ["CMD", "etcdctl", "endpoint", "health"]
      interval: 30s
      timeout: 20s
      retries: 3

  minio:
    container_name: milvus-minio
    image: minio/minio:RELEASE.2023-03-13T19-46-17Z
    environment:
      MINIO_ACCESS_KEY: minioadmin
      MINIO_SECRET_KEY: minioadmin
    ports:
      - "9001:9001"
      - "9000:9000"
    volumes:
      - /opt/milvus/minio:/minio_data
    command: minio server /minio_data --console-address ":9001"
    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.4.0
    command: ["milvus", "run", "standalone"]
    security_opt：
      - seccomp:unconfined
    environment:
      ETCD_ENDPOINTS: etcd:2379
      MINIO_ADDRESS: minio:9000
    volumes:
      - /opt/milvus/milvus:/var/lib/milvus
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:9091/healthz"]
      interval: 30s
      start_period: 90s
      timeout: 20s
      retries: 3
    ports:
      - "19530:19530"
      - "9091:9091"
    depends_on:
      - "etcd"
      - "minio"
    deploy:
      resources:
        reservations:
          devices:
            - capabilities: [gpu]  # 启用 GPU 访问
```

### 启动 Milvus

```bash
cd /opt/milvus
docker compose up -d

# 等待服务启动（约 60 秒）
sleep 60

# 检查所有服务是否健康
docker compose ps

# 检查 Milvus 健康状态
curl http://localhost:9091/healthz
# 预期：{"status":"ok"}

# 查看日志
docker compose logs -f standalone --tail 50
```

***

## 第 4 步——安装 Python 客户端

```bash
pip install pymilvus sentence-transformers numpy tqdm

# 验证连接
python3 << 'EOF'
from pymilvus import connections, utility

connections.connect("default", host="localhost", port="19530")
print(f"Milvus 已连接！")
print(f"版本：{utility.get_server_version()}")
EOF
```

***

## 第 5 步——创建集合

在 Milvus 中， **集合** 类似于数据库表。它具有带类型字段的模式，包括向量字段。

```python
from pymilvus import (
    connections,
    FieldSchema,
    CollectionSchema,
    DataType,
    Collection,
    utility
)

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

# 定义模式
fields = [
    FieldSchema(
        name="id",
        dtype=DataType.INT64,
        is_primary=True,
        auto_id=True           # 自动生成 ID
    ),
    FieldSchema(
        name="text",
        dtype=DataType.VARCHAR,
        max_length=2048        # 最大文本长度
    ),
    FieldSchema(
        name="source",
        dtype=DataType.VARCHAR,
        max_length=256
    ),
    FieldSchema(
        name="category",
        dtype=DataType.VARCHAR,
        max_length=128
    ),
    FieldSchema(
        name="year",
        dtype=DataType.INT32
    ),
    FieldSchema(
        name="embedding",
        dtype=DataType.FLOAT_VECTOR,
        dim=384                # 你的嵌入模型维度
    )
]

schema = CollectionSchema(
    fields=fields,
    description="用于语义搜索的文档嵌入",
    enable_dynamic_field=True  # 允许添加不在模式中的字段
)

# 创建集合
collection_name = "documents"
if utility.has_collection(collection_name):
    utility.drop_collection(collection_name)

collection = Collection(
    name=collection_name,
    schema=schema,
    using="default"
)
print(f"集合 '{collection_name}' 已创建！")
```

***

## 第 6 步——创建索引

在加载数据进行搜索之前，先创建合适的索引：

```python
from pymilvus import Collection

collection = Collection("documents")

# HNSW 索引（适用于大多数场景，低延迟）
hnsw_params = {
    "metric_type": "COSINE",     # 余弦、L2 或 IP（内积）
    "index_type": "HNSW",
    "params": {
        "M": 16,                 # HNSW 图连接度（8-64）
        "efConstruction": 200    # 构建时搜索深度
    }
}

# IVF_FLAT 索引（CPU，适合大集合）
ivf_params = {
    "metric_type": "COSINE",
    "index_type": "IVF_FLAT",
    "params": {
        "nlist": 1024            # 聚类数量（通常取数据规模的平方根）
    }
}

# GPU_IVF_FLAT 索引（需要 CUDA GPU——批量查询最快）
gpu_ivf_params = {
    "metric_type": "L2",
    "index_type": "GPU_IVF_FLAT",
    "params": {
        "nlist": 1024,
        "cache_dataset_on_device": True
    }
}

# 在 embedding 字段上创建索引
collection.create_index(
    field_name="embedding",
    index_params=hnsw_params,
    index_name="embedding_idx"
)

# 为过滤搜索创建标量索引
collection.create_index(field_name="category", index_name="category_idx")
collection.create_index(field_name="year", index_name="year_idx")

print("索引已创建！")
collection.load()  # 加载到内存中以便搜索
```

***

## 第 7 步——插入数据

```python
from pymilvus import Collection
from sentence_transformers import SentenceTransformer
import tqdm

collection = Collection("documents")
model = SentenceTransformer("all-MiniLM-L6-v2", device="cuda")

# 你的文档
documents = [
    {
        "text": "Milvus 是一个面向可扩展 AI 应用的开源向量数据库。",
        "source": "文档",
        "category": "数据库",
        "year": 2024
    },
    {
        "text": "HNSW 通过高召回率提供快速的近似最近邻搜索。",
        "source": "研究",
        "category": "算法",
        "year": 2023
    },
    {
        "text": "GPU 加速索引可显著减少大规模向量集合的构建时间。",
        "source": "博客",
        "category": "性能",
        "year": 2024
    },
    # 在这里添加更多数千条文档
]

def insert_batch(docs: list, batch_size: int = 1000):
    texts = [d["text"] for d in docs]
    
    # GPU 加速嵌入
    embeddings = model.encode(
        texts,
        batch_size=256,
        show_progress_bar=False,
        normalize_embeddings=True
    )
    
    # 插入到 Milvus
    data = {
        "text": [d["text"] for d in docs],
        "source": [d["source"] for d in docs],
        "category": [d["category"] for d in docs],
        "year": [d["year"] for d in docs],
        "embedding": embeddings.tolist()
    }
    
    result = collection.insert(data)
    return result.insert_count

# 分批插入
BATCH_SIZE = 1000
total_inserted = 0

for i in range(0, len(documents), BATCH_SIZE):
    batch = documents[i:i + BATCH_SIZE]
    count = insert_batch(batch)
    total_inserted += count
    print(f"已插入 {total_inserted}/{len(documents)} 篇文档")

# 刷新以确保数据已持久化并建立索引
collection.flush()
print(f"已插入并刷新总数：{total_inserted}")
```

***

## 第 8 步——搜索与查询

### 基础语义搜索

```python
from pymilvus import Collection
from sentence_transformers import SentenceTransformer

collection = Collection("documents")
collection.load()

model = SentenceTransformer("all-MiniLM-L6-v2", device="cuda")

def search(query: str, top_k: int = 10):
    query_embedding = model.encode(
        [query],
        normalize_embeddings=True
    )[0].tolist()
    
    results = collection.search(
        data=[query_embedding],
        anns_field="embedding",
        param={
            "metric_type": "COSINE",
            "params": {"ef": 64}    # HNSW 搜索时参数（ef >= top_k）
        },
        limit=top_k,
        output_fields=["text", "source", "category", "year"]
    )
    
    return results[0]

# 搜索
hits = search("向量相似度搜索是如何工作的")
for hit in hits:
    print(f"得分：{hit.score:.4f}")
    print(f"文本：{hit.entity.get('text')[:100]}")
    print(f"来源：{hit.entity.get('source')}")
    print()
```

### 过滤搜索

```python
from pymilvus import Collection

collection = Collection("documents")

# 带元数据过滤的搜索（布尔表达式）
results = collection.search(
    data=[query_embedding],
    anns_field="embedding",
    param={"metric_type": "COSINE", "params": {"ef": 64}},
    limit=10,
    expr='category == "database" and year >= 2023',  # 布尔过滤
    output_fields=["text", "category", "year"]
)
```

### 混合搜索（稠密 + 稀疏）

```python
# Milvus 2.4+ 支持稠密 + 稀疏混合搜索
from pymilvus import AnnSearchRequest, WeightedRanker, Collection

collection = Collection("documents")

# 稠密搜索请求
dense_req = AnnSearchRequest(
    data=[dense_embedding],
    anns_field="embedding",
    param={"metric_type": "COSINE", "params": {"ef": 64}},
    limit=20
)

# 稀疏搜索请求（需要稀疏向量字段）
sparse_req = AnnSearchRequest(
    data=[sparse_embedding],
    anns_field="sparse_embedding",
    param={"metric_type": "IP"},
    limit=20
)

# 结合互惠排序融合
results = collection.hybrid_search(
    [dense_req, sparse_req],
    rerank=WeightedRanker(0.7, 0.3),  # 70% 稠密，30% 稀疏
    limit=10,
    output_fields=["text"]
)
```

***

## 第 9 步——构建 RAG 服务

```bash
pip install fastapi uvicorn openai

cat > /workspace/milvus_rag.py << 'EOF'
from fastapi import FastAPI
from pydantic import BaseModel
from pymilvus import Collection, connections
from sentence_transformers import SentenceTransformer
from openai import OpenAI
import os

app = FastAPI(title="Milvus RAG API")

# 在启动时初始化
connections.connect("default", host="localhost", port="19530")
collection = Collection("documents")
collection.load()
embedder = SentenceTransformer("all-MiniLM-L6-v2", device="cuda")
llm = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

class QueryRequest(BaseModel):
    question: str
    n_results: int = 5

@app.get("/health")
async def health():
    return {"status": "ok", "vectors": collection.num_entities}

@app.post("/search")
async def semantic_search(req: QueryRequest):
    embedding = embedder.encode(
        [req.question],
        normalize_embeddings=True
    )[0].tolist()
    
    results = collection.search(
        data=[embedding],
        anns_field="embedding",
        param={"metric_type": "COSINE", "params": {"ef": 64}},
        limit=req.n_results,
        output_fields=["text", "source", "category"]
    )
    
    return {
        "results": [
            {
                "text": hit.entity.get("text"),
                "source": hit.entity.get("source"),
                "score": hit.score
            }
            for hit in results[0]
        ]
    }

@app.post("/rag")
async def rag(req: QueryRequest):
    embedding = embedder.encode([req.question], normalize_embeddings=True)[0].tolist()
    
    hits = collection.search(
        data=[embedding],
        anns_field="embedding",
        param={"metric_type": "COSINE", "params": {"ef": 64}},
        limit=req.n_results,
        output_fields=["text", "source"]
    )[0]
    
    context = "\n\n".join([
        f"[{hit.entity.get('source')}]: {hit.entity.get('text')}"
        for hit in hits if hit.score > 0.4
    ])
    
    response = llm.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "请根据上下文回答。简明扼要。"},
            {"role": "user", "content": f"上下文:\n{context}\n\n问题: {req.question}"}
        ]
    )
    
    return {"answer": response.choices[0].message.content, "context_used": len(hits)}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)
EOF

python3 /workspace/milvus_rag.py
```

***

## 第10步 — 监控和管理

```python
from pymilvus import connections, utility, Collection

connections.connect("default", host="localhost", port="19530")

# 列出所有集合
print("Collections:", utility.list_collections())

# 集合统计信息
col = Collection("documents")
print(f"实体数量: {col.num_entities:,}")
print(f"模式: {col.schema}")

# 分区管理
col.create_partition("2024_docs")
col.create_partition("2023_docs")

# 按分区插入
col.insert(data, partition_name="2024_docs")

# 搜索特定分区
results = col.search(
    data=[query_vec],
    anns_field="embedding",
    param={"metric_type": "COSINE", "params": {"ef": 64}},
    limit=10,
    partition_names=["2024_docs"]  # 仅搜索此分区
)
```

***

## 故障排查

### 服务未启动

```bash
# 查看容器日志
docker compose logs etcd
docker compose logs minio
docker compose logs standalone

# 检查磁盘空间
df -h /opt/milvus

# 重启服务
docker compose restart
```

### 19530 端口连接被拒绝

```bash
# 验证 Milvus 正在监听
netstat -tlnp | grep 19530

# 检查健康状态
curl http://localhost:9091/healthz

# 允许启动所需时间（90 秒）
docker compose logs standalone | tail -20
```

### 大集合索引构建超时

```python
# 为大型索引构建增加超时时间
from pymilvus import Collection

collection = Collection("documents")
collection.create_index(
    field_name="embedding",
    index_params=hnsw_params,
    timeout=3600  # 1 小时超时
)
```

### 内存使用过高

```bash
# 在 docker-compose.yml 中配置 Milvus 内存限制
# 添加到 standalone 服务：
deploy:
  resources:
    limits:
      memory: 16g
```

***

## 索引类型选择指南

| 索引类型           | 最适合            | 内存     | 速度 | 需要 GPU |
| -------------- | -------------- | ------ | -- | ------ |
| FLAT           | 小型（<100万），精确搜索 | 高      | 慢  | 否      |
| IVF\_FLAT      | 中型（100万–1000万） | 中等     | 好  | 否      |
| HNSW           | 低延迟，<1亿        | 高      | 优秀 | 否      |
| IVF\_SQ8       | 压缩型，大规模        | 低      | 好  | 否      |
| GPU\_IVF\_FLAT | 快速批量查询         | GPU+内存 | 最佳 | 是      |
| DISKANN        | 十亿级规模          | 低（磁盘）  | 好  | 否      |

***

## 性能基准

| 集合大小    | 索引             | GPU      | QPS      |
| ------- | -------------- | -------- | -------- |
| 100万向量  | HNSW           | RTX 3090 | \~8,000  |
| 1000万向量 | IVF\_FLAT      | RTX 4090 | \~2,500  |
| 1000万向量 | GPU\_IVF\_FLAT | A100     | \~12,000 |
| 1亿向量    | DISKANN        | A100     | \~1,200  |

***

## 更多资源

* [Milvus 文档](https://milvus.io/docs)
* [Milvus GitHub](https://github.com/milvus-io/milvus)
* [PyMilvus 文档](https://milvus.io/api-reference/pymilvus/v2.4.x/About.md)
* [Milvus 训练营](https://github.com/milvus-io/bootcamp) — 示例应用
* [Zilliz Cloud](https://cloud.zilliz.com/) — 托管版 Milvus
* [向量数据库对比](https://milvus.io/docs/benchmark.md)
* [Attu 图形界面](https://github.com/zilliztech/attu) — 用于 Milvus 管理的 Web 界面

***

*Clore.ai 上的 Milvus 是 AI 应用的理想解决方案，尤其适用于需要扩展到数亿以上向量的场景。结合 GPU 加速的嵌入生成，您可以以远低于托管云服务的成本构建世界级的语义搜索和 RAG 系统。*

***

## Clore.ai GPU 推荐

| 使用场景    | 推荐 GPU         | Clore.ai 预计成本     |
| ------- | -------------- | ----------------- |
| 开发/测试   | RTX 3090（24GB） | $0.07–0.21/gpu/hr |
| 生产级向量搜索 | RTX 3090（24GB） | $0.07–0.21/gpu/hr |
| 高吞吐量嵌入  | RTX 4090（24GB） | $0.14–0.42/gpu/hr |

> 💡 本指南中的所有示例都可以部署在 [Clore.ai](https://clore.ai/marketplace) GPU 服务器上。浏览可用 GPU 并按小时租用——无需承诺，拥有完整 root 访问权限。


---

# Agent Instructions
This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com.

## Querying This Documentation
If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter, and the optional `goal` query parameter:

```
GET https://docs.clore.ai/guides/guides_v2-zh/rag-yu-xiang-liang-shu-ju-ku/milvus.md?ask=<question>&goal=<endgoal>
```

`ask` is the immediate question: it should be specific, self-contained, and written in natural language.
`goal` is optional and describes the broader end goal you are ultimately trying to accomplish on behalf of the user. GitBook uses it to tailor the answer towards what is most useful for that goal.

The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.