pyvastbase 0.2.5

Python SDK for Vastbase V3 (PyMilvus-style API)

pyvastbase

Python SDK for Vastbase V3 — 对标 PyMilvus 接口风格，面向 Vastbase 向量引擎。支持 Vastbase 3.0.8+，对 3.0.9+ 新特性自动版本检测与功能门控。

安装

pip install pyvastbase

快速开始

方式一：VastbaseClient（pymilvus 风格）

from pyvastbase import VastbaseClient, DataType

# 使用URI连接字符串（pymilvus风格）
client = VastbaseClient(uri="postgresql://username:password@host:port/vastbase")

# 创建集合
client.create_collection("my_vectors", dimension=128)

# 插入数据
client.insert("my_vectors", [
    {"id": 1, "vector": [0.1] * 128, "text": "你好"},
    {"id": 2, "vector": [0.2] * 128, "text": "世界"},
])

# 向量搜索
results = client.search("my_vectors", data=[0.15] * 128, limit=5)

# 新增的 pymilvus 兼容方法
db_info = client.describe_database("vastbase")
print(db_info)  # {'name': 'vastbase', 'table_count': 5, ...}

results = client.hybrid_search("my_vectors", data=[0.15] * 128, filter="price < 1000", limit=20)

方式二：传统 Collection 风格

from pyvastbase import (
    connect, Collection, CollectionSchema, FieldSchema, DataType
)

# 连接 Vastbase
connect(
    host="your_host",
    port=5432,
    database="vastbase",
    user="your_user",
    password="your_password",
)

# 定义 schema
schema = CollectionSchema(
    name="my_vectors",
    fields=[
        FieldSchema(name="id", dtype=DataType.INT64, is_primary_key=True),
        FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=128),
        FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=65535),
    ]
)

# 创建并使用集合
col = Collection("my_vectors", schema=schema)
col.create()

# 插入数据
col.insert([
    {"id": 1, "embedding": [0.1] * 128, "text": "hello world"},
    {"id": 2, "embedding": [0.2] * 128, "text": "goodbye"},
])

# 向量搜索
results = col.search(
    data=[[0.15] * 128],
    anns_field="embedding",
    limit=5,
    expr="id < 100",
)
for result in results[0]:
    print(f"ID: {result.id}, Distance: {result.distance}")

支持的向量类型

类型	DataType 常量	SQL 类型	说明	最低版本
floatvector	DataType.FLOAT_VECTOR	floatvector(N)	4字节浮点向量	3.0.8 patch 3
int8vector	DataType.INT8_VECTOR	int8vector(N)	1字节有符号整型（节省约 3/4 存储）	3.0.9+
halfvector	DataType.FLOAT16_VECTOR	halfvector(N)	2字节半精度浮点 fp16	3.0.9+
sparsevector	DataType.SPARSE_FLOAT_VECTOR	sparsevector	稀疏浮点向量（BGE-M3, SPLADE）	3.0.9+

向后兼容别名: HALF_VECTOR -> FLOAT16_VECTOR, SPARSE_VECTOR -> SPARSE_FLOAT_VECTOR。

# int8vector — 紧凑型 1 字节整型向量
schema.add_field(FieldSchema("vec", DataType.INT8_VECTOR, dim=256))

# halfvector (fp16)
schema.add_field(FieldSchema("vec", DataType.FLOAT16_VECTOR, dim=512))

# sparsevector — 无需指定维度参数
schema.add_field(FieldSchema("vec", DataType.SPARSE_FLOAT_VECTOR))

# 稀疏向量使用 indices+values 字典格式插入
col.insert([{
    "id": 1,
    "vec": {"indices": [0, 5, 100], "values": [0.5, 0.8, 0.3]},
}])

版本检测与自动门控

SDK 通过 SELECT vb_version() 自动检测已连接的 Vastbase 版本，并对 3.0.9+ 特性进行功能门控。版本信息按连接缓存——仅查询一次，后续所有检查复用缓存结果。

双重校验：Build + Patch（来自 vb_version()）

Vastbase 通过 Build 号和 Patch 级别共同标识版本：

vb_version() 输出:
  version:V3.0 (Build 9)    -> build = 9
  patch:1                   -> patch_level = 1

功能门控同时检查两个条件：build >= 9 AND patch_level >= 1。

from pyvastbase import get_vb_version, check_vb_version

ver = get_vb_version()
print(ver["major"])              # 3
print(ver["minor"])              # 0
print(ver["build"])              # 9  (来自 version:V3.0 (Build 9))
print(ver["patch_level"])        # 1  (来自 patch:1)
print(ver["has_sparsevector"])   # True (build>=9 AND patch_level>=1)
print(ver["has_int8vector"])     # True

