# pbatis-milvus

**Repository Path**: pbatis/pbatis-milvus

## Basic Information

- **Project Name**: pbatis-milvus
- **Description**: No description available
- **Primary Language**: Unknown
- **License**: MIT
- **Default Branch**: main
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-11-04
- **Last Updated**: 2026-11-04

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

<div align="center">
  <img src="docs/pbatis-logo.png" alt="PyBatis Logo" width="220"/>

  <h1>pbatis-milvus</h1>

  <p><a href="https://gitee.com/pbatis/pbatis">PyBatis</a> 的 <a href="https://milvus.io">Milvus</a> 向量数据库方言插件，基于 <code>milvusql</code> DBAPI。</p>

  [![License](https://img.shields.io/github/license/pbatis/pbatis-milvus)](LICENSE)

  [PyBatis 主页](https://gitee.com/pbatis/pbatis) · [milvusql DBAPI](https://github.com/Callix-Tools/milvusql)
</div>

---

`select()`、DDL（`CREATE TABLE` / `CREATE INDEX`）、反射（reflection）、Alembic 迁移——全部走标准 PyBatis API。

向量检索就是 `.order_by(column.cosine_distance(q))`，写法和 [pgvector](https://github.com/pgvector/pgvector-python) 的比较器完全一致。JOIN、分组聚合、子查询、关系对象的 `.any()` / `.has()`（相关 `EXISTS`，去相关成 semi/anti join）都会经过 DBAPI 的关系型查询计划器。`sa.Text` 列就是 Milvus 的全文输入（带分析器），可直接承载 BM25 生成的稀疏字段。

## 安装

```bash
pip install pbatis-milvus
```

依赖：

```bash
pip install pbatis        # ORM 核心（>=1.1.3）
pip install milvusql      # Milvus DBAPI（>=0.1.0,<0.2）
```

## 快速开始

```python
from pbatis import create_engine, select
from pbatis_milvus.types import VECTOR

engine = create_engine("milvusql:///items.db")  # Milvus Lite，或真实服务端的 URI

with engine.connect() as conn:  # items: 带 embedding VECTOR 列的 Table
    rows = conn.execute(
        select(items.c.id, items.c.category)
        .order_by(items.c.embedding.l2_distance([0.1] * 768))
        .limit(5)
    ).all()
```

异步版本：

```python
from pbatis.ext.asyncio import create_async_engine

engine = create_async_engine("milvusql+aio:///items.db")

async with engine.connect() as conn:
    rows = (
        await conn.execute(
            select(items.c.id).order_by(items.c.embedding.cosine_distance(q)).limit(5)
        )
    ).all()
```

## 连接串

```
milvusql:///items.db                                 # Milvus Lite —— 相对路径
milvusql:////abs/path/items.db                      # Milvus Lite —— 绝对路径
milvusql://user:password@host:19530/db_name         # 真实服务端
milvusql://user:password@host:19530/db_name?secure=1  # TLS (https)
milvusql+aio://...                                  # 同上 URL 形式，异步引擎
```

## API

### `VECTOR(dim)` / `SPARSEVEC`

```python
from pbatis import Column, Table, BigInteger
from pbatis_milvus.types import VECTOR, SPARSEVEC

items = Table(
    "items", metadata,
    Column("id", BigInteger, primary_key=True, autoincrement=True),
    Column("embedding", VECTOR(768)),
    Column("sparse", SPARSEVEC),
    milvusql_shards=1,
    milvusql_consistency_level="Strong",
)
```

| 比较器方法 | MilvusQL 运算符 | 适用类型 |
|---|:---:|---|
| `.l2_distance(other)` | `<->` | `VECTOR` |
| `.cosine_distance(other)` | `<=>` | `VECTOR` |
| `.l1_distance(other)` | `<+>` | `VECTOR` |
| `.max_inner_product(other)` | `<#>` | `VECTOR`、`SPARSEVEC` |

绑定值直接以 `list[float]` 传入——不会被字符串化拼进 SQL 文本。

### `hybrid_search()`

加权多向量检索，渲染成 MilvusQL 的 `HYBRID SEARCH ... RERANK ...`，用在 `ORDER BY` 位置：

```python
from pbatis_milvus.hybrid import hybrid_search, weighted

select(items.c.id).order_by(
    hybrid_search(
        weighted(items.c.embedding.cosine_distance(dense_q), 0.7),
        weighted(items.c.sparse.max_inner_product(sparse_q), 0.3),
        rerank="RRF", k=60,
    )
).limit(10)
```

### DDL 选项

方言专属的 `Table` / `Index` 参数，机制与 `mysql_engine=...` 一致：

```python
Table(
    "items", metadata, ...,
    milvusql_shards=2,
    milvusql_consistency_level="Bounded",
    milvusql_partition_key="category",
)

Index(
    "idx_embedding", items.c.embedding,
    milvusql_using="HNSW",
    milvusql_with={"metric_type": "COSINE", "M": 16, "ef_construction": 200},
)
```

### 一致性级别

语句自身的 `CONSISTENCY LEVEL` 子句优先；否则回退到连接的隔离级别——即 Milvus 自身的级别（`Strong`、`Bounded`、`Eventually`、`Session`、`Customized`），用标准 PyBatis 方式设置：

```python
conn = engine.connect().execution_options(isolation_level="Bounded")
```

### Alembic

导入本方言会自动把 `milvusql` 的 `DefaultImpl` 注册到 Alembic——`alembic upgrade` / `downgrade` 对 `milvusql://...` URL 直接可用，无需额外配置。

注意 `transactional_ddl = False`：Milvus 没有多语句回滚，迁移失败后 Alembic 无法撤销——请写只进不退（forward-only）的迁移脚本。

## 常见操作

```python
from pbatis import select, func, desc, text, DDL

# 条件查询 + 排序
conn.execute(
    select(items.c.id, items.c.category)
    .where(items.c.category == "books")
    .order_by(desc(items.c.created_at))
    .limit(20)
).all()

# 聚合
conn.execute(
    select(items.c.category, func.count(items.c.id).label("n"))
    .group_by(items.c.category)
).all()

# 原生 SQL
conn.execute(text("SELECT count(*) FROM items")).scalar()

# DDL
conn.execute(DDL("CREATE INDEX idx_cat ON items(category)"))
```

## 注意事项

| 项目 | 说明 |
|---|---|
| 事务 | Milvus 不支持常规事务，批量写入靠 DBAPI 的批处理语义 |
| 主键 | Milvus 主键需显式指定，常见做法用雪花 ID（见 `pybatis.ext.snowflake`） |
| `VECTOR` 维度 | 建表后维度不可更改，插入时长度必须与定义一致 |
| 索引 | HNSW / IVF 系列索引在建表后通过 `CREATE INDEX` 创建 |
| 稀疏向量 | `SPARSEVEC` 需要 BM25 等算法生成，配合 `Text` 列做混合检索 |

## 相关项目

- PyBatis 核心：https://gitee.com/pbatis/pbatis
- milvusql DBAPI：https://github.com/Callix-Tools/milvusql
- Milvus 官方文档：https://milvus.io/docs

## 版本与许可

当前版本：1.0.0

协议：MIT（详见 `LICENSE` 文件）

---

上游来源：`milvusql-sqlalchemy` 1.0.0（MIT）