upstream-archive/qdrant
1
0
Fork 0
mirror of https://github.com/qdrant/qdrant.git synced 2026-08-06 18:10:58 -05:00
Code Issues Packages Projects Releases Wiki Activity
Files
94de0e2d2dcf997bf26bec34dad407d8de9a9abb
qdrant/lib/edge/python/qdrant_edge.pyi
qdrant-cloud-bot 7e15d343c2 Introduce EdgeShardConfig for edge shard (#8322) 
* Introduce EdgeShardConfig for edge shard

- Add EdgeShardConfig and EdgeOptimizersConfig in lib/edge/src/config.rs
  - Segment config (vector_data, sparse_vector_data, payload_storage_type)
  - Global hnsw_config and per-vector HNSW in segment config
  - Optimizer params: deleted_threshold, vacuum_min_vector_number,
    default_segment_number, max_segment_size, indexing_threshold,
    prevent_unoptimized (excludes memmap_threshold, flush_interval_sec,
    max_optimization_threads)
- Persist/load as edge_config.json in shard path
- EdgeShard uses RwLock<EdgeShardConfig>; load() accepts Option<EdgeShardConfig>,
  falls back to file or infer from segments; compatibility checked on load
- load_with_segment_config() for backward compatibility (SegmentConfig -> EdgeShardConfig)
- optimize() uses EdgeShardConfig for hnsw and optimizer thresholds
- Public methods: set_hnsw_config(), set_vector_hnsw_config(), set_optimizers_config()
  (update and persist)
- Python and examples use load_with_segment_config with existing config API

Made-with: Cursor

* Refactor EdgeShardConfig: user-facing params only, config module

- Replace SegmentConfig inside EdgeShardConfig with user-facing fields:
  - on_disk_payload (bool) instead of payload_storage_type
  - vectors: HashMap<VectorNameBuf, EdgeVectorParams> with on_disk per vector,
    no per-vector quantization; global quantization_config only
  - sparse_vectors: HashMap<VectorNameBuf, EdgeSparseVectorParams> with on_disk
- EdgeVectorParams / EdgeSparseVectorParams use on_disk (bool) instead of
  storage_type; conversion to VectorDataConfig/SparseVectorDataConfig in
  to_segment_config()
- Add config module: mod.rs, optimizers.rs, vectors.rs, shard.rs
- from_segment_config(&SegmentConfig) fills all inferrable params
- to_segment_config() builds SegmentConfig for segments and optimize()
- load_with_segment_config takes Option<SegmentConfig>, uses from_segment_config

Made-with: Cursor

* Move optimizer threshold helpers to shard crate

- Add get_number_segments, get_indexing_threshold_kb, get_max_segment_size_kb,
  get_deferred_points_threshold_bytes in shard::optimizers::config
- Collection OptimizersConfig and edge EdgeOptimizersConfig delegate to these
- Single place for threshold logic; collection and edge use shard helpers

Made-with: Cursor

* Use destructuring in config conversions to avoid missing new fields

- EdgeVectorParams: destructure VectorDataConfig in from_*, destructure self in to_vector_data_config
- EdgeSparseVectorParams: destructure SparseVectorDataConfig and SparseIndexConfig in from_*, destructure self in to_sparse_vector_data_config
- EdgeShardConfig: destructure SegmentConfig in from_segment_config, destructure self in to_segment_config
Adding new fields to source structs will now cause compile errors until conversions are updated.

Made-with: Cursor

* refactor: centralize on_disk_payload→payload_storage_type, on_disk→storage_type, and appendable quantization logic

- PayloadStorageType::from_on_disk_payload(bool) in segment (Mmap/InRamMmap)
- VectorStorageType::from_on_disk(bool) in segment (ChunkedMmap/InRamChunkedMmap)
- QuantizationConfig::for_appendable_segment(Option<&Self>) in segment (feature flag + supports_appendable)
- collection: use from_on_disk_payload in non-rocksdb branch
- edge shard/vectors: use new helpers; remove duplicated conditionals
- shard optimizers: use from_on_disk and for_appendable_segment

Made-with: Cursor

* refactor(edge): use EdgeShardConfig directly, drop segment_config

- Add plain_segment_config() for create_appendable_segment (no HNSW)
- Add segment_optimizer_config() built from EdgeShardConfig for blocking optimizers
- Add vector_data_config(name) for query/MMR
- build_blocking_optimizers: use segment_optimizer_config() instead of SegmentConfig
- create_appendable_segment: use plain_segment_config()
- search/query: use config().vectors and vector_data_config() instead of segment_config()
- Remove segment_config() from EdgeShardConfig and EdgeShard
- Add to_plain_vector_data_config on EdgeVectorParams

Made-with: Cursor

* [manual] review changes

* refactor(edge-py): wrap EdgeShardConfig, add EdgeVectorParams/EdgeSparseVectorParams

- PyEdgeConfig now wraps EdgeShardConfig (vectors, sparse_vectors, on_disk_payload, etc.)
- PyEdgeVectorParams / PyEdgeSparseVectorParams wrap edge config types
- PyEdgeOptimizersConfig for optional optimizer settings
- EdgeShard.load() uses EdgeShardConfig; edge::config made pub for Python crate
- cargo fmt + clippy (remove map_identity)

Made-with: Cursor

* refactor(edge-py): simplify config API, remove unused Py* types, add EdgeConfig

- Remove unused PyPayloadStorageType, PyVectorDataConfig, PyVectorStorageType,
  PySparseVectorDataConfig, PySparseVectorStorageType from Python bindings
- Move PyEdgeOptimizersConfig to lib/edge/python/src/config/optimizers.rs
- Update qdrant_edge.pyi: EdgeConfig with vectors/sparse_vectors,
  EdgeVectorParams, EdgeSparseVectorParams, EdgeOptimizersConfig
- Update examples (common.py, repr.py) to use new config API
- Run cargo fmt

Made-with: Cursor

* [manual] review changes

* [manual] review changes

* [manual] fix test

* Address CodeRabbit review comments for PR 8322 (#8324)

* Address CodeRabbit review comments for PR 8322

- Python examples: explicit imports (repr.py, common.py) and new EdgeConfig API
- HnswIndexConfig: add max_indexing_threads param and property in .pyi and Rust bindings
- EdgeConfig: make vectors optional for sparse-only configs; validate at least one of vectors/sparse_vectors
- EdgeShardConfig::load: use try_exists(), propagate I/O errors
- from_segment_config: infer hnsw_config from per-vector HNSW when all agree
- EdgeShard setters: atomic clone-mutate-save-then-replace; persist config save errors
- Segment compat: prefix vector name in error messages; resolve None datatype to Float32
- max_indexing_threads: preserve 0 (auto) sentinel in trait default; remove per-optimizer overrides
- SegmentOptimizerConfig:🆕 build plain and optimizer maps in single pass
- config_mismatch_optimizer tests: use VectorNameBuf::from() instead of .into()
- vectors.rs: doc updates for per-vector quantization

Made-with: Cursor

* Address @generall review: SaveOnDisk for config, resolve num_rayon_threads in optimizer

- Use SaveOnDisk<EdgeShardConfig> for EdgeShard config (generall: 'We have SaveOnDisk struct for this')
  - Create via SaveOnDisk::new() after resolving config; setters use .write() for atomic persist
  - set_vector_hnsw_config: clone then mutate then write (fallible setter)
- max_indexing_threads: resolve 0 (auto) via num_rayon_threads inside impl (generall: 'proper solution would be to resolve num_rayon_threads inside the optimizer impl')
  - max_indexing_threads_sentinel_aware() now returns Some(num_rayon_threads(raw)) so callers get actual thread count

Made-with: Cursor

* [manual] reorganize num_rayon_threads -> get_num_indexing_threads to better account per-vector configuration

---------

Co-authored-by: Cursor Agent <agent@cursor.com>
Co-authored-by: generall <andrey@vasnetsov.com>

* update docstring and pyi

* fmt

* fmt

* clipy

---------

Co-authored-by: Cursor Agent <agent@cursor.com>
Co-authored-by: generall <andrey@vasnetsov.com>
2026-03-10 00:10:04 +01:00

3191 lines
74 KiB
Python
Raw Blame History

"""
Type stubs for qdrant_edge - Qdrant Edge Python bindings.
This module provides type hints for the qdrant_edge package to enable
IDE autocompletion and type checking.
"""
from enum import Enum
from typing import Any, Dict, List, Optional, Set, Tuple, Union
from uuid import UUID
# Type aliases
PointId = Union[int, UUID, str]
Vector = Union[List[float], List[List[float]], Dict[str, "NamedVector"]]
NamedVector = Union[List[float], "SparseVector", List[List[float]]]
Payload = Dict[str, Any]
JsonPath = str
WithPayloadType = Union[bool, List[str], "PayloadSelector"]
WithVectorType = Union[bool, List[str]]
ScoringQueryType = Union["Query", "Fusion", "OrderBy", "Formula", "Sample", "Mmr"]
ConditionType = Union[
"FieldCondition",
"IsEmptyCondition",
"IsNullCondition",
"HasIdCondition",
"HasVectorCondition",
"NestedCondition",
"Filter",
]
MatchType = Union[
"MatchValue", "MatchText", "MatchTextAny", "MatchPhrase", "MatchAny", "MatchExcept"
]
RangeType = Union["RangeFloat", "RangeDateTime"]
QuantizationConfigType = Union[
"ScalarQuantizationConfig", "ProductQuantizationConfig", "BinaryQuantizationConfig"
]
IndexType = Union["PlainIndexConfig", "HnswIndexConfig"]
StartFromType = Union[int, float, str]
ExpressionType = "Expression"
# ============================================================================
# Main EdgeShard Class
# ============================================================================
class EdgeShard:
"""
The main class representing a Qdrant Edge shard.
A shard is a self-contained unit of storage that can be loaded, queried,
and updated independently.
"""
def __init__(
self,
path: str,
config: Optional["EdgeConfig"] = None,
) -> None:
"""
Load or create a Qdrant Edge shard.
Args:
path: Path to the shard directory.
config: Optional configuration for creating a new shard.
"""
...
def flush(self) -> None:
"""Flush all pending changes to disk."""
...
def close(self) -> None:
"""Close the shard and release all resources."""
...
def optimize(self) -> bool:
"""
Run segment optimizers in-process, blocking until no more optimizations are planned.
Returns:
True if any segments were optimized, False if already optimal.
"""
...
def update(self, operation: "UpdateOperation") -> None:
"""
Apply an update operation to the shard.
Args:
operation: The update operation to apply.
"""
...
def query(self, query: "QueryRequest") -> List["ScoredPoint"]:
"""
Execute a query against the shard.
Args:
query: The query request.
Returns:
List of scored points matching the query.
"""
...
def search(self, search: "SearchRequest") -> List["ScoredPoint"]:
"""
Execute a search against the shard.
Args:
search: The search request.
Returns:
List of scored points matching the search.
"""
...
def scroll(
self, scroll: "ScrollRequest"
) -> Tuple[List["Record"], Optional[PointId]]:
"""
Scroll through points in the shard.
Args:
scroll: The scroll request.
Returns:
Tuple of (points, next_offset).
"""
...
def count(self, count: "CountRequest") -> int:
"""
Count points in the shard.
Args:
count: The count request.
Returns:
Number of points matching the filter.
"""
...
def facet(self, facet: "FacetRequest") -> "FacetResponse":
"""
Get facets for a payload field.
Args:
facet: The facet request.
Returns:
Facet response with hits and counts.
"""
...
def retrieve(
self,
point_ids: List[PointId],
with_payload: Optional[WithPayloadType] = None,
with_vector: Optional[WithVectorType] = None,
) -> List["Record"]:
"""
Retrieve specific points by their IDs.
Args:
point_ids: List of point IDs to retrieve.
with_payload: Whether to include payload in results.
with_vector: Whether to include vectors in results.
Returns:
List of records.
"""
...
def info(self) -> "ShardInfo":
"""
Get information about the shard.
Returns:
Shard information.
"""
...
@staticmethod
def unpack_snapshot(snapshot_path: str, target_path: str) -> None:
"""
Unpack a snapshot to a target directory.
Args:
snapshot_path: Path to the snapshot file.
target_path: Path to extract the snapshot to.
"""
...
def snapshot_manifest(self) -> Any:
"""
Get the snapshot manifest.
Returns:
Snapshot manifest as a JSON-like value.
"""
...
def update_from_snapshot(
self,
snapshot_path: str,
tmp_dir: Optional[str] = None,
) -> None:
"""
Update the shard from a snapshot.
Args:
snapshot_path: Path to the snapshot file.
tmp_dir: Optional temporary directory for extraction.
"""
...
# ============================================================================
# Configuration Classes
# ============================================================================
class EdgeConfig:
"""Configuration for creating a new Qdrant Edge shard."""
def __init__(
self,
vectors: Optional[Union["EdgeVectorParams", Dict[str, "EdgeVectorParams"]]] = None,
sparse_vectors: Optional[Dict[str, "EdgeSparseVectorParams"]] = None,
on_disk_payload: bool = True,
hnsw_config: Optional["HnswIndexConfig"] = None,
quantization_config: Optional[QuantizationConfigType] = None,
optimizers: Optional["EdgeOptimizersConfig"] = None,
) -> None:
"""
Create an EdgeConfig.
Args:
vectors: Dense vector configuration. Can be a single EdgeVectorParams for
the default vector (name "") or a dict of name -> EdgeVectorParams.
Optional if sparse_vectors is provided (sparse-only config).
sparse_vectors: Optional sparse vector configurations.
on_disk_payload: If True, store payload on disk (mmap); otherwise in RAM.
hnsw_config: Optional global HNSW config (used when building HNSW index).
quantization_config: Optional global quantization config.
optimizers: Optional optimizer settings.
"""
...
@property
def vectors(self) -> Dict[str, "EdgeVectorParams"]:
"""Dense vector configurations."""
...
@property
def sparse_vectors(self) -> Dict[str, "EdgeSparseVectorParams"]:
"""Sparse vector configurations."""
...
@property
def on_disk_payload(self) -> bool:
"""Whether payload is stored on disk."""
...
@property
def hnsw_config(self) -> "HnswIndexConfig":
"""Global HNSW config."""
...
@property
def quantization_config(self) -> Optional[QuantizationConfigType]:
"""Global quantization config."""
...
@property
def optimizers(self) -> "EdgeOptimizersConfig":
"""Optimizer settings."""
...
class EdgeVectorParams:
"""Dense vector parameters for EdgeConfig."""
def __init__(
self,
size: int,
distance: "Distance",
on_disk: Optional[bool] = None,
multivector_config: Optional["MultiVectorConfig"] = None,
datatype: Optional["VectorStorageDatatype"] = None,
quantization_config: Optional[QuantizationConfigType] = None,
hnsw_config: Optional["HnswIndexConfig"] = None,
) -> None:
"""
Create EdgeVectorParams.
Args:
size: Dimension of vectors.
distance: Distance metric.
on_disk: If True, store vectors on disk (mmap); otherwise in RAM.
multivector_config: Optional multi-vector configuration.
datatype: Optional storage datatype.
quantization_config: Optional per-vector quantization override.
hnsw_config: Optional per-vector HNSW config override.
"""
...
@property
def size(self) -> int:
"""Vector dimension."""
...
@property
def distance(self) -> "Distance":
"""Distance metric."""
...
@property
def on_disk(self) -> Optional[bool]:
"""Whether vector storage is on disk."""
...
@property
def multivector_config(self) -> Optional["MultiVectorConfig"]:
"""Multi-vector configuration."""
...
@property
def datatype(self) -> Optional["VectorStorageDatatype"]:
"""Storage datatype."""
...
@property
def quantization_config(self) -> Optional[QuantizationConfigType]:
"""Quantization configuration."""
...
@property
def hnsw_config(self) -> Optional["HnswIndexConfig"]:
"""HNSW config override."""
...
class EdgeSparseVectorParams:
"""Sparse vector parameters for EdgeConfig."""
def __init__(
self,
full_scan_threshold: Optional[int] = None,
on_disk: Optional[bool] = None,
modifier: Optional["Modifier"] = None,
datatype: Optional["VectorStorageDatatype"] = None,
) -> None:
"""
Create EdgeSparseVectorParams.
Args:
full_scan_threshold: Threshold for full scan vs index search.
on_disk: If True, sparse index on disk; otherwise in RAM.
modifier: Optional modifier (e.g., IDF).
datatype: Storage datatype.
"""
...
@property
def full_scan_threshold(self) -> Optional[int]:
"""Full scan threshold."""
...
@property
def on_disk(self) -> Optional[bool]:
"""Whether sparse index is on disk."""
...
@property
def modifier(self) -> Optional["Modifier"]:
"""Modifier."""
...
@property
def datatype(self) -> Optional["VectorStorageDatatype"]:
"""Storage datatype."""
...
class EdgeOptimizersConfig:
"""Optimizer-related configuration for EdgeConfig."""
def __init__(
self,
deleted_threshold: Optional[float] = None,
vacuum_min_vector_number: Optional[int] = None,
default_segment_number: Optional[int] = None,
max_segment_size: Optional[int] = None,
indexing_threshold: Optional[int] = None,
prevent_unoptimized: Optional[bool] = None,
) -> None:
"""
Create EdgeOptimizersConfig.
Args:
deleted_threshold: Min fraction of deleted vectors to run vacuum (default 0.2).
vacuum_min_vector_number: Min vectors in segment to run vacuum (default 1000).
default_segment_number: Target number of segments (0 = auto).
max_segment_size: Max segment size in KB.
indexing_threshold: Indexing threshold in KB.
prevent_unoptimized: Block updates when unoptimized segments exceed threshold.
"""
...
@property
def deleted_threshold(self) -> Optional[float]:
"""Deleted threshold."""
...
@property
def vacuum_min_vector_number(self) -> Optional[int]:
"""Vacuum min vector number."""
...
@property
def default_segment_number(self) -> Optional[int]:
"""Default segment number."""
...
@property
def max_segment_size(self) -> Optional[int]:
"""Max segment size in KB."""
...
@property
def indexing_threshold(self) -> Optional[int]:
"""Indexing threshold in KB."""
...
@property
def prevent_unoptimized(self) -> Optional[bool]:
"""Prevent unoptimized flag."""
...
class PlainIndexConfig:
"""Configuration for plain (brute-force) index."""
def __init__(self) -> None:
"""Create a PlainIndexConfig."""
...
class HnswIndexConfig:
"""Configuration for HNSW index."""
def __init__(
self,
m: int,
ef_construct: int,
full_scan_threshold: int,
max_indexing_threads: int = 0,
on_disk: Optional[bool] = None,
payload_m: Optional[int] = None,
inline_storage: Optional[bool] = None,
) -> None:
"""
Create an HnswIndexConfig.
Args:
m: Number of edges per node.
ef_construct: Number of candidates during index construction.
full_scan_threshold: Threshold for full scan.
max_indexing_threads: Max threads for HNSW indexing (0 = auto).
on_disk: Whether to store on disk.
payload_m: Payload index m value.
inline_storage: Whether to use inline storage.
"""
...
@property
def m(self) -> int:
"""Number of edges per node."""
...
@property
def ef_construct(self) -> int:
"""ef_construct value."""
...
@property
def full_scan_threshold(self) -> int:
"""Full scan threshold."""
...
@property
def max_indexing_threads(self) -> int:
"""Max indexing threads (0 = auto)."""
...
@property
def on_disk(self) -> Optional[bool]:
"""On-disk flag."""
...
@property
def payload_m(self) -> Optional[int]:
"""Payload m value."""
...
@property
def inline_storage(self) -> Optional[bool]:
"""Inline storage flag."""
...
class MultiVectorConfig:
"""Configuration for multi-vector storage."""
def __init__(self, comparator: "MultiVectorComparator") -> None:
"""
Create a MultiVectorConfig.
Args:
comparator: Multi-vector comparator.
"""
...
@property
def comparator(self) -> "MultiVectorComparator":
"""Comparator."""
...
# ============================================================================
# Quantization Configuration
# ============================================================================
class ScalarQuantizationConfig:
"""Configuration for scalar quantization."""
def __init__(
self,
type: "ScalarType",
quantile: Optional[float] = None,
always_ram: Optional[bool] = None,
) -> None:
"""
Create a ScalarQuantizationConfig.
Args:
type: Scalar type (e.g., Int8).
quantile: Quantile for normalization.
always_ram: Whether to keep in RAM.
"""
...
@property
def type(self) -> "ScalarType":
"""Scalar type."""
...
@property
def quantile(self) -> Optional[float]:
"""Quantile."""
...
@property
def always_ram(self) -> Optional[bool]:
"""Always RAM flag."""
...
class ProductQuantizationConfig:
"""Configuration for product quantization."""
def __init__(
self,
compression: "CompressionRatio",
always_ram: Optional[bool] = None,
) -> None:
"""
Create a ProductQuantizationConfig.
Args:
compression: Compression ratio.
always_ram: Whether to keep in RAM.
"""
...
@property
def compression(self) -> "CompressionRatio":
"""Compression ratio."""
...
@property
def always_ram(self) -> Optional[bool]:
"""Always RAM flag."""
...
class BinaryQuantizationConfig:
"""Configuration for binary quantization."""
def __init__(
self,
always_ram: Optional[bool] = None,
encoding: Optional["BinaryQuantizationEncoding"] = None,
query_encoding: Optional["BinaryQuantizationQueryEncoding"] = None,
) -> None:
"""
Create a BinaryQuantizationConfig.
Args:
always_ram: Whether to keep in RAM.
encoding: Binary encoding type.
query_encoding: Query encoding type.
"""
...
@property
def always_ram(self) -> Optional[bool]:
"""Always RAM flag."""
...
@property
def encoding(self) -> Optional["BinaryQuantizationEncoding"]:
"""Encoding."""
...
@property
def query_encoding(self) -> Optional["BinaryQuantizationQueryEncoding"]:
"""Query encoding."""
...
# ============================================================================
# Enums
# ============================================================================
class Distance(Enum):
"""Distance metrics for vector comparison."""
Cosine = ...
Euclid = ...
Dot = ...
Manhattan = ...
class VectorStorageDatatype(Enum):
"""Vector storage data types."""
Float32 = ...
Float16 = ...
Uint8 = ...
class MultiVectorComparator(Enum):
"""Multi-vector comparison methods."""
MaxSim = ...
class ScalarType(Enum):
"""Scalar quantization types."""
Int8 = ...
class CompressionRatio(Enum):
"""Product quantization compression ratios."""
X4 = ...
X8 = ...
X16 = ...
X32 = ...
X64 = ...
class BinaryQuantizationEncoding(Enum):
"""Binary quantization encoding types."""
OneBit = ...
TwoBits = ...
OneAndHalfBits = ...
class BinaryQuantizationQueryEncoding(Enum):
"""Binary quantization query encoding types."""
Default = ...
Binary = ...
Scalar4Bits = ...
Scalar8Bits = ...
class Modifier(Enum):
"""Sparse vector modifiers."""
# Note: Python reserved word 'None' cannot be used as enum member
Idf = ...
class UpdateMode(Enum):
"""Defines the mode of the upsert operation."""
Upsert = ...
"""Default mode - insert new points, update existing points."""
InsertOnly = ...
"""Only insert new points, do not update existing points."""
UpdateOnly = ...
"""Only update existing points, do not insert new points."""
class Direction(Enum):
"""Sort direction."""
Asc = ...
Desc = ...
class Sample(Enum):
"""Sampling methods."""
Random = ...
class DecayKind(Enum):
"""Decay function kinds for scoring formulas."""
Lin = ...
Gauss = ...
Exp = ...
class PayloadSchemaType(Enum):
"""Payload field schema types."""
Keyword = ...
Integer = ...
Float = ...
Geo = ...
Text = ...
Bool = ...
Datetime = ...
Uuid = ...
# ============================================================================
# Data Types
# ============================================================================
class Point:
"""A point with ID, vector(s), and optional payload."""
def __init__(
self,
id: PointId,
vector: Vector,
payload: Optional[Payload] = None,
) -> None:
"""
Create a Point.
Args:
id: Point ID (integer or UUID).
vector: Vector data.
payload: Optional payload dictionary.
"""
...
@property
def id(self) -> PointId:
"""Point ID."""
...
@property
def vector(self) -> Vector:
"""Vector data."""
...
@property
def payload(self) -> Optional[Payload]:
"""Payload."""
...
class PointVectors:
"""Point ID with associated vectors for update operations."""
def __init__(self, id: PointId, vector: Vector) -> None:
"""
Create a PointVectors.
Args:
id: Point ID.
vector: Vector data.
"""
...
@property
def id(self) -> PointId:
"""Point ID."""
...
@property
def vector(self) -> Vector:
"""Vector data."""
...
class SparseVector:
"""A sparse vector representation."""
def __init__(self, indices: List[int], values: List[float]) -> None:
"""
Create a SparseVector.
Args:
indices: Non-zero dimension indices.
values: Values at the non-zero dimensions.
"""
...
@property
def indices(self) -> List[int]:
"""Non-zero dimension indices."""
...
@property
def values(self) -> List[float]:
"""Values at non-zero dimensions."""
...
class ScoredPoint:
"""A point with a similarity score."""
@property
def id(self) -> PointId:
"""Point ID."""
...
@property
def version(self) -> int:
"""Point version."""
...
@property
def score(self) -> float:
"""Similarity score."""
...
@property
def vector(self) -> Optional[Vector]:
"""Vector data (if requested)."""
...
@property
def payload(self) -> Optional[Payload]:
"""Payload (if requested)."""
...
@property
def order_value(self) -> Optional[Union[int, float]]:
"""Order value for order_by queries."""
...
class Record:
"""A retrieved point record."""
@property
def id(self) -> PointId:
"""Point ID."""
...
@property
def vector(self) -> Optional[Vector]:
"""Vector data (if requested)."""
...
@property
def payload(self) -> Optional[Payload]:
"""Payload (if requested)."""
...
@property
def order_value(self) -> Optional[Union[int, float]]:
"""Order value for order_by queries."""
...
class ShardInfo:
"""Information about a shard."""
@property
def segments_count(self) -> int:
"""Number of segments."""
...
@property
def points_count(self) -> int:
"""Number of points."""
...
@property
def indexed_vectors_count(self) -> int:
"""Number of indexed vectors."""
...
@property
def payload_schema(self) -> Dict[str, "PayloadIndexInfo"]:
"""Payload schema information."""
...
class PayloadIndexInfo:
"""Information about a payload index."""
@property
def data_type(self) -> PayloadSchemaType:
"""Data type."""
...
@property
def params(self) -> Optional[PayloadSchemaParams]:
"""Index parameters."""
...
@property
def points(self) -> int:
"""Number of points with this field."""
...
# ============================================================================
# Payload Index Schema Parameters
# ============================================================================
PayloadSchemaParams = Union[
"KeywordIndexParams", "IntegerIndexParams", "FloatIndexParams", "GeoIndexParams",
"TextIndexParams", "BoolIndexParams", "DatetimeIndexParams", "UuidIndexParams"
]
class KeywordIndexParams:
"""Index parameters for keyword fields."""
def __init__(
self,
is_tenant: Optional[bool] = None,
on_disk: Optional[bool] = None,
enable_hnsw: Optional[bool] = None,
) -> None:
"""
Create KeywordIndexParams.
Args:
is_tenant: Whether this field is used for tenant separation.
on_disk: Whether to store index on disk.
enable_hnsw: Whether to enable HNSW index for this field.
"""
...
@property
def is_tenant(self) -> Optional[bool]:
"""Whether this field is used for tenant separation."""
...
@property
def on_disk(self) -> Optional[bool]:
"""Whether to store index on disk."""
...
@property
def enable_hnsw(self) -> Optional[bool]:
"""Whether to enable HNSW index."""
...
class IntegerIndexParams:
"""Index parameters for integer fields."""
def __init__(
self,
lookup: Optional[bool] = None,
range: Optional[bool] = None,
is_principal: Optional[bool] = None,
on_disk: Optional[bool] = None,
enable_hnsw: Optional[bool] = None,
) -> None:
"""
Create IntegerIndexParams.
Args:
lookup: Enable exact match filtering.
range: Enable range filtering.
is_principal: Whether this field is a principal identifier.
on_disk: Whether to store index on disk.
enable_hnsw: Whether to enable HNSW index for this field.
"""
...
@property
def lookup(self) -> Optional[bool]:
"""Enable exact match filtering."""
...
@property
def range(self) -> Optional[bool]:
"""Enable range filtering."""
...
@property
def is_principal(self) -> Optional[bool]:
"""Whether this field is a principal identifier."""
...
@property
def on_disk(self) -> Optional[bool]:
"""Whether to store index on disk."""
...
@property
def enable_hnsw(self) -> Optional[bool]:
"""Whether to enable HNSW index."""
...
class FloatIndexParams:
"""Index parameters for float fields."""
def __init__(
self,
is_principal: Optional[bool] = None,
on_disk: Optional[bool] = None,
enable_hnsw: Optional[bool] = None,
) -> None:
"""
Create FloatIndexParams.
Args:
is_principal: Whether this field is a principal identifier.
on_disk: Whether to store index on disk.
enable_hnsw: Whether to enable HNSW index for this field.
"""
...
@property
def is_principal(self) -> Optional[bool]:
"""Whether this field is a principal identifier."""
...
@property
def on_disk(self) -> Optional[bool]:
"""Whether to store index on disk."""
...
@property
def enable_hnsw(self) -> Optional[bool]:
"""Whether to enable HNSW index."""
...
class GeoIndexParams:
"""Index parameters for geo fields."""
def __init__(
self,
on_disk: Optional[bool] = None,
enable_hnsw: Optional[bool] = None,
) -> None:
"""
Create GeoIndexParams.
Args:
on_disk: Whether to store index on disk.
enable_hnsw: Whether to enable HNSW index for this field.
"""
...
@property
def on_disk(self) -> Optional[bool]:
"""Whether to store index on disk."""
...
@property
def enable_hnsw(self) -> Optional[bool]:
"""Whether to enable HNSW index."""
...
class BoolIndexParams:
"""Index parameters for boolean fields."""
def __init__(
self,
on_disk: Optional[bool] = None,
enable_hnsw: Optional[bool] = None,
) -> None:
"""
Create BoolIndexParams.
Args:
on_disk: Whether to store index on disk.
enable_hnsw: Whether to enable HNSW index for this field.
"""
...
@property
def on_disk(self) -> Optional[bool]:
"""Whether to store index on disk."""
...
@property
def enable_hnsw(self) -> Optional[bool]:
"""Whether to enable HNSW index."""
...
class DatetimeIndexParams:
"""Index parameters for datetime fields."""
def __init__(
self,
is_principal: Optional[bool] = None,
on_disk: Optional[bool] = None,
enable_hnsw: Optional[bool] = None,
) -> None:
"""
Create DatetimeIndexParams.
Args:
is_principal: Whether this field is a principal identifier.
on_disk: Whether to store index on disk.
enable_hnsw: Whether to enable HNSW index for this field.
"""
...
@property
def is_principal(self) -> Optional[bool]:
"""Whether this field is a principal identifier."""
...
@property
def on_disk(self) -> Optional[bool]:
"""Whether to store index on disk."""
...
@property
def enable_hnsw(self) -> Optional[bool]:
"""Whether to enable HNSW index."""
...
class UuidIndexParams:
"""Index parameters for UUID fields."""
def __init__(
self,
is_tenant: Optional[bool] = None,
on_disk: Optional[bool] = None,
enable_hnsw: Optional[bool] = None,
) -> None:
"""
Create UuidIndexParams.
Args:
is_tenant: Whether this field is used for tenant separation.
on_disk: Whether to store index on disk.
enable_hnsw: Whether to enable HNSW index for this field.
"""
...
@property
def is_tenant(self) -> Optional[bool]:
"""Whether this field is used for tenant separation."""
...
@property
def on_disk(self) -> Optional[bool]:
"""Whether to store index on disk."""
...
@property
def enable_hnsw(self) -> Optional[bool]:
"""Whether to enable HNSW index."""
...
class TextIndexParams:
"""Index parameters for text fields."""
def __init__(
self,
tokenizer: Optional["TokenizerType"] = None,
min_token_len: Optional[int] = None,
max_token_len: Optional[int] = None,
lowercase: Optional[bool] = None,
ascii_folding: Optional[bool] = None,
phrase_matching: Optional[bool] = None,
stopwords: Optional["Stopwords"] = None,
on_disk: Optional[bool] = None,
stemmer: Optional["StemmingAlgorithm"] = None,
enable_hnsw: Optional[bool] = None,
) -> None:
"""
Create TextIndexParams.
Args:
tokenizer: Tokenizer type.
min_token_len: Minimum token length.
max_token_len: Maximum token length.
lowercase: Convert to lowercase.
ascii_folding: Apply ASCII folding.
phrase_matching: Enable phrase matching.
stopwords: Stopwords configuration.
on_disk: Whether to store index on disk.
stemmer: Stemming algorithm.
enable_hnsw: Whether to enable HNSW index for this field.
"""
...
@property
def tokenizer(self) -> "TokenizerType":
"""Tokenizer type."""
...
@property
def min_token_len(self) -> Optional[int]:
"""Minimum token length."""
...
@property
def max_token_len(self) -> Optional[int]:
"""Maximum token length."""
...
@property
def lowercase(self) -> Optional[bool]:
"""Convert to lowercase."""
...
@property
def ascii_folding(self) -> Optional[bool]:
"""Apply ASCII folding."""
...
@property
def phrase_matching(self) -> Optional[bool]:
"""Enable phrase matching."""
...
@property
def stopwords(self) -> Optional["Stopwords"]:
"""Stopwords configuration."""
...
@property
def on_disk(self) -> Optional[bool]:
"""Whether to store index on disk."""
...
@property
def stemmer(self) -> Optional["StemmingAlgorithm"]:
"""Stemming algorithm."""
...
@property
def enable_hnsw(self) -> Optional[bool]:
"""Whether to enable HNSW index."""
...
class TokenizerType(Enum):
"""Text tokenizer types."""
Prefix = ...
Whitespace = ...
Word = ...
Multilingual = ...
Stopwords = Union["Language", "StopwordsSet"]
"""Stopwords configuration - either a language or a custom set."""
class Language(Enum):
"""Predefined stopword languages."""
Arabic = ...
Azerbaijani = ...
Basque = ...
Bengali = ...
Catalan = ...
Chinese = ...
Danish = ...
Dutch = ...
English = ...
Finnish = ...
French = ...
German = ...
Greek = ...
Hebrew = ...
Hinglish = ...
Hungarian = ...
Indonesian = ...
Italian = ...
Japanese = ...
Kazakh = ...
Nepali = ...
Norwegian = ...
Portuguese = ...
Romanian = ...
Russian = ...
Slovene = ...
Spanish = ...
Swedish = ...
Tajik = ...
Turkish = ...
class StopwordsSet:
"""Custom stopwords set."""
def __init__(
self,
languages: Optional[Set["Language"]] = None,
custom: Optional[Set[str]] = None,
) -> None:
"""
Create a StopwordsSet.
Args:
languages: Predefined language stopwords to include.
custom: Custom stopwords to add.
"""
...
@property
def languages(self) -> Optional[Set["Language"]]:
"""Predefined language stopwords."""
...
@property
def custom(self) -> Optional[Set[str]]:
"""Custom stopwords."""
...
StemmingAlgorithm = Union["SnowballParams"]
class SnowballParams:
"""Snowball stemming algorithm parameters."""
def __init__(self, language: "SnowballLanguage") -> None:
"""
Create SnowballParams.
Args:
language: Snowball language.
"""
...
@property
def language(self) -> "SnowballLanguage":
"""Snowball language."""
...
class SnowballLanguage(Enum):
"""Snowball stemmer languages."""
Arabic = ...
Armenian = ...
Danish = ...
Dutch = ...
English = ...
Finnish = ...
French = ...
German = ...
Greek = ...
Hungarian = ...
Italian = ...
Norwegian = ...
Portuguese = ...
Romanian = ...
Russian = ...
Spanish = ...
Swedish = ...
Tamil = ...
Turkish = ...
# ============================================================================
# Request Classes
# ============================================================================
class QueryRequest:
"""Request for query operation."""
def __init__(
self,
limit: int,
offset: Optional[int] = None,
query: Optional[ScoringQueryType] = None,
prefetches: Optional[List["Prefetch"]] = None,
with_vector: Optional[WithVectorType] = None,
with_payload: Optional[WithPayloadType] = None,
filter: Optional["Filter"] = None,
score_threshold: Optional[float] = None,
params: Optional["SearchParams"] = None,
) -> None:
"""
Create a QueryRequest.
Args:
limit: Maximum number of results.
offset: Number of results to skip.
query: Scoring query (vector, fusion, order_by, etc.).
prefetches: Prefetch stages for multi-stage queries.
with_vector: Whether to include vectors.
with_payload: Whether to include payload.
filter: Filter conditions.
score_threshold: Minimum score threshold.
params: Search parameters.
"""
...
@property
def prefetches(self) -> List["Prefetch"]:
"""Prefetch stages."""
...
@property
def query(self) -> Optional[ScoringQueryType]:
"""Scoring query."""
...
@property
def filter(self) -> Optional["Filter"]:
"""Filter."""
...
@property
def score_threshold(self) -> Optional[float]:
"""Score threshold."""
...
@property
def limit(self) -> int:
"""Result limit."""
...
@property
def offset(self) -> int:
"""Result offset."""
...
@property
def params(self) -> Optional["SearchParams"]:
"""Search parameters."""
...
@property
def with_vector(self) -> WithVectorType:
"""With vector flag."""
...
@property
def with_payload(self) -> WithPayloadType:
"""With payload flag."""
...
class Prefetch:
"""A prefetch stage for multi-stage queries."""
def __init__(
self,
limit: int,
query: Optional[ScoringQueryType] = None,
prefetches: Optional[List["Prefetch"]] = None,
params: Optional["SearchParams"] = None,
filter: Optional["Filter"] = None,
score_threshold: Optional[float] = None,
) -> None:
"""
Create a Prefetch stage.
Args:
limit: Maximum number of results for this stage.
query: Scoring query.
prefetches: Nested prefetch stages.
params: Search parameters.
filter: Filter conditions.
score_threshold: Minimum score threshold.
"""
...
@property
def prefetches(self) -> List["Prefetch"]:
"""Nested prefetch stages."""
...
@property
def query(self) -> Optional[ScoringQueryType]:
"""Scoring query."""
...
@property
def limit(self) -> int:
"""Result limit."""
...
@property
def params(self) -> Optional["SearchParams"]:
"""Search parameters."""
...
@property
def filter(self) -> Optional["Filter"]:
"""Filter."""
...
@property
def score_threshold(self) -> Optional[float]:
"""Score threshold."""
...
class SearchRequest:
"""Request for search operation."""
def __init__(
self,
query: "Query",
limit: int,
offset: Optional[int] = None,
filter: Optional["Filter"] = None,
params: Optional["SearchParams"] = None,
with_vector: Optional[WithVectorType] = None,
with_payload: Optional[WithPayloadType] = None,
score_threshold: Optional[float] = None,
) -> None:
"""
Create a SearchRequest.
Args:
query: Query (vector-based).
limit: Maximum number of results.
offset: Number of results to skip.
filter: Filter conditions.
params: Search parameters.
with_vector: Whether to include vectors.
with_payload: Whether to include payload.
score_threshold: Minimum score threshold.
"""
...
@property
def query(self) -> "Query":
"""Query."""
...
@property
def filter(self) -> Optional["Filter"]:
"""Filter."""
...
@property
def params(self) -> Optional["SearchParams"]:
"""Search parameters."""
...
@property
def limit(self) -> int:
"""Result limit."""
...
@property
def offset(self) -> int:
"""Result offset."""
...
@property
def with_vector(self) -> Optional[WithVectorType]:
"""With vector flag."""
...
@property
def with_payload(self) -> Optional[WithPayloadType]:
"""With payload flag."""
...
@property
def score_threshold(self) -> Optional[float]:
"""Score threshold."""
...
class ScrollRequest:
"""Request for scroll operation."""
def __init__(
self,
offset: Optional[PointId] = None,
limit: Optional[int] = None,
filter: Optional["Filter"] = None,
with_payload: Optional[WithPayloadType] = None,
with_vector: Optional[WithVectorType] = None,
order_by: Optional["OrderBy"] = None,
) -> None:
"""
Create a ScrollRequest.
Args:
offset: Starting point ID.
limit: Maximum number of results.
filter: Filter conditions.
with_payload: Whether to include payload.
with_vector: Whether to include vectors.
order_by: Order by configuration.
"""
...
@property
def offset(self) -> Optional[PointId]:
"""Offset point ID."""
...
@property
def limit(self) -> Optional[int]:
"""Result limit."""
...
@property
def filter(self) -> Optional["Filter"]:
"""Filter."""
...
@property
def with_payload(self) -> Optional[WithPayloadType]:
"""With payload flag."""
...
@property
def with_vector(self) -> WithVectorType:
"""With vector flag."""
...
@property
def order_by(self) -> Optional["OrderBy"]:
"""Order by configuration."""
...
class CountRequest:
"""Request for count operation."""
def __init__(
self,
exact: bool = True,
filter: Optional["Filter"] = None,
) -> None:
"""
Create a CountRequest.
Args:
exact: Whether to count exactly or estimate.
filter: Filter conditions.
"""
...
@property
def filter(self) -> Optional["Filter"]:
"""Filter."""
...
@property
def exact(self) -> bool:
"""Exact count flag."""
...
class FacetRequest:
"""Request for facet operation."""
def __init__(
self,
key: JsonPath,
limit: int = 10,
exact: bool = False,
filter: Optional["Filter"] = None,
) -> None:
"""
Create a FacetRequest.
Args:
key: Payload field key to facet on.
limit: Maximum number of facet hits to return.
exact: Whether to count exactly or estimate.
filter: Filter conditions.
"""
...
@property
def key(self) -> str:
"""Facet key."""
...
@property
def limit(self) -> int:
"""Result limit."""
...
@property
def exact(self) -> bool:
"""Exact count flag."""
...
@property
def filter(self) -> Optional["Filter"]:
"""Filter."""
...
class FacetHit:
"""A facet hit with value and count."""
@property
def value(self) -> Union[str, int, bool]:
"""Facet value."""
...
@property
def count(self) -> int:
"""Count of points with this value."""
...
class FacetResponse:
"""Response for facet operation."""
@property
def hits(self) -> List["FacetHit"]:
"""Facet hits."""
...
def __len__(self) -> int:
"""Number of hits."""
...
def __iter__(self) -> Any:
"""Iterate over hits."""
...
class SearchParams:
"""Parameters for search operations."""
def __init__(
self,
hnsw_ef: Optional[int] = None,
exact: bool = False,
quantization: Optional["QuantizationSearchParams"] = None,
indexed_only: bool = False,
acorn: Optional["AcornSearchParams"] = None,
) -> None:
"""
Create SearchParams.
Args:
hnsw_ef: ef parameter for HNSW search.
exact: Whether to use exact search.
quantization: Quantization search parameters.
indexed_only: Whether to search only indexed vectors.
acorn: Acorn search parameters.
"""
...
@property
def hnsw_ef(self) -> Optional[int]:
"""HNSW ef parameter."""
...
@property
def exact(self) -> bool:
"""Exact search flag."""
...
@property
def quantization(self) -> Optional["QuantizationSearchParams"]:
"""Quantization parameters."""
...
@property
def indexed_only(self) -> bool:
"""Indexed only flag."""
...
@property
def acorn(self) -> Optional["AcornSearchParams"]:
"""Acorn parameters."""
...
class QuantizationSearchParams:
"""Parameters for quantization during search."""
def __init__(
self,
ignore: bool = False,
rescore: Optional[bool] = None,
oversampling: Optional[float] = None,
) -> None:
"""
Create QuantizationSearchParams.
Args:
ignore: Whether to ignore quantization.
rescore: Whether to rescore with original vectors.
oversampling: Oversampling factor.
"""
...
@property
def ignore(self) -> bool:
"""Ignore quantization flag."""
...
@property
def rescore(self) -> Optional[bool]:
"""Rescore flag."""
...
@property
def oversampling(self) -> Optional[float]:
"""Oversampling factor."""
...
class AcornSearchParams:
"""Parameters for Acorn filtered search."""
def __init__(
self,
enable: bool = False,
max_selectivity: Optional[float] = None,
) -> None:
"""
Create AcornSearchParams.
Args:
enable: Whether to enable Acorn.
max_selectivity: Maximum filter selectivity for Acorn.
"""
...
@property
def enable(self) -> bool:
"""Enable flag."""
...
@property
def max_selectivity(self) -> Optional[float]:
"""Maximum selectivity."""
...
# ============================================================================
# Query Types
# ============================================================================
class Query(Enum):
"""Query types for vector search."""
@staticmethod
def Nearest(
query: NamedVector, using: Optional[str] = None
) -> "Query":
"""Create a nearest neighbor query."""
...
@staticmethod
def RecommendBestScore(
query: "RecommendQuery", using: Optional[str] = None
) -> "Query":
"""Create a recommend query using best score."""
...
@staticmethod
def RecommendSumScores(
query: "RecommendQuery", using: Optional[str] = None
) -> "Query":
"""Create a recommend query using sum of scores."""
...
@staticmethod
def Discover(
query: "DiscoverQuery", using: Optional[str] = None
) -> "Query":
"""Create a discover query."""
...
@staticmethod
def Context(
query: "ContextQuery", using: Optional[str] = None
) -> "Query":
"""Create a context query."""
...
@staticmethod
def FeedbackNaive(
query: "FeedbackNaiveQuery", using: Optional[str] = None
) -> "Query":
"""Create a feedback naive query."""
...
class Fusion:
"""Fusion methods for combining multiple prefetch results."""
class Rrf:
"""
RRF (Reciprocal Rank Fusion) with given parameters.
Args:
k: The RRF k parameter.
weights: Optional weights for each prefetch source.
Higher weight gives more influence on the final ranking.
If not specified, all prefetches are weighted equally.
Examples:
# Basic RRF with k=2
Fusion.Rrf(k=2)
# Weighted RRF - first prefetch has 3x weight
Fusion.Rrf(k=2, weights=[3.0, 1.0])
"""
def __init__(self, k: int, weights: Optional[List[float]] = None) -> None: ...
@property
def k(self) -> int: ...
@property
def weights(self) -> Optional[List[float]]: ...
class Dbsf:
"""DBSF (Distribution-Based Score Fusion)."""
def __init__(self) -> None: ...
class OrderBy:
"""Order results by a payload field."""
def __init__(
self,
key: JsonPath,
direction: Optional[Direction] = None,
start_from: Optional[StartFromType] = None,
) -> None:
"""
Create an OrderBy.
Args:
key: Payload field path.
direction: Sort direction.
start_from: Starting value.
"""
...
@property
def key(self) -> str:
"""Field key."""
...
@property
def direction(self) -> Optional[Direction]:
"""Sort direction."""
...
@property
def start_from(self) -> Optional[StartFromType]:
"""Starting value."""
...
class Mmr:
"""Maximal Marginal Relevance for result diversification."""
def __init__(
self,
vector: NamedVector,
lambda_: float,
candidates_limit: int,
using: Optional[str] = None,
) -> None:
"""
Create an MMR query.
Args:
vector: Query vector.
lambda_: Balance between relevance and diversity (0-1).
candidates_limit: Number of candidates to consider.
using: Named vector to use.
"""
...
@property
def vector(self) -> NamedVector:
"""Query vector."""
...
@property
def lambda_(self) -> float:
"""Balance between relevance and diversity."""
...
@property
def using(self) -> str:
"""Named vector."""
...
# Note: 'lambda' is Python reserved word, using 'lambda_' in __init__
# but the property may be named differently
@property
def candidates_limit(self) -> int:
"""Candidates limit."""
...
class RecommendQuery:
"""Query for recommendation based on positive and negative examples."""
def __init__(
self,
positives: List[NamedVector],
negatives: List[NamedVector],
) -> None:
"""
Create a RecommendQuery.
Args:
positives: Positive example vectors.
negatives: Negative example vectors.
"""
...
@property
def positives(self) -> List[NamedVector]:
"""Positive examples."""
...
@property
def negatives(self) -> List[NamedVector]:
"""Negative examples."""
...
class DiscoverQuery:
"""Query for discovery using a target and context pairs."""
def __init__(
self,
target: NamedVector,
pairs: List["ContextPair"],
) -> None:
"""
Create a DiscoverQuery.
Args:
target: Target vector.
pairs: Context pairs.
"""
...
@property
def target(self) -> NamedVector:
"""Target vector."""
...
@property
def pairs(self) -> List["ContextPair"]:
"""Context pairs."""
...
class ContextQuery:
"""Query based on context pairs only."""
def __init__(self, pairs: List["ContextPair"]) -> None:
"""
Create a ContextQuery.
Args:
pairs: Context pairs.
"""
...
@property
def pairs(self) -> List["ContextPair"]:
"""Context pairs."""
...
class ContextPair:
"""A positive/negative pair for context-based queries."""
def __init__(
self,
positive: NamedVector,
negative: NamedVector,
) -> None:
"""
Create a ContextPair.
Args:
positive: Positive example.
negative: Negative example.
"""
...
@property
def positive(self) -> NamedVector:
"""Positive example."""
...
@property
def negative(self) -> NamedVector:
"""Negative example."""
...
class FeedbackNaiveQuery:
"""Query using naive feedback approach."""
def __init__(
self,
target: NamedVector,
feedback: List["FeedbackItem"],
strategy: "NaiveFeedbackStrategy",
) -> None:
"""
Create a FeedbackNaiveQuery.
Args:
target: Target vector.
feedback: Feedback items with scores.
strategy: Feedback coefficients.
"""
...
@property
def target(self) -> NamedVector:
"""Target vector."""
...
@property
def feedback(self) -> List["FeedbackItem"]:
"""Feedback items."""
...
@property
def coefficients(self) -> "NaiveFeedbackStrategy":
"""Coefficients."""
...
class FeedbackItem:
"""A feedback item with vector and score."""
def __init__(self, vector: NamedVector, score: float) -> None:
"""
Create a FeedbackItem.
Args:
vector: Feedback vector.
score: Feedback score.
"""
...
@property
def vector(self) -> NamedVector:
"""Feedback vector."""
...
@property
def score(self) -> float:
"""Feedback score."""
...
class NaiveFeedbackStrategy:
"""Coefficients for naive feedback query."""
def __init__(self, a: float, b: float, c: float) -> None:
"""
Create NaiveFeedbackStrategy coefficients.
Args:
a: Coefficient a.
b: Coefficient b.
c: Coefficient c.
"""
...
@property
def a(self) -> float:
"""Coefficient a."""
...
@property
def b(self) -> float:
"""Coefficient b."""
...
@property
def c(self) -> float:
"""Coefficient c."""
...
# ============================================================================
# Formula Classes
# ============================================================================
class Formula:
"""A scoring formula for custom ranking."""
def __init__(
self,
formula: ExpressionType,
defaults: Optional[Dict[str, Any]] = None,
) -> None:
"""
Create a Formula.
Args:
formula: Expression tree.
defaults: Default variable values.
"""
...
class Expression(Enum):
"""Expression types for formulas."""
@staticmethod
def Constant(val: float) -> "Expression":
"""Create a constant expression."""
...
@staticmethod
def Variable(var: str) -> "Expression":
"""Create a variable expression."""
...
@staticmethod
def Condition(cond: ConditionType) -> "Expression":
"""Create a condition expression (returns 1 if true, 0 if false)."""
...
@staticmethod
def GeoDistance(origin: "GeoPoint", to: JsonPath) -> "Expression":
"""Create a geo distance expression."""
...
@staticmethod
def Datetime(date_time: str) -> "Expression":
"""Create a datetime constant expression."""
...
@staticmethod
def DatetimeKey(path: JsonPath) -> "Expression":
"""Create a datetime field expression."""
...
@staticmethod
def Mult(exprs: List["Expression"]) -> "Expression":
"""Create a multiplication expression."""
...
@staticmethod
def Sum(exprs: List["Expression"]) -> "Expression":
"""Create a sum expression."""
...
@staticmethod
def Neg(expr: "Expression") -> "Expression":
"""Create a negation expression."""
...
@staticmethod
def Div(
left: "Expression",
right: "Expression",
by_zero_default: Optional[float] = None,
) -> "Expression":
"""Create a division expression."""
...
@staticmethod
def Sqrt(expr: "Expression") -> "Expression":
"""Create a square root expression."""
...
@staticmethod
def Pow(base: "Expression", exponent: "Expression") -> "Expression":
"""Create a power expression."""
...
@staticmethod
def Exp(expr: "Expression") -> "Expression":
"""Create an exponential expression."""
...
@staticmethod
def Log10(expr: "Expression") -> "Expression":
"""Create a log10 expression."""
...
@staticmethod
def Ln(expr: "Expression") -> "Expression":
"""Create a natural log expression."""
...
@staticmethod
def Abs(expr: "Expression") -> "Expression":
"""Create an absolute value expression."""
...
@staticmethod
def Decay(
kind: DecayKind,
x: "Expression",
target: Optional["Expression"] = None,
midpoint: Optional[float] = None,
scale: Optional[float] = None,
) -> "Expression":
"""Create a decay expression."""
...
# ============================================================================
# Filter Classes
# ============================================================================
class Filter:
"""Filter conditions for queries."""
def __init__(
self,
must: Optional[List[ConditionType]] = None,
should: Optional[List[ConditionType]] = None,
must_not: Optional[List[ConditionType]] = None,
min_should: Optional["MinShould"] = None,
) -> None:
"""
Create a Filter.
Args:
must: Conditions that must all match.
should: Conditions where at least one should match.
must_not: Conditions that must not match.
min_should: Minimum number of should conditions to match.
"""
...
@property
def must(self) -> Optional[List[ConditionType]]:
"""Must conditions."""
...
@property
def should(self) -> Optional[List[ConditionType]]:
"""Should conditions."""
...
@property
def must_not(self) -> Optional[List[ConditionType]]:
"""Must not conditions."""
...
@property
def min_should(self) -> Optional["MinShould"]:
"""Minimum should configuration."""
...
class MinShould:
"""Minimum number of should conditions that must match."""
def __init__(
self,
conditions: List[ConditionType],
min_count: int,
) -> None:
"""
Create a MinShould.
Args:
conditions: List of conditions.
min_count: Minimum number that must match.
"""
...
@property
def conditions(self) -> List[ConditionType]:
"""Conditions."""
...
@property
def min_count(self) -> int:
"""Minimum count."""
...
class FieldCondition:
"""Condition on a payload field."""
def __init__(
self,
key: JsonPath,
match: Optional[MatchType] = None,
range: Optional[RangeType] = None,
geo_bounding_box: Optional["GeoBoundingBox"] = None,
geo_radius: Optional["GeoRadius"] = None,
geo_polygon: Optional["GeoPolygon"] = None,
values_count: Optional["ValuesCount"] = None,
is_empty: Optional[bool] = None,
is_null: Optional[bool] = None,
) -> None:
"""
Create a FieldCondition.
Args:
key: Payload field path.
match: Match condition.
range: Range condition.
geo_bounding_box: Geo bounding box condition.
geo_radius: Geo radius condition.
geo_polygon: Geo polygon condition.
values_count: Values count condition.
is_empty: Check if empty.
is_null: Check if null.
"""
...
@property
def key(self) -> str:
"""Field key."""
...
@property
def match(self) -> Optional[MatchType]:
"""Match condition."""
...
@property
def range(self) -> Optional[RangeType]:
"""Range condition."""
...
@property
def geo_bounding_box(self) -> Optional["GeoBoundingBox"]:
"""Geo bounding box."""
...
@property
def geo_radius(self) -> Optional["GeoRadius"]:
"""Geo radius."""
...
@property
def geo_polygon(self) -> Optional["GeoPolygon"]:
"""Geo polygon."""
...
@property
def values_count(self) -> Optional["ValuesCount"]:
"""Values count."""
...
@property
def is_empty(self) -> Optional[bool]:
"""Is empty flag."""
...
@property
def is_null(self) -> Optional[bool]:
"""Is null flag."""
...
class IsEmptyCondition:
"""Check if a field is empty."""
def __init__(self, key: JsonPath) -> None:
"""
Create an IsEmptyCondition.
Args:
key: Payload field path.
"""
...
@property
def key(self) -> str:
"""Field key."""
...
class IsNullCondition:
"""Check if a field is null."""
def __init__(self, key: JsonPath) -> None:
"""
Create an IsNullCondition.
Args:
key: Payload field path.
"""
...
@property
def key(self) -> str:
"""Field key."""
...
class HasIdCondition:
"""Check if point ID is in a set."""
def __init__(self, point_ids: Set[PointId]) -> None:
"""
Create a HasIdCondition.
Args:
point_ids: Set of point IDs.
"""
...
@property
def point_ids(self) -> Set[PointId]:
"""Point IDs."""
...
class HasVectorCondition:
"""Check if point has a specific vector."""
def __init__(self, vector: str) -> None:
"""
Create a HasVectorCondition.
Args:
vector: Vector name.
"""
...
@property
def vector(self) -> str:
"""Vector name."""
...
class NestedCondition:
"""Condition on nested objects."""
def __init__(self, key: JsonPath, filter: Filter) -> None:
"""
Create a NestedCondition.
Args:
key: Path to nested array.
filter: Filter to apply to nested objects.
"""
...
@property
def key(self) -> str:
"""Nested field key."""
...
@property
def filter(self) -> Filter:
"""Nested filter."""
...
# ============================================================================
# Match Conditions
# ============================================================================
class MatchValue:
"""Match exact value."""
def __init__(self, value: Union[str, int, bool]) -> None:
"""
Create a MatchValue.
Args:
value: Value to match.
"""
...
@property
def value(self) -> Union[str, int, bool]:
"""Value."""
...
class MatchText:
"""Full-text match."""
def __init__(self, text: str) -> None:
"""
Create a MatchText.
Args:
text: Text to search for.
"""
...
@property
def text(self) -> str:
"""Text."""
...
class MatchTextAny:
"""Match any of the words in text."""
def __init__(self, text_any: str) -> None:
"""
Create a MatchTextAny.
Args:
text_any: Space-separated words to match any of.
"""
...
@property
def text_any(self) -> str:
"""Text."""
...
class MatchPhrase:
"""Match exact phrase."""
def __init__(self, phrase: str) -> None:
"""
Create a MatchPhrase.
Args:
phrase: Phrase to match.
"""
...
@property
def phrase(self) -> str:
"""Phrase."""
...
class MatchAny:
"""Match any of the values."""
def __init__(self, any: Union[List[str], List[int]]) -> None:
"""
Create a MatchAny.
Args:
any: List of values to match any of.
"""
...
@property
def value(self) -> Union[List[str], List[int]]:
"""Values."""
...
class MatchExcept:
"""Match any value except these."""
def __init__(self, except_: Union[List[str], List[int]]) -> None:
"""
Create a MatchExcept.
Args:
except_: List of values to exclude.
"""
...
@property
def value(self) -> Union[List[str], List[int]]:
"""Excluded values."""
...
# ============================================================================
# Range Conditions
# ============================================================================
class RangeFloat:
"""Range condition for float values."""
def __init__(
self,
gte: Optional[float] = None,
gt: Optional[float] = None,
lte: Optional[float] = None,
lt: Optional[float] = None,
) -> None:
"""
Create a RangeFloat.
Args:
gte: Greater than or equal.
gt: Greater than.
lte: Less than or equal.
lt: Less than.
"""
...
@property
def gte(self) -> Optional[float]:
"""Greater than or equal."""
...
@property
def gt(self) -> Optional[float]:
"""Greater than."""
...
@property
def lte(self) -> Optional[float]:
"""Less than or equal."""
...
@property
def lt(self) -> Optional[float]:
"""Less than."""
...
class RangeDateTime:
"""Range condition for datetime values."""
def __init__(
self,
gte: Optional[str] = None,
gt: Optional[str] = None,
lte: Optional[str] = None,
lt: Optional[str] = None,
) -> None:
"""
Create a RangeDateTime.
Args:
gte: Greater than or equal (ISO 8601 string).
gt: Greater than (ISO 8601 string).
lte: Less than or equal (ISO 8601 string).
lt: Less than (ISO 8601 string).
"""
...
@property
def gte(self) -> Optional[str]:
"""Greater than or equal."""
...
@property
def gt(self) -> Optional[str]:
"""Greater than."""
...
@property
def lte(self) -> Optional[str]:
"""Less than or equal."""
...
@property
def lt(self) -> Optional[str]:
"""Less than."""
...
class ValuesCount:
"""Condition on count of values in array field."""
def __init__(
self,
lt: Optional[int] = None,
gt: Optional[int] = None,
lte: Optional[int] = None,
gte: Optional[int] = None,
) -> None:
"""
Create a ValuesCount.
Args:
lt: Less than.
gt: Greater than.
lte: Less than or equal.
gte: Greater than or equal.
"""
...
@property
def lt(self) -> Optional[int]:
"""Less than."""
...
@property
def gt(self) -> Optional[int]:
"""Greater than."""
...
@property
def lte(self) -> Optional[int]:
"""Less than or equal."""
...
@property
def gte(self) -> Optional[int]:
"""Greater than or equal."""
...
# ============================================================================
# Geo Types
# ============================================================================
class GeoPoint:
"""A geographic point."""
def __init__(self, lon: float, lat: float) -> None:
"""
Create a GeoPoint.
Args:
lon: Longitude (-180 to 180).
lat: Latitude (-90 to 90).
"""
...
@property
def lon(self) -> float:
"""Longitude."""
...
@property
def lat(self) -> float:
"""Latitude."""
...
class GeoBoundingBox:
"""A geographic bounding box."""
def __init__(self, top_left: GeoPoint, bottom_right: GeoPoint) -> None:
"""
Create a GeoBoundingBox.
Args:
top_left: Top-left corner.
bottom_right: Bottom-right corner.
"""
...
@property
def top_left(self) -> GeoPoint:
"""Top-left corner."""
...
@property
def bottom_right(self) -> GeoPoint:
"""Bottom-right corner."""
...
class GeoRadius:
"""A geographic circle."""
def __init__(self, center: GeoPoint, radius: float) -> None:
"""
Create a GeoRadius.
Args:
center: Center point.
radius: Radius in meters.
"""
...
@property
def center(self) -> GeoPoint:
"""Center point."""
...
@property
def radius(self) -> float:
"""Radius in meters."""
...
class GeoPolygon:
"""A geographic polygon."""
def __init__(
self,
exterior: List[GeoPoint],
interiors: Optional[List[List[GeoPoint]]] = None,
) -> None:
"""
Create a GeoPolygon.
Args:
exterior: Exterior ring points.
interiors: Optional interior rings (holes).
"""
...
@property
def exterior(self) -> List[GeoPoint]:
"""Exterior ring."""
...
@property
def interiors(self) -> Optional[List[List[GeoPoint]]]:
"""Interior rings (holes)."""
...
# ============================================================================
# Payload Selector
# ============================================================================
class PayloadSelector(Enum):
"""Select specific payload fields."""
@staticmethod
def Include(keys: List[str]) -> "PayloadSelector":
"""Include only specified fields."""
...
@staticmethod
def Exclude(keys: List[str]) -> "PayloadSelector":
"""Exclude specified fields."""
...
# ============================================================================
# Update Operation
# ============================================================================
class UpdateOperation:
"""Operations for updating shard data."""
@staticmethod
def upsert_points(
points: List[Point],
condition: Optional[Filter] = None,
update_mode: Optional[UpdateMode] = None,
) -> "UpdateOperation":
"""
Insert or update points.
Args:
points: Points to upsert.
condition: Optional condition for conditional upsert.
update_mode: Optional mode of the upsert operation:
- UpdateMode.Upsert (default): insert new points, update existing points
- UpdateMode.InsertOnly: only insert new points, do not update existing points
- UpdateMode.UpdateOnly: only update existing points, do not insert new points
"""
...
@staticmethod
def delete_points(point_ids: List[PointId]) -> "UpdateOperation":
"""
Delete points by ID.
Args:
point_ids: IDs of points to delete.
"""
...
@staticmethod
def delete_points_by_filter(filter: Filter) -> "UpdateOperation":
"""
Delete points matching a filter.
Args:
filter: Filter for points to delete.
"""
...
@staticmethod
def update_vectors(
point_vectors: List[PointVectors],
condition: Optional[Filter] = None,
) -> "UpdateOperation":
"""
Update vectors of existing points.
Args:
point_vectors: Point IDs with new vectors.
condition: Optional filter condition.
"""
...
@staticmethod
def delete_vectors(
point_ids: List[PointId],
vector_names: List[str],
) -> "UpdateOperation":
"""
Delete specific vectors from points.
Args:
point_ids: Point IDs.
vector_names: Names of vectors to delete.
"""
...
@staticmethod
def delete_vectors_by_filter(
filter: Filter,
vector_names: List[str],
) -> "UpdateOperation":
"""
Delete vectors from points matching a filter.
Args:
filter: Filter for points.
vector_names: Names of vectors to delete.
"""
...
@staticmethod
def set_payload(
point_ids: List[PointId],
payload: Payload,
key: Optional[str] = None,
) -> "UpdateOperation":
"""
Set payload fields on points.
Args:
point_ids: Point IDs.
payload: Payload to set.
key: Optional nested key path.
"""
...
@staticmethod
def set_payload_by_filter(
filter: Filter,
payload: Payload,
key: Optional[str] = None,
) -> "UpdateOperation":
"""
Set payload on points matching a filter.
Args:
filter: Filter for points.
payload: Payload to set.
key: Optional nested key path.
"""
...
@staticmethod
def delete_payload(
point_ids: List[PointId],
keys: List[str],
) -> "UpdateOperation":
"""
Delete payload fields from points.
Args:
point_ids: Point IDs.
keys: Payload field keys to delete.
"""
...
@staticmethod
def delete_payload_by_filter(
filter: Filter,
keys: List[str],
) -> "UpdateOperation":
"""
Delete payload fields from points matching a filter.
Args:
filter: Filter for points.
keys: Payload field keys to delete.
"""
...
@staticmethod
def clear_payload(point_ids: List[PointId]) -> "UpdateOperation":
"""
Clear all payload from points.
Args:
point_ids: Point IDs.
"""
...
@staticmethod
def clear_payload_by_filter(filter: Filter) -> "UpdateOperation":
"""
Clear all payload from points matching a filter.
Args:
filter: Filter for points.
"""
...
@staticmethod
def overwrite_payload(
point_ids: List[PointId],
payload: Payload,
key: Optional[str] = None,
) -> "UpdateOperation":
"""
Overwrite entire payload on points.
Args:
point_ids: Point IDs.
payload: New payload.
key: Optional nested key path.
"""
...
@staticmethod
def overwrite_payload_by_filter(
filter: Filter,
payload: Payload,
key: Optional[str] = None,
) -> "UpdateOperation":
"""
Overwrite payload on points matching a filter.
Args:
filter: Filter for points.
payload: New payload.
key: Optional nested key path.
"""
...
@staticmethod
def create_field_index(
field_name: str,
schema: Union[PayloadSchemaType, PayloadSchemaParams],
) -> "UpdateOperation":
"""
Create an index on a payload field.
Args:
field_name: Path to the payload field.
schema: Schema type or index parameters for the field.
"""
...
@staticmethod
def delete_field_index(field_name: str) -> "UpdateOperation":
"""
Delete an index from a payload field.
Args:
field_name: Path to the payload field.
"""
...
Reference in New Issue View Git Blame Copy Permalink