基础语法 #

PyMilvus SDK模块结构：

┌─────────────────────────────────────────┐
│              pymilvus                    │
├─────────────────────────────────────────┤
│  ├── connections      连接管理           │
│  ├── Collection       集合操作           │
│  ├── Partition        分区操作           │
│  ├── Index            索引操作           │
│  ├── utility          工具函数           │
│  ├── db               数据库操作         │
│  ├── schema           Schema定义         │
│  ├── types            类型定义           │
│  └── orm              ORM接口            │
└─────────────────────────────────────────┘

from pymilvus import (
    connections,
    Collection,
    Partition,
    FieldSchema,
    CollectionSchema,
    DataType,
    utility,
    db
)

from pymilvus import connections

connections.connect(
    alias="default",
    host="localhost",
    port="19530",
    user="root",
    password="Milvus",
    timeout=10
)

connections.disconnect("default")

connections.remove_connection("default")

print(connections.list_connections())

print(connections.get_connection_addr("default"))

from pymilvus import db

db.create_database("my_database")

db.using_database("my_database")

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

print(db.list_databases())

db.drop_database("my_database")

from pymilvus import FieldSchema, CollectionSchema, DataType

fields = [
    FieldSchema(
        name="id",
        dtype=DataType.INT64,
        is_primary=True,
        auto_id=False
    ),
    FieldSchema(
        name="title",
        dtype=DataType.VARCHAR,
        max_length=256
    ),
    FieldSchema(
        name="embedding",
        dtype=DataType.FLOAT_VECTOR,
        dim=768
    )
]

schema = CollectionSchema(
    fields=fields,
    description="文档集合",
    enable_dynamic_field=True
)

from pymilvus import Collection

collection = Collection(
    name="documents",
    schema=schema,
    using="default",
    shards_num=2
)

from pymilvus import utility

print(utility.list_collections())

print(collection.schema)

print(collection.description)

print(collection.num_entities)

collection.load()

collection.release()

print(collection.is_loaded)

utility.drop_collection("documents")

data = [
    [1, 2, 3],
    ["文档1", "文档2", "文档3"],
    [[0.1]*768, [0.2]*768, [0.3]*768]
]

result = collection.insert(data)
print(result.insert_count)

entities = [
    {"id": 1, "title": "文档1", "embedding": [0.1]*768},
    {"id": 2, "title": "文档2", "embedding": [0.2]*768}
]

collection.insert(entities)

entities = [
    {
        "id": 1,
        "title": "文档1",
        "embedding": [0.1]*768,
        "author": "张三",
        "tags": ["技术", "AI"]
    }
]

collection.insert(entities)

partition = collection.partition("2024_01")

data = [
    [1, 2],
    ["文档1", "文档2"],
    [[0.1]*768, [0.2]*768]
]

partition.insert(data)

import numpy as np

batch_size = 1000
total = 10000

for i in range(0, total, batch_size):
    batch_data = [
        list(range(i, i + batch_size)),
        [f"文档{j}" for j in range(i, i + batch_size)],
        np.random.rand(batch_size, 768).tolist()
    ]
    collection.insert(batch_data)

collection.load()

search_params = {
    "metric_type": "L2",
    "params": {"nprobe": 10}
}

results = collection.search(
    data=[[0.1]*768],
    anns_field="embedding",
    param=search_params,
    limit=10
)

for hits in results:
    for hit in hits:
        print(f"ID: {hit.id}, Distance: {hit.distance}")

search_params = {
    "metric_type": "L2",
    "params": {
        "nprobe": 16,
        "offset": 0,
        "radius": 1.0,
        "range_filter": 0.5
    }
}

results = collection.search(
    data=[[0.1]*768],
    anns_field="embedding",
    param=search_params,
    limit=10,
    expr='title like "%AI%"',
    output_fields=["title", "author"]
)

results = collection.search(
    data=[[0.1]*768],
    anns_field="embedding",
    param=search_params,
    limit=10,
    partition_names=["2024_01"]
)

query_vectors = [
    [0.1]*768,
    [0.2]*768,
    [0.3]*768
]

results = collection.search(
    data=query_vectors,
    anns_field="embedding",
    param=search_params,
    limit=10
)

