or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

client-setup.mdclustering-sharding.mdcollection-management.mdfastembed-integration.mdindex.mdindexing-optimization.mdsearch-query.mdsnapshots-backup.mdvector-operations.md
tile.json

tessl/pypi-qdrant-client

Client library for the Qdrant vector search engine

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/qdrant-client@1.15.x

To install, run

npx @tessl/cli install tessl/pypi-qdrant-client@1.15.0

index.mddocs/

Qdrant Client

A comprehensive Python client library for the Qdrant vector search engine, enabling developers to interact with Qdrant databases for vector similarity search, neural matching, and AI-powered search applications. The library offers both synchronous and asynchronous API methods with full type hints, supports both REST and gRPC protocols for optimal performance, includes a local mode for development without running a separate server, and provides helper methods for common operations like collection management and bulk data uploading.

Package Information

  • Package Name: qdrant-client
  • Language: Python
  • Installation: pip install qdrant-client
  • Optional Features: pip install qdrant-client[fastembed] for embedding generation

Core Imports

from qdrant_client import QdrantClient

For async operations:

from qdrant_client import AsyncQdrantClient

Import models and types:

from qdrant_client import models
from qdrant_client.models import Distance, VectorParams, PointStruct, Filter

Basic Usage

from qdrant_client import QdrantClient, models
import numpy as np

# Initialize client (local mode for development)
client = QdrantClient(":memory:")  # or QdrantClient(path="./qdrant_storage")

# For remote server
# client = QdrantClient(host="localhost", port=6333)

# Create a collection
client.create_collection(
    collection_name="my_collection",
    vectors_config=models.VectorParams(size=100, distance=models.Distance.COSINE),
)

# Insert vectors
vectors = np.random.rand(10, 100)
client.upsert(
    collection_name="my_collection",
    points=[
        models.PointStruct(
            id=idx,
            vector=vector.tolist(),
            payload={"category": "example", "number": idx}
        )
        for idx, vector in enumerate(vectors)
    ]
)

# Search for similar vectors
query_vector = np.random.rand(100)
hits = client.query_points(
    collection_name="my_collection",
    query=query_vector,
    limit=5
)

# Search with filtering
filtered_hits = client.query_points(
    collection_name="my_collection",
    query=query_vector,
    query_filter=models.Filter(
        must=[
            models.FieldCondition(
                key="number",
                range=models.Range(gte=3)
            )
        ]
    ),
    limit=5
)

Architecture

The Qdrant client library is built around several key components:

  • Client Classes: QdrantClient and AsyncQdrantClient provide the main interface
  • Transport Layer: Automatic REST/gRPC protocol handling with conversion between formats
  • Local Mode: QdrantLocal enables development without running a separate server
  • FastEmbed Integration: Optional automatic embedding generation for text and images
  • Model System: Comprehensive type definitions ensuring compatibility between REST and gRPC APIs
  • Authentication: Bearer token support for secure connections

Capabilities

Client Initialization & Connection

Core client setup, connection management, and configuration options supporting local mode, remote servers, and cloud instances with authentication.

class QdrantClient:
    def __init__(
        self,
        location: Optional[str] = None,
        url: Optional[str] = None,
        port: Optional[int] = 6333,
        grpc_port: Optional[int] = 6334,
        prefer_grpc: bool = False,
        https: Optional[bool] = None,
        api_key: Optional[str] = None,
        prefix: Optional[str] = None,
        timeout: Optional[float] = None,
        host: Optional[str] = None,
        path: Optional[str] = None,
        **kwargs
    ): ...

class AsyncQdrantClient:
    def __init__(self, **kwargs): ...

Client Setup

Collection Management

Collection creation, deletion, configuration, and metadata management including vector space configuration, optimization settings, and aliasing.

