環境配置:可以看上一節
一.數據庫使用
- 連接 Milvus Standalone
- 創建數據庫 my_database_1(無額外屬性)
- 創建數據庫 my_database_2(設置副本數為 3)
- 列出所有數據庫
- 查看默認數據庫(default)詳情
- 修改 my_database_1 屬性:限制最大集合數為 10
- 刪除 my_database_1 的 max.collections 限制
- 切換到 my_database_2(后續所有操作都作用于該庫)
- 刪除數據庫
from pymilvus import MilvusClient, exceptions# ——————————————
# 1. 連接 Milvus Standalone
# ——————————————
# uri: 協議+地址+端口,默認為 http://docker_IP:19530
# token: "用戶名:密碼",默認 root:Milvus
client = MilvusClient(uri="http://IP:19531",token="root:Milvus"
)# ——————————————
# 2. 創建數據庫 my_database_1(無額外屬性)
# ——————————————
try:client.create_database(db_name="my_database_1")print("? my_database_1 創建成功")
except exceptions.MilvusException:print("? my_database_1 已存在")# ——————————————
# 3. 創建數據庫 my_database_2(設置副本數為 3)
# ——————————————
client.create_database(db_name="my_database_2",properties={"database.replica.number": 3}
)
print("? my_database_2 創建成功,副本數=3")# ——————————————
# 4. 列出所有數據庫
# ——————————————
db_list = client.list_databases()
print("當前所有數據庫:", db_list)# ——————————————
# 5. 查看默認數據庫(default)詳情
# ——————————————
default_info = client.describe_database(db_name="default")
print("默認數據庫詳情:", default_info)# ——————————————
# 6. 修改 my_database_1 屬性:限制最大集合數為 10
# ——————————————
client.alter_database_properties(db_name="my_database_1",properties={"database.max.collections": 10}
)
print("? 已為 my_database_1 限制最大集合數為 10")# ——————————————
# 7. 刪除 my_database_1 的 max.collections 限制
# ——————————————
client.drop_database_properties(db_name="my_database_1",property_keys=["database.max.collections"]
)
print("? 已移除 my_database_1 的最大集合數限制")# ——————————————
# 8. 切換到 my_database_2(后續所有操作都作用于該庫)
# ——————————————
client.use_database(db_name="my_database_2")
print("? 已切換當前數據庫為 my_database_2")# ——————————————
# 9. 刪除數據庫 my_database_2
# (注意:如果庫內有 Collection,需先 client.drop_collection() 將其清空)
# ——————————————
client.drop_database(db_name="my_database_2")
print("? my_database_2 已刪除")# ——————————————
# 10. 刪除數據庫 my_database_1
# ——————————————
client.drop_database(db_name="my_database_1")
print("? my_database_1 已刪除")
? my_database_1 創建成功
? my_database_2 創建成功,副本數=3
當前所有數據庫: ['my_database_1', 'my_database_2', 'default']
默認數據庫詳情: {'name': 'default'}
? 已為 my_database_1 限制最大集合數為 10
? 已移除 my_database_1 的最大集合數限制
? 已切換當前數據庫為 my_database_2
? my_database_2 已刪除
? my_database_1 已刪除
二.Collection
Collection
:在 Milvus 中,Collection(集合)是存儲向量和其相關元數據的核心對象,可以類比為傳統數據庫中的“表”。
2.1.TTL
在 Milvus 中,設置 Collection 的 TTL(Time-To-Live)屬性 是為了控制:
插入數據在內存中的生存時間,即:多長時間后這批數據會自動從內存中移除,不再參與檢索。
client.create_collection(collection_name="example_collection",dimension=128,ttl_seconds=60
)
這表示:
- 插入到該集合中的數據,只會在內存中保留 60 秒
- 超過 60 秒后,這部分數據就會自動被 Milvus 標記過期
- 這些數據將不再參與 向量檢索,也不會占用 檢索內存資源
2.2.Milvus 是“冷熱分離”的架構
在 Milvus 中,必須先將數據加載到內存(RAM)中,才能進行向量檢索。
Milvus 為了高性能向量搜索,采用冷熱數據分離設計:
數據狀態 | 存儲位置 | 能否檢索 | 說明 |
---|---|---|---|
未加載 | 磁盤 / 持久化 | ? 不可檢索 | 僅存儲,節省內存資源 |
已加載 | 內存(RAM) | ? 可檢索 | 向量數據加載到內存才能計算相似度 |
2.3.Partition
Partition 是 Collection 內部的數據子集,用于邏輯分組、分批插入、加速查詢。
? 為什么需要 Partition?
如果你要處理 大規模數據(百萬、千萬、甚至上億條向量),把所有數據放在一個 Collection 中不夠靈活:
- 檢索慢(因為全表搜索)
- 不方便管理(比如不同用戶/場景的數據混在一起)
引入 Partition 后,你可以:
- 按條件劃分數據(如按“用戶”“時間”“類別”等)
- 檢索時只針對部分 Partition,提高速度
- 分批管理加載 / 卸載 Partition,節省內存
📦 舉個例子
比如你有一個集合 product_vectors,表示商品的向量數據。
你可以按商品類型創建分區:
Partition 名稱 | 內容說明 |
---|---|
shoes | 所有鞋類商品向量 |
phones | 手機類商品向量 |
books | 圖書類商品向量 |
檢索時只搜索某個分區:
client.search(collection_name="product_vectors",partition_names=["phones"], # 只搜索手機...
)
2.4.Alias
Alias 是指向某個 Collection 的“虛擬名稱”,可以隨時修改指向的真實集合,不影響客戶端代碼邏輯。
📌 舉個例子:
你有兩個 Collection:
products_v1
:舊版本的商品向量集合products_v2
:新版本,已經完成向量更新
此時你可以創建一個別名:
client.create_alias("products", "products_v1")
客戶端代碼里一直用別名 products
:
client.search(collection_name="products", ...)
上線新模型后,只需切換別名指向:
client.alter_alias("products", "products_v2")
這樣做的好處?客戶端代碼不用改,就實現了“無感知熱切換”。
? 注意事項
- Alias 是 一對一映射:一個別名只能指向一個 Collection
- 一個 Collection 可以被多個 alias 指向
- 檢索時如果用 alias,必須確保其所指 Collection 已加載(load_collection)
操作 | 作用 |
---|---|
create_alias(alias, collection) | 創建一個別名,初始指向某個 Collection |
alter_alias(alias, new_collection) | 修改別名指向的 Collection,實現熱切換 |
drop_alias(alias) | 刪除別名 |
list_aliases(collection_name=...) | 查看某集合的所有別名 |
2.5.演示代碼
from pymilvus import MilvusClient# ——————————————
# 0. 連接 Milvus
# ——————————————
client = MilvusClient(uri="http://IP:19530",token="root:Milvus"
)
print("? 已連接 Milvus接口")# ——————————————
# 1. 創建 Collection(快速模式)
# ——————————————
# 檢查并刪除已存在的集合
collection_name = "quick_setup"
if collection_name in client.list_collections():client.drop_collection(collection_name=collection_name)print(f"? 已刪除已存在的集合 {collection_name}")# 沒有定義scheme
# 使用快捷方式創建一個維度為 5 的向量集合(Quick Setup 模式)
client.create_collection(collection_name=collection_name,dimension=5
)
print(f"? {collection_name} 已創建")# ——————————————
# 2. 列出所有 Collections
# ——————————————
cols = client.list_collections()
print("當前所有集合:", cols)# ——————————————
# 3. 查看 Collection 詳情
# ——————————————
info = client.describe_collection(collection_name=collection_name)
print(f"{collection_name} 詳情:", info)# ——————————————
# 4. 重命名 Collection
# ——————————————
new_collection_name = "quick_renamed"
if new_collection_name in client.list_collections():client.drop_collection(collection_name=new_collection_name)print(f"? 已刪除已存在的集合 {new_collection_name}")client.rename_collection(old_name=collection_name,new_name=new_collection_name
)
print(f"? {collection_name} 已重命名為 {new_collection_name}")# ——————————————
# 5. 修改 Collection 屬性(設置 TTL 60 秒)
# ——————————————
client.alter_collection_properties(collection_name=new_collection_name,properties={"collection.ttl.seconds": 60}
)
print(f"? 已為 {new_collection_name} 設置 TTL=60s")# ——————————————
# 6. 刪除 Collection 屬性(TTL)
# ——————————————
client.drop_collection_properties(collection_name=new_collection_name,property_keys=["collection.ttl.seconds"]
)
print(f"? 已刪除 {new_collection_name} 的 TTL 屬性")# ——————————————
# 7. 加載集合(將數據加載到內存中)
# ——————————————
client.load_collection(collection_name=new_collection_name)
state = client.get_load_state(collection_name=new_collection_name) # 檢查加載狀態
print("加載狀態:", state)# ——————————————
# 8. 釋放集合(將數據從內存中釋放)
# ——————————————
client.release_collection(collection_name=new_collection_name)
state = client.get_load_state(collection_name=new_collection_name) # 檢查釋放狀態
print("釋放后狀態:", state)# ——————————————
# 9. 管理 Partition
# ——————————————
# 9.1 列出 Partition(默認只有 "_default")
parts = client.list_partitions(collection_name=new_collection_name)
print("Partition 列表:", parts)# 9.2 創建新 Partition
client.create_partition(collection_name=new_collection_name,partition_name="partA"
)
print("? 已創建 partition partA")
print("更新后 Partition 列表:", client.list_partitions(new_collection_name))# 9.3 檢查 Partition 是否存在
exists = client.has_partition(collection_name=new_collection_name,partition_name="partA"
)
print("partA 存在?", exists)# 9.4 加載 & 釋放 指定 Partition 【存入內存】
client.load_partitions(collection_name=new_collection_name,partition_names=["partA"]
)
print("partA 加載狀態:", client.get_load_state(new_collection_name, partition_name="partA"))
# 釋放 Partition
client.release_partitions(collection_name=new_collection_name,partition_names=["partA"]
)
print("partA 釋放后狀態:", client.get_load_state(new_collection_name, partition_name="partA"))# 9.5 刪除 Partition(需先 release)
client.drop_partition(collection_name=new_collection_name,partition_name="partA"
)
print("? 已刪除 partition partA")
print("最終 Partition 列表:", client.list_partitions(new_collection_name))# ——————————————
# 10. 管理 Alias
# ——————————————
# 10.1 創建 Alias
client.create_alias(collection_name=new_collection_name, alias="alias3")
client.create_alias(collection_name=new_collection_name, alias="alias4")
print("? 已創建 alias3, alias4")# 10.2 列出 Alias
aliases = client.list_aliases(collection_name=new_collection_name)
print("當前 aliases:", aliases)# 10.3 查看 Alias 詳情
desc = client.describe_alias(alias="alias3")
print("alias3 詳情:", desc)# 10.4 重新分配 Alias
client.alter_alias(collection_name=new_collection_name, alias="alias4")
print("? 已將 alias4 重新分配給 quick_renamed")# 10.5 刪除 Alias
client.drop_alias(alias="alias3")
print("? 已刪除 alias3")
print("剩余 aliases:", client.list_aliases(new_collection_name))# ——————————————
# 11. 刪除 Collection
# ——————————————
client.drop_collection(collection_name=new_collection_name)
print(f"? 集合 {new_collection_name} 已刪除")
? 已連接 Milvus接口
? quick_setup 已創建
當前所有集合: ['quick_setup']
quick_setup 詳情: {'collection_name': 'quick_setup', 'auto_id': False, 'num_shards': 1, 'description': '', 'fields': [{'field_id': 100, 'name': 'id', 'description': '', 'type': <DataType.INT64: 5>, 'params': {}, 'is_primary': True}, {'field_id': 101, 'name': 'vector', 'description': '', 'type': <DataType.FLOAT_VECTOR: 101>, 'params': {'dim': 5}}], 'functions': [], 'aliases': [], 'collection_id': 458770152609592009, 'consistency_level': 2, 'properties': {}, 'num_partitions': 1, 'enable_dynamic_field': True}
? quick_setup 已重命名為 quick_renamed
? 已為 quick_renamed 設置 TTL=60s
? 已刪除 quick_renamed 的 TTL 屬性
加載狀態: {'state': <LoadState: Loaded>}
釋放后狀態: {'state': <LoadState: NotLoad>}
Partition 列表: ['_default']
? 已創建 partition partA
更新后 Partition 列表: ['_default', 'partA']
partA 存在? True
partA 加載狀態: {'state': <LoadState: Loaded>}
partA 釋放后狀態: {'state': <LoadState: NotLoad>}
? 已刪除 partition partA
最終 Partition 列表: ['_default']
? 已創建 alias3, alias4
當前 aliases: {'aliases': ['alias3', 'alias4'], 'collection_name': 'quick_renamed', 'db_name': 'default'}
alias3 詳情: {'alias': 'alias3', 'collection_name': 'quick_renamed', 'db_name': 'default'}
? 已將 alias4 重新分配給 quick_renamed
? 已刪除 alias3
剩余 aliases: {'aliases': ['alias4'], 'collection_name': 'quick_renamed', 'db_name': 'default'}
? 集合 quick_renamed 已刪除
三.Scheme
schema
定義了 Collections 的數據結構。創建 Collections 時,需要根據自己的要求設計模式。
類似于,你將怎么定義表的結構(列名)
from pymilvus import MilvusClient, DataType# ——————————————
# 0. 連接 Milvus
# ——————————————
client = MilvusClient(uri="http://IP:19530",token="root:Milvus"
)
print("? 已連接 Milvus接口")# ——————————————
# 1. 創建基本 Schema
# ——————————————
schema = MilvusClient.create_schema()
print("? 已創建空 Schema")# ——————————————
# 2. 添加主鍵字段(Primary Field)
# ——————————————
# 2.1 INT64類型主鍵(手動指定ID)
schema.add_field(field_name="id",datatype=DataType.INT64,is_primary=True, # 設置為主鍵auto_id=False # 不自動生成ID
)# 2.2 VARCHAR類型主鍵(自動生成ID)
# schema.add_field(
# field_name="doc_id",
# datatype=DataType.VARCHAR,
# is_primary=True, # 設置為主鍵
# auto_id=True, # 自動生成ID
# max_length=100 # VARCHAR類型需要指定最大長度
# )
print("? 已添加主鍵字段")# ——————————————
# 3. 添加向量字段(Vector Field)
# ——————————————
# 3.1 Dense Vector (浮點向量)
schema.add_field(field_name="text_vector",datatype=DataType.FLOAT_VECTOR, # 32位浮點向量dim=768 # 向量維度
)# 3.2 Binary Vector (二進制向量)
schema.add_field(field_name="image_vector",datatype=DataType.BINARY_VECTOR, # 二進制向量dim=256 # 維度必須是8的倍數
)
print("? 已添加向量字段")# ——————————————
# 4. 添加標量字段(Scalar Field)
# ——————————————
# 4.1 字符串字段
schema.add_field(field_name="title",datatype=DataType.VARCHAR,max_length=200,# 可以為空且有默認值is_nullable=True,default_value="untitled"
)# 4.2 數值字段
schema.add_field(field_name="age",datatype=DataType.INT32,is_nullable=False # 不可為空
)# 4.3 布爾字段
schema.add_field(field_name="is_active",datatype=DataType.BOOL,default_value=True # 默認值為True
)# 4.4 JSON字段
schema.add_field(field_name="metadata",datatype=DataType.JSON
)# 4.5 數組字段
schema.add_field(field_name="tags",datatype=DataType.ARRAY,element_type=DataType.VARCHAR, # 數組元素類型max_capacity=10, # 數組最大容量max_length=50 # 每個元素最大長度
)
print("? 已添加標量字段")# ——————————————
# 5. 添加動態字段(Dynamic Field)
# ——————————————
# schema.add_field(
# field_name="dynamic_field",
# datatype=DataType.VARCHAR,
# is_dynamic=True, # 設置為動態字段
# max_length=500
# )
print("? 已添加動態字段")# ——————————————
# 6. 使用Schema創建Collection
# ——————————————
collection_name = "document_store10"
client.create_collection(collection_name=collection_name,schema=schema
)
print(f"? 已創建集合 {collection_name}")# ——————————————
# 7. 修改Collection字段
# ——————————————
# 添加新字段
# client.alter_collection_field(
# collection_name=collection_name,
# field_name="tags",
# field_params={
# "max_capacity": 64
# }
# )
# print("? 已添加新字段")# ——————————————
# 8. 查看Collection詳情
# ——————————————
info = client.describe_collection(collection_name=collection_name)
print("Collection詳情:", info)# ——————————————
# 9. 清理
# ——————————————
client.drop_collection(collection_name=collection_name)
print("? 已刪除測試集合")
? 已連接 Milvus接口
? 已創建空 Schema
? 已添加主鍵字段
? 已添加向量字段
? 已添加標量字段
? 已添加動態字段
? 已創建集合 document_store10
Collection詳情: {'collection_name': 'document_store10', 'auto_id': False, 'num_shards': 1, 'description': '', 'fields': [{'field_id': 100, 'name': 'id', 'description': '', 'type': <DataType.INT64: 5>, 'params': {}, 'is_primary': True}, {'field_id': 101, 'name': 'text_vector', 'description': '', 'type': <DataType.FLOAT_VECTOR: 101>, 'params': {'dim': 768}}, {'field_id': 102, 'name': 'image_vector', 'description': '', 'type': <DataType.BINARY_VECTOR: 100>, 'params': {'dim': 256}}, {'field_id': 103, 'name': 'title', 'description': '', 'type': <DataType.VARCHAR: 21>, 'params': {'max_length': 200}, 'default_value': string_data: "untitled"
}, {'field_id': 104, 'name': 'age', 'description': '', 'type': <DataType.INT32: 4>, 'params': {}}, {'field_id': 105, 'name': 'is_active', 'description': '', 'type': <DataType.BOOL: 1>, 'params': {}, 'default_value': bool_data: true
}, {'field_id': 106, 'name': 'metadata', 'description': '', 'type': <DataType.JSON: 23>, 'params': {}}, {'field_id': 107, 'name': 'tags', 'description': '', 'type': <DataType.ARRAY: 22>, 'params': {'max_length': 50, 'max_capacity': 10}, 'element_type': <DataType.VARCHAR: 21>}], 'functions': [], 'aliases': [], 'collection_id': 458770152609609581, 'consistency_level': 2, 'properties': {}, 'num_partitions': 1, 'enable_dynamic_field': False}
? 已刪除測試集合
四.實體:Entity
在 Milvus 中,Entity(實體) 是指 插入到 Collection(集合)中的一條數據記錄
? 舉個例子
你要插入一張商品圖像的向量表示到 Milvus:
entities = [{"product_id": 1, "image_vector": [0.1, 0.2, 0.3, 0.4], "category": "shoe"},{"product_id": 2, "image_vector": [0.9, 0.8, 0.7, 0.6], "category": "book"},
]client.insert(collection_name="products",data=entities
)
五.實體:Upsert
當你需要更新集合中的實體,或者不確定是更新還是插入時,可以嘗試使用 Upsert 操作。
務必確保 Upsert 請求中包含的實體包含主鍵,否則會報錯。
- 檢查Collection的主字段是否啟用了AutoId。
- 如果是,Milvus 會將實體中的主鍵替換為自動生成的主鍵,并插入數據。
- 如果沒有,Milvus 將使用實體攜帶的主鍵來插入數據。
- 根據Upsert請求中包含的Entity的主鍵值執行刪除操作。
from pymilvus import MilvusClient
import random# 連接到 Milvus
client = MilvusClient(uri="http://1.14.201.121:19530",token="root:Milvus"
)# 1. 創建集合
# 檢查集合是否存在,如果存在則刪除
if client.has_collection("quick_setup"):client.drop_collection("quick_setup")# 創建集合
client.create_collection(collection_name="quick_setup",dimension=5, # vector 維度primary_field_name="id",vector_field_name="vector",id_type="int"
)print("集合創建成功")# 2. 插入實體(Entity)
data=[{"id": 0, "vector": [0.3580376395471989, -0.6023495712049978, 0.18414012509913835, -0.26286205330961354, 0.9029438446296592], "color": "pink_8682"},{"id": 1, "vector": [0.19886812562848388, 0.06023560599112088, 0.6976963061752597, 0.2614474506242501, 0.838729485096104], "color": "red_7025"},{"id": 2, "vector": [0.43742130801983836, -0.5597502546264526, 0.6457887650909682, 0.7894058910881185, 0.20785793220625592], "color": "orange_6781"},{"id": 3, "vector": [0.3172005263489739, 0.9719044792798428, -0.36981146090600725, -0.4860894583077995, 0.95791889146345], "color": "pink_9298"},{"id": 4, "vector": [0.4452349528804562, -0.8757026943054742, 0.8220779437047674, 0.46406290649483184, 0.30337481143159106], "color": "red_4794"},{"id": 5, "vector": [0.985825131989184, -0.8144651566660419, 0.6299267002202009, 0.1206906911183383, -0.1446277761879955], "color": "yellow_4222"},{"id": 6, "vector": [0.8371977790571115, -0.015764369584852833, -0.31062937026679327, -0.562666951622192, -0.8984947637863987], "color": "red_9392"},{"id": 7, "vector": [-0.33445148015177995, -0.2567135004164067, 0.8987539745369246, 0.9402995886420709, 0.5378064918413052], "color": "grey_8510"},{"id": 8, "vector": [0.39524717779832685, 0.4000257286739164, -0.5890507376891594, -0.8650502298996872, -0.6140360785406336], "color": "white_9381"},{"id": 9, "vector": [0.5718280481994695, 0.24070317428066512, -0.3737913482606834, -0.06726932177492717, -0.6980531615588608], "color": "purple_4976"}
]res = client.insert(collection_name="quick_setup",data=data
)print(res)# 3. 更新實體
update_data = [{"id": 0, "vector": [random.random() for _ in range(5)], "color": "updated_pink_8682"},{"id": 1, "vector": [random.random() for _ in range(5)], "color": "updated_red_7025"}
]res = client.upsert(collection_name="quick_setup",data=update_data
)
print("\n更新結果:", res)# 4. 刪除實體
res = client.delete(collection_name="quick_setup",ids=[0]
)
print("\n刪除結果:", res)# 5. 查詢實體
res = client.query(collection_name="quick_setup",filter="id in [1,2]",output_fields=["id", "color"]
)
print("\n查詢結果:", res)
{'insert_count': 10, 'ids': [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]}
更新結果: {'upsert_count': 2}
刪除結果: {'delete_count': 1}
查詢結果: data: ["{'id': 1, 'color': 'updated_red_7025'}", "{'id': 2, 'color': 'orange_6781'}"]
六.Function
您可以在 Milvus 創建集合時設置函數來導出字段。例如,全文搜索功能使用用戶自定義函數從特定的 varchar 字段導出稀疏向量字段。
七.Consistency Level(數據一致性)
分布式數據庫系統通常使用一致性級別來定義跨數據節點和副本的數據相同性。
可以在創建集合或在集合內進行相似性搜索時設置單獨的一致性級別。適用的一致性級別包括:Strong, Bounded Staleness, Session, Eventually
client.create_collection(collection_name="my_collection",schema=schema,consistency_level="Bounded", # Defaults to Bounded if not specified
)