qdrant-edge

Qdrant Edge is a lightweight, embedded vector search engine that runs inside the application process — no separate server needed. Data is stored locally on disk, enabling low-latency search with or without internet connectivity.

⚠️ Beta: The API and functionality may change in future releases.

---

Quick Reference

| Topic | Details | |---|---| | Package | qdrant-edge-py (PyPI) | | Core class | EdgeShard | | Storage | Local directory on disk | | Embeddings | Use fastembed for on-device embedding generation | | Sync | Snapshots ↔ Qdrant server (full or partial) |

---

1. Installation

pip install qdrant-edge-py

# For on-device embeddings:
pip install qdrant-edge-py fastembed

---

2. Core Workflow

2.1 Create & Configure an EdgeShard

from pathlib import Path
from qdrant_edge import Distance, EdgeConfig, EdgeShard, VectorDataConfig

SHARD_DIR = "./qdrant-edge-data"
VECTOR_NAME = "my-vector"
VECTOR_DIM = 384  # must match embedding model output dimension

Path(SHARD_DIR).mkdir(parents=True, exist_ok=True)

config = EdgeConfig(
    vector_data={
        VECTOR_NAME: VectorDataConfig(
            size=VECTOR_DIM,
            distance=Distance.Cosine,  # or Distance.Dot, Distance.Euclid
        )
    }
)

shard = EdgeShard(SHARD_DIR, config)

2.2 Insert / Update Points

from qdrant_edge import Point, UpdateOperation

point = Point(
    id=1,                                     # int or UUID string
    vector={VECTOR_NAME: [0.1, 0.2, ...]},   # list of floats, length = VECTOR_DIM
    payload={"text": "hello", "category": "A"}
)

shard.update(UpdateOperation.upsert_points([point]))

2.3 Query (Nearest Neighbor Search)

from qdrant_edge import Query, QueryRequest

results = shard.query(
    QueryRequest(
        query=Query.Nearest([0.2, 0.1, ...], using=VECTOR_NAME),
        limit=10,
        with_vector=False,
        with_payload=True,
    )
)
# results is a list of ScoredPoint objects

2.4 Other Retrieval Methods

# Retrieve by ID
points = shard.retrieve(point_ids=[1, 2, 3], with_payload=True, with_vector=False)

# Scroll (paginate all points)
scroll_result = shard.scroll(limit=100, offset=None, with_payload=True, with_vector=False)

# Count
count = shard.count()

# Metadata
info = shard.info()

2.5 Persist & Reopen

shard.flush()   # force write to disk (optional, close() does this too)
shard.close()   # always call on shutdown

# Reopen existing shard (no config needed — loaded from disk)
shard = EdgeShard(SHARD_DIR)

---

3. On-Device Embeddings with FastEmbed

See references/fastembed.md for full details. Quick example:

from fastembed import TextEmbedding
from qdrant_edge import Point, UpdateOperation, Query, QueryRequest

MODELS_DIR = "./qdrant-edge-data/models"
MODEL_NAME = "BAAI/bge-small-en-v1.5"   # 384-dim, efficient for edge

# Pre-download (run once with internet):
TextEmbedding(model_name=MODEL_NAME, cache_dir=MODELS_DIR)

# At runtime (offline):
model = TextEmbedding(model_name=MODEL_NAME, cache_dir=MODELS_DIR, local_files_only=True)

# Insert
docs = ["Paris is the capital of France", "Berlin is in Germany"]
for i, (doc, emb) in enumerate(zip(docs, model.embed(docs))):
    shard.update(UpdateOperation.upsert_points([
        Point(id=i, vector={VECTOR_NAME: emb.tolist()}, payload={"text": doc})
    ]))

# Query
query_emb = list(model.embed(["European capitals"]))[0]
results = shard.query(QueryRequest(
    query=Query.Nearest(query_emb.tolist(), using=VECTOR_NAME),
    limit=5, with_payload=True, with_vector=False
))

Important: Always use local_files_only=True at runtime on edge devices to avoid network calls.

---

4. Data Synchronization

See references/synchronization.md for full details and code patterns.

Pattern A — Server → Edge (Initialize from Snapshot)

Download a server shard snapshot and unpack it into a local EdgeShard:

import requests, shutil, tempfile
from pathlib import Path
from qdrant_edge import EdgeShard

snapshot_url = f"{QDRANT_URL}/collections/{COLLECTION}/shards/0/snapshot"

with tempfile.TemporaryDirectory() as tmp:
    snap_path = Path(tmp) / "shard.snapshot"
    with requests.get(snapshot_url, headers={"api-key": API_KEY}, stream=True) as r:
        r.raise_for_status()
        snap_path.write_bytes(r.content)

    if Path(SHARD_DIR).exists():
        shutil.rmtree(SHARD_DIR)
    Path(SHARD_DIR).mkdir(parents=True)
    EdgeShard.unpack_snapshot(str(snap_path), SHARD_DIR)

shard = EdgeShard(SHARD_DIR)

Pattern B — Server → Edge (Incremental Partial Snapshot)

Only transfer changed segments — much more efficient for periodic updates:

manifest = shard.snapshot_manifest()
url = f"{QDRANT_URL}/collections/{COLLECTION}/shards/0/snapshot/partial/create"

with tempfile.TemporaryDirectory(dir=SHARD_DIR) as tmp:
    partial_path = Path(tmp) / "partial.snapshot"
    resp = requests.post(url, headers={"api-key": API_KEY}, json=manifest, stream=True)
    resp.raise_for_status()
    partial_path.write_bytes(resp.content)
    shard.update_from_snapshot(str(partial_path))

Pattern C — Edge → Server (Dual-Write + Queue)

Write to EdgeShard immediately; sync to server asynchronously via a queue:

from queue import Queue, Empty
from qdrant_client import QdrantClient, models

server = QdrantClient(url=QDRANT_URL, api_key=API_KEY)
upload_queue = Queue()

def write_point(id, vector, payload):
    # Local write — always succeeds offline
    shard.update(UpdateOperation.upsert_points([
        Point(id=id, vector={VECTOR_NAME: vector}, payload=payload)
    ]))
    # Enqueue for server sync
    upload_queue.put(models.PointStruct(id=id, vector={VECTOR_NAME: vector}, payload=payload))

def flush_to_server(batch_size=10):
    batch = []
    while len(batch) < batch_size:
        try:
            batch.append(upload_queue.get_nowait())
        except Empty:
            break
    if batch:
        server.upsert(collection_name=COLLECTION, points=batch)

---

5. Common Pitfalls & Best Practices

• Always call shard.close() on application shutdown to ensure data is flushed.
• local_files_only=True must be set when loading FastEmbed models on offline devices.
• Vector dimension must match: VectorDataConfig(size=...) must equal the embedding model's output dim.
• IDs: can be int or UUID str. Be consistent within a shard.
• Partial snapshots are preferred over full snapshots for periodic syncs — they transfer only changed segments.
• For production Edge→Server sync, use a persistent queue (e.g., SQLite, Redis) rather than an in-memory Queue to survive restarts.
• Multitenancy: one server collection can serve many edge devices via different shard IDs.

---

6. Reference Files

• references/fastembed.md — Detailed guide for on-device text & image embeddings with FastEmbed
• references/synchronization.md — Full synchronization patterns (Server↔Edge) with complete code

---

7. External Resources

• Qdrant Edge Docs
• Quickstart
• On-Device Embeddings
• Sync Patterns
• Sync Guide
• Python Examples (GitHub)
• qdrant-edge-py on PyPI