# Qdrant

> **用于语义搜索和 RAG 应用的高性能向量数据库 — GPU 加速索引**

Qdrant 是一个用 Rust 编写的开源、生产就绪的向量数据库。它在数十亿向量上提供快速的近似最近邻（ANN）搜索，并具备高级过滤、载荷索引和多向量支持。它是许多生产级 RAG（检索增强生成）管道和语义搜索应用的核心组件。

**GitHub：** [qdrant/qdrant](https://github.com/qdrant/qdrant) — 22K+ ⭐

***

## 为什么选择 Qdrant？

| 功能          | Qdrant | Pinecone | Weaviate | Chroma   |
| ----------- | ------ | -------- | -------- | -------- |
| 开源          | ✅      | ❌        | ✅        | ✅        |
| Rust 性能     | ✅      | —        | ❌ Go     | ❌ Python |
| 查询时过滤       | ✅ 高级   | ✅ 基本     | ✅        | ✅ 基本     |
| 多向量         | ✅      | ❌        | ✅        | ❌        |
| 基于磁盘的 HNSW  | ✅      | ✅        | ✅        | ❌        |
| 载荷索引        | ✅      | 有限       | ✅        | 有限       |
| gRPC + REST | ✅ 两者   | ✅ REST   | ✅        | REST     |
| 自托管         | ✅      | ❌ 仅云端    | ✅        | ✅        |

{% hint style="success" %}
**Qdrant 是用 Rust 编写的** — 在保证内存安全的同时提供接近 C 级的性能。基准测试表明，在高负载场景下 Qdrant 一直 **快 1.5–3 倍** 比基于 Python 的替代方案（如 Chroma）更快。
{% endhint %}

***

## 主要使用场景

* **RAG（检索增强生成）** — 为大型语言模型的提示找到相关上下文
* **语义搜索** — 按含义而非仅按关键词搜索
* **推荐系统** — 通过嵌入相似性查找相似项目
* **重复检测** — 识别近似重复内容
* **异常检测** — 查找远离簇中心的向量
* **图像/音频相似性搜索** — 多模态检索

***

## 先决条件

* 具有 GPU 租用的 Clore.ai 帐户
* 对 REST API 或 Python 的基本熟悉
* 您选择的嵌入模型（OpenAI、SentenceTransformers 等）

***

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

Qdrant 在提供服务时主要受 CPU/RAM 限制，但在以下情况受益于 GPU：

* 在提供服务的同时生成嵌入（嵌入模型与服务在同一服务器上）
* 大规模批量索引操作

1. 前往 [clore.ai](https://clore.ai) → **市场**
2. 对于 **嵌入 + 服务 组合：** 配备 32GB+ 显存的 RTX 3090/4090
3. 对于 **仅提供服务：** 带快速 NVMe 存储的 CPU 优化服务器

{% hint style="info" %}
**内存规划：**

* 每个 1536 维的 float32 向量 = 6KB
* 100 万个向量 = 约 6GB RAM
* 1000 万个向量 = 约 60GB RAM
* 对非常大的集合启用磁盘存储
  {% endhint %}

***

## 步骤 2 — 部署 Qdrant 容器

**Docker 镜像：**

```
qdrant/qdrant:latest
```

**端口：**

```
22
6333
6334
```

* **端口 6333：** REST API（HTTP）
* **端口 6334：** gRPC API（对批量操作具有更高性能）

**环境变量：**

```
QDRANT__SERVICE__HTTP_PORT=6333
QDRANT__SERVICE__GRPC_PORT=6334
QDRANT__LOG_LEVEL=INFO
QDRANT__STORAGE__STORAGE_PATH=/qdrant/storage
```

**卷/持久化存储：** 挂载 `/qdrant/storage` 以实现数据持久化。若不挂载，容器重启后数据将丢失。

***

## 步骤 3 — 验证 Qdrant 是否运行

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

# 检查 Qdrant 是否运行
curl http://localhost:6333/

# 预期响应：
# {"title":"qdrant - vector search engine","version":"..."}

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

# 检查集群信息
curl http://localhost:6333/cluster
```

***

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

```bash
# 安装 Qdrant Python 客户端和嵌入工具
pip install qdrant-client sentence-transformers openai numpy

# 验证连接
python3 << 'EOF'
from qdrant_client import QdrantClient

client = QdrantClient("localhost", port=6333)
print(f"Qdrant connected: {client.get_collections()}")
EOF
```

***

## 步骤 5 — 创建集合

集合是具有固定维度的一组命名向量。

```python
from qdrant_client import QdrantClient
from qdrant_client.models import (
    Distance,
    VectorParams,
    HnswConfigDiff,
    OptimizersConfigDiff,
    QuantizationConfig,
    ScalarQuantizationConfig,
    ScalarType
)

client = QdrantClient("localhost", port=6333)

# 为 OpenAI text-embedding-3-small（1536 维）创建集合
client.create_collection(
    collection_name="documents",
    vectors_config=VectorParams(
        size=1536,           # 向量维度（与您的嵌入模型匹配）
        distance=Distance.COSINE,  # 可选：COSINE、EUCLID、DOT
        on_disk=False        # 对于非常大的集合将其设为 True
    ),
    hnsw_config=HnswConfigDiff(
        m=16,                # HNSW 图连接度（更高 = 更好召回，但占用更多内存）
        ef_construct=100,    # 构建时的搜索深度（更高 = 更好质量，但索引更慢）
        full_scan_threshold=10000  # 在此数量以下使用暴力搜索
    ),
    optimizers_config=OptimizersConfigDiff(
        indexing_threshold=20000  # 在达到该向量数后开始 HNSW 索引
    ),
    quantization_config=QuantizationConfig(
        scalar=ScalarQuantizationConfig(
            type=ScalarType.INT8,  # 将向量压缩为 INT8（内存减少约 4 倍）
            quantile=0.99,
            always_ram=True        # 将量化索引保留在内存中
        )
    )
)

print("Collection created!")
print(client.get_collection("documents"))
```

### 用于 SentenceTransformers 的集合（384 维）

```python
client.create_collection(
    collection_name="embeddings_384",
    vectors_config=VectorParams(
        size=384,              # all-MiniLM-L6-v2 的输出维度
        distance=Distance.COSINE
    )
)
```

***

## 步骤 6 — 索引文档

### 使用 OpenAI 嵌入

```python
from qdrant_client import QdrantClient
from qdrant_client.models import PointStruct
from openai import OpenAI
import uuid

client = QdrantClient("localhost", port=6333)
openai_client = OpenAI(api_key="your-openai-api-key")

def get_embeddings(texts: list[str], batch_size: int = 100) -> list[list[float]]:
    """以批次方式生成嵌入。"""
    all_embeddings = []
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]
        response = openai_client.embeddings.create(
            model="text-embedding-3-small",
            input=batch
        )
        all_embeddings.extend([e.embedding for e in response.data])
    return all_embeddings

# 示例文档
documents = [
    {
        "id": str(uuid.uuid4()),
        "text": "Qdrant is a vector database built in Rust for high performance.",
        "source": "documentation",
        "category": "database",
        "year": 2024
    },
    {
        "id": str(uuid.uuid4()),
        "text": "Machine learning models convert text to dense vector representations.",
        "source": "article",
        "category": "ml",
        "year": 2023
    },
    # 添加更多文档...
]

# 生成嵌入
texts = [doc["text"] for doc in documents]
embeddings = get_embeddings(texts)

# Upsert 到 Qdrant
points = [
    PointStruct(
        id=str(uuid.uuid4()),
        vector=embedding,
        payload={
            "text": doc["text"],
            "source": doc["source"],
            "category": doc["category"],
            "year": doc["year"]
        }
    )
    for doc, embedding in zip(documents, embeddings)
]

client.upsert(
    collection_name="documents",
    points=points,
    wait=True  # 等待索引完成
)

print(f"Indexed {len(points)} documents!")
```

### 使用 SentenceTransformers（本地、GPU 加速）

```python
from sentence_transformers import SentenceTransformer
from qdrant_client import QdrantClient
from qdrant_client.models import PointStruct
import torch
import uuid

# 在 GPU 上加载嵌入模型
model = SentenceTransformer("all-MiniLM-L6-v2", device="cuda")

client = QdrantClient("localhost", port=6333)

documents = [
    {"text": "How do I set up Qdrant on a GPU server?", "tag": "setup"},
    {"text": "Vector databases store high-dimensional embeddings for similarity search.", "tag": "concept"},
    {"text": "HNSW algorithm provides approximate nearest neighbor search.", "tag": "algorithm"},
    # ... 更多文档
]

# GPU 加速的批量编码
texts = [doc["text"] for doc in documents]
embeddings = model.encode(
    texts,
    batch_size=256,       # 为 GPU 提高效率使用较大的批量大小
    show_progress_bar=True,
    normalize_embeddings=True  # 为余弦相似度进行归一化
)

# 在 Qdrant 中索引
points = [
    PointStruct(
        id=str(uuid.uuid4()),
        vector=embedding.tolist(),
        payload=doc
    )
    for doc, embedding in zip(documents, embeddings)
]

# 批量 upsert（更高效）
BATCH_SIZE = 1000
for i in range(0, len(points), BATCH_SIZE):
    batch = points[i:i + BATCH_SIZE]
    client.upsert(collection_name="embeddings_384", points=batch)
    print(f"Indexed {min(i + BATCH_SIZE, len(points))}/{len(points)}")
```

***

## 步骤 7 — 搜索与查询

### 基础语义搜索

```python
from qdrant_client import QdrantClient
from sentence_transformers import SentenceTransformer

client = QdrantClient("localhost", port=6333)
model = SentenceTransformer("all-MiniLM-L6-v2", device="cuda")

def search(query: str, limit: int = 5, collection: str = "embeddings_384"):
    # 生成查询嵌入
    query_vector = model.encode(query, normalize_embeddings=True).tolist()
    
    # 搜索
    results = client.search(
        collection_name=collection,
        query_vector=query_vector,
        limit=limit,
        with_payload=True,
        with_vectors=False    # 不返回向量（节省带宽）
    )
    
    return results

# 测试搜索
results = search("vector database performance")
for r in results:
    print(f"Score: {r.score:.4f} | {r.payload['text'][:100]}")
```

### 带过滤的搜索（元数据 + 向量）

```python
from qdrant_client.models import Filter, FieldCondition, MatchValue, Range

# 使用元数据过滤进行搜索
results = client.search(
    collection_name="documents",
    query_vector=query_vector,
    query_filter=Filter(
        must=[
            FieldCondition(
                key="category",
                match=MatchValue(value="database")
            ),
            FieldCondition(
                key="year",
                range=Range(gte=2023)  # 年份 >= 2023
            )
        ]
    ),
    limit=10,
    with_payload=True
)
```

### 批量/多查询搜索

```python
from qdrant_client.models import SearchRequest

queries = [
    "how to install vector database",
    "machine learning inference optimization",
    "RAG pipeline architecture"
]

query_vectors = model.encode(queries, normalize_embeddings=True)

# 批量搜索（对所有查询一次 API 调用）
results = client.search_batch(
    collection_name="embeddings_384",
    requests=[
        SearchRequest(
            vector=vec.tolist(),
            limit=5,
            with_payload=True
        )
        for vec in query_vectors
    ]
)

for query, res in zip(queries, results):
    print(f"\nQuery: {query}")
    for r in res:
        print(f"  {r.score:.3f}: {r.payload['text'][:80]}")
```

***

## 步骤 8 — 构建 RAG 管道

```python
from qdrant_client import QdrantClient
from sentence_transformers import SentenceTransformer
from openai import OpenAI

# 初始化客户端
qdrant = QdrantClient("localhost", port=6333)
embedder = SentenceTransformer("all-MiniLM-L6-v2", device="cuda")
llm = OpenAI(api_key="your-openai-key")

def rag_query(question: str, n_context: int = 5) -> str:
    # 第 1 步：对问题进行嵌入
    query_vector = embedder.encode(question, normalize_embeddings=True).tolist()
    
    # 第 2 步：从 Qdrant 检索相关上下文
    search_results = qdrant.search(
        collection_name="documents",
        query_vector=query_vector,
        limit=n_context,
        with_payload=True
    )
    
    # 第 3 步：构建上下文字符串
    context = "\n\n".join([
        f"[Source: {r.payload.get('source', 'unknown')}]\n{r.payload['text']}"
        for r in search_results
        if r.score > 0.5  # 过滤低置信度结果
    ])
    
    # 第 4 步：使用 LLM 生成答案
    response = llm.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": "根据提供的上下文回答问题。简明且准确。"
            },
            {
                "role": "user",
                "content": f"Context:\n{context}\n\nQuestion: {question}"
            }
        ],
        temperature=0.1
    )
    
    return response.choices[0].message.content

