Incremental Embedding Updates

from qdrant_client import QdrantClient, models

# Replace url and api_key with your own from https://cloud.qdrant.io
client = QdrantClient(
    url="https://xyz-example.qdrant.io:6333",
    api_key="<your-api-key>",
    cloud_inference=True
)

import { QdrantClient, Schemas } from "@qdrant/js-client-rest";

// Replace url and apiKey with your own from https://cloud.qdrant.io
const client = new QdrantClient({
    url: "https://xyz-example.qdrant.io:6333",
    apiKey: "<your-api-key>",
});

// Replace the URL and API key with your own from https://cloud.qdrant.io
let client = Qdrant::from_url("https://xyz-example.qdrant.io:6334")
    .api_key("<your-api-key>")
    .build()?;

// Replace the host and API key with your own from https://cloud.qdrant.io
static final QdrantClient client =
    new QdrantClient(
        QdrantGrpcClient.newBuilder("xyz-example.qdrant.io", 6334, true)
            .withApiKey("<your-api-key>")
            .build());

// Replace the host and API key with your own from https://cloud.qdrant.io
var client = new QdrantClient(
	host: "xyz-example.qdrant.io",
	https: true,
	apiKey: "<your-api-key>"
);

// Replace the host and API key with your own from https://cloud.qdrant.io
client, err := qdrant.NewClient(&qdrant.Config{
	Host:   "xyz-example.qdrant.io",
	APIKey: "<your-api-key>",
	UseTLS: true,
})

page https://qdrant.tech/documentation/tutorials-operations/secure-qdrant/ (url)
├── section  #prerequisites (anchor)
│   └── chunk_num 0  "Prerequisites - Docker and Docker Compose installed..."
├── section  #secure-a-self-hosted-qdrant-instance (anchor)
│   ├── chunk_num 0  "Secure a Self-Hosted Qdrant Instance | Time: 45 min..."
│   └── chunk_num 1  "Secure a Self-Hosted Qdrant Instance > Qdrant Cloud..."
├── section  #step-1-start-an-unsecured-instance (anchor)
│   └── chunk_num 0  "Step 1: Start an Unsecured Instance Start Qdrant..."
├── section  #step-2-enable-tls (anchor)
│   └── chunk_num 0  "Step 2: Enable TLS Unencrypted connections allow..."
└── ...

normalize(text):
    - remove invisible characters (zero-width spaces, byte-order mark, soft hyphen)
    - collapse any whitespace run into a single space
    - ...

MODEL = "sentence-transformers/all-MiniLM-L6-v2"
PIPELINE = "docs-prep-pipeline-v1"
COLLECTION = "docs-sync-tutorial"

client.create_collection(
    COLLECTION,
    vectors_config=models.VectorParams(
        size=384,  # all-MiniLM-L6-v2 output dimension
        distance=models.Distance.COSINE,
    ),
    metadata={"embedding_model": MODEL, "pipeline_version": PIPELINE},
)

const MODEL = "sentence-transformers/all-MiniLM-L6-v2";
const PIPELINE = "docs-prep-pipeline-v1";
const COLLECTION = "docs-sync-tutorial";

await client.createCollection(COLLECTION, {
    vectors: {
        size: 384, // all-MiniLM-L6-v2 output dimension
        distance: "Cosine",
    },
});

await client.updateCollection(COLLECTION, {
    metadata: { embedding_model: MODEL, pipeline_version: PIPELINE },
});

const MODEL: &str = "sentence-transformers/all-MiniLM-L6-v2";
const PIPELINE: &str = "docs-prep-pipeline-v1";
const COLLECTION: &str = "docs-sync-tutorial";

let mut metadata: HashMap<String, Value> = HashMap::new();
metadata.insert("embedding_model".to_string(), json!(MODEL));
metadata.insert("pipeline_version".to_string(), json!(PIPELINE));

client
    .create_collection(
        CreateCollectionBuilder::new(COLLECTION)
            .vectors_config(VectorParamsBuilder::new(
                384, // all-MiniLM-L6-v2 output dimension
                Distance::Cosine,
            ))
            .metadata(metadata),
    )
    .await?;

static final String MODEL = "sentence-transformers/all-MiniLM-L6-v2";
static final String PIPELINE = "docs-prep-pipeline-v1";
static final String COLLECTION = "docs-sync-tutorial";

