Multimodal RAG Pipeline

from mixpeek import Mixpeek

client = Mixpeek(api_key="YOUR_API_KEY", namespace="rag-kb")

# 1. A bucket for the knowledge base, and a collection that splits every document
#    into paragraphs and embeds each one
bucket = client.buckets.create(
    bucket_name="knowledge-base",
    bucket_schema={"properties": {"document": {"type": "text"}}},
)
collection = client.collections.create(
    collection_name="knowledge-base",
    source={"type": "bucket", "bucket_ids": [bucket["bucket_id"]]},
    feature_extractor={
        "feature_extractor_name": "text_extractor",
        "version": "v1",
        "parameters": {
            "split_by": "paragraphs",
        },
    },
)

# 2. Upload and process
client.buckets.upload(
    bucket["bucket_id"],
    blobs=[{"property": "document", "type": "text", "data": "s3://your-bucket/knowledge-base/board-minutes-2026-08.md"}],
)
client.collections.trigger(collection["collection_id"])

# 3. The retriever finds passages, reranks them and writes the answer. BM25 reads the
#    namespace's text payload indexes, so declare one on text before relying on it.
retriever = client.retrievers.create(
    retriever_name="rag-kb",
    collection_identifiers=["knowledge-base"],
    input_schema={"query": {"type": "text", "required": True}},
    stages=[
        {
            "stage_name": "search",
            "stage_id": "feature_search",
            "parameters": {
                "searches": [
                    {
                        "feature_uri": "mixpeek://text_extractor@v1/multilingual_e5_large_instruct_v1",
                        "query": {"input_mode": "text", "value": "{{INPUT.query}}"},
                        "top_k": 50,
                    },
                    {
                        "feature_uri": "mixpeek://text_extractor@v1/multilingual_e5_large_instruct_v1",
                        "query": {"input_mode": "text", "value": "{{INPUT.query}}"},
                        "top_k": 50,
                        "lexical": True,
                    },
                ],
                "fusion": "rrf",
                "final_top_k": 30,
            },
        },
        {
            "stage_name": "rerank",
            "stage_id": "rerank",
            "parameters": {
                "inference_name": "BAAI__bge_reranker_v2_m3",
                "query": "{{INPUT.query}}",
                "document_field": "text",
                "top_k": 8,
            },
        },
        {
            "stage_name": "answer",
            "stage_id": "summarize",
            "parameters": {
                "prompt": "Answer the question {{INPUT.query}} using only these numbered passages, and cite the passage numbers. {{DOCUMENTS}}",
                "provider": "google",
                "model_name": "gemini-2.5-flash-lite",
                "content_field": "text",
                "output_field": "answer",
                "include_sources": True,
            },
        },
    ],
)

results = client.retrievers.execute(retriever["retriever_id"], inputs={"query": "What were the key decisions from the last board meeting?"})
answer = results["documents"][0]
print(answer["answer"])

# The summary document lists the documents it read; fetch them for citations
for i, document_id in enumerate(answer.get("source_document_ids") or [], 1):
    source = client.documents.get(collection["collection_id"], document_id)
    print(f"[{i}]", source.get("root_object_id"), source.get("text"))