# 测试 RAG 管道
answer = rag_query("What is Qdrant and how does it work?")
print(answer)
```

***

## 步骤 9 — 监控与管理集合

```python
# 集合统计信息
info = client.get_collection("documents")
print(f"Vectors count: {info.vectors_count:,}")
print(f"Points count: {info.points_count:,}")
print(f"Indexed vectors: {info.indexed_vectors_count:,}")
print(f"Status: {info.status}")
print(f"Disk usage: {info.disk_data_size / 1024 / 1024:.1f} MB")

# 列出所有集合
collections = client.get_collections()
for c in collections.collections:
    print(f" - {c.name}")

# 按过滤器删除点
client.delete(
    collection_name="documents",
    points_selector=Filter(
        must=[FieldCondition(key="source", match=MatchValue(value="old_source"))]
    )
)

# 优化集合（强制构建索引）
client.update_collection(
    collection_name="documents",
    optimizer_config=OptimizersConfigDiff(indexing_threshold=0)  # 强制立即索引
)
```

***

## 故障排除

### 连接被拒绝

```bash
# 检查 Qdrant 是否运行
docker ps | grep qdrant
# 或检查进程
ps aux | grep qdrant

# 检查端口是否打开
curl http://localhost:6333/
netstat -tlnp | grep 6333
```

### 搜索性能缓慢

```python
# 优化 HNSW 参数以获得更好的召回率
client.update_collection(
    collection_name="documents",
    hnsw_config=HnswConfigDiff(ef=128)  # 增加搜索时的 ef（默认 100）
)