# 手动版本检查 (major, minor, build, patch_level)
check_vb_version((3, 0, 9, 1))   # True — build 9 >= 9, patch 1 >= 1
check_vb_version((3, 0, 9, 2))   # False — patch 1 < 2
check_vb_version((3, 0, 8, 1))   # True — 服务器版本高于最低要求

所有 3.0.9+ 操作在执行前会自动校验服务器版本。若服务器版本过低，会返回具体的错误信息，明确指出哪个条件未满足。

索引管理

from pyvastbase import IndexParams, IndexType

# Graph 索引 (HNSW)
params = IndexParams.graph_index(
    m=16,
    ef_construction=64,
    parallel_workers=4,
    auto_quantization=True,           # 3.0.9+: 达到 1 万行时自动启用 PQ/RabitQ
)
col.create_index(field_name="embedding", index_params=params)

# 全文索引，支持 BM25 同义词
ft_params = IndexParams.fulltext_index(
    dictionary="cn_tokenizer",
    algorithm="BM25",
    coefficients="b=0.75:k=1.2",
    synonyms={"search_terms": ["搜索", "检索", "查找"]},
)
col.create_index(field_name="content", index_params=ft_params)

BM25 全文搜索

from pyvastbase import bm25_search_highlight

# 直接模式——高亮单个文本的匹配结果
highlighted = bm25_search_highlight(
    "Vastbase is a vector database for AI",
    "vector database",
    start_sel="<b>",
    stop_sel="</b>",
)
# 结果: 'Vastbase is a <b>vector</b> <b>database</b> for AI'

# 表模式——高亮表中所有匹配行
results = bm25_search_highlight(
    query="向量",
    table_name="articles",
    text_column="content",
    dict_name="cn_tokenizer",
)
# 返回: ['<b>向量</b>数据库是AI应用的核心', ...]

混合搜索

将向量相似度与标量字段过滤结合，实现灵活的复合查询：

results = col.hybrid_search(
    data=[0.1] * 128,
    filter={
        "$and": [
            {"price": {"$lte": 1000}},
            {"category": {"$in": ["electronics", "books"]}}
        ]
    },
    limit=20,
)

for result in results[0]:
    print(f"ID: {result.id}, Distance: {result.distance}")
    print(f"Price: {result.data.get('price')}")

缓冲区检查

from pyvastbase import vectorbuffer_inspect, bulkbuffer_inspect

for buf in vectorbuffer_inspect():
    print(buf["buffer_id"], buf["table_name"], buf["status"])

for buf in bulkbuffer_inspect():
    print(buf["table_name"], buf["size"])

MutationResult

所有变更操作返回 MutationResultWrapper，同时支持向后兼容的列表访问和完整的 MutationResult 接口：

result = col.insert([{"id": 1, "embedding": [0.1] * 128, "text": "hello"}])

# 向后兼容（ID 列表）
list(result)                # [1]
len(result)                 # 1

# MutationResult 接口
result.insert_count         # 1
result.primary_keys         # [1]
result.upsert_count         # 0
result.delete_count         # 0

# 删除
result = col.delete(expr="id >= 100")
result.delete_count         # 5

用户与权限管理

from pyvastbase import create_user, drop_user, list_users, grant, revoke

create_user(user="reader", password="securepass")
grant(privilege="SELECT", object_type="COLLECTION", object_name="my_vectors", user="reader")
revoke(privilege="SELECT", object_type="COLLECTION", object_name="my_vectors", user="reader")
drop_user(user="reader")

数据库管理

from pyvastbase import create_database, drop_database, list_databases

create_database(database="newdb")
print(list_databases())    # ['postgres', 'vastbase', 'newdb']
drop_database(database="newdb")

集合生命周期

from pyvastbase import (
    list_collections, has_collection, describe_collection,
    create_collection, drop_collection, rename_collection,
    load_collection, release_collection,
)

create_collection(name="temp", fields=[
    {"name": "id", "dtype": DataType.INT64, "is_primary_key": True},
    {"name": "vec", "dtype": DataType.FLOAT_VECTOR, "dim": 256},
])

print(has_collection("temp"))      # True
info = describe_collection("temp") # 获取 schema 详情
print(info["row_count"], info["columns"])

load_collection("temp")
release_collection("temp")
drop_collection("temp")

分页搜索

# 批量迭代大量搜索结果
for batch in col.search_iterator(
    data=[0.1] * 128,
    limit=1000,
    batch_size=100,
):
    for result in batch:
        process(result)

