*Qdrant* connector - Cocoindex

The qdrant connector provides utilities for writing points to Qdrant vector databases, with support for single vectors, named dense vectors, sparse vectors, and multi-vector configurations.

python

from cocoindex.connectors import qdrant

:::note[Dependencies] This connector requires additional dependencies. Install with:

bash

pip install cocoindex[qdrant]

:::

Connection setup

create_client() creates a Qdrant client connection with optional gRPC support.

python

def create_client(
    url: str,
    *,
    prefer_grpc: bool = True,
    **kwargs: Any,
) -> QdrantClient

Parameters:

url — Qdrant server URL (e.g., "http://localhost:6333").
prefer_grpc — Whether to prefer gRPC over HTTP (default: True).
**kwargs — Additional arguments passed directly to QdrantClient.

Returns: A Qdrant client instance.

Example:

python

client = qdrant.create_client("http://localhost:6333")

As target

The qdrant connector provides target state APIs for writing points to collections. CocoIndex tracks what points should exist and automatically handles upserts and deletions.

Declaring target states

Setting up a connection

Create a ContextKey[QdrantClient] to identify your Qdrant client, then provide it in your lifespan:

:::note The key name is load-bearing across runs — it's the stable identity CocoIndex uses to track managed collections. See ContextKey as stable identity before renaming. :::

python

from qdrant_client import QdrantClient
import cocoindex as coco

QDRANT_DB = coco.ContextKey[QdrantClient]("my_vectors")

@coco.lifespan
async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]:
    client = qdrant.create_client(QDRANT_URL)
    builder.provide(QDRANT_DB, client)
    yield

Collections (parent state)

Declares a collection as a target state. Returns a CollectionTarget for declaring points.

python

def declare_collection_target(
    db: ContextKey[QdrantClient],
    collection_name: str,
    schema: CollectionSchema,
    *,
    managed_by: Literal["system", "user"] = "system",
) -> CollectionTarget[coco.PendingS]

Parameters:

db — A ContextKey[QdrantClient] identifying the Qdrant client to use.
collection_name — Name of the collection.
schema — Schema definition specifying vector configurations (see Collection Schema).
managed_by — Whether CocoIndex manages the collection lifecycle ("system") or assumes it exists ("user").

Returns: A pending CollectionTarget. Use the convenience wrapper await qdrant.mount_collection_target(QDRANT_DB, collection_name, schema) to resolve.

Points (child states)

Once a CollectionTarget is resolved, declare points to be upserted using qdrant.PointStruct, which is an alias of qdrant_client.http.models.PointStruct:

python

def CollectionTarget.declare_point(
    self,
    point: qdrant.PointStruct,
) -> None

Parameters:

point — A qdrant.PointStruct (alias of qdrant_client.http.models.PointStruct) containing:
- id — Point ID (str, int, or UUID)
- vector — Vector data (single vector or dict of named vectors)
- payload — Optional metadata as a JSON-serializable dict

Collection schema

Define vector configurations for a collection using CollectionSchema. Unlike row-oriented databases, Qdrant uses a point-oriented model where each point has schemaless payload and one or more vectors with predefined dimensions.

python

class CollectionSchema:
    @classmethod
    async def create(
        cls,
        vectors: QdrantVectorDef
        | dict[str, QdrantVectorDef | QdrantSparseVectorDef]
        | None = None,
    ) -> CollectionSchema

Parameters:

vectors — Either:
- A single QdrantVectorDef for an unnamed dense vector
- A dict mapping vector names to QdrantVectorDef (dense) or QdrantSparseVectorDef (sparse)

Dense and sparse vectors share one namespace in Qdrant (the server rejects duplicate names across the two kinds), so both live in the same dict — a name collision is impossible to express. Sparse vectors are always named.

QdrantVectorDef

Specifies vector configuration including dimension, distance metric, and multi-vector settings:

python

class QdrantVectorDef(NamedTuple):
    schema: VectorSchemaProvider | MultiVectorSchemaProvider
    distance: Literal["cosine", "dot", "euclid"] = "cosine"
    multivector_comparator: Literal["max_sim"] = "max_sim"

Parameters:

schema — A VectorSchemaProvider or MultiVectorSchemaProvider that defines vector dimensions
distance — Distance metric for similarity search (default: "cosine")
multivector_comparator — Comparator for multi-vector fields (only applies to MultiVectorSchemaProvider)

QdrantSparseVectorDef

Specifies a named sparse vector configuration:

python

class QdrantSparseVectorDef(NamedTuple):
    modifier: Literal["idf"] | None = None