results = collection.query(
    expr='id in [1, 2, 3]',
    output_fields=["id", "title", "embedding"]
)

for result in results:
    print(result)

results = collection.query(
    expr='title like "%AI%" and id > 10',
    output_fields=["id", "title"]
)

results = collection.query(
    expr='author in ["张三", "李四"]',
    output_fields=["id", "title", "author"]
)

results = collection.query(
    expr='id > 0',
    output_fields=["id", "title"],
    offset=0,
    limit=100
)

results = collection.query(
    expr='id > 0',
    output_fields=["id", "title"],
    partition_names=["2024_01"]
)

data = [
    [1],
    ["更新后的标题"],
    [[0.5]*768]
]

collection.upsert(data)

expr = 'id in [1, 2, 3]'
collection.delete(expr)

expr = 'title like "%test%"'
collection.delete(expr)

index_params = {
    "metric_type": "L2",
    "index_type": "IVF_FLAT",
    "params": {"nlist": 128}
}

collection.create_index(
    field_name="embedding",
    index_params=index_params,
    index_name="embedding_idx"
)

print(collection.indexes)

for index in collection.indexes:
    print(f"Field: {index.field_name}")
    print(f"Params: {index.params}")

collection.drop_index(index_name="embedding_idx")

collection.create_partition("2024_01", description="2024年1月数据")

print(collection.partitions)

for partition in collection.partitions:
    print(f"Name: {partition.name}")
    print(f"Description: {partition.description}")
    print(f"Num entities: {partition.num_entities}")

collection.drop_partition("2024_01")

expr = 'id == 1'
expr = 'id != 1'
expr = 'id > 10'
expr = 'id >= 10'
expr = 'id < 100'
expr = 'id <= 100'

expr = 'id > 10 and id < 100'
expr = 'id < 10 or id > 100'
expr = 'not (id == 50)'

expr = 'id in [1, 2, 3]'
expr = 'id not in [1, 2, 3]'

expr = 'title like "%AI%"'
expr = 'title like "AI%"'
expr = 'title like "%AI"'

expr = 'metadata["category"] == "tech"'
expr = 'json_contains(tags, "AI")'
expr = 'json_contains_all(tags, ["AI", "ML"])'
expr = 'json_contains_any(tags, ["AI", "ML"])'

expr = 'array_length(tags) > 2'
expr = 'array_contains(tags, "AI")'

规范：
- 只包含字母、数字、下划线
- 以字母或下划线开头
- 长度1-255字符
- 不区分大小写

正确示例：
- products
- user_profiles
- _temp_collection

错误示例：
- 123collection
- my-collection
- collection name

规范：
- 只包含字母、数字、下划线
- 以字母开头
- 长度1-255字符
- 不能与保留字冲突

正确示例：
- id
- user_name
- embedding_768

错误示例：
- 123field
- field-name
- from (保留字)

规范：
- 只包含字母、数字、下划线
- 以字母或下划线开头
- 长度1-255字符
- 不能使用 _default (保留)

正确示例：
- 2024_01
- region_beijing
- category_electronics

from pymilvus import MilvusException

try:
    collection = Collection("non_existent")
except MilvusException as e:
    print(f"Error code: {e.code}")
    print(f"Error message: {e.message}")

from pymilvus import connections
from contextlib import contextmanager

@contextmanager
def milvus_connection(alias, host, port):
    try:
        connections.connect(alias, host=host, port=port)
        yield
    finally:
        connections.disconnect(alias)

with milvus_connection("default", "localhost", "19530"):
    pass

def batch_insert(collection, data, batch_size=1000):
    total = len(data[0])
    for i in range(0, total, batch_size):
        batch = [field[i:i+batch_size] for field in data]
        collection.insert(batch)
    collection.flush()

from pymilvus import MilvusException
import time

def search_with_retry(collection, vectors, max_retries=3):
    for attempt in range(max_retries):
        try:
            return collection.search(
                data=vectors,
                anns_field="embedding",
                param={"metric_type": "L2", "params": {"nprobe": 10}},
                limit=10
            )
        except MilvusException as e:
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)