def create_collection(
    self,
    collection_name: str,
    vectors_config: Union[VectorParams, VectorsConfig],
    shard_number: Optional[int] = None,
    replication_factor: Optional[int] = None,
    write_consistency_factor: Optional[int] = None,
    on_disk_payload: Optional[bool] = None,
    hnsw_config: Optional[HnswConfig] = None,
    optimizers_config: Optional[OptimizersConfig] = None,
    wal_config: Optional[WalConfig] = None,
    quantization_config: Optional[QuantizationConfig] = None,
    init_from: Optional[InitFrom] = None,
    timeout: Optional[int] = None,
) -> bool: ...

def delete_collection(self, collection_name: str, timeout: Optional[int] = None) -> bool: ...
def get_collection(self, collection_name: str) -> CollectionInfo: ...
def get_collections(self) -> CollectionsResponse: ...
def collection_exists(self, collection_name: str) -> bool: ...

Collection Management

Vector Operations

Point insertion, updating, deletion, and retrieval operations including batch processing and bulk upload utilities.

def upsert(
    self,
    collection_name: str,
    points: Iterable[PointStruct],
    wait: bool = True,
    ordering: Optional[WriteOrdering] = None,
    shard_key_selector: Optional[ShardKeySelector] = None,
    **kwargs
) -> UpdateResult: ...

def upload_points(
    self,
    collection_name: str,
    points: Iterable[PointStruct],
    batch_size: int = 100,
    parallel: int = 1,
    max_retries: int = 3,
    wait: bool = True,
    shard_key_selector: Optional[ShardKeySelector] = None,
) -> None: ...

def delete(
    self,
    collection_name: str,
    points_selector: Union[PointIdsList, FilterSelector],
    wait: bool = True,
    ordering: Optional[WriteOrdering] = None,
    shard_key_selector: Optional[ShardKeySelector] = None,
    **kwargs
) -> UpdateResult: ...

Vector Operations

Search & Query

Vector similarity search, recommendations, discovery, and hybrid search capabilities with filtering and result ranking.

def query_points(
    self,
    collection_name: str,
    query: Union[QueryRequest, NumpyArray, QueryResponse, Document, str, List[float]],
    query_filter: Optional[Filter] = None,
    search_params: Optional[SearchParams] = None,
    limit: int = 10,
    offset: Optional[int] = None,
    with_payload: Union[bool, List[str], PayloadSelector] = True,
    with_vectors: Union[bool, List[str]] = False,
    score_threshold: Optional[float] = None,
    using: Optional[str] = None,
    timeout: Optional[int] = None,
    shard_key_selector: Optional[ShardKeySelector] = None,
    **kwargs
) -> QueryResponse: ...

def recommend(
    self,
    collection_name: str,
    positive: Optional[List[PointId]] = None,
    negative: Optional[List[PointId]] = None,
    query_filter: Optional[Filter] = None,
    search_params: Optional[SearchParams] = None,
    limit: int = 10,
    offset: int = 0,
    with_payload: Union[bool, List[str], PayloadSelector] = True,
    with_vectors: Union[bool, List[str]] = False,
    score_threshold: Optional[float] = None,
    using: Optional[str] = None,
    lookup_from: Optional[LookupLocation] = None,
    timeout: Optional[int] = None,
    shard_key_selector: Optional[ShardKeySelector] = None,
    **kwargs
) -> List[ScoredPoint]: ...

def discover(
    self,
    collection_name: str,
    target: Optional[PointId] = None,
    context: Optional[List[ContextExamplePair]] = None,
    query_filter: Optional[Filter] = None,
    search_params: Optional[SearchParams] = None,
    limit: int = 10,
    offset: int = 0,
    with_payload: Union[bool, List[str], PayloadSelector] = True,
    with_vectors: Union[bool, List[str]] = False,
    using: Optional[str] = None,
    lookup_from: Optional[LookupLocation] = None,
    timeout: Optional[int] = None,
    shard_key_selector: Optional[ShardKeySelector] = None,
    **kwargs
) -> List[ScoredPoint]: ...

Search & Query

FastEmbed Integration

Automatic embedding generation for text and images using the FastEmbed library, enabling semantic search without manual vector creation.

