Multimodal Hybrid Search Pipeline

from mixpeek import Mixpeek

client = Mixpeek(api_key="YOUR_API_KEY", namespace="multimodal-search")

# 1. Two collections, one extractor each: video and images through multimodal_extractor, documents through text_extractor
media = client.buckets.create(
    bucket_name="media",
    bucket_schema={
        "properties": {
            "asset": {
                "type": "video",
            },
        },
    },
)
docs = client.buckets.create(
    bucket_name="documents",
    bucket_schema={
        "properties": {
            "body": {
                "type": "text",
            },
        },
    },
)
media_collection = client.collections.create(
    collection_name="media",
    source={"type": "bucket", "bucket_ids": [media["bucket_id"]]},
    feature_extractor={
        "feature_extractor_name": "multimodal_extractor",
        "version": "v1",
    },
)
text_collection = client.collections.create(
    collection_name="documents",
    source={"type": "bucket", "bucket_ids": [docs["bucket_id"]]},
    feature_extractor={
        "feature_extractor_name": "text_extractor",
        "version": "v1",
        "parameters": {
            "split_by": "paragraphs",
            "chunk_size": 3,
        },
    },
)

# 2. Upload and process
client.buckets.upload(
    media["bucket_id"],
    blobs=[{"property": "asset", "type": "video", "data": "s3://your-bucket/media/q3-review.mp4"}],
)
client.buckets.upload(
    docs["bucket_id"],
    blobs=[{"property": "body", "type": "text", "data": "s3://your-bucket/docs/q3-report.md"}],
)
client.collections.trigger(media_collection["collection_id"])
client.collections.trigger(text_collection["collection_id"])

# 3. Dense search over both collections plus BM25 over document text (it reads the namespace text payload indexes), fused with RRF, then a rerank
retriever = client.retrievers.create(
    retriever_name="hybrid-multimodal",
    collection_identifiers=["media", "documents"],
    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": 100,
                    },
                    {
                        "feature_uri": "mixpeek://text_extractor@v1/multilingual_e5_large_instruct_v1",
                        "query": {
                            "input_mode": "text",
                            "value": "{{INPUT.query}}",
                        },
                        "top_k": 100,
                    },
                    {
                        "feature_uri": "mixpeek://text_extractor@v1/multilingual_e5_large_instruct_v1",
                        "query": {
                            "input_mode": "text",
                            "value": "{{INPUT.query}}",
                        },
                        "top_k": 100,
                        "lexical": True,
                    },
                ],
                "fusion": "rrf",
                "final_top_k": 100,
            },
        },
        {
            "stage_name": "rerank",
            "stage_id": "rerank",
            "parameters": {
                "inference_name": "BAAI__bge_reranker_v2_m3",
                "query": "{{INPUT.query}}",
                "document_field": "text",
                "top_k": 20,
            },
        },
    ],
)

# 4. Search
results = client.retrievers.execute(
    retriever["retriever_id"],
    inputs={
    "query": "quarterly revenue growth chart",
},
)

for doc in results["documents"]:
    print(doc["document_id"], doc["score"])