Sparse vectors use Qdrant's native sparse-vector storage and dot-product scoring. The optional modifier="idf" enables Qdrant's IDF modifier for sparse vectors.

Single (unnamed) vector

For collections with a single unnamed vector:

python

from cocoindex.ops.sentence_transformers import SentenceTransformerEmbedder

embedder = SentenceTransformerEmbedder("sentence-transformers/all-MiniLM-L6-v2")

schema = await qdrant.CollectionSchema.create(
    vectors=qdrant.QdrantVectorDef(schema=embedder)
)

Points use the vector directly:

python

point = qdrant.PointStruct(
    id=123,  # unsigned int or UUID — see "Point IDs" below
    vector=embedding.tolist(),  # Single vector
    payload={"text": "...", "metadata": {...}},
)

Named vectors

For collections with multiple named vectors:

python

from cocoindex.resources.schema import VectorSchema
import numpy as np

schema = await qdrant.CollectionSchema.create(
    vectors={
        "text_embedding": qdrant.QdrantVectorDef(
            schema=VectorSchema(dtype=np.float32, size=384),
            distance="cosine",
        ),
        "image_embedding": qdrant.QdrantVectorDef(
            schema=VectorSchema(dtype=np.float32, size=512),
            distance="dot",
        ),
    }
)

Points use a dict of vectors:

python

point = qdrant.PointStruct(
    id=123,  # unsigned int or UUID — see "Point IDs" below
    vector={
        "text_embedding": text_vec.tolist(),
        "image_embedding": image_vec.tolist(),
    },
    payload={"text": "...", "metadata": {...}},
)

Dense + sparse vectors

For hybrid retrieval, declare a named dense vector and a named sparse vector in the same collection:

python

from qdrant_client.http import models as qdrant_models
from cocoindex.resources.schema import VectorSchema
import numpy as np

schema = await qdrant.CollectionSchema.create(
    vectors={
        "dense": qdrant.QdrantVectorDef(
            schema=VectorSchema(dtype=np.float32, size=384),
            distance="cosine",
        ),
        "sparse": qdrant.QdrantSparseVectorDef(modifier="idf"),
    },
)

Points put both vector types in PointStruct.vector. The sparse vector is a native Qdrant SparseVector, not payload:

python

point = qdrant.PointStruct(
    id=123,
    vector={
        "dense": dense_embedding.tolist(),
        "sparse": qdrant_models.SparseVector(
            indices=sparse_indices,
            values=sparse_values,
        ),
    },
    payload={"text": text},
)
target.declare_point(point)

VectorSchemaProvider

The schema field of QdrantVectorDef accepts a VectorSchemaProvider, a ContextKey, or an explicit VectorSchema to specify the vector dimension and dtype. See Vector Schema for details.

Multi-vector support

For multi-vector configurations (multiple vectors per point stored together):

python

from cocoindex.resources.schema import MultiVectorSchema, VectorSchema
import numpy as np

schema = await qdrant.CollectionSchema.create(
    vectors=qdrant.QdrantVectorDef(
        schema=MultiVectorSchema(
            vector_schema=VectorSchema(dtype=np.float32, size=384)
        ),
        multivector_comparator="max_sim",
    )
)

Distance metrics

The distance parameter in QdrantVectorDef specifies the similarity metric:

"cosine" — Cosine similarity (default, normalized dot product)
"dot" — Dot product similarity
"euclid" — Euclidean distance (L2)

Example: single vector

python

from qdrant_client import QdrantClient
import cocoindex as coco
from cocoindex.connectors import qdrant
from cocoindex.ops.sentence_transformers import SentenceTransformerEmbedder
from typing import AsyncIterator

QDRANT_URL = "http://localhost:6333"
QDRANT_DB = coco.ContextKey[QdrantClient]("main_vectors")

embedder = SentenceTransformerEmbedder("sentence-transformers/all-MiniLM-L6-v2")

@coco.lifespan
async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]:
    client = qdrant.create_client(QDRANT_URL)
    builder.provide(QDRANT_DB, client)
    yield

@coco.fn
async def process_document(
    doc_id: str,
    text: str,
    target: qdrant.CollectionTarget,
) -> None:
    embedding = await embedder.embed(text)

    point = qdrant.PointStruct(
        id=doc_id,
        vector=embedding.tolist(),
        payload={"text": text},
    )
    target.declare_point(point)

