Hierarchical Classification

import requests

from mixpeek import Mixpeek

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

# 1. One exemplar collection per node of the hierarchy
safe_bucket = client.buckets.create(
    bucket_name="safe-examples",
    bucket_schema={"properties": {"image": {"type": "image"}}},
)
safe = client.collections.create(
    collection_name="safe-examples",
    source={"type": "bucket", "bucket_ids": [safe_bucket["bucket_id"]]},
    feature_extractor={
        "feature_extractor_name": "multimodal_extractor",
        "version": "v1",
    },
)
edu_bucket = client.buckets.create(
    bucket_name="educational-examples",
    bucket_schema={"properties": {"image": {"type": "image"}}},
)
edu = client.collections.create(
    collection_name="educational-examples",
    source={"type": "bucket", "bucket_ids": [edu_bucket["bucket_id"]]},
    feature_extractor={
        "feature_extractor_name": "multimodal_extractor",
        "version": "v1",
    },
)
client.buckets.upload(
    safe_bucket["bucket_id"],
    blobs=[{"property": "image", "type": "image", "data": "s3://your-bucket/examples/safe/0001.jpg"}],
)
client.buckets.upload(
    edu_bucket["bucket_id"],
    blobs=[{"property": "image", "type": "image", "data": "s3://your-bucket/examples/educational/0001.jpg"}],
)
client.collections.trigger(safe["collection_id"])
client.collections.trigger(edu["collection_id"])

# 2. The retriever that matches a document against the exemplars
matcher = client.retrievers.create(
    retriever_name="example-matcher",
    collection_identifiers=["safe-examples", "educational-examples"],
    input_schema={"query": {"type": "text", "required": True}},
    stages=[
        {
            "stage_name": "search",
            "stage_id": "feature_search",
            "parameters": {
                "searches": [
                    {
                        "feature_uri": "mixpeek://multimodal_extractor@v1/vertex_multimodal_embedding",
                        "query": {"input_mode": "text", "value": "{{INPUT.query}}"},
                        "top_k": 3,
                    },
                ],
                "final_top_k": 3,
            },
        },
    ],
)

# 3. A hierarchical taxonomy: Educational sits under Safe. The SDK has no
#    taxonomies resource, so this part is REST.
taxonomy = requests.post(API + "/taxonomies", headers=HEADERS, json={
    "taxonomy_name": "content-classification",
    "config": {
        "taxonomy_type": "hierarchical",
        "retriever_id": matcher["retriever_id"],
        "input_mappings": [{"input_key": "query", "source_type": "payload", "path": "description"}],
        "hierarchical_nodes": [
            {"collection_id": safe["collection_id"], "label": "Safe"},
            {"collection_id": edu["collection_id"], "parent_collection_id": safe["collection_id"], "label": "Educational"},
        ],
    },
}).json()

# 4. The content collection writes one set of fields per tier onto every document
bucket = client.buckets.create(
    bucket_name="videos",
    bucket_schema={"properties": {"video": {"type": "video"}}},
)
videos = client.collections.create(
    collection_name="videos",
    source={"type": "bucket", "bucket_ids": [bucket["bucket_id"]]},
    feature_extractor={
        "feature_extractor_name": "multimodal_extractor",
        "version": "v1",
        "parameters": {
            "split_method": "scene",
            "run_video_description": True,
        },
    },
    taxonomy_applications=[{"taxonomy_id": taxonomy["taxonomy_id"], "execution_mode": "materialize", "hierarchical_enrichment_style": "full_chain"}],
)
client.buckets.upload(
    bucket["bucket_id"],
    blobs=[{"property": "video", "type": "video", "data": "s3://your-bucket/videos/chemistry-lesson.mp4"}],
)
client.collections.trigger(videos["collection_id"])

docs = client.documents.list(videos["collection_id"], page_size=20)
for doc in docs["results"]:
    print(doc["document_id"], {k: v for k, v in doc.items() if "tier" in k})