# 使用 INT8 量化以将更多向量放入内存
```

### 高内存使用

```python
# 对于大型集合启用磁盘存储
client.create_collection(
    collection_name="large_collection",
    vectors_config=VectorParams(
        size=1536,
        distance=Distance.COSINE,
        on_disk=True  # 将向量存储在磁盘上而不是内存中
    )
)
```

***

## REST API 快速参考

```bash
# 列出集合
curl http://localhost:6333/collections

# 创建集合
curl -X PUT http://localhost:6333/collections/my_collection \
    -H "Content-Type: application/json" \
    -d '{"vectors": {"size": 384, "distance": "Cosine"}}'

# 计数点数
curl http://localhost:6333/collections/my_collection/points/count

# 搜索
curl -X POST http://localhost:6333/collections/my_collection/points/search \
    -H "Content-Type: application/json" \
    -d '{
        "vector": [0.1, 0.2, ...],
        "limit": 5,
        "with_payload": true
    }'

# 删除集合
curl -X DELETE http://localhost:6333/collections/my_collection
```

***

## Clore.ai 上的成本估算

| 设置     | 服务器              | 每月费用       | 容量       |
| ------ | ---------------- | ---------- | -------- |
| 小型 RAG | RTX 3090，32GB 内存 | \~$60–80   | \~5M 向量  |
| 中等搜索   | RTX 4090，64GB 内存 | \~$120–150 | \~15M 向量 |
| 大规模    | A100，128GB 内存    | \~$250–350 | \~30M 向量 |

***

## 其他资源

* [Qdrant 文档](https://qdrant.tech/documentation/)
* [Qdrant GitHub](https://github.com/qdrant/qdrant)
* [Qdrant Python 客户端](https://github.com/qdrant/qdrant-client)
* [Qdrant 示例](https://github.com/qdrant/examples)
* [向量数据库基准](https://qdrant.tech/benchmarks/)
* [Sentence Transformers](https://www.sbert.net/)

***

*在 Clore.ai 上运行 Qdrant 可为您提供自托管的高性能向量数据库，免除像 Pinecone 或 Weaviate Cloud 那样的按查询费用。非常适合处理数百万文档的 RAG 管道。*

***

## Clore.ai 的 GPU 建议

| 在 Clore.ai 上的预估费用 | 开发/测试 | RTX 3090（24GB） |
| ----------------- | ----- | -------------- |
| \~$0.12/每 GPU/每小时 | 生产    | RTX 4090（24GB） |
| 生产级向量搜索           | 生产    | RTX 4090（24GB） |
| 高吞吐量嵌入            | 大规模   | A100 80GB      |

> GPU 服务器上。浏览可用 GPU 并按小时租用 — 无需承诺，提供完整的 root 访问权限。 [Clore.ai](https://clore.ai/marketplace) GPU 服务器。浏览可用 GPU 并按小时租用 — 无需承诺，提供完整的 root 访问权限。