def get_embedding_size(self, model_name: str) -> int: ...

def upload_collection(
    self,
    collection_name: str,
    vectors: Union[
        Iterable[VectorStruct],
        Iterable[PointStruct], 
        Iterable[Record],
        Iterable[Document],
        Iterable[ImageDocument]
    ],
    ids: Optional[Iterable[PointId]] = None,
    batch_size: int = 100,
    parallel: int = 1,
    max_retries: int = 3,
    payload: Optional[Iterable[Payload]] = None,
    wait: bool = True,
    shard_key_selector: Optional[ShardKeySelector] = None,
) -> None: ...

def query_qdrant(
    self,
    collection_name: str,
    query: Union[str, Document, ImageDocument],
    query_filter: Optional[Filter] = None,
    limit: int = 10,
    search_params: Optional[SearchParams] = None,
    **kwargs
) -> List[QueryResponse]: ...

FastEmbed Integration

Indexing & Optimization

Payload field indexing, collection optimization, and performance tuning capabilities.

def create_payload_index(
    self,
    collection_name: str,
    field_name: str,
    field_schema: Optional[PayloadFieldSchema] = None,
    wait: bool = True,
    ordering: Optional[WriteOrdering] = None,
    **kwargs
) -> UpdateResult: ...

def delete_payload_index(
    self,
    collection_name: str,
    field_name: str,
    wait: bool = True,
    ordering: Optional[WriteOrdering] = None,
    **kwargs
) -> UpdateResult: ...

Indexing & Optimization

Clustering & Sharding

Distributed operation support including shard key management, cluster operations, and multi-tenant configurations.

def create_shard_key(
    self,
    collection_name: str,
    shard_key: ShardKey,
    shards_number: Optional[int] = None,
    replication_factor: Optional[int] = None,
    placement: Optional[List[int]] = None,
    timeout: Optional[int] = None,
    **kwargs
) -> bool: ...

def delete_shard_key(
    self,
    collection_name: str,
    shard_key: ShardKey,
    timeout: Optional[int] = None,
    **kwargs
) -> bool: ...

Clustering & Sharding

Snapshots & Backup

Collection and full database snapshot creation, management, and restoration capabilities for backup and disaster recovery.

def create_snapshot(
    self,
    collection_name: str,
    wait: bool = True,
    **kwargs
) -> SnapshotDescription: ...

def create_full_snapshot(self, wait: bool = True, **kwargs) -> SnapshotDescription: ...
def list_snapshots(self, collection_name: str, **kwargs) -> List[SnapshotDescription]: ...
def delete_snapshot(
    self,
    collection_name: str,
    snapshot_name: str,
    wait: bool = True,
    **kwargs
) -> bool: ...

Snapshots & Backup

Core Types

class Distance(str, Enum):
    COSINE = "Cosine"
    EUCLID = "Euclid" 
    DOT = "Dot"
    MANHATTAN = "Manhattan"

class VectorParams(BaseModel):
    size: int
    distance: Distance
    hnsw_config: Optional[HnswConfig] = None
    quantization_config: Optional[QuantizationConfig] = None
    on_disk: Optional[bool] = None

class PointStruct(BaseModel):
    id: PointId
    vector: Union[VectorStruct, Dict[str, VectorStruct]]
    payload: Optional[Payload] = None

class Filter(BaseModel):
    must: Optional[List[Condition]] = None
    must_not: Optional[List[Condition]] = None
    should: Optional[List[Condition]] = None
    min_should: Optional[MinShould] = None

class ScoredPoint(BaseModel):
    id: PointId
    version: int
    score: float
    payload: Optional[Payload] = None
    vector: Optional[Union[List[float], Dict[str, List[float]]]] = None
    shard_key: Optional[ShardKey] = None
    order_value: Optional[OrderValue] = None

# Type aliases
PointId = Union[str, int]
Payload = Dict[str, Any]
VectorStruct = List[float]