static void createCollection() throws Exception {
    client.createCollectionAsync(
        CreateCollection.newBuilder()
            .setCollectionName(COLLECTION)
            .setVectorsConfig(
                VectorsConfig.newBuilder()
                    .setParams(
                        VectorParams.newBuilder()
                            .setSize(384) // all-MiniLM-L6-v2 output dimension
                            .setDistance(Distance.Cosine)
                            .build())
                    .build())
            .putAllMetadata(
                Map.of(
                    "embedding_model", value(MODEL),
                    "pipeline_version", value(PIPELINE)))
            .build()).get();
}

var MODEL = "sentence-transformers/all-MiniLM-L6-v2";
var PIPELINE = "docs-prep-pipeline-v1";
var COLLECTION = "docs-sync-tutorial";

await client.CreateCollectionAsync(
	collectionName: COLLECTION,
	vectorsConfig: new VectorParams
	{
		Size = 384, // all-MiniLM-L6-v2 output dimension
		Distance = Distance.Cosine
	},
	metadata: new()
	{
		["embedding_model"] = MODEL,
		["pipeline_version"] = PIPELINE
	}
);

MODEL := "sentence-transformers/all-MiniLM-L6-v2"
PIPELINE := "docs-prep-pipeline-v1"
COLLECTION := "docs-sync-tutorial"

client.CreateCollection(context.Background(), &qdrant.CreateCollection{
	CollectionName: COLLECTION,
	VectorsConfig: qdrant.NewVectorsConfig(&qdrant.VectorParams{
		Size:     384, // all-MiniLM-L6-v2 output dimension
		Distance: qdrant.Distance_Cosine,
	}),
	Metadata: qdrant.NewValueMap(map[string]any{
		"embedding_model":  MODEL,
		"pipeline_version": PIPELINE,
	}),
})

def check_gate():
    # compare this pipeline's constants against what the collection records about itself
    meta = client.get_collection(COLLECTION).config.metadata or {}

    if meta.get("embedding_model") != MODEL or meta.get("pipeline_version") != PIPELINE:
        raise RuntimeError(f"collection was built by {meta}: full re-embed into a fresh collection required")

async function checkGate() {
    // compare this pipeline's constants against what the collection records about itself
    const meta = ((await client.getCollection(COLLECTION)).config.metadata ??
        {}) as Record<string, unknown>;

    if (meta.embedding_model !== MODEL || meta.pipeline_version !== PIPELINE) {
        throw new Error(`collection was built by ${JSON.stringify(meta)}: full re-embed into a fresh collection required`);
    }
}

async fn check_gate(client: &Qdrant) -> anyhow::Result<()> {
    // compare this pipeline's constants against what the collection records about itself
    let meta = client
        .collection_info(COLLECTION)
        .await?
        .result
        .and_then(|info| info.config)
        .map(|config| config.metadata)
        .unwrap_or_default();

    if meta.get("embedding_model").and_then(|v| v.as_str()).map(String::as_str) != Some(MODEL)
        || meta.get("pipeline_version").and_then(|v| v.as_str()).map(String::as_str)
            != Some(PIPELINE)
    {
        anyhow::bail!(
            "collection was built by {meta:?}: full re-embed into a fresh collection required"
        );
    }
    Ok(())
}

static void checkGate() throws Exception {
    // compare this pipeline's constants against what the collection records about itself
    Map<String, Value> meta =
        client.getCollectionInfoAsync(COLLECTION).get().getConfig().getMetadataMap();

    Value model = meta.get("embedding_model");
    Value pipeline = meta.get("pipeline_version");
    if (model == null || !MODEL.equals(model.getStringValue())
            || pipeline == null || !PIPELINE.equals(pipeline.getStringValue())) {
        throw new RuntimeException(
            "collection was built by " + meta + ": full re-embed into a fresh collection required");
    }
}

async Task CheckGate()
{
	// compare this pipeline's constants against what the collection records about itself
	var meta = (await client.GetCollectionInfoAsync(COLLECTION)).Config.Metadata;
	var model = meta.GetValueOrDefault("embedding_model")?.StringValue;
	var pipeline = meta.GetValueOrDefault("pipeline_version")?.StringValue;

	if (model != MODEL || pipeline != PIPELINE)
		throw new InvalidOperationException(
			$"collection was built by {model}/{pipeline}: full re-embed into a fresh collection required");
}