# 批量查询迭代
for batch in col.query(batch_size=100):
    for record in batch:
        process(record)

异常处理

from pyvastbase import (
    ConnectionError, CollectionNotExistsError, VectorDimensionError,
    DataError, SchemaError
)

try:
    col = Collection("my_vectors")
    col.search(data=[[0.1] * 128], limit=5)
except CollectionNotExistsError:
    print("集合不存在")
except VectorDimensionError as e:
    print(f"维度不匹配: 期望 {e.expected}, 实际 {e.actual}")
except ConnectionError:
    print("数据库连接失败")

API 参考

完整 API 参考见: docs/api-reference.md（英文）| docs/api-reference_zh.md（简体中文）

模块	说明
Connection	connect, get_connection, get_vb_version, check_vb_version, health_check
Schema	DataType, FieldSchema, CollectionSchema, create_collection_schema
Collection	insert, delete, upsert, search, hybrid_search, query, get, 索引管理
Index	IndexType, DistanceType, IndexParams, IndexBuilder, QuantizerType
Search	SearchParams, SearchResult, SearchQueryBuilder, FulltextSearchBuilder
Utility	服务器信息, 用户/数据库管理, flush/compact, 3.0.9+ 新特性
ORM	Entity, Model, Field, VectorField
Exceptions	VastbaseException 继承层次
Async API	AsyncCollection, AsyncConnections
MutationResult	MutationResult, MutationResultWrapper

快速 API 参考

连接

connect(host, port, database, user, password, alias="default", pool_size=10)
get_connection(alias="default")
remove_connection(alias)
close_all()
list_connections()
health_check()
get_server_version()
get_vb_version()                           # 查询 vb_version()，按连接缓存
check_vb_version((3, 0, 9, 1))             # 检查 (major, minor, build, patch_level)

Schema

FieldSchema(name, dtype, dim=None, is_primary_key=False,
            max_length=None, description="", default_value=None)
CollectionSchema(name, fields, description="", auto_id=False)
create_collection_schema(name="my_coll", dim=128, metric_type=DistanceType.L2)

数据类型

# 标量类型
DataType.BOOL, .INT16, .INT32, .INT64, .FLOAT, .DOUBLE,
DataType.VARCHAR, .TEXT, .TIMESTAMP, .DATE, .JSON, .UUID,

# 向量类型（对标 pymilvus 规范名称）
DataType.FLOAT_VECTOR          # 所有版本
DataType.INT8_VECTOR           # 3.0.9+
DataType.FLOAT16_VECTOR        # 3.0.9+
DataType.SPARSE_FLOAT_VECTOR   # 3.0.9+

# 辅助方法
DataType.is_vector_type(dtype)
DataType.is_sparse_type(dtype)
DataType.is_dimensionless_type(dtype)

Collection

col = Collection(name, schema=None, using="default")
col.create()
col.drop()
col.exists()
col.truncate()
col.insert(data, normalize_vectors=False)          # -> MutationResultWrapper
col.batch_insert(data, batch_size=100)
col.delete(expr="id > 10") | delete(pks=[1, 2, 3])
col.upsert(data)
col.search(data, anns_field, param, limit, expr, output_fields)
col.hybrid_search(data, anns_field, param, limit, filter, output_fields)
col.query(expr="id < 100", limit=10, offset=0, batch_size=None)
col.get(ids=[1, 2, 3])
col.num_entities / col.is_empty
col.create_index(field_name, index_params)
col.drop_index(field_name)
col.has_index(field_name=None)
col.list_indexes()

索引与搜索常量

IndexType.GRAPH_INDEX, .HYBRID_ANN, .HNSW, .IVFFLAT, .IVFPQ, .FULLTEXT
DistanceType.L2, .COSINE, .IP
QuantizerType.NONE, .PQ, .RABITQ
SearchType.VECTOR, .HYBRID, .FULLTEXT, .RANGE, .KNN

ORM

from pyvastbase.orm import Entity, Model, Field, VectorField

@Model(collection="documents")
class Document(Entity):
    id = Field(primary_key=True, auto_increment=True)
    title = StringField(max_length=255)
    content = TextField()
    embedding = VectorField(dim=128)

# 从 Entity 生成 schema
schema = Document.to_schema()

# 便捷字段工厂
StringField, TextField, IntField, FloatField, BoolField, DateField, TimestampField

模块配置

configure(timeout=30.0, pool_size=10, max_overflow=10)
get_config()  # -> {"default_connection": "default", "timeout": 30.0, ...}

