From 420aa3490204c0130698fbe560188e27ef6e6f22 Mon Sep 17 00:00:00 2001 From: Cheney Zhang Date: Mon, 6 Jul 2026 11:24:58 +0000 Subject: [PATCH 1/2] feat(storage): add Milvus vector adapter --- docs/en/guides/01-configuration.md | 34 +- docs/zh/guides/01-configuration.md | 34 +- .../storage/vectordb_adapters/README.md | 28 +- .../storage/vectordb_adapters/__init__.py | 2 + .../storage/vectordb_adapters/factory.py | 2 + .../vectordb_adapters/milvus_adapter.py | 1644 +++++++++++++++++ .../utils/config/vectordb_config.py | 76 +- pyproject.toml | 3 + tests/storage/test_milvus_adapter.py | 367 ++++ uv.lock | 361 ++-- 10 files changed, 2376 insertions(+), 175 deletions(-) create mode 100644 openviking/storage/vectordb_adapters/milvus_adapter.py create mode 100644 tests/storage/test_milvus_adapter.py diff --git a/docs/en/guides/01-configuration.md b/docs/en/guides/01-configuration.md index 67a4080a38..fcc41f5719 100644 --- a/docs/en/guides/01-configuration.md +++ b/docs/en/guides/01-configuration.md @@ -1209,7 +1209,7 @@ Vector database storage configuration | Parameter | Type | Description | Default | |-----------|------|-------------|---------| -| `backend` | str | VectorDB backend type: 'local' (file-based), 'http' (remote service), 'volcengine' (cloud VikingDB), 'vikingdb' (private deployment), 'cuvs' (local storage + GPU dense search), 'qdrant', or 'opengauss' | "local" | +| `backend` | str | VectorDB backend type: 'local' (file-based), 'http' (remote service), 'volcengine' (cloud VikingDB), 'vikingdb' (private deployment), 'cuvs' (local storage + GPU dense search), 'qdrant', 'milvus', or 'opengauss' | "local" | | `name` | str | VectorDB collection name | "context" | | `url` | str | Remote service URL for 'http' type (e.g., 'http://localhost:5000') | null | | `project_name` | str | Project name (alias project) | "default" | @@ -1220,6 +1220,7 @@ Vector database storage configuration | `vikingdb` | object | 'vikingdb' type private deployment configuration | - | | `cuvs` | object | NVIDIA cuVS configuration for the 'cuvs' backend and the opt-in memory-aware auto mode on 'local'; see the [cuVS guide](./16-cuvs.md) | - | | `qdrant` | object | 'qdrant' type Qdrant configuration | - | +| `milvus` | object | 'milvus' type Milvus or Zilliz Cloud configuration | - | | `opengauss` | object | 'opengauss' native vector backend configuration | - | Default local mode @@ -1254,6 +1255,37 @@ Supports cloud-deployed VikingDB on Volcengine ``` +
+Milvus + +Install the `milvus` optional extra before using this backend. The default `uri` +uses Milvus Lite and stores data in a local `milvus.db` file. Use an HTTP URI for +self-hosted Milvus, or a cloud endpoint plus `token` for Zilliz Cloud. + +```json +{ + "storage": { + "vectordb": { + "name": "context", + "backend": "milvus", + "project": "default", + "distance_metric": "cosine", + "dimension": 1024, + "milvus": { + "uri": "./milvus.db", + "token": null, + "db_name": null, + "consistency_level": "Session" + } + } + } +} +``` + +For a self-hosted server, set `"uri": "http://localhost:19530"`. For Zilliz Cloud, +set `"uri"` to the cloud endpoint and provide `"token"`. +
+
openGauss diff --git a/docs/zh/guides/01-configuration.md b/docs/zh/guides/01-configuration.md index bfe1080785..033072d575 100644 --- a/docs/zh/guides/01-configuration.md +++ b/docs/zh/guides/01-configuration.md @@ -1181,7 +1181,7 @@ RAGFS 默认使用 Rust binding 模式，通过 Rust 实现直接访问文件系 | 参数 | 类型 | 说明 | 默认值 | |------|------|------|--------| -| `backend` | str | VectorDB 后端类型: 'local'（基于文件）, 'http'（远程服务）, 'volcengine'（云上 VikingDB）, 'vikingdb'（私有部署）, 'cuvs'（本地存储 + GPU dense search）, 'qdrant' 或 'opengauss' | "local" | +| `backend` | str | VectorDB 后端类型: 'local'（基于文件）, 'http'（远程服务）, 'volcengine'（云上 VikingDB）, 'vikingdb'（私有部署）, 'cuvs'（本地存储 + GPU dense search）, 'qdrant'、'milvus' 或 'opengauss' | "local" | | `name` | str | VectorDB 的集合名称 | "context" | | `url` | str | 'http' 类型的远程服务 URL（例如 'http://localhost:5000'） | null | | `project_name` | str | 项目名称（别名 project） | "default" | @@ -1192,6 +1192,7 @@ RAGFS 默认使用 Rust binding 模式，通过 Rust 实现直接访问文件系 | `vikingdb` | object | 'vikingdb' 类型的私有部署配置 | - | | `cuvs` | object | NVIDIA cuVS 配置，也用于在 'local' 下显式开启显存感知自动模式，参见 [cuVS 使用指南](./16-cuvs.md) | - | | `qdrant` | object | 'qdrant' 类型的 Qdrant 配置 | - | +| `milvus` | object | 'milvus' 类型的 Milvus 或 Zilliz Cloud 配置 | - | | `opengauss` | object | 'opengauss' 原生向量后端配置 | - | 默认使用本地模式 @@ -1226,6 +1227,37 @@ RAGFS 默认使用 Rust binding 模式，通过 Rust 实现直接访问文件系 ```
+
+Milvus + +使用该后端前需要安装 `milvus` 可选依赖。默认 `uri` 使用 Milvus Lite， +并把数据写入本地 `milvus.db` 文件。自托管 Milvus 使用 HTTP URI； +Zilliz Cloud 使用云端 endpoint 并配置 `token`。 + +```json +{ + "storage": { + "vectordb": { + "name": "context", + "backend": "milvus", + "project": "default", + "distance_metric": "cosine", + "dimension": 1024, + "milvus": { + "uri": "./milvus.db", + "token": null, + "db_name": null, + "consistency_level": "Session" + } + } + } +} +``` + +如果连接自托管服务，可设置 `"uri": "http://localhost:19530"`。如果使用 +Zilliz Cloud，把 `"uri"` 设置为云端 endpoint，并填写 `"token"`。 +
+
openGauss diff --git a/openviking/storage/vectordb_adapters/README.md b/openviking/storage/vectordb_adapters/README.md index 4158306c47..a3f40099c0 100644 --- a/openviking/storage/vectordb_adapters/README.md +++ b/openviking/storage/vectordb_adapters/README.md @@ -146,6 +146,32 @@ 1. `backend`: 填写 Adapter 类的完整 Python 路径（例如 `my_project.adapters.MyAdapter`）。 2. `custom_params`: 这是一个字典，你可以放入任何自定义参数，Adapter 的 `from_config` 方法可以通过 `config.custom_params` 获取这些值。 +内置 Milvus 后端可直接使用 registry 名称，并通过 `milvus` 配置块传递连接信息： + +```json +{ + "storage": { + "vectordb": { + "backend": "milvus", + "name": "context", + "project": "default", + "distance_metric": "cosine", + "dimension": 1024, + "milvus": { + "uri": "./milvus.db", + "token": null, + "db_name": null, + "consistency_level": "Session" + } + } + } +} +``` + +`uri` 默认为 Milvus Lite 本地文件；也可以设置为自建 Milvus 服务 +（如 `http://localhost:19530`）或 Zilliz Cloud endpoint，认证统一使用 +`token` 字段。 + --- @@ -238,4 +264,4 @@ class ThirdPartyCollectionAdapter(CollectionAdapter): - `backend=thirdparty` 可正常初始化。 - create 后可完成 upsert/get/query/delete/count 全流程。 - 不改上层业务调用方式即可参与 `find/search` 检索链路。 -- 后端差异全部封装在 adapter 层。 \ No newline at end of file +- 后端差异全部封装在 adapter 层。 diff --git a/openviking/storage/vectordb_adapters/__init__.py b/openviking/storage/vectordb_adapters/__init__.py index bb33a3a368..b97ea18e40 100644 --- a/openviking/storage/vectordb_adapters/__init__.py +++ b/openviking/storage/vectordb_adapters/__init__.py @@ -6,6 +6,7 @@ from .factory import create_collection_adapter from .http_adapter import HttpCollectionAdapter from .local_adapter import CuVSCollectionAdapter, LocalCollectionAdapter +from .milvus_adapter import MilvusCollectionAdapter from .opengauss_adapter import OpenGaussCollectionAdapter from .qdrant_adapter import QdrantCollectionAdapter from .vikingdb_private_adapter import VikingDBPrivateCollectionAdapter @@ -16,6 +17,7 @@ "LocalCollectionAdapter", "CuVSCollectionAdapter", "HttpCollectionAdapter", + "MilvusCollectionAdapter", "OpenGaussCollectionAdapter", "QdrantCollectionAdapter", "VolcengineCollectionAdapter", diff --git a/openviking/storage/vectordb_adapters/factory.py b/openviking/storage/vectordb_adapters/factory.py index 24e570e6c7..bde47f2248 100644 --- a/openviking/storage/vectordb_adapters/factory.py +++ b/openviking/storage/vectordb_adapters/factory.py @@ -9,6 +9,7 @@ from .base import CollectionAdapter from .http_adapter import HttpCollectionAdapter from .local_adapter import CuVSCollectionAdapter, LocalCollectionAdapter +from .milvus_adapter import MilvusCollectionAdapter from .opengauss_adapter import OpenGaussCollectionAdapter from .qdrant_adapter import QdrantCollectionAdapter from .vikingdb_private_adapter import VikingDBPrivateCollectionAdapter @@ -18,6 +19,7 @@ "local": LocalCollectionAdapter, "cuvs": CuVSCollectionAdapter, "http": HttpCollectionAdapter, + "milvus": MilvusCollectionAdapter, "opengauss": OpenGaussCollectionAdapter, "qdrant": QdrantCollectionAdapter, "volcengine": VolcengineCollectionAdapter, diff --git a/openviking/storage/vectordb_adapters/milvus_adapter.py b/openviking/storage/vectordb_adapters/milvus_adapter.py new file mode 100644 index 0000000000..6f7915382d --- /dev/null +++ b/openviking/storage/vectordb_adapters/milvus_adapter.py @@ -0,0 +1,1644 @@ +# Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd. +# SPDX-License-Identifier: AGPL-3.0 +"""Milvus-backed vector collection adapter.""" + +from __future__ import annotations + +import datetime as dt +import json +import math +import re +from typing import Any, Dict, Iterable, List, Optional, Sequence + +from openviking.storage.expr import ( + And, + Contains, + Eq, + FilterExpr, + In, + Or, + PathScope, + Range, + RawDSL, + TimeRange, +) +from openviking.storage.vectordb.collection.collection import Collection, ICollection +from openviking.storage.vectordb.collection.result import ( + AggregateResult, + DataItem, + FetchDataInCollectionResult, + SearchItemResult, + SearchResult, +) +from openviking.storage.vectordb.index.index import IIndex +from openviking.storage.vectordb.store.data import DeltaRecord +from openviking.storage.vectordb_adapters.base import CollectionAdapter +from openviking_cli.utils import get_logger + +logger = get_logger(__name__) + +_DEFAULT_URI = "./milvus.db" +_DEFAULT_TIMEOUT_SECONDS = 30 +_DEFAULT_QUERY_LIMIT = 10_000 +_MILVUS_MAX_COLLECTION_NAME_LENGTH = 255 +_MILVUS_VARCHAR_MAX_LENGTH = 65_535 +_ID_MAX_LENGTH = 512 +_URI_MAX_LENGTH = 4096 +_FIELD_NAME_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") +_COLLECTION_NAME_RE = re.compile(r"[^A-Za-z0-9_]+") +_VECTOR_FIELD_TYPES = {"vector", "float_vector"} +_LIST_STRING_FIELD_TYPES = {"list", "array"} +_STRING_FIELD_TYPES = {"string", "path", "text", "date_time"} +_INT_FIELD_TYPES = {"int64", "int32", "integer", "long"} +_FLOAT_FIELD_TYPES = {"float", "double"} +_BOOL_FIELD_TYPES = {"bool", "boolean"} +_META_PROPERTY_KEY = "openviking_meta" +_INDEX_META_PROPERTY_PREFIX = "openviking_index_" +_META_COLLECTION_NAME = "ov_openviking_milvus_meta" +_META_VECTOR_FIELD = "meta_vector" + + +def _import_pymilvus(): + try: + import pymilvus # type: ignore # noqa: PLC0415 + + return pymilvus + except ImportError as exc: # pragma: no cover - exercised only without optional driver + raise ImportError( + "The Milvus backend requires pymilvus with Milvus Lite support. " + "Install the `openviking[milvus]` optional extra." + ) from exc + + +def _safe_collection_name(*parts: Any, prefix: str = "ov") -> str: + raw = "_".join(str(part or "") for part in parts) + normalized = _COLLECTION_NAME_RE.sub("_", raw).strip("_") + if not normalized: + normalized = "default" + if normalized[0].isdigit(): + normalized = f"{prefix}_{normalized}" + elif prefix and not normalized.startswith(f"{prefix}_"): + normalized = f"{prefix}_{normalized}" + if len(normalized) <= _MILVUS_MAX_COLLECTION_NAME_LENGTH: + return normalized + + import hashlib + + digest = hashlib.sha1(normalized.encode("utf-8")).hexdigest()[:10] + keep = _MILVUS_MAX_COLLECTION_NAME_LENGTH - len(digest) - 1 + return f"{normalized[:keep]}_{digest}" + + +def _normalize_distance(distance: str) -> str: + value = (distance or "cosine").strip().lower() + if value not in {"cosine", "l2", "ip"}: + raise ValueError( + f"Milvus backend supports only cosine, l2, and ip distance metrics; got {distance!r}" + ) + return value + + +def _milvus_metric(distance: str) -> str: + return {"cosine": "COSINE", "l2": "L2", "ip": "IP"}[_normalize_distance(distance)] + + +def _json_default(value: Any) -> Any: + if isinstance(value, (dt.datetime, dt.date)): + return value.isoformat() + return str(value) + + +def _json_dumps(value: Any) -> str: + return json.dumps(value, ensure_ascii=False, sort_keys=True, default=_json_default) + + +def _json_loads(value: Any) -> Any: + if not isinstance(value, str): + return value + try: + return json.loads(value) + except (TypeError, ValueError): + return value + + +def _truncate_utf8(value: str, byte_limit: int) -> str: + encoded = value.encode("utf-8") + if len(encoded) <= byte_limit: + return value + cut = byte_limit + while cut > 0 and (encoded[cut] & 0xC0) == 0x80: + cut -= 1 + return encoded[:cut].decode("utf-8") + + +def _encode_scope_roots(value: Any) -> str: + roots = value if isinstance(value, list) else [value] + normalized = [str(root) for root in roots if root is not None] + return "\n" + "\n".join(normalized) + "\n" if normalized else "\n" + + +def _sparse_dot(left: Optional[Dict[str, float]], right: Optional[Dict[str, float]]) -> float: + if not left or not right: + return 0.0 + total = 0.0 + for key, raw_value in left.items(): + try: + total += float(raw_value) * float(right.get(key, 0.0)) + except (TypeError, ValueError): + continue + return total + + +def _coerce_datetime_value(value: Any) -> Any: + if isinstance(value, dt.datetime): + if value.tzinfo is None: + value = value.replace(tzinfo=dt.timezone.utc) + return value.isoformat() + if isinstance(value, dt.date): + return value.isoformat() + return value + + +def _format_number(value: Any) -> str: + if isinstance(value, bool): + raise ValueError("Boolean values are not valid numeric filter operands") + number = float(value) + if not math.isfinite(number): + raise ValueError(f"Invalid numeric filter value: {value!r}") + if number.is_integer(): + return str(int(number)) + return format(number, ".12g") + + +def _quote_value(value: Any) -> str: + value = _coerce_datetime_value(value) + if value is None: + return "null" + if isinstance(value, bool): + return "true" if value else "false" + if isinstance(value, (int, float)) and not isinstance(value, bool): + return _format_number(value) + text = ( + str(value) + .replace("\\", "\\\\") + .replace("\n", "\\n") + .replace("\r", "\\r") + .replace("\t", "\\t") + .replace('"', '\\"') + ) + return f'"{text}"' + + +def _format_value_list(values: Iterable[Any]) -> str: + return "[" + ", ".join(_quote_value(value) for value in values) + "]" + + +def _score_from_hit(hit: Dict[str, Any], distance_metric: str) -> float: + raw_score = ( + hit.get("score") + if hit.get("score") is not None + else hit.get("distance", hit.get("_distance", 0.0)) + ) + try: + score = float(raw_score) + except (TypeError, ValueError): + return 0.0 + if not math.isfinite(score): + return 0.0 + if distance_metric == "l2": + return 1.0 / (1.0 + max(score, 0.0)) + return score + + +class MilvusIndex(IIndex): + """Metadata-only logical index facade for Milvus.""" + + def __init__(self, collection: "MilvusCollection", index_name: str, meta: Dict[str, Any]): + super().__init__(meta=meta) + self._collection = collection + self._index_name = index_name + self._meta = dict(meta) + + def upsert_data(self, delta_list: List[DeltaRecord]): + raise NotImplementedError("MilvusIndex.upsert_data is managed at collection level") + + def delete_data(self, delta_list: List[DeltaRecord]): + raise NotImplementedError("MilvusIndex.delete_data is managed at collection level") + + def search( + self, + query_vector: Optional[List[float]], + limit: int = 10, + filters: Optional[Dict[str, Any]] = None, + sparse_raw_terms: Optional[List[str]] = None, + sparse_values: Optional[List[float]] = None, + ): + raise NotImplementedError("MilvusIndex.search is not exposed via raw index interface") + + def aggregate(self, filters: Optional[Dict[str, Any]] = None): + raise NotImplementedError("MilvusIndex.aggregate is not exposed via raw index interface") + + def update( + self, scalar_index: Optional[Dict[str, Any]] = None, description: Optional[str] = None + ): + self._collection.update_index( + index_name=self._index_name, + scalar_index=scalar_index, + description=description, + ) + self._meta = self._collection.get_index_meta_data(self._index_name) or self._meta + + def get_meta_data(self): + return dict(self._meta) + + def close(self): + return None + + def drop(self): + self._collection.drop_index(self._index_name) + + +class MilvusCollection(ICollection): + """A single OpenViking collection stored in Milvus.""" + + INTERNAL_PATH_FIELDS = { + "parent_uri": "path", + "scope_roots": "string", + "uri_depth": "int64", + } + + def __init__( + self, + *, + client: Any, + logical_collection_name: str, + physical_collection_name: str, + project_name: str, + dense_vector_name: str, + sparse_vector_name: str, + distance_metric: str, + timeout_seconds: int, + meta: Optional[Dict[str, Any]] = None, + ) -> None: + super().__init__() + self._client = client + self._logical_collection_name = logical_collection_name + self._physical_collection_name = physical_collection_name + self._project_name = project_name + self._dense_vector_name = dense_vector_name + self._sparse_vector_name = sparse_vector_name + self._distance_metric = _normalize_distance(distance_metric) + self._timeout_seconds = int(timeout_seconds) + self._meta = dict(meta or {}) + self._field_types = self._build_field_type_map(self._meta) + self._varchar_lengths = self._build_varchar_length_map() + self._vector_dim = self._extract_vector_dim(self._meta) + + @property + def collection_name(self) -> str: + return self._physical_collection_name + + @staticmethod + def _extract_vector_dim(meta: Dict[str, Any]) -> int: + for field in meta.get("Fields", []) or []: + if str(field.get("FieldType") or "").lower() in _VECTOR_FIELD_TYPES: + try: + return int(field.get("Dim") or 0) + except (TypeError, ValueError): + return 0 + return 0 + + @classmethod + def _build_field_type_map(cls, meta: Dict[str, Any]) -> Dict[str, str]: + mapping: Dict[str, str] = {} + for field in meta.get("Fields", []) or []: + name = field.get("FieldName") + field_type = field.get("FieldType") + if name and field_type: + mapping[str(name)] = str(field_type).lower() + mapping.setdefault("id", "string") + mapping.setdefault("vector", "vector") + mapping.setdefault("sparse_vector", "json") + mapping.update(cls.INTERNAL_PATH_FIELDS) + return mapping + + def _build_varchar_length_map(self) -> Dict[str, int]: + lengths: Dict[str, int] = { + "id": _ID_MAX_LENGTH, + "uri": _URI_MAX_LENGTH, + "parent_uri": _URI_MAX_LENGTH, + "scope_roots": _MILVUS_VARCHAR_MAX_LENGTH, + } + for field_name, field_type in self._field_types.items(): + if field_type in _STRING_FIELD_TYPES: + lengths.setdefault(field_name, _MILVUS_VARCHAR_MAX_LENGTH) + return lengths + + def collection_exists(self) -> bool: + return bool( + self._client.has_collection( + collection_name=self._physical_collection_name, + timeout=self._timeout_seconds, + ) + ) + + def _collection_properties(self) -> Dict[str, Any]: + try: + desc = self._client.describe_collection( + collection_name=self._physical_collection_name, + timeout=self._timeout_seconds, + ) + except Exception: + return {} + props = desc.get("properties") if isinstance(desc, dict) else None + return dict(props or {}) + + def _ensure_meta_collection(self) -> None: + try: + if self._client.has_collection( + collection_name=_META_COLLECTION_NAME, + timeout=self._timeout_seconds, + ): + return + pymilvus = _import_pymilvus() + DataType = pymilvus.DataType + schema = self._client.create_schema(auto_id=False, enable_dynamic_field=False) + schema.add_field( + field_name="id", + datatype=DataType.VARCHAR, + is_primary=True, + max_length=_MILVUS_MAX_COLLECTION_NAME_LENGTH, + ) + schema.add_field( + field_name="meta_json", + datatype=DataType.VARCHAR, + max_length=_MILVUS_VARCHAR_MAX_LENGTH, + ) + schema.add_field(field_name="indexes_json", datatype=DataType.JSON, nullable=True) + schema.add_field(field_name=_META_VECTOR_FIELD, datatype=DataType.FLOAT_VECTOR, dim=1) + self._client.create_collection( + collection_name=_META_COLLECTION_NAME, + schema=schema, + timeout=self._timeout_seconds, + ) + except Exception as exc: + logger.warning("Failed to ensure Milvus metadata collection: %s", exc) + + def _load_meta_record(self) -> Dict[str, Any]: + self._ensure_meta_collection() + try: + rows = self._client.get( + collection_name=_META_COLLECTION_NAME, + ids=[self._physical_collection_name], + output_fields=["meta_json", "indexes_json"], + timeout=self._timeout_seconds, + ) + except Exception: + return {} + return dict(rows[0]) if rows else {} + + def _save_meta_record(self, *, meta: Optional[Dict[str, Any]] = None) -> None: + self._ensure_meta_collection() + existing = self._load_meta_record() + meta_json = _json_dumps(meta if meta is not None else self._meta) + indexes_json = existing.get("indexes_json") if existing else {} + try: + self._client.upsert( + collection_name=_META_COLLECTION_NAME, + data=[ + { + "id": self._physical_collection_name, + "meta_json": meta_json, + "indexes_json": indexes_json if isinstance(indexes_json, dict) else {}, + _META_VECTOR_FIELD: [0.0], + } + ], + timeout=self._timeout_seconds, + ) + except Exception as exc: + logger.warning("Failed to persist Milvus metadata record: %s", exc) + + def load_remote_meta(self) -> Optional[Dict[str, Any]]: + record = self._load_meta_record() + raw_meta = record.get("meta_json") + if isinstance(raw_meta, str): + try: + meta = json.loads(raw_meta) + except (TypeError, ValueError): + meta = None + if isinstance(meta, dict): + self._meta = meta + self._field_types = self._build_field_type_map(meta) + self._varchar_lengths = self._build_varchar_length_map() + self._vector_dim = self._extract_vector_dim(meta) + return meta + + props = self._collection_properties() + raw_meta = props.get(_META_PROPERTY_KEY) + if isinstance(raw_meta, str): + try: + meta = json.loads(raw_meta) + except (TypeError, ValueError): + meta = None + if isinstance(meta, dict): + self._meta = meta + self._field_types = self._build_field_type_map(meta) + self._varchar_lengths = self._build_varchar_length_map() + self._vector_dim = self._extract_vector_dim(meta) + return meta + + return None + + def create_remote_collection( + self, + meta_data: Dict[str, Any], + *, + consistency_level: Optional[str] = None, + ) -> None: + self._meta = dict(meta_data) + self._field_types = self._build_field_type_map(self._meta) + self._varchar_lengths = self._build_varchar_length_map() + self._vector_dim = self._extract_vector_dim(self._meta) + if self._vector_dim <= 0: + raise ValueError("Milvus collection requires a positive dense vector dimension") + + pymilvus = _import_pymilvus() + schema = self._build_schema(pymilvus) + create_kwargs: Dict[str, Any] = {} + if consistency_level: + create_kwargs["consistency_level"] = consistency_level + self._client.create_collection( + collection_name=self._physical_collection_name, + schema=schema, + timeout=self._timeout_seconds, + **create_kwargs, + ) + self._save_collection_meta() + + def _build_schema(self, pymilvus: Any): + DataType = pymilvus.DataType + schema = self._client.create_schema( + auto_id=False, + enable_dynamic_field=True, + description=self._meta.get("Description") or "", + ) + seen = set() + for field in self._iter_schema_fields(): + field_name = str(field["FieldName"]) + if field_name in seen: + continue + seen.add(field_name) + field_type = str(field.get("FieldType") or "").lower() + kwargs: Dict[str, Any] = {} + if field_name == "id": + kwargs.update(is_primary=True, max_length=_ID_MAX_LENGTH) + datatype = DataType.VARCHAR + elif field_type in _VECTOR_FIELD_TYPES: + dim = int(field.get("Dim") or self._vector_dim) + if dim <= 0: + raise ValueError("Milvus vector field requires Dim") + datatype = DataType.FLOAT_VECTOR + kwargs["dim"] = dim + elif field_name == self._sparse_vector_name or field_type in {"json", "sparse_vector"}: + datatype = DataType.JSON + kwargs["nullable"] = True + elif field_type in _LIST_STRING_FIELD_TYPES: + datatype = DataType.ARRAY + kwargs.update( + element_type=DataType.VARCHAR, + max_capacity=1024, + max_length=1024, + nullable=True, + ) + elif field_type in _INT_FIELD_TYPES: + datatype = DataType.INT64 + kwargs["nullable"] = True + elif field_type in _FLOAT_FIELD_TYPES: + datatype = DataType.DOUBLE + kwargs["nullable"] = True + elif field_type in _BOOL_FIELD_TYPES: + datatype = DataType.BOOL + kwargs["nullable"] = True + else: + datatype = DataType.VARCHAR + kwargs.update( + max_length=self._varchar_lengths.get(field_name, _MILVUS_VARCHAR_MAX_LENGTH), + nullable=True, + ) + schema.add_field(field_name=field_name, datatype=datatype, **kwargs) + return schema + + def _iter_schema_fields(self) -> List[Dict[str, Any]]: + fields = [dict(field) for field in self._meta.get("Fields", []) or []] + names = {field.get("FieldName") for field in fields} + for field_name, field_type in self.INTERNAL_PATH_FIELDS.items(): + if field_name not in names: + fields.append({"FieldName": field_name, "FieldType": field_type}) + return fields + + def _save_collection_meta(self) -> None: + self._save_meta_record(meta=self._meta) + try: + self._client.alter_collection_properties( + collection_name=self._physical_collection_name, + properties={_META_PROPERTY_KEY: _json_dumps(self._meta)}, + timeout=self._timeout_seconds, + ) + except Exception as exc: + logger.debug("Milvus collection properties are not available: %s", exc) + + def update(self, fields: Optional[Dict[str, Any]] = None, description: Optional[str] = None): + if fields: + self._meta.update(fields) + if description is not None: + self._meta["Description"] = description + self._save_collection_meta() + return dict(self._meta) + + def get_meta_data(self): + if not self._meta: + self.load_remote_meta() + return dict(self._meta) + + def close(self): + return None + + def drop(self): + if self.collection_exists(): + self._client.drop_collection( + collection_name=self._physical_collection_name, + timeout=self._timeout_seconds, + ) + try: + self._client.delete( + collection_name=_META_COLLECTION_NAME, + ids=[self._physical_collection_name], + timeout=self._timeout_seconds, + ) + except Exception: + pass + + def create_index(self, index_name: str, meta_data: Dict[str, Any]) -> IIndex: + meta = dict(meta_data or {}) + vector_meta = dict(meta.get("VectorIndex") or {}) + metric_type = _milvus_metric(vector_meta.get("Distance") or self._distance_metric) + existing_indexes = set(self.list_indexes() or []) + if index_name in existing_indexes or self._dense_vector_name in existing_indexes: + meta["VectorIndex"] = { + **vector_meta, + "IndexType": "AUTOINDEX", + "Distance": self._distance_metric, + } + self._save_index_meta(index_name, meta) + return MilvusIndex(self, index_name, meta) + + index_params = self._client.prepare_index_params() + index_params.add_index( + field_name=self._dense_vector_name, + index_name=index_name, + index_type="AUTOINDEX", + metric_type=metric_type, + ) + try: + self._client.create_index( + collection_name=self._physical_collection_name, + index_params=index_params, + timeout=self._timeout_seconds, + ) + except Exception as exc: + if "index already" not in str(exc).lower(): + raise + try: + self._client.load_collection( + collection_name=self._physical_collection_name, + timeout=self._timeout_seconds, + ) + except Exception as exc: + logger.debug("Milvus collection load skipped or failed: %s", exc) + meta["VectorIndex"] = { + **vector_meta, + "IndexType": "AUTOINDEX", + "Distance": self._distance_metric, + } + self._save_index_meta(index_name, meta) + return MilvusIndex(self, index_name, meta) + + def _save_index_meta(self, index_name: str, meta: Dict[str, Any]) -> None: + self._ensure_meta_collection() + record = self._load_meta_record() + indexes = record.get("indexes_json") if record else {} + if not isinstance(indexes, dict): + indexes = {} + indexes[index_name] = meta + try: + self._client.upsert( + collection_name=_META_COLLECTION_NAME, + data=[ + { + "id": self._physical_collection_name, + "meta_json": record.get("meta_json") or _json_dumps(self._meta), + "indexes_json": indexes, + _META_VECTOR_FIELD: [0.0], + } + ], + timeout=self._timeout_seconds, + ) + except Exception as exc: + logger.warning("Failed to persist Milvus index metadata: %s", exc) + try: + self._client.alter_collection_properties( + collection_name=self._physical_collection_name, + properties={f"{_INDEX_META_PROPERTY_PREFIX}{index_name}": _json_dumps(meta)}, + timeout=self._timeout_seconds, + ) + except Exception as exc: + logger.debug("Milvus index properties are not available: %s", exc) + + def has_index(self, index_name: str) -> bool: + return self.get_index_meta_data(index_name) is not None or index_name in ( + self.list_indexes() or [] + ) + + def get_index(self, index_name: str) -> Optional[IIndex]: + meta = self.get_index_meta_data(index_name) + return MilvusIndex(self, index_name, meta) if meta else None + + def update_index( + self, + index_name: str, + scalar_index: Optional[Dict[str, Any]] = None, + description: Optional[str] = None, + ): + meta = self.get_index_meta_data(index_name) or {"IndexName": index_name} + if scalar_index is not None: + meta["ScalarIndex"] = ( + list(scalar_index.keys()) if isinstance(scalar_index, dict) else list(scalar_index) + ) + if description is not None: + meta["Description"] = description + self._save_index_meta(index_name, meta) + return meta + + def get_index_meta_data(self, index_name: str): + record = self._load_meta_record() + indexes = record.get("indexes_json") if record else {} + if isinstance(indexes, dict) and isinstance(indexes.get(index_name), dict): + return indexes[index_name] + + props = self._collection_properties() + raw_meta = props.get(f"{_INDEX_META_PROPERTY_PREFIX}{index_name}") + if not isinstance(raw_meta, str): + return None + try: + meta = json.loads(raw_meta) + except (TypeError, ValueError): + return None + return meta if isinstance(meta, dict) else None + + def list_indexes(self): + try: + return list( + self._client.list_indexes( + collection_name=self._physical_collection_name, + timeout=self._timeout_seconds, + ) + or [] + ) + except Exception: + return [] + + def drop_index(self, index_name: str): + try: + self._client.release_collection( + collection_name=self._physical_collection_name, + timeout=self._timeout_seconds, + ) + except Exception: + pass + for remote_name in list(self.list_indexes() or []): + if remote_name == index_name or str(remote_name).startswith(f"{index_name}_"): + try: + self._client.drop_index( + collection_name=self._physical_collection_name, + index_name=remote_name, + timeout=self._timeout_seconds, + ) + except Exception as exc: + logger.warning("Failed to drop Milvus index %s: %s", remote_name, exc) + try: + self._client.drop_collection_properties( + collection_name=self._physical_collection_name, + property_keys=[f"{_INDEX_META_PROPERTY_PREFIX}{index_name}"], + timeout=self._timeout_seconds, + ) + except Exception: + pass + try: + record = self._load_meta_record() + indexes = record.get("indexes_json") if record else {} + if isinstance(indexes, dict) and index_name in indexes: + indexes.pop(index_name, None) + self._client.upsert( + collection_name=_META_COLLECTION_NAME, + data=[ + { + "id": self._physical_collection_name, + "meta_json": record.get("meta_json") or _json_dumps(self._meta), + "indexes_json": indexes, + _META_VECTOR_FIELD: [0.0], + } + ], + timeout=self._timeout_seconds, + ) + except Exception: + pass + + def _prepare_record_for_write(self, record: Dict[str, Any]) -> Dict[str, Any]: + prepared: Dict[str, Any] = {} + for field_name, value in record.items(): + if value is None: + continue + field_type = self._field_types.get(field_name, "") + if field_name == "id": + text = str(value) + if len(text.encode("utf-8")) > _ID_MAX_LENGTH: + raise ValueError("Milvus record id exceeds 512 bytes") + prepared[field_name] = text + elif field_name == self._dense_vector_name: + prepared[field_name] = self._coerce_dense_vector(value) + elif field_name == "scope_roots": + prepared[field_name] = _encode_scope_roots(value) + elif field_name == self._sparse_vector_name or field_type == "sparse_vector": + prepared[field_name] = self._coerce_sparse_vector(value) + elif field_type in _LIST_STRING_FIELD_TYPES: + prepared[field_name] = [str(item) for item in (value or []) if item is not None] + elif field_type in _INT_FIELD_TYPES: + prepared[field_name] = int(value) + elif field_type in _FLOAT_FIELD_TYPES: + number = float(value) + prepared[field_name] = number if math.isfinite(number) else 0.0 + elif field_type in _BOOL_FIELD_TYPES: + prepared[field_name] = bool(value) + elif field_type == "date_time": + prepared[field_name] = str(_coerce_datetime_value(value)) + elif isinstance(value, str): + limit = self._varchar_lengths.get(field_name) + prepared[field_name] = _truncate_utf8(value, limit) if limit else value + else: + prepared[field_name] = value + return prepared + + def _coerce_dense_vector(self, value: Any) -> List[float]: + if not isinstance(value, Sequence) or isinstance(value, (str, bytes)): + raise ValueError("Milvus dense vector must be a sequence of floats") + vector = [] + for item in value: + number = float(item) + vector.append(number if math.isfinite(number) else 0.0) + if self._vector_dim > 0 and len(vector) != self._vector_dim: + raise ValueError( + f"Milvus dense vector dimension mismatch: expected {self._vector_dim}, " + f"got {len(vector)}" + ) + return vector + + @staticmethod + def _coerce_sparse_vector(value: Any) -> Dict[str, float]: + if value in (None, ""): + return {} + if isinstance(value, str): + decoded = _json_loads(value) + value = decoded if isinstance(decoded, dict) else {} + if not isinstance(value, dict): + return {} + result: Dict[str, float] = {} + for key, raw_value in value.items(): + try: + number = float(raw_value) + except (TypeError, ValueError): + continue + if math.isfinite(number): + result[str(key)] = number + return result + + def _record_from_entity(self, entity: Dict[str, Any]) -> tuple[Any, Dict[str, Any]]: + record = dict(entity or {}) + record_id = record.pop("id", None) + if record_id is None: + record_id = entity.get("pk") or entity.get("primary_key") + return record_id, self._decode_record(record) + + def _decode_record(self, record: Dict[str, Any]) -> Dict[str, Any]: + decoded = dict(record) + sparse = decoded.get(self._sparse_vector_name) + if isinstance(sparse, str): + parsed = _json_loads(sparse) + decoded[self._sparse_vector_name] = parsed if isinstance(parsed, dict) else sparse + return decoded + + def _select_output_fields( + self, + output_fields: Optional[List[str]], + *, + include_vector: bool = False, + include_sparse: bool = False, + ) -> List[str]: + if output_fields: + fields = [field for field in output_fields if field != "id"] + else: + fields = [field for field in self._field_types if field != "id"] + if not include_vector: + fields = [field for field in fields if field != self._dense_vector_name] + if not include_sparse: + fields = [field for field in fields if field != self._sparse_vector_name] + return list(dict.fromkeys(fields)) + + def search_by_vector( + self, + index_name: str, + dense_vector: Optional[List[float]] = None, + limit: int = 10, + offset: int = 0, + filters: Optional[str] = None, + sparse_vector: Optional[Dict[str, float]] = None, + output_fields: Optional[List[str]] = None, + ) -> SearchResult: + del index_name + if limit <= 0: + return SearchResult() + if dense_vector is None: + return self._search_by_sparse(sparse_vector, limit, offset, filters, output_fields) + + fetch_limit = max(limit + offset, limit) + fields = self._select_output_fields( + output_fields, + include_vector=False, + include_sparse=bool(sparse_vector), + ) + raw_results = self._client.search( + collection_name=self._physical_collection_name, + data=[self._coerce_dense_vector(dense_vector)], + anns_field=self._dense_vector_name, + filter=filters or "", + limit=fetch_limit, + output_fields=fields, + search_params={"metric_type": _milvus_metric(self._distance_metric)}, + timeout=self._timeout_seconds, + ) + hits = raw_results[0] if raw_results else [] + items: List[SearchItemResult] = [] + for hit in hits: + entity = hit.get("entity") if isinstance(hit, dict) else None + entity = dict(entity or {}) + if "id" not in entity and isinstance(hit, dict): + entity["id"] = hit.get("id") + record_id, payload = self._record_from_entity(entity) + score = _score_from_hit(hit, self._distance_metric) if isinstance(hit, dict) else 0.0 + if sparse_vector: + sparse_payload = payload.pop(self._sparse_vector_name, None) + score += _sparse_dot( + sparse_vector, sparse_payload if isinstance(sparse_payload, dict) else None + ) + items.append(SearchItemResult(id=record_id, fields=payload, score=score)) + if sparse_vector: + items.sort(key=lambda item: item.score or 0.0, reverse=True) + return SearchResult(data=items[offset : offset + limit]) + + def _search_by_sparse( + self, + sparse_vector: Optional[Dict[str, float]], + limit: int, + offset: int, + filters: Optional[str], + output_fields: Optional[List[str]], + ) -> SearchResult: + if not sparse_vector: + return SearchResult() + fields = self._select_output_fields(output_fields, include_sparse=True) + rows = self._client.query( + collection_name=self._physical_collection_name, + filter=filters or "", + output_fields=fields, + limit=max(limit + offset, _DEFAULT_QUERY_LIMIT), + timeout=self._timeout_seconds, + ) + items = [] + for row in rows: + record_id, payload = self._record_from_entity(row) + sparse_payload = payload.pop(self._sparse_vector_name, None) + score = _sparse_dot( + sparse_vector, sparse_payload if isinstance(sparse_payload, dict) else None + ) + if score > 0: + items.append(SearchItemResult(id=record_id, fields=payload, score=score)) + items.sort(key=lambda item: item.score or 0.0, reverse=True) + return SearchResult(data=items[offset : offset + limit]) + + def search_by_keywords( + self, + index_name: str, + keywords: Optional[List[str]] = None, + query: Optional[str] = None, + limit: int = 10, + offset: int = 0, + filters: Optional[str] = None, + output_fields: Optional[List[str]] = None, + ) -> SearchResult: + del index_name + query_text = query or " ".join(keywords or []) + if not query_text.strip(): + return SearchResult() + compiler = MilvusFilterCompiler(self._field_types) + text_filter = compiler.compile_legacy_filter( + { + "op": "or", + "conds": [ + {"op": "contains", "field": field, "substring": query_text} + for field in ("name", "description", "abstract", "tags", "content") + if field in self._field_types + ], + } + ) + combined = ( + f"({filters}) and ({text_filter})" + if filters and text_filter + else filters or text_filter + ) + return self.search_by_random("", limit, offset, combined, output_fields) + + def search_by_id( + self, + index_name: str, + id: Any, + limit: int = 10, + offset: int = 0, + filters: Optional[str] = None, + output_fields: Optional[List[str]] = None, + ) -> SearchResult: + rows = self._client.get( + collection_name=self._physical_collection_name, + ids=[str(id)], + output_fields=self._select_output_fields( + None, + include_vector=True, + include_sparse=True, + ), + timeout=self._timeout_seconds, + ) + if not rows: + return SearchResult() + dense_vector = rows[0].get(self._dense_vector_name) + sparse_vector = rows[0].get(self._sparse_vector_name) + result = self.search_by_vector( + index_name=index_name, + dense_vector=dense_vector, + sparse_vector=sparse_vector if isinstance(sparse_vector, dict) else None, + limit=limit + offset + 1, + offset=0, + filters=filters, + output_fields=output_fields, + ) + data = [item for item in result.data if str(item.id) != str(id)] + return SearchResult(data=data[offset : offset + limit]) + + def search_by_multimodal( + self, + index_name: str, + text: Optional[str], + image: Optional[Any], + video: Optional[Any], + limit: int = 10, + offset: int = 0, + filters: Optional[str] = None, + output_fields: Optional[List[str]] = None, + ) -> SearchResult: + raise NotImplementedError("MilvusCollection.search_by_multimodal is not supported") + + def search_by_random( + self, + index_name: str, + limit: int = 10, + offset: int = 0, + filters: Optional[str] = None, + output_fields: Optional[List[str]] = None, + ) -> SearchResult: + del index_name + rows = self._client.query( + collection_name=self._physical_collection_name, + filter=filters or "", + output_fields=self._select_output_fields(output_fields), + limit=limit, + offset=offset, + timeout=self._timeout_seconds, + ) + items = [] + for row in rows: + record_id, payload = self._record_from_entity(row) + items.append(SearchItemResult(id=record_id, fields=payload, score=1.0)) + return SearchResult(data=items) + + def search_by_scalar( + self, + index_name: str, + field: str, + order: Optional[str] = "desc", + limit: int = 10, + offset: int = 0, + filters: Optional[str] = None, + output_fields: Optional[List[str]] = None, + ) -> SearchResult: + del index_name + fields = self._select_output_fields(output_fields) + if field not in fields: + fields.append(field) + rows = self._client.query( + collection_name=self._physical_collection_name, + filter=filters or "", + output_fields=fields, + limit=max(limit + offset, _DEFAULT_QUERY_LIMIT), + timeout=self._timeout_seconds, + ) + reverse = (order or "desc").lower() == "desc" + rows.sort(key=lambda row: (row.get(field) is None, row.get(field)), reverse=reverse) + items = [] + for row in rows[offset : offset + limit]: + record_id, payload = self._record_from_entity(row) + score = ( + payload.pop(field, None) + if output_fields and field not in output_fields + else payload.get(field) + ) + items.append( + SearchItemResult( + id=record_id, + fields=payload, + score=score if isinstance(score, (int, float)) else None, + ) + ) + return SearchResult(data=items) + + def upsert_data(self, data_list: List[Dict[str, Any]], ttl=0): + del ttl + if not data_list: + return [] + records = [self._prepare_record_for_write(record) for record in data_list] + self._client.upsert( + collection_name=self._physical_collection_name, + data=records, + timeout=self._timeout_seconds, + ) + return [record.get("id") for record in records if record.get("id") is not None] + + def update_data(self, data_list: List[Dict[str, Any]]): + updated_records: List[Dict[str, Any]] = [] + updated_ids: List[Any] = [] + for raw_data in data_list: + if "id" not in raw_data or raw_data.get("id") in (None, ""): + raise ValueError("Milvus update requires id") + record_id = str(raw_data["id"]) + existing = self.fetch_data([record_id]).items + if not existing: + raise ValueError(f"Milvus entity does not exist for update: {record_id}") + merged = dict(existing[0].fields or {}) + merged["id"] = existing[0].id + merged.update(raw_data) + updated_records.append(self._prepare_record_for_write(merged)) + updated_ids.append(record_id) + if updated_records: + self._client.upsert( + collection_name=self._physical_collection_name, + data=updated_records, + timeout=self._timeout_seconds, + ) + return updated_ids + + def fetch_data(self, primary_keys: List[Any]): + if not primary_keys: + return FetchDataInCollectionResult() + rows = self._client.get( + collection_name=self._physical_collection_name, + ids=[str(pk) for pk in primary_keys], + output_fields=self._select_output_fields( + None, + include_vector=True, + include_sparse=True, + ), + timeout=self._timeout_seconds, + ) + items = [] + found_ids = set() + for row in rows: + record_id, payload = self._record_from_entity(row) + if record_id is not None: + found_ids.add(str(record_id)) + items.append(DataItem(id=record_id, fields=payload)) + return FetchDataInCollectionResult( + items=items, + ids_not_exist=[pk for pk in primary_keys if str(pk) not in found_ids], + ) + + def delete_data(self, primary_keys: List[Any]): + if not primary_keys: + return None + self._client.delete( + collection_name=self._physical_collection_name, + ids=[str(pk) for pk in primary_keys], + timeout=self._timeout_seconds, + ) + return None + + def delete_all_data(self): + self._client.delete( + collection_name=self._physical_collection_name, + filter='id != ""', + timeout=self._timeout_seconds, + ) + + def aggregate_data( + self, + index_name: str, + op: str = "count", + field: Optional[str] = None, + filters: Optional[str] = None, + cond: Optional[Dict[str, Any]] = None, + ) -> AggregateResult: + del index_name + if op != "count": + return AggregateResult(agg={}, op=op, field=field) + if not field: + try: + rows = self._client.query( + collection_name=self._physical_collection_name, + filter=filters or "", + output_fields=["count(*)"], + timeout=self._timeout_seconds, + ) + total = int((rows[0] if rows else {}).get("count(*)", 0)) + except Exception: + rows = self._client.query( + collection_name=self._physical_collection_name, + filter=filters or "", + output_fields=["id"], + limit=_DEFAULT_QUERY_LIMIT, + timeout=self._timeout_seconds, + ) + total = len(rows) + return AggregateResult(agg={"_total": total}, op=op, field=None) + + rows = self._client.query( + collection_name=self._physical_collection_name, + filter=filters or "", + output_fields=[field], + limit=_DEFAULT_QUERY_LIMIT, + timeout=self._timeout_seconds, + ) + grouped: Dict[Any, int] = {} + for row in rows: + value = row.get(field) + if value is not None: + grouped[value] = grouped.get(value, 0) + 1 + if cond: + grouped = { + key: value + for key, value in grouped.items() + if (cond.get("gt") is None or value > cond["gt"]) + and (cond.get("gte") is None or value >= cond["gte"]) + and (cond.get("lt") is None or value < cond["lt"]) + and (cond.get("lte") is None or value <= cond["lte"]) + } + return AggregateResult(agg=grouped, op=op, field=field) + + +class MilvusFilterCompiler: + """Compile OpenViking filters to safe Milvus boolean expressions.""" + + def __init__(self, field_types: Optional[Dict[str, str]] = None) -> None: + self._field_types = field_types or {} + + def compile(self, expr: FilterExpr | Dict[str, Any] | str | None) -> str: + if expr is None: + return "" + if isinstance(expr, str): + return expr.strip() + if isinstance(expr, dict): + if "op" in expr: + return self.compile_legacy_filter(expr) + return self._compile_mapping(expr) + if isinstance(expr, RawDSL): + payload = expr.payload + if isinstance(payload, dict) and "expr" in payload: + return str(payload["expr"]).strip() + return self.compile(payload) + if isinstance(expr, And): + return self._join("and", [self.compile(cond) for cond in expr.conds if cond]) + if isinstance(expr, Or): + return self._join("or", [self.compile(cond) for cond in expr.conds if cond]) + if isinstance(expr, Eq): + return self._eq(expr.field, expr.value) + if isinstance(expr, In): + return self._in(expr.field, list(expr.values)) + if isinstance(expr, Range): + return self._range( + expr.field, + gte=expr.gte, + gt=expr.gt, + lte=expr.lte, + lt=expr.lt, + ) + if isinstance(expr, TimeRange): + return self._range( + expr.field, + gte=_coerce_datetime_value(expr.start), + lt=_coerce_datetime_value(expr.end), + ) + if isinstance(expr, Contains): + return self._contains(expr.field, expr.substring) + if isinstance(expr, PathScope): + path = MilvusCollectionAdapter._normalize_path( + CollectionAdapter._encode_uri_field_value(expr.path) + if expr.field in CollectionAdapter._URI_FIELD_NAMES + else expr.path + ) + if expr.depth == 0: + return self._eq(expr.field, path) + if expr.depth == 1: + return self._eq("parent_uri", path) + if expr.depth == -1: + return self._contains("scope_roots", f"\n{path}\n") + raise ValueError( + f"Milvus adapter only supports PathScope depth 0/1/-1, got {expr.depth}" + ) + raise TypeError(f"Unsupported filter expr type: {type(expr)!r}") + + def compile_legacy_filter(self, payload: Dict[str, Any]) -> str: + op = str(payload.get("op") or "").lower() + if not op: + return self._compile_mapping(payload) + if op in {"and", "or"}: + return self._join( + op, + [self.compile_legacy_filter(cond) for cond in payload.get("conds", []) if cond], + ) + if op == "must": + field = payload.get("field") + values = payload.get("conds", []) or [] + if not values: + return "" + if field in CollectionAdapter._URI_FIELD_NAMES: + values = [ + MilvusCollectionAdapter._normalize_path( + CollectionAdapter._encode_uri_field_value(value) + ) + for value in values + ] + return ( + self._in(str(field), list(values)) + if len(values) > 1 + else self._eq(field, values[0]) + ) + if op == "must_not": + field = payload.get("field") + values = payload.get("conds", []) or [] + if not values: + return "" + expr = ( + self._in(str(field), list(values)) + if len(values) > 1 + else self._eq(field, values[0]) + ) + return f"not ({expr})" if expr else "" + if op in {"range", "time_range"}: + return self._range( + str(payload.get("field")), + gte=payload.get("gte"), + gt=payload.get("gt"), + lte=payload.get("lte"), + lt=payload.get("lt"), + ) + if op == "range_out": + field = str(payload.get("field")) + branches = [] + if payload.get("gte") is not None: + branches.append(self._range(field, lt=payload["gte"])) + if payload.get("lte") is not None: + branches.append(self._range(field, gt=payload["lte"])) + return self._join("or", branches) + if op == "contains": + return self._contains(str(payload.get("field")), str(payload.get("substring", ""))) + if op == "prefix": + field = str(payload.get("field")) + prefix = str(payload.get("prefix", "")) + if field in CollectionAdapter._URI_FIELD_NAMES: + return self.compile(PathScope(field, prefix, depth=-1)) + return self._like(field, f"{prefix}%") + return self._compile_mapping(payload) + + def _compile_mapping(self, payload: Dict[str, Any]) -> str: + return self._join("and", [self._eq(str(key), value) for key, value in payload.items()]) + + @staticmethod + def _join(op: str, exprs: Iterable[str]) -> str: + items = [expr for expr in exprs if expr] + if not items: + return "" + if len(items) == 1: + return items[0] + return f" {op} ".join(f"({item})" for item in items) + + def _validate_field(self, field: Any) -> str: + if not isinstance(field, str) or not _FIELD_NAME_RE.match(field): + raise ValueError(f"Invalid Milvus filter field: {field!r}") + return field + + def _eq(self, field: Any, value: Any) -> str: + field_name = self._validate_field(field) + field_type = self._field_types.get(field_name, "") + if value is None: + return f"{field_name} is null" + if field_type in _LIST_STRING_FIELD_TYPES: + return f"ARRAY_CONTAINS({field_name}, {_quote_value(value)})" + return f"{field_name} == {_quote_value(value)}" + + def _in(self, field: str, values: List[Any]) -> str: + field_name = self._validate_field(field) + if not values: + return "" + field_type = self._field_types.get(field_name, "") + if field_type in _LIST_STRING_FIELD_TYPES: + return self._join( + "or", + [f"ARRAY_CONTAINS({field_name}, {_quote_value(value)})" for value in values], + ) + non_null = [value for value in values if value is not None] + expr = f"{field_name} in {_format_value_list(non_null)}" if non_null else "" + if any(value is None for value in values): + null_expr = f"{field_name} is null" + return self._join("or", [expr, null_expr]) + return expr + + def _range(self, field: str, **bounds: Any) -> str: + field_name = self._validate_field(field) + parts = [] + operators = {"gte": ">=", "gt": ">", "lte": "<=", "lt": "<"} + for key, operator in operators.items(): + value = bounds.get(key) + if value is not None: + parts.append(f"{field_name} {operator} {_quote_value(value)}") + return " and ".join(parts) + + def _contains(self, field: str, substring: str) -> str: + field_name = self._validate_field(field) + field_type = self._field_types.get(field_name, "") + if field_type in _LIST_STRING_FIELD_TYPES: + return f"ARRAY_CONTAINS({field_name}, {_quote_value(substring)})" + return self._like(field_name, f"%{substring}%") + + def _like(self, field: str, pattern: str) -> str: + field_name = self._validate_field(field) + return f"{field_name} like {_quote_value(pattern)}" + + +class MilvusCollectionAdapter(CollectionAdapter): + """CollectionAdapter backed by Milvus or Zilliz Cloud.""" + + mode = "milvus" + INTERNAL_PATH_FIELDS = ["parent_uri", "scope_roots", "uri_depth"] + + def __init__( + self, + *, + uri: str, + token: Optional[str], + db_name: Optional[str], + consistency_level: Optional[str], + timeout_seconds: int, + project_name: str, + collection_name: str, + index_name: str, + distance_metric: str, + dense_vector_name: str, + sparse_vector_name: str, + ) -> None: + super().__init__(collection_name=collection_name, index_name=index_name) + self._uri = uri + self._token = token + self._db_name = db_name + self._consistency_level = consistency_level + self._timeout_seconds = int(timeout_seconds) + self._project_name = project_name + self._distance_metric = _normalize_distance(distance_metric) + self._dense_vector_name = dense_vector_name + self._sparse_vector_name = sparse_vector_name + self._client = None + + @classmethod + def from_config(cls, config: Any): + cfg = getattr(config, "milvus", None) + params = dict(getattr(config, "custom_params", {}) or {}) + uri = getattr(cfg, "uri", None) or getattr(config, "url", None) or params.get("uri") + token = getattr(cfg, "token", None) or params.get("token") + db_name = getattr(cfg, "db_name", None) or params.get("db_name") + consistency_level = getattr(cfg, "consistency_level", None) or params.get( + "consistency_level" + ) + return cls( + uri=str(uri or _DEFAULT_URI), + token=str(token) if token else None, + db_name=str(db_name) if db_name else None, + consistency_level=str(consistency_level) if consistency_level else None, + timeout_seconds=int( + getattr(cfg, "timeout_seconds", None) + or params.get("timeout_seconds") + or _DEFAULT_TIMEOUT_SECONDS + ), + project_name=config.project_name or "default", + collection_name=config.name or "context", + index_name=config.index_name or "default", + distance_metric=config.distance_metric or "cosine", + dense_vector_name=str( + getattr(cfg, "dense_vector_name", None) + or params.get("dense_vector_name") + or "vector" + ), + sparse_vector_name=str( + getattr(cfg, "sparse_vector_name", None) + or params.get("sparse_vector_name") + or "sparse_vector" + ), + ) + + @property + def physical_collection_name(self) -> str: + return _safe_collection_name(self._project_name, self._collection_name) + + def _connect(self): + if self._client is not None: + return self._client + pymilvus = _import_pymilvus() + kwargs: Dict[str, Any] = { + "uri": self._uri, + "timeout": self._timeout_seconds, + } + if self._token: + kwargs["token"] = self._token + if self._db_name: + kwargs["db_name"] = self._db_name + self._client = pymilvus.MilvusClient(**kwargs) + return self._client + + def _new_collection(self, meta: Optional[Dict[str, Any]] = None) -> MilvusCollection: + return MilvusCollection( + client=self._connect(), + logical_collection_name=self._collection_name, + physical_collection_name=self.physical_collection_name, + project_name=self._project_name, + dense_vector_name=self._dense_vector_name, + sparse_vector_name=self._sparse_vector_name, + distance_metric=self._distance_metric, + timeout_seconds=self._timeout_seconds, + meta=meta, + ) + + def _load_existing_collection_if_needed(self) -> None: + if self._collection is not None: + return + raw_collection = self._new_collection() + if not raw_collection.collection_exists(): + return + meta = raw_collection.load_remote_meta() + if not meta: + raise RuntimeError( + "Milvus collection exists but OpenViking metadata is missing: " + f"{self.physical_collection_name}. Use a different project/name, restore metadata, " + "or drop the stale Milvus collection." + ) + self._collection = Collection(raw_collection) + + def _create_backend_collection(self, meta: Dict[str, Any]) -> Collection: + raw_collection = self._new_collection(meta) + raw_collection.create_remote_collection(meta, consistency_level=self._consistency_level) + return Collection(raw_collection) + + def close(self) -> None: + super().close() + if self._client is not None: + close = getattr(self._client, "close", None) + if callable(close): + close() + self._client = None + + def _sanitize_scalar_index_fields( + self, + scalar_index_fields: list[str], + fields_meta: list[dict[str, Any]], + ) -> list[str]: + del fields_meta + return list(dict.fromkeys(list(scalar_index_fields) + self.INTERNAL_PATH_FIELDS)) + + def _build_default_index_meta( + self, + *, + index_name: str, + distance: str, + use_sparse: bool, + sparse_weight: float, + scalar_index_fields: list[str], + ) -> Dict[str, Any]: + if use_sparse: + logger.warning( + "Milvus adapter stores sparse vectors but currently searches dense vectors first; " + "sparse scores are only applied to returned dense candidates." + ) + return { + "IndexName": index_name, + "VectorIndex": { + "IndexType": "AUTOINDEX", + "Distance": _normalize_distance(distance), + "Quant": "int8", + "EnableSparse": bool(use_sparse), + "SearchWithSparseLogitAlpha": sparse_weight, + }, + "ScalarIndex": scalar_index_fields, + } + + @staticmethod + def _normalize_path(path: str) -> str: + stripped = (path or "").strip() + if not stripped: + return "/" + if not stripped.startswith("/"): + stripped = f"/{stripped}" + if len(stripped) > 1: + stripped = stripped.rstrip("/") + return stripped or "/" + + @classmethod + def _compute_parent_uri(cls, uri: str) -> str: + normalized = cls._normalize_path(uri) + if normalized == "/": + return "/" + parts = normalized.strip("/").split("/") + if len(parts) <= 1: + return "/" + return "/" + "/".join(parts[:-1]) + + @classmethod + def _compute_scope_roots(cls, uri: str) -> List[str]: + normalized = cls._normalize_path(uri) + if normalized == "/": + return ["/"] + parts = normalized.strip("/").split("/") + roots = ["/"] + current_parts: List[str] = [] + for part in parts[:-1]: + current_parts.append(part) + roots.append("/" + "/".join(current_parts)) + return roots + + def _normalize_record_for_write(self, record: Dict[str, Any]) -> Dict[str, Any]: + normalized = dict(super()._normalize_record_for_write(record)) + raw_uri = normalized.get("uri") + if isinstance(raw_uri, str): + normalized_uri = self._normalize_path(raw_uri) + normalized["uri"] = normalized_uri + normalized["parent_uri"] = self._compute_parent_uri(normalized_uri) + normalized["scope_roots"] = self._compute_scope_roots(normalized_uri) + normalized["uri_depth"] = len( + [part for part in normalized_uri.strip("/").split("/") if part] + ) + return normalized + + def _normalize_record_for_read(self, record: Dict[str, Any]) -> Dict[str, Any]: + normalized = super()._normalize_record_for_read(record) + for field_name in self.INTERNAL_PATH_FIELDS: + normalized.pop(field_name, None) + return normalized + + def _field_types_for_filter(self) -> Dict[str, str]: + try: + collection = self.get_collection() + meta = collection.get_meta_data() or {} + field_types = MilvusCollection._build_field_type_map(meta) + except Exception: + field_types = { + "id": "string", + self._dense_vector_name: "vector", + self._sparse_vector_name: "sparse_vector", + "uri": "path", + "parent_uri": "path", + "scope_roots": "string", + "uri_depth": "int64", + "level": "int64", + "active_count": "int64", + "search_tags": "list", + } + return field_types + + def _compile_filter(self, expr: FilterExpr | Dict[str, Any] | str | None) -> str: + return MilvusFilterCompiler(self._field_types_for_filter()).compile(expr) + + def update_data(self, data_list: List[Dict[str, Any]]): + collection = self.get_collection() + normalized = [self._normalize_record_for_write(item) for item in data_list] + result = collection.update_data(normalized) + return [str(item) for item in (result or []) if item is not None] diff --git a/openviking_cli/utils/config/vectordb_config.py b/openviking_cli/utils/config/vectordb_config.py index 36d97e4417..cee25b12b0 100644 --- a/openviking_cli/utils/config/vectordb_config.py +++ b/openviking_cli/utils/config/vectordb_config.py @@ -151,6 +151,59 @@ class CuVSConfig(BaseModel): model_config = {"extra": "forbid"} +class MilvusConfig(BaseModel): + """Configuration for Milvus backend.""" + + uri: str = Field( + default="./milvus.db", + description=( + "Milvus URI. Use a local .db path for Milvus Lite, " + "'http://localhost:19530' for self-hosted Milvus, or a Zilliz Cloud endpoint." + ), + ) + token: Optional[str] = Field( + default=None, + description="Optional token for authenticated Milvus or Zilliz Cloud deployments.", + ) + db_name: Optional[str] = Field( + default=None, + description="Optional Milvus database name for server or cloud deployments.", + ) + consistency_level: Optional[str] = Field( + default=None, + description="Optional Milvus consistency level: Strong, Session, Bounded, or Eventually.", + ) + timeout_seconds: int = Field(default=30, description="Milvus client timeout in seconds") + dense_vector_name: str = Field(default="vector", description="Dense vector field name") + sparse_vector_name: str = Field( + default="sparse_vector", description="Sparse vector JSON field name" + ) + + model_config = {"extra": "forbid"} + + @model_validator(mode="after") + def validate_milvus(self): + self.uri = (self.uri or "./milvus.db").strip() + if not self.uri: + raise ValueError("Milvus uri must not be empty") + if self.uri.startswith(("http://", "https://")): + self.uri = self.uri.rstrip("/") + if self.consistency_level: + allowed = {"Strong", "Session", "Bounded", "Eventually"} + normalized = self.consistency_level.strip() + title_value = normalized[:1].upper() + normalized[1:].lower() + if title_value not in allowed: + raise ValueError( + "Milvus consistency_level must be one of: " + f"{sorted(allowed)}; got {self.consistency_level!r}" + ) + self.consistency_level = title_value + self.dense_vector_name = (self.dense_vector_name or "vector").strip() + self.sparse_vector_name = (self.sparse_vector_name or "sparse_vector").strip() + if self.timeout_seconds <= 0: + raise ValueError("Milvus timeout_seconds must be positive") + return self + _OPENGAUSS_MODES = {"standalone", "distributed"} @@ -214,7 +267,7 @@ class VectorDBBackendConfig(BaseModel): description=( "VectorDB backend type: 'local', 'cuvs', 'http', " "'volcengine' (AK/SK signed or API key data-plane only), " - "'vikingdb' (private deployment), 'qdrant', or 'opengauss'" + "'vikingdb' (private deployment), 'qdrant', 'milvus', or 'opengauss'" ), ) @@ -278,6 +331,11 @@ class VectorDBBackendConfig(BaseModel): description="NVIDIA cuVS dense-vector search configuration for the 'cuvs' backend", ) + milvus: Optional[MilvusConfig] = Field( + default_factory=MilvusConfig, + description="Milvus configuration for 'milvus' type", + ) + opengauss: Optional[OpenGaussConfig] = Field( default_factory=OpenGaussConfig, description="openGauss configuration for 'opengauss' type", @@ -300,6 +358,7 @@ def validate_config(self): "volcengine", "vikingdb", "qdrant", + "milvus", "opengauss", ] @@ -364,6 +423,21 @@ def validate_config(self): if self.url: self.url = self.url.strip().rstrip("/") + elif self.backend == "milvus": + milvus_uri = ( + (self.milvus.uri if self.milvus else None) + or self.url + or self.custom_params.get("uri") + or "./milvus.db" + ) + if self.milvus is None: + self.milvus = MilvusConfig() + self.milvus.uri = str(milvus_uri).strip() + if self.milvus.uri.startswith(("http://", "https://")): + self.milvus.uri = self.milvus.uri.rstrip("/") + if self.url: + self.url = self.url.strip().rstrip("/") + elif self.backend == "opengauss": if self.opengauss is None: self.opengauss = OpenGaussConfig() diff --git a/pyproject.toml b/pyproject.toml index 6d932adecb..0258978d25 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -106,6 +106,9 @@ test = [ opengauss = [ "psycopg2-binary>=2.9", ] +milvus = [ + "pymilvus[milvus_lite]>=2.6.0", +] dev = [ "mypy>=1.0.0", "ruff>=0.1.0", diff --git a/tests/storage/test_milvus_adapter.py b/tests/storage/test_milvus_adapter.py new file mode 100644 index 0000000000..237265c6aa --- /dev/null +++ b/tests/storage/test_milvus_adapter.py @@ -0,0 +1,367 @@ +# Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd. +# SPDX-License-Identifier: AGPL-3.0 + +from __future__ import annotations + +import importlib.util +import uuid +from datetime import datetime, timezone + +import pytest + +from openviking.storage.expr import And, Contains, Eq, PathScope, TimeRange +from openviking.storage.vectordb_adapters.factory import create_collection_adapter +from openviking.storage.vectordb_adapters.milvus_adapter import ( + MilvusCollection, + MilvusCollectionAdapter, + MilvusFilterCompiler, + _encode_scope_roots, + _normalize_distance, + _safe_collection_name, +) +from openviking_cli.utils.config.vectordb_config import VectorDBBackendConfig + + +def _build_config() -> VectorDBBackendConfig: + return VectorDBBackendConfig.model_validate( + { + "backend": "milvus", + "project": "default", + "name": "context", + "index_name": "default", + "distance_metric": "cosine", + "milvus": { + "uri": "./milvus.db", + "token": "test-token", + "db_name": "default", + "consistency_level": "session", + "timeout_seconds": 7, + "dense_vector_name": "vector", + "sparse_vector_name": "sparse_vector", + }, + } + ) + + +def _schema() -> dict: + return { + "CollectionName": "context", + "Description": "test collection", + "Fields": [ + {"FieldName": "id", "FieldType": "string", "IsPrimaryKey": True}, + {"FieldName": "uri", "FieldType": "path"}, + {"FieldName": "vector", "FieldType": "vector", "Dim": 2}, + {"FieldName": "sparse_vector", "FieldType": "sparse_vector"}, + {"FieldName": "abstract", "FieldType": "string"}, + {"FieldName": "level", "FieldType": "int64"}, + {"FieldName": "updated_at", "FieldType": "date_time"}, + {"FieldName": "search_tags", "FieldType": "list"}, + {"FieldName": "account_id", "FieldType": "string"}, + ], + "ScalarIndex": ["uri", "level", "updated_at", "search_tags", "account_id"], + } + + +def test_milvus_backend_config_validation(): + config = _build_config() + + assert config.backend == "milvus" + assert config.milvus is not None + assert config.milvus.uri == "./milvus.db" + assert config.milvus.token == "test-token" + assert config.milvus.db_name == "default" + assert config.milvus.consistency_level == "Session" + + +def test_factory_creates_milvus_adapter_without_connecting(): + adapter = create_collection_adapter(_build_config()) + + assert isinstance(adapter, MilvusCollectionAdapter) + assert adapter.mode == "milvus" + assert adapter.collection_name == "context" + assert adapter.index_name == "default" + assert adapter.physical_collection_name == "ov_default_context" + + +def test_augments_path_fields_on_write_and_hides_them_on_read(): + adapter = MilvusCollectionAdapter.from_config(_build_config()) + source_record = { + "id": "1", + "uri": "viking://resources/acme/docs/a.md", + "vector": [0.1, 0.2], + } + + normalized = adapter._normalize_record_for_write(source_record) + + assert normalized["uri"] == "/resources/acme/docs/a.md" + assert normalized["parent_uri"] == "/resources/acme/docs" + assert normalized["scope_roots"] == [ + "/", + "/resources", + "/resources/acme", + "/resources/acme/docs", + ] + assert normalized["uri_depth"] == 4 + assert source_record["uri"] == "viking://resources/acme/docs/a.md" + + public_record = adapter.normalize_record_for_read(normalized) + assert public_record["uri"] == "viking://resources/acme/docs/a.md" + assert "parent_uri" not in public_record + assert "scope_roots" not in public_record + assert "uri_depth" not in public_record + + +def test_compiles_filter_exprs(): + compiler = MilvusFilterCompiler( + { + "account_id": "string", + "scope_roots": "string", + "updated_at": "date_time", + "abstract": "string", + } + ) + + compiled = compiler.compile( + And( + [ + Eq("account_id", "acme"), + PathScope("uri", "viking://resources/acme/docs", depth=-1), + TimeRange( + "updated_at", + start=datetime(2026, 5, 1, tzinfo=timezone.utc), + end=datetime(2026, 6, 1, tzinfo=timezone.utc), + ), + Contains("abstract", "quarterly report"), + ] + ) + ) + + assert compiled == ( + '(account_id == "acme") and ' + '(scope_roots like "%\\n/resources/acme/docs\\n%") and ' + '(updated_at >= "2026-05-01T00:00:00+00:00" and ' + 'updated_at < "2026-06-01T00:00:00+00:00") and ' + '(abstract like "%quarterly report%")' + ) + + +def test_compiles_legacy_dict_filters(): + compiler = MilvusFilterCompiler( + { + "account_id": "string", + "updated_at": "date_time", + "scope_roots": "string", + } + ) + + compiled = compiler.compile( + { + "op": "and", + "conds": [ + {"op": "must", "field": "account_id", "conds": ["acme"]}, + { + "op": "time_range", + "field": "updated_at", + "gte": "2026-05-01T00:00:00Z", + "lt": "2026-06-01T00:00:00Z", + }, + {"op": "prefix", "field": "uri", "prefix": "viking://resources/acme/docs"}, + ], + } + ) + + assert compiled == ( + '(account_id == "acme") and ' + '(updated_at >= "2026-05-01T00:00:00Z" and ' + 'updated_at < "2026-06-01T00:00:00Z") and ' + '(scope_roots like "%\\n/resources/acme/docs\\n%")' + ) + + +def test_vector_literal_and_collection_name_safety(): + name = _safe_collection_name("Project/With Space", "Context.Table") + + assert name.startswith("ov_Project_With_Space_Context_Table") + assert len(name) <= 255 + assert _normalize_distance("ip") == "ip" + + with pytest.raises(ValueError, match="supports only cosine, l2, and ip"): + _normalize_distance("dot") + + +def test_scope_roots_encoding_is_token_safe(): + encoded = _encode_scope_roots(["/a", "/a/b"]) + + assert encoded == "\n/a\n/a/b\n" + assert "\n/a\n" in encoded + assert "\n/a/b\n" in encoded + assert "\n/a/c\n" not in encoded + + +class _FakeSchema: + def __init__(self) -> None: + self.fields = [] + + def add_field(self, **kwargs): + self.fields.append(kwargs) + + +class _FakeIndexParams: + def __init__(self) -> None: + self.indexes = [] + + def add_index(self, **kwargs): + self.indexes.append(kwargs) + + +class _FakeMilvusClient: + def __init__(self) -> None: + self.schema = _FakeSchema() + self.index_params = _FakeIndexParams() + self.created_collection = None + self.created_index = None + self.properties = {} + self.loaded = False + + def create_schema(self, **kwargs): + self.schema_kwargs = kwargs + return self.schema + + def create_collection(self, **kwargs): + self.created_collection = kwargs + + def alter_collection_properties(self, collection_name, properties, timeout=None): + self.properties.update(properties) + + def prepare_index_params(self): + return self.index_params + + def create_index(self, **kwargs): + self.created_index = kwargs + + def load_collection(self, collection_name, timeout=None): + self.loaded = True + + +def test_collection_creation_uses_explicit_schema_and_autoindex(): + client = _FakeMilvusClient() + collection = MilvusCollection( + client=client, + logical_collection_name="context", + physical_collection_name="ov_default_context", + project_name="default", + dense_vector_name="vector", + sparse_vector_name="sparse_vector", + distance_metric="cosine", + timeout_seconds=7, + meta=_schema(), + ) + + collection.create_remote_collection(_schema(), consistency_level="Session") + collection.create_index( + "default", + { + "IndexName": "default", + "VectorIndex": {"IndexType": "AUTOINDEX", "Distance": "cosine"}, + "ScalarIndex": ["uri", "level", "parent_uri", "scope_roots"], + }, + ) + + assert client.schema_kwargs["auto_id"] is False + assert client.schema_kwargs["enable_dynamic_field"] is True + field_names = {field["field_name"] for field in client.schema.fields} + assert {"id", "uri", "vector", "sparse_vector", "parent_uri", "scope_roots"} <= field_names + vector_field = next(field for field in client.schema.fields if field["field_name"] == "vector") + assert vector_field["dim"] == 2 + assert client.created_collection["collection_name"] == "ov_default_context" + assert client.created_collection["consistency_level"] == "Session" + assert client.index_params.indexes[0] == { + "field_name": "vector", + "index_name": "default", + "index_type": "AUTOINDEX", + "metric_type": "COSINE", + } + assert client.loaded is True + + +@pytest.mark.skipif( + importlib.util.find_spec("milvus_lite") is None, + reason="milvus_lite is not installed", +) +def test_milvus_lite_adapter_integration_smoke(tmp_path): + pytest.importorskip("pymilvus") + + suffix = uuid.uuid4().hex[:8] + uri = str(tmp_path / "milvus.db") + project_name = f"pytest_{suffix}" + + def _new_adapter() -> MilvusCollectionAdapter: + return MilvusCollectionAdapter( + uri=uri, + token=None, + db_name=None, + consistency_level="Strong", + timeout_seconds=30, + project_name=project_name, + collection_name="context", + index_name="default", + distance_metric="cosine", + dense_vector_name="vector", + sparse_vector_name="sparse_vector", + ) + + adapter = _new_adapter() + + try: + assert adapter.create_collection( + "context", + _schema(), + distance="cosine", + sparse_weight=0.0, + index_name="default", + ) + adapter.upsert( + [ + { + "id": "doc-1", + "uri": "viking://resources/acme/docs/a.md", + "vector": [1.0, 0.0], + "sparse_vector": {"quarter": 1.0}, + "abstract": "quarterly report", + "level": 1, + "updated_at": "2026-05-15T00:00:00+00:00", + "search_tags": ["finance"], + "account_id": "acme", + }, + { + "id": "doc-2", + "uri": "viking://resources/acme/notes/b.md", + "vector": [0.0, 1.0], + "sparse_vector": {"notes": 1.0}, + "abstract": "meeting notes", + "level": 2, + "updated_at": "2026-05-16T00:00:00+00:00", + "search_tags": ["notes"], + "account_id": "acme", + }, + ] + ) + + adapter.close() + adapter = _new_adapter() + assert adapter.collection_exists() is True + + result = adapter.query( + query_vector=[1.0, 0.0], + limit=1, + filter=PathScope("uri", "viking://resources/acme/docs", depth=-1), + output_fields=["id", "uri", "abstract", "level"], + ) + assert [item["id"] for item in result] == ["doc-1"] + assert adapter.count(Eq("account_id", "missing")) == 0 + assert adapter.count() == 2 + assert adapter.delete(ids=["doc-2"]) == 1 + assert adapter.count() == 1 + finally: + adapter.drop_collection() + adapter.close() diff --git a/uv.lock b/uv.lock index 0be3eae181..e8ecbbd7a8 100644 --- a/uv.lock +++ b/uv.lock @@ -526,6 +526,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/c5/0d/84a4380f930db0010168e0aa7b7a8fed9ba1835a8fbb1472bc6d0201d529/build-1.4.0-py3-none-any.whl", hash = "sha256:6a07c1b8eb6f2b311b96fcbdbce5dab5fe637ffda0fd83c9cac622e927501596", size = 24141, upload-time = "2026-01-08T16:41:46.453Z" }, ] +[[package]] +name = "cachetools" +version = "7.1.4" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/f4/8b/0d3945a13955303b81272f759a0331e54c5c793da455e6f5706b89d2639c/cachetools-7.1.4.tar.gz", hash = "sha256:437f55a4e0c1b01a4f3077cc470e6991d47430970e36fbcb77e2be0df4fc1cd6", size = 40085, upload-time = "2026-05-21T22:40:43.376Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/8c/7b/1fc1c09cc0756cf25861a3be10565915953876da48bb228fb9a672b20a42/cachetools-7.1.4-py3-none-any.whl", hash = "sha256:323dc4127934744db5b54eb4924482d7edafbf9554e820d1531c2e08c0e4ef54", size = 16761, upload-time = "2026-05-21T22:40:41.845Z" }, +] + [[package]] name = "certifi" version = "2026.2.25" @@ -1228,7 +1237,7 @@ name = "exceptiongroup" version = "1.3.1" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "typing-extensions", marker = "python_full_version < '3.11'" }, + { name = "typing-extensions" }, ] sdist = { url = "https://files.pythonhosted.org/packages/50/79/66800aadf48771f6b62f7eb014e352e5d06856655206165d775e675a02c9/exceptiongroup-1.3.1.tar.gz", hash = "sha256:8b412432c6055b0b7d14c310000ae93352ed6754f70fa8f7c34141f91c4e3219", size = 30371, upload-time = "2025-11-21T23:01:54.787Z" } wheels = [ @@ -1244,6 +1253,24 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/ab/84/02fc1827e8cdded4aa65baef11296a9bbe595c474f0d6d758af082d849fd/execnet-2.1.2-py3-none-any.whl", hash = "sha256:67fba928dd5a544b783f6056f449e5e3931a5c378b128bc18501f7ea79e296ec", size = 40708, upload-time = "2025-11-12T09:56:36.333Z" }, ] +[[package]] +name = "faiss-cpu" +version = "1.14.3" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, + { name = "numpy", version = "2.4.3", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, + { name = "packaging" }, +] +wheels = [ + { url = "https://files.pythonhosted.org/packages/83/b0/48c083d01b7b68c463c1d56507147a9d733f791e1c469a77215a872a9fb5/faiss_cpu-1.14.3-cp310-abi3-macosx_14_0_arm64.whl", hash = "sha256:a9369863290a3f0e033757e4c10577b6ef7431f1cede394dabd0a137e4e2ed45", size = 4768290, upload-time = "2026-06-13T02:19:03.427Z" }, + { url = "https://files.pythonhosted.org/packages/ab/34/6b04ef5bae3eada6b5a9457d7875cce041c040d53c890815cbd1e9821c65/faiss_cpu-1.14.3-cp310-abi3-macosx_15_0_x86_64.whl", hash = "sha256:f9d0e84d909194f63f027bbd3c1e35e905e48c9345c2db6e6f24da09a6bc5906", size = 6925734, upload-time = "2026-06-13T02:19:05.232Z" }, + { url = "https://files.pythonhosted.org/packages/7c/8a/b451af4b3c6dd18749ecfb58ccb68503b77e49c9aa4a89d950e9d521e058/faiss_cpu-1.14.3-cp310-abi3-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:1d734cfa9ac90b6a5dfed3a27cb706d05f22824703dafc3969b4e2071877a31c", size = 9661210, upload-time = "2026-06-13T02:19:06.99Z" }, + { url = "https://files.pythonhosted.org/packages/a0/ed/57335bc18c9e18677587bec9bf070c675b29c8e683e13f4def0440731ca0/faiss_cpu-1.14.3-cp310-abi3-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:8780b526c06e57aad90a8c4655dfba2ff1b3195bd600ff4499752fc45159c9fc", size = 18506292, upload-time = "2026-06-13T02:19:09.621Z" }, + { url = "https://files.pythonhosted.org/packages/e7/d3/c6ca8c44a63b909e78aa8a69e14501c79c89613e62261d58455ea603d710/faiss_cpu-1.14.3-cp310-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:b28ba083e8c02f2c9be03783402537fd3f00d27e68799c44bf88931500ee12ec", size = 11238800, upload-time = "2026-06-13T02:19:12.821Z" }, + { url = "https://files.pythonhosted.org/packages/93/5f/b405692913a301251749cb175cb3f564ed257fdaa80a22c9a36444d0095d/faiss_cpu-1.14.3-cp310-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:cdcb90850cb4b7c27d270839b37bcc0dc8d1fb6a62d1e13053e51ac95061ca25", size = 19237637, upload-time = "2026-06-13T02:19:15.531Z" }, +] + [[package]] name = "fastapi" version = "0.135.1" @@ -2192,13 +2219,13 @@ name = "langchain-classic" version = "1.0.2" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "langchain-core", marker = "python_full_version >= '3.11'" }, - { name = "langchain-text-splitters", marker = "python_full_version >= '3.11'" }, - { name = "langsmith", marker = "python_full_version >= '3.11'" }, - { name = "pydantic", marker = "python_full_version >= '3.11'" }, - { name = "pyyaml", marker = "python_full_version >= '3.11'" }, - { name = "requests", marker = "python_full_version >= '3.11'" }, - { name = "sqlalchemy", marker = "python_full_version >= '3.11'" }, + { name = "langchain-core" }, + { name = "langchain-text-splitters" }, + { name = "langsmith" }, + { name = "pydantic" }, + { name = "pyyaml" }, + { name = "requests" }, + { name = "sqlalchemy" }, ] sdist = { url = "https://files.pythonhosted.org/packages/21/6f/59da67274d8ceea16d0610142af33e348a24750894f08c0688de01504ff2/langchain_classic-1.0.2.tar.gz", hash = "sha256:bbf686613d0051905794f2646ecb6a79fa398db399750a4af039107d93054335", size = 10533928, upload-time = "2026-03-06T20:19:46.176Z" } wheels = [ @@ -2213,18 +2240,18 @@ resolution-markers = [ "python_full_version < '3.11'", ] dependencies = [ - { name = "aiohttp", marker = "python_full_version < '3.11'" }, - { name = "dataclasses-json", marker = "python_full_version < '3.11'" }, - { name = "httpx-sse", marker = "python_full_version < '3.11'" }, - { name = "langchain", marker = "python_full_version < '3.11'" }, - { name = "langchain-core", marker = "python_full_version < '3.11'" }, - { name = "langsmith", marker = "python_full_version < '3.11'" }, - { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, - { name = "pydantic-settings", marker = "python_full_version < '3.11'" }, - { name = "pyyaml", marker = "python_full_version < '3.11'" }, - { name = "requests", marker = "python_full_version < '3.11'" }, - { name = "sqlalchemy", marker = "python_full_version < '3.11'" }, - { name = "tenacity", marker = "python_full_version < '3.11'" }, + { name = "aiohttp" }, + { name = "dataclasses-json" }, + { name = "httpx-sse" }, + { name = "langchain" }, + { name = "langchain-core" }, + { name = "langsmith" }, + { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" } }, + { name = "pydantic-settings" }, + { name = "pyyaml" }, + { name = "requests" }, + { name = "sqlalchemy" }, + { name = "tenacity" }, ] sdist = { url = "https://files.pythonhosted.org/packages/83/49/2ff5354273809e9811392bc24bcffda545a196070666aef27bc6aacf1c21/langchain_community-0.3.31.tar.gz", hash = "sha256:250e4c1041539130f6d6ac6f9386cb018354eafccd917b01a4cff1950b80fd81", size = 33241237, upload-time = "2025-10-07T20:17:57.857Z" } wheels = [ @@ -2250,18 +2277,18 @@ resolution-markers = [ "python_full_version == '3.11.*' and sys_platform != 'emscripten' and sys_platform != 'win32'", ] dependencies = [ - { name = "aiohttp", marker = "python_full_version >= '3.11'" }, - { name = "dataclasses-json", marker = "python_full_version >= '3.11'" }, - { name = "httpx-sse", marker = "python_full_version >= '3.11'" }, - { name = "langchain-classic", marker = "python_full_version >= '3.11'" }, - { name = "langchain-core", marker = "python_full_version >= '3.11'" }, - { name = "langsmith", marker = "python_full_version >= '3.11'" }, - { name = "numpy", version = "2.4.3", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, - { name = "pydantic-settings", marker = "python_full_version >= '3.11'" }, - { name = "pyyaml", marker = "python_full_version >= '3.11'" }, - { name = "requests", marker = "python_full_version >= '3.11'" }, - { name = "sqlalchemy", marker = "python_full_version >= '3.11'" }, - { name = "tenacity", marker = "python_full_version >= '3.11'" }, + { name = "aiohttp" }, + { name = "dataclasses-json" }, + { name = "httpx-sse" }, + { name = "langchain-classic" }, + { name = "langchain-core" }, + { name = "langsmith" }, + { name = "numpy", version = "2.4.3", source = { registry = "https://pypi.org/simple" } }, + { name = "pydantic-settings" }, + { name = "pyyaml" }, + { name = "requests" }, + { name = "sqlalchemy" }, + { name = "tenacity" }, ] sdist = { url = "https://files.pythonhosted.org/packages/53/97/a03585d42b9bdb6fbd935282d6e3348b10322a24e6ce12d0c99eb461d9af/langchain_community-0.4.1.tar.gz", hash = "sha256:f3b211832728ee89f169ddce8579b80a085222ddb4f4ed445a46e977d17b1e85", size = 33241144, upload-time = "2025-10-27T15:20:32.504Z" } wheels = [ @@ -2306,7 +2333,7 @@ name = "langchain-text-splitters" version = "1.1.1" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "langchain-core", marker = "python_full_version >= '3.11'" }, + { name = "langchain-core" }, ] sdist = { url = "https://files.pythonhosted.org/packages/85/38/14121ead61e0e75f79c3a35e5148ac7c2fe754a55f76eab3eed573269524/langchain_text_splitters-1.1.1.tar.gz", hash = "sha256:34861abe7c07d9e49d4dc852d0129e26b32738b60a74486853ec9b6d6a8e01d2", size = 279352, upload-time = "2026-02-18T23:02:42.798Z" } wheels = [ @@ -2702,7 +2729,7 @@ resolution-markers = [ "python_full_version < '3.11'", ] dependencies = [ - { name = "mdurl", marker = "python_full_version < '3.11'" }, + { name = "mdurl" }, ] sdist = { url = "https://files.pythonhosted.org/packages/38/71/3b932df36c1a044d397a1f92d1cf91ee0a503d91e470cbd670aa66b07ed0/markdown-it-py-3.0.0.tar.gz", hash = "sha256:e3f60a94fa066dc52ec76661e37c851cb232d92f9886b15cb560aaada2df8feb", size = 74596, upload-time = "2023-06-03T06:41:14.443Z" } wheels = [ @@ -2728,7 +2755,7 @@ resolution-markers = [ "python_full_version == '3.11.*' and sys_platform != 'emscripten' and sys_platform != 'win32'", ] dependencies = [ - { name = "mdurl", marker = "python_full_version >= '3.11'" }, + { name = "mdurl" }, ] sdist = { url = "https://files.pythonhosted.org/packages/5b/f5/4ec618ed16cc4f8fb3b701563655a69816155e79e24a17b651541804721d/markdown_it_py-4.0.0.tar.gz", hash = "sha256:cb0a2b4aa34f932c007117b194e945bd74e0ec24133ceb5bac59009cda1cb9f3", size = 73070, upload-time = "2025-08-11T12:57:52.854Z" } wheels = [ @@ -2879,6 +2906,23 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/b3/38/89ba8ad64ae25be8de66a6d463314cf1eb366222074cfda9ee839c56a4b4/mdurl-0.1.2-py3-none-any.whl", hash = "sha256:84008a41e51615a49fc9966191ff91509e3c40b939176e643fd50a5c2196b8f8", size = 9979, upload-time = "2022-08-14T12:40:09.779Z" }, ] +[[package]] +name = "milvus-lite" +version = "3.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "faiss-cpu" }, + { name = "grpcio" }, + { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, + { name = "numpy", version = "2.4.3", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, + { name = "pyarrow" }, + { name = "tomli", marker = "python_full_version < '3.11'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/a3/9a/d80d260e6fe1246818a8ef782c374ba9c6ca46ca3b987c14eabe914ef805/milvus_lite-3.0.tar.gz", hash = "sha256:2c35d0d046b1faae3402cde1fb73d65f51ee8c6aba65f54de1dda46f7bb18b9b", size = 589749, upload-time = "2026-05-13T07:14:05.827Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/59/42/dee08ae3bfc1731572f193d00b248c9370b0b9dff12becb0ffd8b2ee8d56/milvus_lite-3.0-py3-none-any.whl", hash = "sha256:d9a094eab84bdaa4253da3721482282c939da1cce6f4e1759f947e8d3e53406e", size = 230490, upload-time = "2026-05-13T07:14:00.816Z" }, +] + [[package]] name = "msgpack" version = "1.1.2" @@ -3164,12 +3208,12 @@ resolution-markers = [ "python_full_version < '3.11'", ] dependencies = [ - { name = "docutils", version = "0.21.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, - { name = "jinja2", marker = "python_full_version < '3.11'" }, - { name = "markdown-it-py", version = "3.0.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, - { name = "mdit-py-plugins", marker = "python_full_version < '3.11'" }, - { name = "pyyaml", marker = "python_full_version < '3.11'" }, - { name = "sphinx", version = "8.1.3", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, + { name = "docutils", version = "0.21.2", source = { registry = "https://pypi.org/simple" } }, + { name = "jinja2" }, + { name = "markdown-it-py", version = "3.0.0", source = { registry = "https://pypi.org/simple" } }, + { name = "mdit-py-plugins" }, + { name = "pyyaml" }, + { name = "sphinx", version = "8.1.3", source = { registry = "https://pypi.org/simple" } }, ] sdist = { url = "https://files.pythonhosted.org/packages/66/a5/9626ba4f73555b3735ad86247a8077d4603aa8628537687c839ab08bfe44/myst_parser-4.0.1.tar.gz", hash = "sha256:5cfea715e4f3574138aecbf7d54132296bfd72bb614d31168f48c477a830a7c4", size = 93985, upload-time = "2025-02-12T10:53:03.833Z" } wheels = [ @@ -3195,12 +3239,12 @@ resolution-markers = [ "python_full_version == '3.11.*' and sys_platform != 'emscripten' and sys_platform != 'win32'", ] dependencies = [ - { name = "docutils", version = "0.22.4", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, - { name = "jinja2", marker = "python_full_version >= '3.11'" }, - { name = "markdown-it-py", version = "4.0.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, - { name = "mdit-py-plugins", marker = "python_full_version >= '3.11'" }, - { name = "pyyaml", marker = "python_full_version >= '3.11'" }, - { name = "sphinx", version = "9.0.4", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.11.*'" }, + { name = "docutils", version = "0.22.4", source = { registry = "https://pypi.org/simple" } }, + { name = "jinja2" }, + { name = "markdown-it-py", version = "4.0.0", source = { registry = "https://pypi.org/simple" } }, + { name = "mdit-py-plugins" }, + { name = "pyyaml" }, + { name = "sphinx", version = "9.0.4", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.12'" }, { name = "sphinx", version = "9.1.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.12'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/33/fa/7b45eef11b7971f0beb29d27b7bfe0d747d063aa29e170d9edd004733c8a/myst_parser-5.0.0.tar.gz", hash = "sha256:f6f231452c56e8baa662cc352c548158f6a16fcbd6e3800fc594978002b94f3a", size = 98535, upload-time = "2026-01-15T09:08:18.036Z" } @@ -3679,35 +3723,6 @@ benchmark = [ { name = "tiktoken" }, ] bot = [ - { name = "beautifulsoup4" }, - { name = "croniter" }, - { name = "ddgs" }, - { name = "gradio" }, - { name = "html2text" }, - { name = "httpx", extra = ["socks"] }, - { name = "mcp" }, - { name = "msgpack" }, - { name = "prompt-toolkit" }, - { name = "py-machineid" }, - { name = "pydantic-settings" }, - { name = "pygments" }, - { name = "python-engineio" }, - { name = "python-socketio" }, - { name = "python-socks", extra = ["asyncio"] }, - { name = "readability-lxml" }, - { name = "rich" }, - { name = "socksio" }, - { name = "tavily-python" }, - { name = "websocket-client" }, - { name = "websockets" }, -] -bot-dingtalk = [ - { name = "dingtalk-stream" }, -] -bot-feishu = [ - { name = "lark-oapi" }, -] -bot-full = [ { name = "agent-sandbox" }, { name = "beautifulsoup4" }, { name = "croniter" }, @@ -3741,29 +3756,6 @@ bot-full = [ { name = "websocket-client" }, { name = "websockets" }, ] -bot-fuse = [ - { name = "fusepy" }, -] -bot-langfuse = [ - { name = "langfuse" }, -] -bot-opencode = [ - { name = "opencode-ai" }, -] -bot-qq = [ - { name = "qq-botpy" }, -] -bot-sandbox = [ - { name = "agent-sandbox" }, - { name = "opensandbox" }, - { name = "opensandbox-server" }, -] -bot-slack = [ - { name = "slack-sdk" }, -] -bot-telegram = [ - { name = "python-telegram-bot", extra = ["socks"] }, -] build = [ { name = "build" }, { name = "cmake" }, @@ -3808,6 +3800,9 @@ langgraph = [ local-embed = [ { name = "llama-cpp-python" }, ] +milvus = [ + { name = "pymilvus", extra = ["milvus-lite"] }, +] ocr = [ { name = "pytesseract" }, ] @@ -3835,7 +3830,7 @@ dev = [ [package.metadata] requires-dist = [ - { name = "agent-sandbox", marker = "extra == 'bot-sandbox'", specifier = ">=0.0.23" }, + { name = "agent-sandbox", marker = "extra == 'bot'", specifier = ">=0.0.23" }, { name = "anyio", marker = "extra == 'gemini-async'", specifier = ">=4.0.0" }, { name = "apscheduler", specifier = ">=3.11.0" }, { name = "argon2-cffi", specifier = ">=23.0.0" }, @@ -3852,11 +3847,11 @@ requires-dist = [ { name = "ddgs", marker = "extra == 'bot'", specifier = ">=9.0.0" }, { name = "defusedxml", specifier = ">=0.7.1" }, { name = "diff-match-patch", marker = "extra == 'test'", specifier = ">=20200713" }, - { name = "dingtalk-stream", marker = "extra == 'bot-dingtalk'", specifier = ">=0.4.0" }, + { name = "dingtalk-stream", marker = "extra == 'bot'", specifier = ">=0.4.0" }, { name = "ebooklib", specifier = ">=0.18.0" }, { name = "fastapi", specifier = ">=0.128.0" }, { name = "feedparser", specifier = ">=6.0.0" }, - { name = "fusepy", marker = "extra == 'bot-fuse'", specifier = ">=3.0.1" }, + { name = "fusepy", marker = "extra == 'bot'", specifier = ">=3.0.1" }, { name = "google-genai", marker = "extra == 'gemini'", specifier = ">=1.0.0" }, { name = "google-genai", marker = "extra == 'gemini-async'", specifier = ">=1.0.0" }, { name = "gradio", marker = "extra == 'bot'", specifier = ">=6.6.0" }, @@ -3872,11 +3867,11 @@ requires-dist = [ { name = "langchain-core", marker = "extra == 'langchain'", specifier = ">=1.0.0,<2.0.0" }, { name = "langchain-core", marker = "extra == 'langgraph'", specifier = ">=1.0.0,<2.0.0" }, { name = "langchain-openai", marker = "extra == 'benchmark'", specifier = ">=1.0.0" }, - { name = "langfuse", marker = "extra == 'bot-langfuse'", specifier = ">=3.0.0" }, + { name = "langfuse", marker = "extra == 'bot'", specifier = ">=3.0.0" }, { name = "langgraph", marker = "extra == 'langgraph'", specifier = ">=1.0.0,<2.0.0" }, { name = "lark-oapi", specifier = ">=1.5.3" }, - { name = "lark-oapi", marker = "extra == 'bot-feishu'", specifier = ">=1.0.0" }, - { name = "litellm", specifier = ">=1.83.7,<1.89.3" }, + { name = "lark-oapi", marker = "extra == 'bot'", specifier = ">=1.0.0" }, + { name = "litellm", specifier = ">=1.83.7,<1.90.3" }, { name = "llama-cpp-python", marker = "extra == 'local-embed'", specifier = ">=0.3.0" }, { name = "loguru", specifier = ">=0.7.3" }, { name = "mcp", specifier = ">=1.27.0" }, @@ -3886,16 +3881,15 @@ requires-dist = [ { name = "myst-parser", marker = "extra == 'doc'", specifier = ">=2.0.0" }, { name = "olefile", specifier = ">=0.47" }, { name = "openai", specifier = ">=1.0.0" }, - { name = "opencode-ai", marker = "extra == 'bot-opencode'", specifier = ">=0.1.0a0" }, + { name = "opencode-ai", marker = "extra == 'bot'", specifier = ">=0.1.0a0" }, { name = "openpyxl", specifier = ">=3.0.0" }, - { name = "opensandbox", marker = "extra == 'bot-sandbox'", specifier = ">=0.1.0" }, - { name = "opensandbox-server", marker = "extra == 'bot-sandbox'", specifier = ">=0.1.0" }, + { name = "opensandbox", marker = "extra == 'bot'", specifier = ">=0.1.0" }, + { name = "opensandbox-server", marker = "extra == 'bot'", specifier = ">=0.1.0" }, { name = "opentelemetry-api", specifier = ">=1.14" }, { name = "opentelemetry-exporter-otlp-proto-grpc", specifier = ">=1.14" }, { name = "opentelemetry-exporter-otlp-proto-http", specifier = ">=1.14" }, { name = "opentelemetry-instrumentation-asyncio", specifier = ">=0.61b0" }, { name = "opentelemetry-sdk", specifier = ">=1.14" }, - { name = "openviking", extras = ["bot", "bot-dingtalk", "bot-feishu", "bot-fuse", "bot-langfuse", "bot-opencode", "bot-qq", "bot-sandbox", "bot-slack", "bot-telegram"], marker = "extra == 'bot-full'" }, { name = "openviking-sdk", specifier = ">=0.1.1" }, { name = "pandas", marker = "extra == 'benchmark'", specifier = ">=2.0.0" }, { name = "pandas", marker = "extra == 'eval'", specifier = ">=2.0.0" }, @@ -3910,6 +3904,7 @@ requires-dist = [ { name = "pydantic", specifier = ">=2.0.0" }, { name = "pydantic-settings", marker = "extra == 'bot'", specifier = ">=2.0.0" }, { name = "pygments", marker = "extra == 'bot'", specifier = ">=2.16.0" }, + { name = "pymilvus", extras = ["milvus-lite"], marker = "extra == 'milvus'", specifier = ">=2.6.0" }, { name = "pytesseract", marker = "extra == 'ocr'", specifier = ">=0.3.10" }, { name = "pytest", marker = "extra == 'test'", specifier = ">=7.0.0" }, { name = "pytest-asyncio", marker = "extra == 'test'", specifier = ">=0.21.0" }, @@ -3921,9 +3916,9 @@ requires-dist = [ { name = "python-pptx", specifier = ">=1.0.0" }, { name = "python-socketio", marker = "extra == 'bot'", specifier = ">=5.16.2" }, { name = "python-socks", extras = ["asyncio"], marker = "extra == 'bot'", specifier = ">=2.4.0" }, - { name = "python-telegram-bot", extras = ["socks"], marker = "extra == 'bot-telegram'", specifier = ">=21.0" }, + { name = "python-telegram-bot", extras = ["socks"], marker = "extra == 'bot'", specifier = ">=21.0" }, { name = "pyyaml", specifier = ">=6.0" }, - { name = "qq-botpy", marker = "extra == 'bot-qq'", specifier = ">=1.0.0" }, + { name = "qq-botpy", marker = "extra == 'bot'", specifier = ">=1.0.0" }, { name = "ragas", marker = "extra == 'eval'", specifier = ">=0.1.0" }, { name = "ragas", marker = "extra == 'test'", specifier = ">=0.1.0" }, { name = "readability-lxml", marker = "extra == 'bot'", specifier = ">=0.8.0" }, @@ -3934,7 +3929,7 @@ requires-dist = [ { name = "setuptools", marker = "extra == 'build'", specifier = ">=61.0" }, { name = "setuptools-scm", marker = "extra == 'build'", specifier = ">=8.0" }, { name = "setuptools-scm", marker = "extra == 'dev'", specifier = ">=10.0.0" }, - { name = "slack-sdk", marker = "extra == 'bot-slack'", specifier = ">=3.26.0" }, + { name = "slack-sdk", marker = "extra == 'bot'", specifier = ">=3.26.0" }, { name = "socksio", marker = "extra == 'bot'", specifier = ">=1.0.0" }, { name = "sphinx", marker = "extra == 'doc'", specifier = ">=7.0.0" }, { name = "sphinx-rtd-theme", marker = "extra == 'doc'", specifier = ">=1.3.0" }, @@ -3965,7 +3960,7 @@ requires-dist = [ { name = "xlrd", specifier = ">=2.0.1" }, { name = "xxhash", specifier = ">=3.0.0" }, ] -provides-extras = ["test", "opengauss", "dev", "doc", "eval", "gemini", "gemini-async", "ocr", "build", "bot", "bot-langfuse", "bot-telegram", "bot-feishu", "bot-dingtalk", "bot-slack", "bot-qq", "bot-sandbox", "bot-fuse", "bot-opencode", "bot-full", "benchmark", "langchain", "langgraph", "local-embed"] +provides-extras = ["test", "opengauss", "milvus", "dev", "doc", "eval", "gemini", "gemini-async", "ocr", "build", "bot", "benchmark", "langchain", "langgraph", "local-embed"] [package.metadata.requires-dev] dev = [{ name = "pytest", specifier = ">=9.0.2" }] @@ -4136,10 +4131,10 @@ resolution-markers = [ "python_full_version < '3.11'", ] dependencies = [ - { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, - { name = "python-dateutil", marker = "python_full_version < '3.11'" }, - { name = "pytz", marker = "python_full_version < '3.11'" }, - { name = "tzdata", marker = "python_full_version < '3.11'" }, + { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" } }, + { name = "python-dateutil" }, + { name = "pytz" }, + { name = "tzdata" }, ] sdist = { url = "https://files.pythonhosted.org/packages/33/01/d40b85317f86cf08d853a4f495195c73815fdf205eef3993821720274518/pandas-2.3.3.tar.gz", hash = "sha256:e05e1af93b977f7eafa636d043f9f94c7ee3ac81af99c13508215942e64c993b", size = 4495223, upload-time = "2025-09-29T23:34:51.853Z" } wheels = [ @@ -4211,9 +4206,9 @@ resolution-markers = [ "python_full_version == '3.11.*' and sys_platform != 'emscripten' and sys_platform != 'win32'", ] dependencies = [ - { name = "numpy", version = "2.4.3", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, - { name = "python-dateutil", marker = "python_full_version >= '3.11'" }, - { name = "tzdata", marker = "(python_full_version >= '3.11' and sys_platform == 'emscripten') or (python_full_version >= '3.11' and sys_platform == 'win32')" }, + { name = "numpy", version = "2.4.3", source = { registry = "https://pypi.org/simple" } }, + { name = "python-dateutil" }, + { name = "tzdata", marker = "sys_platform == 'emscripten' or sys_platform == 'win32'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/2e/0c/b28ed414f080ee0ad153f848586d61d1878f91689950f037f976ce15f6c8/pandas-3.0.1.tar.gz", hash = "sha256:4186a699674af418f655dbd420ed87f50d56b4cd6603784279d9eef6627823c8", size = 4641901, upload-time = "2026-02-17T22:20:16.434Z" } wheels = [ @@ -5010,6 +5005,30 @@ crypto = [ { name = "cryptography" }, ] +[[package]] +name = "pymilvus" +version = "3.0.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "cachetools" }, + { name = "grpcio" }, + { name = "orjson" }, + { name = "pandas", version = "2.3.3", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, + { name = "pandas", version = "3.0.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, + { name = "protobuf" }, + { name = "python-dotenv" }, + { name = "requests" }, + { name = "setuptools" }, +] +wheels = [ + { url = "https://files.pythonhosted.org/packages/ac/c1/01647e61f3a82fd881382746b6dde3401d65b88cd4f75bd059901fb2392b/pymilvus-3.0.0-1-py3-none-any.whl", hash = "sha256:57c8e7c87fbbf579f122b4df893949bc78e50bca2988527864891bd544817b05", size = 344817, upload-time = "2026-05-07T14:57:45.235Z" }, +] + +[package.optional-dependencies] +milvus-lite = [ + { name = "milvus-lite", marker = "sys_platform != 'win32'" }, +] + [[package]] name = "pyopenssl" version = "26.3.0" @@ -5971,23 +5990,23 @@ resolution-markers = [ "python_full_version < '3.11'", ] dependencies = [ - { name = "alabaster", marker = "python_full_version < '3.11'" }, - { name = "babel", marker = "python_full_version < '3.11'" }, - { name = "colorama", marker = "python_full_version < '3.11' and sys_platform == 'win32'" }, - { name = "docutils", version = "0.21.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, - { name = "imagesize", marker = "python_full_version < '3.11'" }, - { name = "jinja2", marker = "python_full_version < '3.11'" }, - { name = "packaging", marker = "python_full_version < '3.11'" }, - { name = "pygments", marker = "python_full_version < '3.11'" }, - { name = "requests", marker = "python_full_version < '3.11'" }, - { name = "snowballstemmer", marker = "python_full_version < '3.11'" }, - { name = "sphinxcontrib-applehelp", marker = "python_full_version < '3.11'" }, - { name = "sphinxcontrib-devhelp", marker = "python_full_version < '3.11'" }, - { name = "sphinxcontrib-htmlhelp", marker = "python_full_version < '3.11'" }, - { name = "sphinxcontrib-jsmath", marker = "python_full_version < '3.11'" }, - { name = "sphinxcontrib-qthelp", marker = "python_full_version < '3.11'" }, - { name = "sphinxcontrib-serializinghtml", marker = "python_full_version < '3.11'" }, - { name = "tomli", marker = "python_full_version < '3.11'" }, + { name = "alabaster" }, + { name = "babel" }, + { name = "colorama", marker = "sys_platform == 'win32'" }, + { name = "docutils", version = "0.21.2", source = { registry = "https://pypi.org/simple" } }, + { name = "imagesize" }, + { name = "jinja2" }, + { name = "packaging" }, + { name = "pygments" }, + { name = "requests" }, + { name = "snowballstemmer" }, + { name = "sphinxcontrib-applehelp" }, + { name = "sphinxcontrib-devhelp" }, + { name = "sphinxcontrib-htmlhelp" }, + { name = "sphinxcontrib-jsmath" }, + { name = "sphinxcontrib-qthelp" }, + { name = "sphinxcontrib-serializinghtml" }, + { name = "tomli" }, ] sdist = { url = "https://files.pythonhosted.org/packages/6f/6d/be0b61178fe2cdcb67e2a92fc9ebb488e3c51c4f74a36a7824c0adf23425/sphinx-8.1.3.tar.gz", hash = "sha256:43c1911eecb0d3e161ad78611bc905d1ad0e523e4ddc202a58a821773dc4c927", size = 8184611, upload-time = "2024-10-13T20:27:13.93Z" } wheels = [ @@ -6004,23 +6023,23 @@ resolution-markers = [ "python_full_version == '3.11.*' and sys_platform != 'emscripten' and sys_platform != 'win32'", ] dependencies = [ - { name = "alabaster", marker = "python_full_version == '3.11.*'" }, - { name = "babel", marker = "python_full_version == '3.11.*'" }, - { name = "colorama", marker = "python_full_version == '3.11.*' and sys_platform == 'win32'" }, - { name = "docutils", version = "0.22.4", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.11.*'" }, - { name = "imagesize", marker = "python_full_version == '3.11.*'" }, - { name = "jinja2", marker = "python_full_version == '3.11.*'" }, - { name = "packaging", marker = "python_full_version == '3.11.*'" }, - { name = "pygments", marker = "python_full_version == '3.11.*'" }, - { name = "requests", marker = "python_full_version == '3.11.*'" }, - { name = "roman-numerals", marker = "python_full_version == '3.11.*'" }, - { name = "snowballstemmer", marker = "python_full_version == '3.11.*'" }, - { name = "sphinxcontrib-applehelp", marker = "python_full_version == '3.11.*'" }, - { name = "sphinxcontrib-devhelp", marker = "python_full_version == '3.11.*'" }, - { name = "sphinxcontrib-htmlhelp", marker = "python_full_version == '3.11.*'" }, - { name = "sphinxcontrib-jsmath", marker = "python_full_version == '3.11.*'" }, - { name = "sphinxcontrib-qthelp", marker = "python_full_version == '3.11.*'" }, - { name = "sphinxcontrib-serializinghtml", marker = "python_full_version == '3.11.*'" }, + { name = "alabaster" }, + { name = "babel" }, + { name = "colorama", marker = "sys_platform == 'win32'" }, + { name = "docutils", version = "0.22.4", source = { registry = "https://pypi.org/simple" } }, + { name = "imagesize" }, + { name = "jinja2" }, + { name = "packaging" }, + { name = "pygments" }, + { name = "requests" }, + { name = "roman-numerals" }, + { name = "snowballstemmer" }, + { name = "sphinxcontrib-applehelp" }, + { name = "sphinxcontrib-devhelp" }, + { name = "sphinxcontrib-htmlhelp" }, + { name = "sphinxcontrib-jsmath" }, + { name = "sphinxcontrib-qthelp" }, + { name = "sphinxcontrib-serializinghtml" }, ] sdist = { url = "https://files.pythonhosted.org/packages/42/50/a8c6ccc36d5eacdfd7913ddccd15a9cee03ecafc5ee2bc40e1f168d85022/sphinx-9.0.4.tar.gz", hash = "sha256:594ef59d042972abbc581d8baa577404abe4e6c3b04ef61bd7fc2acbd51f3fa3", size = 8710502, upload-time = "2025-12-04T07:45:27.343Z" } wheels = [ @@ -6043,23 +6062,23 @@ resolution-markers = [ "python_full_version == '3.12.*' and sys_platform != 'emscripten' and sys_platform != 'win32'", ] dependencies = [ - { name = "alabaster", marker = "python_full_version >= '3.12'" }, - { name = "babel", marker = "python_full_version >= '3.12'" }, - { name = "colorama", marker = "python_full_version >= '3.12' and sys_platform == 'win32'" }, - { name = "docutils", version = "0.22.4", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.12'" }, - { name = "imagesize", marker = "python_full_version >= '3.12'" }, - { name = "jinja2", marker = "python_full_version >= '3.12'" }, - { name = "packaging", marker = "python_full_version >= '3.12'" }, - { name = "pygments", marker = "python_full_version >= '3.12'" }, - { name = "requests", marker = "python_full_version >= '3.12'" }, - { name = "roman-numerals", marker = "python_full_version >= '3.12'" }, - { name = "snowballstemmer", marker = "python_full_version >= '3.12'" }, - { name = "sphinxcontrib-applehelp", marker = "python_full_version >= '3.12'" }, - { name = "sphinxcontrib-devhelp", marker = "python_full_version >= '3.12'" }, - { name = "sphinxcontrib-htmlhelp", marker = "python_full_version >= '3.12'" }, - { name = "sphinxcontrib-jsmath", marker = "python_full_version >= '3.12'" }, - { name = "sphinxcontrib-qthelp", marker = "python_full_version >= '3.12'" }, - { name = "sphinxcontrib-serializinghtml", marker = "python_full_version >= '3.12'" }, + { name = "alabaster" }, + { name = "babel" }, + { name = "colorama", marker = "sys_platform == 'win32'" }, + { name = "docutils", version = "0.22.4", source = { registry = "https://pypi.org/simple" } }, + { name = "imagesize" }, + { name = "jinja2" }, + { name = "packaging" }, + { name = "pygments" }, + { name = "requests" }, + { name = "roman-numerals" }, + { name = "snowballstemmer" }, + { name = "sphinxcontrib-applehelp" }, + { name = "sphinxcontrib-devhelp" }, + { name = "sphinxcontrib-htmlhelp" }, + { name = "sphinxcontrib-jsmath" }, + { name = "sphinxcontrib-qthelp" }, + { name = "sphinxcontrib-serializinghtml" }, ] sdist = { url = "https://files.pythonhosted.org/packages/cd/bd/f08eb0f4eed5c83f1ba2a3bd18f7745a2b1525fad70660a1c00224ec468a/sphinx-9.1.0.tar.gz", hash = "sha256:7741722357dd75f8190766926071fed3bdc211c74dd2d7d4df5404da95930ddb", size = 8718324, upload-time = "2025-12-31T15:09:27.646Z" } wheels = [ From 4871c23d692e3e5bdaeb6603b3e7c8d83e3cbd1e Mon Sep 17 00:00:00 2001 From: Cheney Zhang Date: Wed, 8 Jul 2026 10:38:13 +0000 Subject: [PATCH 2/2] fix(storage): normalize Milvus cosine scores Signed-off-by: Cheney Zhang --- openviking/storage/vectordb_adapters/milvus_adapter.py | 2 ++ tests/storage/test_milvus_adapter.py | 8 ++++++++ 2 files changed, 10 insertions(+) diff --git a/openviking/storage/vectordb_adapters/milvus_adapter.py b/openviking/storage/vectordb_adapters/milvus_adapter.py index 6f7915382d..d20a99d766 100644 --- a/openviking/storage/vectordb_adapters/milvus_adapter.py +++ b/openviking/storage/vectordb_adapters/milvus_adapter.py @@ -205,6 +205,8 @@ def _score_from_hit(hit: Dict[str, Any], distance_metric: str) -> float: return 0.0 if not math.isfinite(score): return 0.0 + if distance_metric == "cosine": + return 1.0 - score if distance_metric == "l2": return 1.0 / (1.0 + max(score, 0.0)) return score diff --git a/tests/storage/test_milvus_adapter.py b/tests/storage/test_milvus_adapter.py index 237265c6aa..4f7462f6b2 100644 --- a/tests/storage/test_milvus_adapter.py +++ b/tests/storage/test_milvus_adapter.py @@ -198,6 +198,13 @@ def test_scope_roots_encoding_is_token_safe(): assert "\n/a/c\n" not in encoded +def test_score_from_cosine_distance_is_higher_is_better(): + from openviking.storage.vectordb_adapters.milvus_adapter import _score_from_hit + + assert _score_from_hit({"distance": 0.0}, "cosine") == pytest.approx(1.0) + assert _score_from_hit({"distance": 1.0}, "cosine") == pytest.approx(0.0) + + class _FakeSchema: def __init__(self) -> None: self.fields = [] @@ -358,6 +365,7 @@ def _new_adapter() -> MilvusCollectionAdapter: output_fields=["id", "uri", "abstract", "level"], ) assert [item["id"] for item in result] == ["doc-1"] + assert result[0]["_score"] == pytest.approx(1.0) assert adapter.count(Eq("account_id", "missing")) == 0 assert adapter.count() == 2 assert adapter.delete(ids=["doc-2"]) == 1