checkGate := func() {
	// compare this pipeline's constants against what the collection records about itself
	info, err := client.GetCollectionInfo(context.Background(), COLLECTION)
	meta := info.GetConfig().GetMetadata()

	if meta["embedding_model"].GetStringValue() != MODEL || meta["pipeline_version"].GetStringValue() != PIPELINE {
		panic(fmt.Sprintf("collection was built by %v: full re-embed into a fresh collection required", meta))
	}
}

import hashlib
import uuid
from datetime import datetime, timezone

def content_hash(text):
    return hashlib.sha256(text.encode()).hexdigest()

def point_id(url, anchor, num):
    # NAMESPACE_URL is a fixed constant uuid5 requires; it marks the input as a URL-like name
    return str(uuid.uuid5(uuid.NAMESPACE_URL, f"{url}#{anchor}::{num}"))

def prepare_chunks_for_sync(chunks):
    """Derive both values (and the section address) for every raw chunk."""
    out = []
    for c in chunks:
        text = normalize(c["text"])
        out.append({
            **c,
            "text": text,
            "section_url": f"{c['url']}#{c['anchor']}" if c["anchor"] else c["url"],
            "content_hash": content_hash(text),
            "point_id": point_id(c["url"], c["anchor"], c["chunk_num"]),
        })
    return out

import { createHash } from "node:crypto";

type RawChunk = { url: string; anchor: string; chunk_num: number; text: string };
type SyncChunk = RawChunk & { section_url: string; content_hash: string; point_id: string };

function contentHash(text: string): string {
    return createHash("sha256").update(text).digest("hex");
}

// NAMESPACE_URL is a fixed constant name-based (v5) UUIDs require; it marks the input as a URL-like name
function pointId(url: string, anchor: string, num: number): string {
    // Qdrant accepts any well-formed UUID as a point ID:
    // hash the address, format the digest as a UUID, and the same address always yields the same ID
    const hex = createHash("sha256").update(`${url}#${anchor}::${num}`).digest("hex");
    return `${hex.slice(0, 8)}-${hex.slice(8, 12)}-${hex.slice(12, 16)}-${hex.slice(16, 20)}-${hex.slice(20, 32)}`;
}

// Derive both values (and the section address) for every raw chunk.
function prepareChunksForSync(chunks: RawChunk[]): SyncChunk[] {
    return chunks.map((c) => {
        const text = normalize(c.text);
        return {
            ...c,
            text,
            section_url: c.anchor ? `${c.url}#${c.anchor}` : c.url,
            content_hash: contentHash(text),
            point_id: pointId(c.url, c.anchor, c.chunk_num),
        };
    });
}

fn content_hash(text: &str) -> String {
    Sha256::digest(text.as_bytes())
        .iter()
        .map(|byte| format!("{byte:02x}"))
        .collect()
}

fn point_id(url: &str, anchor: &str, num: u32) -> String {
    // NAMESPACE_URL is a fixed constant uuid5 requires; it marks the input as a URL-like name
    uuid::Uuid::new_v5(
        &uuid::Uuid::NAMESPACE_URL,
        format!("{url}#{anchor}::{num}").as_bytes(),
    )
    .to_string()
}

/// Derive both values (and the section address) for every raw chunk.
fn prepare_chunks_for_sync(chunks: &[Chunk]) -> Vec<Chunk> {
    chunks
        .iter()
        .map(|c| {
            let text = normalize(&c.text);
            Chunk {
                text: text.clone(),
                section_url: if c.anchor.is_empty() {
                    c.url.clone()
                } else {
                    format!("{}#{}", c.url, c.anchor)
                },
                content_hash: content_hash(&text),
                point_id: point_id(&c.url, &c.anchor, c.chunk_num),
                ..c.clone()
            }
        })
        .collect()
}

static String contentHash(String text) throws Exception {
    byte[] digest = MessageDigest.getInstance("SHA-256")
        .digest(text.getBytes(StandardCharsets.UTF_8));
    return String.format("%064x", new BigInteger(1, digest));
}

