九有数据库PYUDTV使用

from pymilvus import MilvusClient # UDBClient的异名类
import numpy as np
VECTOR_FIELD = 'vector'
user_name = "unvdb"
password = "unvdb"
host = "localhost"
port = 5678
db_name = "test"
# unvdb://是实例化UNVDB后端标志
client = MilvusClient(f"unvdb://{user_name}:{password}@{host}:{port}/{db_name}")
# client._wrapped_class 查看原始的类 MilvusClient或者UDBClientBase
print(f"{client._wrapped_class} 连接成功！")

# 创建表, 极简模式，默认创建vector字段为向量字段。 创建成功之后，可以在数据库找到test这张表
COLLECTION_NAME = 'test'
DIM = 6
client.create_collection(
            collection_name=COLLECTION_NAME, # 表名
            dimension=DIM
        )
print("集合创建成功！")

# 生成测试数据
num_entities = 100
rng = np.random.default_rng(seed=58)
entities = []
for i in range(num_entities):
    ent_ = {
        "id": i + 1,
        VECTOR_FIELD: rng.random((1, DIM))[0].tolist(),
        "tag": f"tag_{i % 10}"
    }
    entities.append(ent_)
# 插入数据
insert_result = client.insert(
        collection_name=COLLECTION_NAME,
        data=entities,
        progress_bar=True
    )

print("插入数据成功")

# 加载表， UNVDB 后端可选
client.load_collection(COLLECTION_NAME)

# 查找数据
query_vector = rng.random((1, DIM))[0].tolist()
search_results = client.search(
    collection_name=COLLECTION_NAME,
    data=[query_vector],
    limit=5,
    output_fields=["id", "tag", "vector"],
    anns_field=VECTOR_FIELD,
    search_params={"nprobe": 10} # 查找聚类中心数
)
print(type(search_results), type(search_results[0]))
print("搜索结果:")
for i, hits in enumerate(search_results):
    print(f"查询 {i+1} 的结果:")
    for hit in hits:
        print(f"  ID: {hit['id']}, 距离: {hit['distance']:.4f}, 标签: {hit.get('tag')}, 值: {hit.get('vector')}")

# 删除测试表
client.drop_collection(COLLECTION_NAME)
print(f"删除测试表：{COLLECTION_NAME}")

def create_schema(cls, **kwargs) -> CollectionSchema

def create_collection(
    self,
    collection_name: str,
    dimension: Optional[int] = None,
    primary_field_name: str = "id",
    id_type: str = "int",
    vector_field_name: str = "vector",
    metric_type: str = "COSINE",
    auto_id: bool = True,
    timeout: Optional[float] = None,
    schema: Optional[CollectionSchema] = None,
    **kwargs,
)

def insert(
    self,
    collection_name: str,
    data: Union[Dict, List[Dict]],
    timeout: Optional[float] = None,
    partition_name: Optional[str] = "",
    **kwargs,
) -> Dict

def upsert(
    self,
    collection_name: str,
    data: Union[Dict, List[Dict]],
    timeout: Optional[float] = None,
    partition_name: Optional[str] = "",
    **kwargs,
) -> Dict

def search(
    self,
    collection_name: str,
    data: Union[List[List[float]], List[float]],
    limit: int = 10,
    output_fields: Optional[List[str]] = None,
    filter: Optional[str] = "",
    timeout: Optional[float] = None,
    anns_field: str = "",
    search_params: Optional[dict] = None,
    **kwargs,
) -> SearchResult

def query(
    self,
    collection_name: str,
    filter: Optional[str] = "",
    output_fields: Optional[List[str]] = None,
    timeout: Optional[float] = None,
    limit: Optional[int] = None,
    **kwargs,
) -> HybridExtraList

def delete(
    self,
    collection_name: str,
    ids: Optional[List[int]] = None,
    timeout: Optional[float] = None,
    filter: Optional[str] = "",
    **kwargs,
) -> Dict

def update(
    self,
    collection_name: str,
    data: Dict,
    ids: List[int] = None,
    timeout: Optional[float] = None,
    filter: Optional[str] = "",
    **kwargs,
) -> Dict

def hybrid_search(
    self,
    collection_name: str,
    reqs: List[AnnSearchRequest],
    ranker: Union[WeightedRanker, RRFRanker],
    limit: int = 10,
    output_fields: Optional[List[str]] = None,
    timeout: Optional[float] = None,
    **kwargs,
) -> HybridHits

def create_index(
    self,
    collection_name: str,
    index_params: Optional[dict] = None,
    timeout: Optional[float] = None,
    field_name: Optional[str] = "",
    **kwargs,
)

def describe_index(
    self,
    collection_name: str,
    index_name: Optional[str] = "",
    timeout: Optional[float] = None,
    **kwargs,
) -> dict

def drop_index(
    self,
    collection_name: str,
    index_name: Optional[str] = "",
    timeout: Optional[float] = None,
    field_name: Optional[str] = "",
    **kwargs,
)

def list_indexes(
    self,
    collection_name: str,
    field_name: Optional[str] = "",
    **kwargs,
) -> List[str]

def has_collection(
    self,
    collection_name: str,
    timeout: Optional[float] = None,
    **kwargs,
) -> bool

def list_collections(self, **kwargs) -> List[str]

def describe_collection(
    self,
    collection_name: str,
    timeout: Optional[float] = None,
    **kwargs,
) -> dict

def drop_collection(
    self,
    collection_name: str,
    timeout: Optional[float] = None,
    **kwargs,
)

def load_collection(
    self,
    collection_name: str,
    *args,
    **kwargs,
)

def release_collection(self, *args, **kwargs)

def close(self)