一、是什么

Qdrant 是一个专门用于存储和检索 向量数据（Vector） 的向量数据库。

传统数据库主要通过：精确匹配 = > < LIKE 等进行数据查询。

Qdrant 主要通过： 向量相似度 进行数据检索。

例如：

有一段文本：今天天气非常好，经过 Embedding 模型转换后，可能变成：[0.12, 0.85, 0.33, 0.67, …]

另一段文本：今天阳光明媚，经过 Embedding 模型转换，转换后可能是：[0.11, 0.82, 0.35, 0.69, …]

Qdrant 通过对比两者的向量距离，如果比较近就可以认为：“今天天气非常好” ≈ “今天阳光明媚”

二、为什么

Qdrant 非常适合：

  • AI 知识库
  • RAG
  • 语义搜索
  • 图片相似度搜索
  • 推荐系统
  • 人脸/图像特征检索
  • 文档搜索
  • 多模态搜索

三、怎么做

Qdrant 的核心概念

Qdrant 中最重要的几个概念如下：

1
2
3
4
5
6
7
8
9
Qdrant
  │
  └── Collection（集合）
        │
        └── Point（数据点）
              │
              ├── ID
              ├── Vector（向量）
              └── Payload（附加数据）

例如：

1
2
3
4
5
6
7
8
{
  "id": 1,
  "vector": [0.1, 0.2, 0.3, 0.4],
  "payload": {
    "color": "red",
    "rand_number": 5
  }
}

对应：

概念 说明
Collection 类似 MySQL 的表
Point 一条数据
Vector 向量数据
Payload 附加的结构化数据
ID 数据唯一标识

安装与连接 Qdrant

1. 安装

1
pip install qdrant-client

2. 连接本地 Qdrant

1
2
3
4
5
6
7
from qdrant_client import QdrantClient

client = QdrantClient(
    host="localhost",
    port=6333,
    check_compatibility=False  # 检查客户端版本和本地数据库版本一致性，若不一致需设为false
)

3. 连接 Qdrant Cloud

url、api_key 可在 Qdrant 官方申请，官方提供免费额度

Nodes Disk RAM vCPUs
1 4 GiB 1 GiB 0.5 
1
2
3
4
5
client = QdrantClient(
    url="your-qdrant-cloud-access-link",
    api_key="your-api-key",
    cloud_inference=True
)

新增数据

1. 新增 Collection

在 Qdrant 中，首先需要创建集合

1
2
3
4
5
6
7
8
9
10
11
12
13
from qdrant_client.models import VectorParams, Distance

COLLECTION_NAME = "collection_name"
VECTOR_SIZE = 4

if not client.collection_exists(COLLECTION_NAME):
    client.create_collection(
        collection_name=COLLECTION_NAME,
        vectors_config=VectorParams(
            size=VECTOR_SIZE,
            distance=Distance.COSINE
        )
    )

size参数含义

size 表示向量维度。例如 VECTOR_SIZE = 4 表示 [0.1, 0.2, 0.3, 0.4]

同一集合内每个点的向量必须具有相同的维度，并且必须使用单一指标进行比较。可以使用命名向量在单个点中包含多个向量，其中每个向量都可以有自己的维度和指标要求。

命名向量

一个记录中可以包含多个向量。此功能允许每个集合拥有多个向量存储。为了区分同一记录中的向量，它们必须具有唯一的名称。在此模式下，每个命名向量都有其自己的距离指标和维度。

1
2
3
4
5
6
7
8
9
from qdrant_client import QdrantClient, models

client.create_collection(
    collection_name=COLLECTION_NAME,
    vectors_config={
        "image": models.VectorParams(size=4, distance=models.Distance.DOT),
        "text": models.VectorParams(size=8, distance=models.Distance.COSINE),
    },
)

distance参数含义

distance指标用于衡量向量之间的相似度。指标的选择取决于获取向量的方式，特别是取决于神经网络编码器的训练方法。

Qdrant 支持以下最常用的指标类型

  1. Cosine
1
Distance.COSINE

余弦相似度。适合：

  • 文本向量
  • Embedding
  • 语义搜索

通常 RAG 项目使用

  1. Dot
1
Distance.DOT

点积。适合：

  • 已经归一化的向量
  • 某些推荐算法
  • 特定模型输出
  1. Euclid
1
Distance.EUCLID

欧氏距离。适合：

  • 图像特征
  • 空间距离
  • 数值型特征

2. 新增 Point

单次插入

Qdrant 的一条数据称为：Point

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from qdrant_client.models import PointStruct

client.upsert(
    collection_name=COLLECTION_NAME,
    points=[
        PointStruct(
            id=str(uuid.uuid4()),
            vector=[0.1, 0.2, 0.3, 0.4],
            payload={
                "color": "black",
                "rand_number": 999
            }
        )
    ]
)