static String pointId(String url, String anchor, int num) {
    // name-based UUID (version 3); the same address always yields the same ID
    return UUID.nameUUIDFromBytes(
        (url + "#" + anchor + "::" + num).getBytes(StandardCharsets.UTF_8)).toString();
}

// Derive both values (and the section address) for every raw chunk.
static List<Chunk> prepareChunksForSync(List<Chunk> chunks) throws Exception {
    List<Chunk> out = new ArrayList<>();
    for (Chunk c : chunks) {
        String text = normalize(c.text);
        Chunk prepared = new Chunk(c.url, c.anchor, c.chunkNum, text);
        prepared.sectionUrl = !c.anchor.isEmpty() ? c.url + "#" + c.anchor : c.url;
        prepared.contentHash = contentHash(text);
        prepared.pointId = pointId(c.url, c.anchor, c.chunkNum);
        out.add(prepared);
    }
    return out;
}

string ContentHash(string text) =>
	Convert.ToHexString(SHA256.HashData(Encoding.UTF8.GetBytes(text))).ToLowerInvariant();

// Qdrant accepts any well-formed UUID as a point ID:
// a Guid built from the first 16 bytes of the address hash, so the same address always yields the same ID
string PointIdFor(string url, string anchor, int num) =>
	new Guid(SHA256.HashData(Encoding.UTF8.GetBytes($"{url}#{anchor}::{num}")).AsSpan(0, 16)).ToString();

// Derive both values (and the section address) for every raw chunk.
List<Chunk> PrepareChunksForSync(List<Chunk> chunks)
{
	var prepared = new List<Chunk>();
	foreach (var c in chunks)
	{
		var text = Normalize(c.Text);
		prepared.Add(c with
		{
			Text = text,
			SectionUrl = c.Anchor != "" ? $"{c.Url}#{c.Anchor}" : c.Url,
			ContentHash = ContentHash(text),
			PointId = PointIdFor(c.Url, c.Anchor, c.ChunkNum),
		});
	}
	return prepared;
}

contentHash := func(text string) string {
	sum := sha256.Sum256([]byte(text))
	return hex.EncodeToString(sum[:])
}

pointID := func(url, anchor string, num int) string {
	// NewSHA1 with a namespace is UUIDv5; NameSpaceURL is a fixed constant it requires,
	// marking the input as a URL-like name
	return uuid.NewSHA1(uuid.NameSpaceURL, []byte(fmt.Sprintf("%s#%s::%d", url, anchor, num))).String()
}

// derive both values (and the section address) for every raw chunk
prepareChunksForSync := func(chunks []Chunk) []Chunk {
	out := make([]Chunk, 0, len(chunks))
	for _, c := range chunks {
		c.Text = normalize(c.Text)
		c.SectionURL = c.URL
		if c.Anchor != "" {
			c.SectionURL = c.URL + "#" + c.Anchor
		}
		c.ContentHash = contentHash(c.Text)
		c.PointID = pointID(c.URL, c.Anchor, c.ChunkNum)
		out = append(out, c)
	}
	return out
}

point ID: 2ff5204a-0353-5991-... # UUID(url + "#" + anchor + "::" + chunk_num)
text (to vectorize):         Prerequisites - Docker and Docker Compose...
content_hash:                27d55e75b962f1d5... # sha256(text)

for field in ("content_hash", "url", "section_url"):
    client.create_payload_index(COLLECTION, field, models.PayloadSchemaType.KEYWORD)

for (const field of ["content_hash", "url", "section_url"]) {
    await client.createPayloadIndex(COLLECTION, {
        field_name: field,
        field_schema: "keyword",
    });
}

for field in ["content_hash", "url", "section_url"] {
    client
        .create_field_index(CreateFieldIndexCollectionBuilder::new(
            COLLECTION,
            field,
            FieldType::Keyword,
        ))
        .await?;
}

static void createPayloadIndexes() throws Exception {
    for (String field : List.of("content_hash", "url", "section_url")) {
        client.createPayloadIndexAsync(
            COLLECTION, field, PayloadSchemaType.Keyword, null, null, null, null).get();
    }
}

foreach (var field in new[] { "content_hash", "url", "section_url" })
	await client.CreatePayloadIndexAsync(COLLECTION, field, PayloadSchemaType.Keyword);