@coco.fn
async def app_main() -> None:
    # Declare collection target state
    collection = await qdrant.mount_collection_target(
        QDRANT_DB,
        "documents",
        await qdrant.CollectionSchema.create(
            vectors=qdrant.QdrantVectorDef(schema=embedder)
        ),
    )

    # Declare points
    for doc_id, text in documents:
        await coco.mount(
            coco.component_subpath("doc", doc_id),
            process_document,
            doc_id,
            text,
            collection,
        )

Example: named vectors

python

from cocoindex.resources.schema import VectorSchema
import numpy as np

@coco.fn
async def app_main() -> None:
    collection = await qdrant.mount_collection_target(
        QDRANT_DB,
        "multimodal_docs",
        await qdrant.CollectionSchema.create(
            vectors={
                "text": qdrant.QdrantVectorDef(
                    schema=text_embedder,
                    distance="cosine",
                ),
                "image": qdrant.QdrantVectorDef(
                    schema=VectorSchema(dtype=np.float32, size=512),
                    distance="dot",
                ),
            }
        ),
    )

    # Declare points with named vectors
    for doc in documents:
        point = qdrant.PointStruct(
            id=doc.id,
            vector={
                "text": doc.text_embedding.tolist(),
                "image": doc.image_embedding.tolist(),
            },
            payload={"title": doc.title, "url": doc.url},
        )
        collection.declare_point(point)

Example: dense + sparse hybrid vectors

python

from qdrant_client.http import models as qdrant_models
from cocoindex.resources.schema import VectorSchema
import numpy as np

@coco.fn
async def app_main() -> None:
    collection = await qdrant.mount_collection_target(
        QDRANT_DB,
        "hybrid_docs",
        await qdrant.CollectionSchema.create(
            vectors={
                "dense": qdrant.QdrantVectorDef(
                    schema=VectorSchema(dtype=np.float32, size=384),
                    distance="cosine",
                ),
                "sparse": qdrant.QdrantSparseVectorDef(modifier="idf"),
            },
        ),
    )

    for chunk in chunks:
        point = qdrant.PointStruct(
            id=chunk.id,
            vector={
                "dense": chunk.dense_embedding.tolist(),
                "sparse": qdrant_models.SparseVector(
                    indices=chunk.sparse_indices,
                    values=chunk.sparse_values,
                ),
            },
            payload={"text": chunk.text},
        )
        collection.declare_point(point)

Point IDs

Qdrant only accepts two point ID types — this is enforced by the Qdrant server itself (see Qdrant's Point IDs documentation):

int — unsigned 64-bit integers (0 <= id < 2**64)
str — UUID strings, in any standard textual form (hyphenated, 32-character hex, or URN)

uuid.UUID instances are also accepted and converted to strings. declare_point validates the ID eagerly and raises ValueError for anything else, rather than letting the write fail later at the Qdrant server.

To derive a stable point ID from an arbitrary string key (e.g. a document path or chunk key), use a deterministic UUID:

python

import uuid

point_id = str(uuid.uuid5(uuid.NAMESPACE_URL, f"doc/{chunk_key}"))

Payloads

Point payloads are schemaless JSON objects. Any JSON-serializable Python data structure can be used:

python

payload = {
    "text": "Document content",
    "metadata": {
        "author": "Alice",
        "tags": ["machine-learning", "nlp"],
        "published": "2024-01-15",
    },
    "stats": {
        "views": 1500,
        "likes": 42,
    },
}

Vector search

The connector focuses on writing points to Qdrant. For vector search, use the Qdrant client directly:

python

from qdrant_client.http import models as qdrant_models

# Get the registered client
client = qdrant.create_client("http://localhost:6333")

# Perform search
results = client.query_points(
    collection_name="documents",
    query=query_embedding.tolist(),
    limit=10,
    with_payload=True,
)

for point in results.points:
    print(f"Score: {point.score}, ID: {point.id}")
    print(f"Payload: {point.payload}")

For named vectors:

python

results = client.query_points(
    collection_name="documents",
    query=query_embedding.tolist(),
    using="text",  # Search using the "text" vector
    limit=10,
)

For hybrid dense + sparse retrieval, use Qdrant's query API with prefetches and reciprocal rank fusion:

python

results = client.query_points(
    collection_name="hybrid_docs",
    prefetch=[
        qdrant_models.Prefetch(
            query=query_dense_embedding.tolist(),
            using="dense",
            limit=100,
        ),
        qdrant_models.Prefetch(
            query=qdrant_models.SparseVector(
                indices=query_sparse_indices,
                values=query_sparse_values,
            ),
            using="sparse",
            limit=100,
        ),
    ],
    query=qdrant_models.FusionQuery(fusion=qdrant_models.Fusion.RRF),
    limit=10,
    with_payload=True,
)