批量插入

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
client.upsert(
    collection_name=COLLECTION_NAME,
    wait=True, # 重要数据：wait=True; 大批量导入：wait=False
    points=[
        PointStruct(
            id=idx,
            vector=vector.tolist(),
            payload={
                "color": "red",
                "rand_number": idx % 10
            }
        )
        for idx, vector in enumerate(vectors)
    ]
)

查询数据

1
2
3
4
5
hits = client.query_points(
    collection_name=COLLECTION_NAME,
    query=query_vector,
    limit=5
)

Qdrant 会根据 query_vector 在 Collection 中搜索最相似的数据。

1. 根据 ID 查询

1
2
3
4
5
point = client.retrieve(
    COLLECTION_NAME,
    [point_id],
    with_vectors=True  # 如果向量维度非常大，建议设置成False，减少数据传输
)

2. 根据 过滤条件 查询

过滤数值

1
2
3
4
5
6
7
8
9
10
11
12
13
client.query_points(
    collection_name=COLLECTION_NAME,
    query=query_vector,
    query_filter=Filter(
        must=[
            FieldCondition(
                key="rand_number",
                range=Range(gte=3, lte=5)
            )
        ]
    ),
    limit=5
)

上述代码意思为：先筛选 5 >= rand_number >= 3，然后在符合条件的数据中进行向量相似度搜索，返回最相似的 5 条

过滤字符串

1
2
3
4
5
6
7
8
9
10
from qdrant_client.models import MatchValue

query_filter = Filter(
    must=[
        FieldCondition(
            key="color",
            match=MatchValue(value="red")
        )
    ]
)

上述代码意思为：先筛选 color == red，然后在符合条件的数据中进行向量相似度搜索

多个条件 AND

1
2
3
4
5
6
7
8
9
10
11
12
Filter(
    must=[
        FieldCondition(
            key="color",
            match=MatchValue(value="red")
        ),
        FieldCondition(
            key="age",
            range=Range(gte=18)
        )
    ]
)

修改数据

1. 使用 upsert

1
2
3
4
5
6
7
8
9
10
client.upsert(
    collection_name=COLLECTION_NAME,
    points=[
        PointStruct(
            id=point_id,
            vector=new_vector,
            payload=new_payload
        )
    ]
)

注：upsert 中 Payload 参数需写全，会整体覆盖

2. 只更新 Payload

1
2
3
4
5
6
7
client.set_payload(
    collection_name=COLLECTION_NAME,
    payload={
        "age": 21
    },
    points=[point_id]
)

这样就只会修改 age 字段，其他 Payload 保留

删除数据

1. 删除 Payload 字段

1
2
3
4
5
client.delete_payload(
    collection_name=COLLECTION_NAME,
    keys=["age"],
    points=[1]
)

2. 根据 ID 删除

1
2
3
4
client.delete(
    collection_name=COLLECTION_NAME,
    points_selector=[point_id1, point_id2, point_id3]
)

3. 根据条件删除

删除所有 color = red 的数据

可以使用过滤条件：

1
2
3
4
5
6
7
8
9
10
11
client.delete(
    collection_name=COLLECTION_NAME,
    points_selector=Filter(
        must=[
            FieldCondition(
                key="color",
                match=MatchValue(value="red")
            )
        ]
    )
)

4. 删除整个 Collection

1
client.delete_collection(COLLECTION_NAME)

这会删除 Collection 中所有向量、所有 Payload、所有索引

Python代码示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
import uuid

import numpy as np
from qdrant_client import QdrantClient
from qdrant_client.models import PointStruct, VectorParams, Distance, Filter, FieldCondition, Range

# connect to Qdrant Local
# client = QdrantClient(host="localhost", port=6333)
# connect to Qdrant Cloud
client = QdrantClient(
    url="your-qdrant-cloud-access-link",
    api_key="your-api-key",
    cloud_inference=True
)

COLLECTION_NAME = "test_collection"
VECTOR_SIZE = 4


def init_collection():
    """初始化集合，如果不存在则创建"""
    if not client.collection_exists(COLLECTION_NAME):
        client.create_collection(
            collection_name=COLLECTION_NAME,
            vectors_config=VectorParams(size=VECTOR_SIZE, distance=Distance.DOT),
        )
        print(f"集合 '{COLLECTION_NAME}' 创建成功")
    else:
        print(f"集合 '{COLLECTION_NAME}' 已存在")


def create_points():
    init_collection()
    vector = np.random.rand(VECTOR_SIZE)
    client.upsert(
        collection_name=COLLECTION_NAME,
        points=[
            PointStruct(
                id=str(uuid.uuid4()),
                vector=vector,
                payload={"color": "black", "rand_number": 999}
            )
        ]
    )
    print(f"成功插入数据")


def create_points_batch(count=100):
    """增加：批量插入向量数据"""
    init_collection()
    vectors = np.random.rand(count, VECTOR_SIZE)
    client.upsert(
        collection_name=COLLECTION_NAME,
        wait=True,
        points=[
            PointStruct(
                id=idx,
                vector=vector.tolist(),
                payload={"color": "red", "rand_number": idx % 10}
            )
            for idx, vector in enumerate(vectors)
        ]
    )
    print(f"成功插入 {count} 条数据")