for _, field := range []string{"content_hash", "url", "section_url"} {
	client.CreateFieldIndex(context.Background(), &qdrant.CreateFieldIndexCollection{
		CollectionName: COLLECTION,
		FieldName:      field,
		FieldType:      qdrant.FieldType_FieldTypeKeyword.Enum(),
	})
}

client.upsert(COLLECTION, points=[
    models.PointStruct(
        id=c["point_id"],
        vector=models.Document(text=c["text"], model=MODEL),
        payload=payload(c),
    )
    for c in prepare_chunks_for_sync(CHUNKS)
], wait=True)

await client.upsert(COLLECTION, {
    points: prepareChunksForSync(CHUNKS).map((c) => ({
        id: c.point_id,
        vector: { text: c.text, model: MODEL },
        payload: payload(c),
    })),
    wait: true,
});

let points: Vec<PointStruct> = prepare_chunks_for_sync(&chunks)
    .iter()
    .map(|c| {
        Ok(PointStruct::new(
            c.point_id.clone(),
            Document::new(&c.text, MODEL),
            payload(c, None)?,
        ))
    })
    .collect::<anyhow::Result<_>>()?;

client
    .upsert_points(UpsertPointsBuilder::new(COLLECTION, points).wait(true))
    .await?;

static void populate() throws Exception {
    List<PointStruct> points = new ArrayList<>();
    for (Chunk c : prepareChunksForSync(CHUNKS)) {
        points.add(
            PointStruct.newBuilder()
                .setId(id(UUID.fromString(c.pointId)))

                .setVectors(
                    vectors(
                        vector(
                            Document.newBuilder()
                                .setText(c.text)
                                .setModel(MODEL)
                                .build())))
                .putAllPayload(payload(c, null))
                .build());
    }
    client.upsertAsync(COLLECTION, points).get();
}

await client.UpsertAsync(
	collectionName: COLLECTION,
	points: PrepareChunksForSync(CHUNKS).Select(c => new PointStruct
	{
		Id = new PointId { Uuid = c.PointId },
		Vectors = new Document { Text = c.Text, Model = MODEL },
		Payload = { Payload(c) },
	}).ToList(),
	wait: true
);

var points []*qdrant.PointStruct
for _, c := range prepareChunksForSync(CHUNKS) {
	points = append(points, &qdrant.PointStruct{
		Id: qdrant.NewID(c.PointID),

		Vectors: qdrant.NewVectorsDocument(&qdrant.Document{Text: c.Text, Model: MODEL}),
		Payload: qdrant.NewValueMap(payload(c, "")),
	})
}
client.Upsert(context.Background(), &qdrant.UpsertPoints{
	CollectionName: COLLECTION,
	Points:         points,
	Wait:           qdrant.PtrOf(true),
})

incoming chunk
├─ ID found in the collection?
│   ├─ yes: fingerprint equal?
│   │   ├─ yes -> unchanged: the point stays as is
│   │   └─ no  -> content changed: re-embed in place
│   └─ no: identical fingerprint under another ID?
│       ├─ yes -> address changed: reuse the vector, create a new point
│       └─ no  -> new: embed and insert a new point
└─ stored point whose ID is absent from the incoming list
                -> gone: delete (last, after all writes)

def split_by_state(latest_chunks):
    """Compare the incoming chunk list to the collection: who is unchanged, changed, or unknown."""
    incoming = {c["point_id"]: c for c in latest_chunks}

    stored = {}
    points = client.retrieve(
        COLLECTION,
        ids=list(incoming),
        with_payload=["content_hash"],
        with_vectors=False,
    )
    for p in points:
        stored[str(p.id)] = p.payload["content_hash"]

    unchanged, content_changed, unknown_ids = [], [], []
    for pid, c in incoming.items():
        if stored.get(pid) == c["content_hash"]:
            unchanged.append(c)
        elif pid in stored:
            content_changed.append(c)
        else:
            unknown_ids.append(c)

    return incoming, unchanged, content_changed, unknown_ids

incoming_ids, unchanged, content_changed, unknown_ids = split_by_state(LATEST_CHUNKS)

