# Qdrant connector

> **CocoIndex v1.** This page documents CocoIndex **v1** — a ground-up redesign from v0. When writing code, ignore any v0 flow-builder DSL or deprecated decorators.
>
> Source: https://cocoindex.io/docs/connectors/qdrant/ · Docs index: https://cocoindex.io/docs/llms.txt · Agent skill: https://cocoindex.io/docs/skill.md
>
> v0→v1 quick map — if you reach for these v0 symbols, stop and use the v1 form: `@cocoindex.flow_def`/`FlowBuilder` → `coco.App` + a `@coco.fn` main function; `add_collector()`/`collect()`/`export()` → declare target states (`declare_row`, `declare_file`); `cocoindex.sources/functions/targets.*` → connector APIs (`localfs.walk_dir`, `coco.ops.*`, `postgres.declare_table_target`). Full mapping + API reference: https://cocoindex.io/docs/skill.md.

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](/docs/programming_guide/context#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](#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`](/docs/common_resources/vector_schema#vectorschemaprovider), a `ContextKey`, or an explicit `VectorSchema` to specify the vector dimension and dtype. See [Vector Schema](/docs/common_resources/vector_schema#vectorschemaprovider) 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](https://qdrant.tech/documentation/manage-data/points/#point-ids)):

- `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,
)
```