def delete_points(point_ids=None):
    """删除：根据ID删除数据"""
    if point_ids:
        client.delete(
            collection_name=COLLECTION_NAME,
            points_selector=point_ids
        )
        print(f"已删除 ID 为 {point_ids} 的数据")
    else:
        # 删除所有数据
        client.delete_collection(COLLECTION_NAME)
        print(f"已删除集合 '{COLLECTION_NAME}'")


def update_points(point_id, new_payload=None):
    """修改：更新指定点的payload"""
    if new_payload is None:
        new_payload = {"color": "blue", "rand_number": 99}

    # 先获取该点的向量
    point = client.retrieve(COLLECTION_NAME, [point_id], with_vectors=True)
    if not point:
        print(f"ID {point_id} 不存在")
        return

    # 更新点
    client.upsert(
        collection_name=COLLECTION_NAME,
        points=[
            PointStruct(
                id=point_id,
                vector=point[0].vector,
                payload=new_payload
            )
        ]
    )
    print(f"已更新 ID {point_id} 的数据，新payload: {new_payload}")


def query_points(with_filter=False):
    """查询：搜索向量数据"""
    query_vector = np.random.rand(VECTOR_SIZE)
    print(f"查询条件：{query_vector}")

    if with_filter:
        hits = client.query_points(
            collection_name=COLLECTION_NAME,
            query=query_vector,
            query_filter=Filter(
                must=[
                    FieldCondition(
                        key='rand_number',
                        range=Range(gte=3)
                    )
                ]
            ),
            limit=5
        )
        print("带过滤条件的查询结果 (rand_number >= 3):")
    else:
        hits = client.query_points(
            collection_name=COLLECTION_NAME,
            query=query_vector,
            limit=5
        )
        print("查询结果:")

    for hit in hits.points:
        print(f"  ID: {hit.id}, Score: {hit.score: .4f}, Payload: {hit.payload}")


def show_menu():
    """显示菜单"""
    print("\n" + "=" * 40)
    print("Qdrant 增删改查操作菜单")
    print("=" * 40)
    print("1. 增加 (Create) - 批量插入数据")
    print("2. 删除 (Delete) - 删除数据")
    print("3. 修改 (Update) - 更新数据")
    print("4. 查询 (Query)  - 搜索数据")
    print("5. 查询(带过滤)   - 带条件搜索")
    print("6. 初始化集合     - 创建集合")
    print("0. 退出")
    print("=" * 40)


def main():
    """主循环：自动选择操作"""
    while True:
        show_menu()
        choice = input("请选择操作 (0-6): ").strip()

        if choice == "1":
            count = input("输入要插入的数据条数 (默认100): ").strip()
            if count:
                count = int(count)
                create_points_batch(count)
            else:
                create_points()
        elif choice == "2":
            ids_input = input("输入要删除的ID (多个用逗号分隔，留空删除集合): ").strip()
            if ids_input:
                point_ids = [int(x.strip()) for x in ids_input.split(",")]
                delete_points(point_ids)
            else:
                confirm = input("确认删除整个集合? (y/n): ").strip().lower()
                if confirm == "y":
                    delete_points()
        elif choice == "3":
            point_id = input("输入要更新的ID: ").strip()
            if point_id:
                update_points(int(point_id))
        elif choice == "4":
            query_points(with_filter=False)
        elif choice == "5":
            query_points(with_filter=True)
        elif choice == "6":
            init_collection()
        elif choice == "0":
            print("退出程序")
            break
        else:
            print("无效选择，请重新输入")


if __name__ == "__main__":
    main()

总结

代码中核心 CRUD 方法可以记成：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# 创建 / 更新
client.upsert()

# 查询
client.retrieve()
client.query_points()

# 更新 Payload
client.set_payload()

# 删除
client.delete()

# 删除集合
client.delete_collection()

其中最重要的设计思想是：

Vector 用来做相似度搜索，Payload 用来存储业务数据和做条件过滤。

比如一个 RAG 知识库中的数据可以设计为：

1
2
3
4
5
6
7
8
9
10
11
{
    "id": "doc_001_chunk_001",
    "vector": [0.12, 0.53, 0.87, "..."],
    "payload": {
        "document_id": "doc_001",
        "title": "Qdrant 使用指南",
        "content": "Qdrant 是一个向量数据库...",
        "category": "技术文档",
        "tenant_id": 10001
    }
}

查询时：

1
2
3
4
5
6
7
8
9
10
11
用户问题
    ↓
Embedding
    ↓
生成 Query Vector
    ↓
Qdrant 向量搜索
    ↓
Payload 过滤
    ↓
返回最相似的文档片段

这就是 Qdrant 在 AI 知识库和 RAG 系统中的最典型使用方式。