Semantic Drift Detection

import time
import requests

from mixpeek import Mixpeek

client = Mixpeek(api_key="YOUR_API_KEY", namespace="drift-monitor")
API = "https://api.mixpeek.com/v1"
HEADERS = {"Authorization": "Bearer YOUR_API_KEY", "X-Namespace": "drift-monitor"}

# 1. The training images, embedded
bucket = client.buckets.create(
    bucket_name="training-data",
    bucket_schema={"properties": {"image": {"type": "image"}}},
)
collection = client.collections.create(
    collection_name="training-data",
    source={"type": "bucket", "bucket_ids": [bucket["bucket_id"]]},
    feature_extractor={
        "feature_extractor_name": "multimodal_extractor",
        "version": "v1",
    },
)
client.buckets.upload(
    bucket["bucket_id"],
    blobs=[{"property": "image", "type": "image", "data": "s3://your-bucket/training/batch-01/img-0001.jpg"}],
)
client.collections.trigger(collection["collection_id"])

# 2. One cluster definition, run whenever you want a snapshot of the embedding space.
#    The SDK has no clusters resource, so this part is REST.
cluster = requests.post(API + "/clusters", headers=HEADERS, json={
    "cluster_name": "training-drift",
    "collection_ids": [collection["collection_id"]],
    "cluster_type": "vector",
    "vector_config": {
        "feature_uris": ["mixpeek://multimodal_extractor@v1/vertex_multimodal_embedding"],
        "clustering_method": "hdbscan",
        "algorithm_params": {"min_cluster_size": 20},
    },
}).json()


def snapshot():
    task = requests.post(API + "/clusters/" + cluster["cluster_id"] + "/execute", headers=HEADERS, json={}).json()
    while client.tasks.get(task["task_id"])["status"] not in ("COMPLETED", "COMPLETED_WITH_ERRORS", "FAILED"):
        time.sleep(15)
    return requests.get(API + "/clusters/" + cluster["cluster_id"] + "/executions", headers=HEADERS).json()


baseline = snapshot()

# 3. After new data lands and is processed, snapshot again and compare the runs
client.buckets.upload(
    bucket["bucket_id"],
    blobs=[{"property": "image", "type": "image", "data": "s3://your-bucket/training/batch-02/img-0001.jpg"}],
)
client.collections.trigger(collection["collection_id"])
current = snapshot()

for name in ("noise_ratio", "silhouette_score", "mean_cosine_to_centroid"):
    before = (baseline.get("metrics") or {}).get(name)
    after = (current.get("metrics") or {}).get(name)
    print(name, before, "->", after)
print("clusters:", baseline["num_clusters"], "->", current["num_clusters"])

# Every run stays in the history
history = requests.post(API + "/clusters/" + cluster["cluster_id"] + "/executions/list", headers=HEADERS, json={}).json()
for run in history["results"]:
    print(run["run_id"], run["num_clusters"], (run.get("metrics") or {}).get("noise_ratio"))