// Compare the incoming chunk list to the collection: who is unchanged, changed, or unknown.
async function splitByState(latestChunks: SyncChunk[]) {
    const incoming = new Map(latestChunks.map((c) => [c.point_id, c]));

    const stored = new Map<string, string>();
    const points = await client.retrieve(COLLECTION, {
        ids: [...incoming.keys()],
        with_payload: ["content_hash"],
        with_vector: false,
    });
    for (const p of points) {
        stored.set(String(p.id), p.payload?.content_hash as string);
    }

    const unchanged: SyncChunk[] = [];
    const contentChanged: SyncChunk[] = [];
    const unknownIds: SyncChunk[] = [];
    for (const [pid, c] of incoming) {
        if (stored.get(pid) === c.content_hash) {
            unchanged.push(c);
        } else if (stored.has(pid)) {
            contentChanged.push(c);
        } else {
            unknownIds.push(c);
        }
    }

    return { incoming, unchanged, contentChanged, unknownIds };
}

const { incoming, unchanged, contentChanged, unknownIds } = await splitByState(LATEST_CHUNKS);

/// Compare the incoming chunk list to the collection: who is unchanged, changed, or unknown.
async fn split_by_state(
    client: &Qdrant,
    latest_chunks: &[Chunk],
) -> anyhow::Result<(HashMap<String, Chunk>, Vec<Chunk>, Vec<Chunk>, Vec<Chunk>)> {
    let incoming: HashMap<String, Chunk> = latest_chunks
        .iter()
        .map(|c| (c.point_id.clone(), c.clone()))
        .collect();

    let ids: Vec<PointId> = incoming.keys().map(|id| id.as_str().into()).collect();
    let points = client
        .get_points(
            GetPointsBuilder::new(COLLECTION, ids)
                .with_payload(PayloadIncludeSelector::new(vec!["content_hash".to_string()]))
                .with_vectors(false),
        )
        .await?;

    let mut stored: HashMap<String, String> = HashMap::new();
    for p in points.result {
        let hash = p.get("content_hash").as_str().cloned();
        if let (Some(PointIdOptions::Uuid(id)), Some(hash)) =
            (p.id.and_then(|i| i.point_id_options), hash)
        {
            stored.insert(id, hash);
        }
    }

    let (mut unchanged, mut content_changed, mut unknown_ids) =
        (Vec::new(), Vec::new(), Vec::new());
    for (pid, c) in &incoming {
        if stored.get(pid) == Some(&c.content_hash) {
            unchanged.push(c.clone());
        } else if stored.contains_key(pid) {
            content_changed.push(c.clone());
        } else {
            unknown_ids.push(c.clone());
        }
    }

    Ok((incoming, unchanged, content_changed, unknown_ids))
}

let (incoming_ids, unchanged, content_changed, unknown_ids) =
    split_by_state(&client, &latest_chunks).await?;

static class SyncState {
    Map<String, Chunk> incoming = new LinkedHashMap<>();
    List<Chunk> unchanged = new ArrayList<>();
    List<Chunk> contentChanged = new ArrayList<>();
    List<Chunk> unknownIds = new ArrayList<>();
}

// Compare the incoming chunk list to the collection: who is unchanged, changed, or unknown.
static SyncState splitByState(List<Chunk> latestChunks) throws Exception {
    SyncState state = new SyncState();
    for (Chunk c : latestChunks) {
        state.incoming.put(c.pointId, c);
    }

    Map<String, String> stored = new HashMap<>();
    var points = client.retrieveAsync(
        COLLECTION,
        state.incoming.keySet().stream()
            .map(pid -> id(UUID.fromString(pid)))
            .collect(Collectors.toList()),
        WithPayloadSelectorFactory.include(List.of("content_hash")),
        WithVectorsSelectorFactory.enable(false),
        null).get();
    for (var p : points) {
        stored.put(p.getId().getUuid(), p.getPayloadMap().get("content_hash").getStringValue());
    }

    for (Map.Entry<String, Chunk> e : state.incoming.entrySet()) {
        String pid = e.getKey();
        Chunk c = e.getValue();
        if (c.contentHash.equals(stored.get(pid))) {
            state.unchanged.add(c);
        } else if (stored.containsKey(pid)) {
            state.contentChanged.add(c);
        } else {
            state.unknownIds.add(c);
        }
    }

    return state;
}