pymilvus 兼容性

已对齐

以下功能已与 pymilvus API 对齐：

• DataType IntEnum — 所有标量和向量类型使用 IntEnum，与 pymilvus 枚举值兼容
• MutationResult — insert/upsert/delete 返回 MutationResultWrapper，同时支持列表访问和结构化字段（insert_count, delete_count, primary_keys）
• SearchResult — 搜索结果结构与 pymilvus 一致，支持 result.id, result.distance, result.data
• connect(uri) — 支持通过 URI 连接字符串连接（postgresql://user:pass@host:port/db）
• hybrid_search — SDK 层多路向量召回 + RRF/WeightedRRF/WeightedRanker 重排序（pymilvus 兼容 API）
• get_entity_by_id 弃用 — 已弃用 get_entity_by_id，请使用 get(ids=...)
• metric_type/search_params 参数名 — 索引和搜索参数命名与 pymilvus 对齐
• VastbaseClient / MilvusClient — VastbaseClient 为高级入口；MilvusClient 别名用于导入兼容

新增实现方法（15 个原 stub 方法现已完整实现）

以下 15 个 VastbaseClient 方法此前为 NotSupportedError 占位 stub，现已通过三层架构（operations -> utility -> client）完整实现：

高级查询委托：

• hybrid_search — 多路向量召回与重排序
• query_iterator — 大结果集分页查询迭代
• search_iterator — 大结果集分页搜索迭代

数据库与用户信息查询：

• describe_index — 按集合和字段名获取索引详情
• describe_database — 获取数据库元信息（名称、表数量等）
• describe_user — 获取用户信息及角色详情
• update_password — 修改用户密码

别名管理：

• list_aliases — 列出所有集合别名
• describe_alias — 获取指定别名的详细信息

索引与集合属性管理：

• alter_index_properties — 修改索引级别属性
• drop_index_properties — 删除索引级别属性
• alter_collection_properties — 修改集合级别属性
• drop_collection_properties — 删除集合级别属性
• add_collection_field — 向已有集合添加新字段

Flush 操作：

• flush_all — 将所有集合数据刷入持久存储
• get_flush_all_state — 检查 flush_all 操作的完成状态

扩展功能

以下功能为 Vastbase 独有扩展，pymilvus 中不可用：

• normalize_vectors — insert() 和 upsert() 支持自动向量归一化
• pks on delete — delete(pks=[...]) 支持按主键列表删除
• limit/offset on query — query() 原生支持分页参数
• hybrid_ann_search — 混合 ANN 搜索，将向量相似度与标量过滤结合
• bm25_search_highlight — BM25 全文搜索高亮，支持自定义标签和中文分词
• buffer inspection — vectorbuffer_inspect / bulkbuffer_inspect 缓冲区检查

不支持

以下 pymilvus 功能在 Vastbase 中不支持：

• partition_names — PostgreSQL 不支持 Milvus 风格的分区，传入时静默忽略
• ranker/highlighter — 使用时抛出 ParamError
• FLAT 索引类型 — 不支持 FLAT 索引（Vastbase 使用 Graph/HNSW 索引）

架构

pyvastbase/
├── core/                  # 核心类型与抽象
│   ├── constants.py       # IndexType, DistanceType, 版本常量, 算子映射
│   ├── connections.py     # ConnectionConfig, VastbaseConnection (含版本缓存)
│   ├── exceptions.py      # 19 个异常类
│   ├── mutation_result.py # MutationResult, MutationResultWrapper
│   └── schema.py          # FieldSchema, CollectionSchema, DataType
├── collection.py          # Collection 类，完整的 CRUD、搜索、索引操作
├── utility.py             # 50+ 工具函数：用户/数据库/权限, flush/compact, 3.0.9+特性
├── index/
│   └── index_builder.py   # IndexParams (7 种索引类型), IndexBuilder, 工厂方法
├── search/
│   └── search_builder.py  # SearchParams, SearchQueryBuilder, FulltextSearchBuilder
├── utils/                 # SQL 生成, 向量/距离/过滤/结果工具
├── orm/                   # Entity/Model ORM 层 (Django 风格装饰器)
├── async_impl/            # AsyncCollection 等异步对应实现
├── tests/                 # 1100+ 测试覆盖所有模块
└── docs/
    ├── api-reference.md   # 完整 API 参考（英文）
    └── api-reference_zh.md # 完整 API 参考（简体中文）

环境要求

• Python >= 3.9
• psycopg[binary] >= 3.2
• Vastbase >= 3.0.8（新向量类型和特性需要 3.0.9+）

License

MIT