Mosaic SDK

Production-grade, modular RAG platform and SDK.

Mosaic lets you assemble every stage of a Retrieval-Augmented Generation pipeline from interchangeable, independently testable modules. Swap any component without touching adjacent layers.

You pick each pipeline component, or let the Advisor recommend a complete pipeline from a few plain questions. Either way, Mosaic validates compatibility (embedding dimensions, hybrid/sparse requirements, multimodal flags, etc.) before anything runs, and says how to fix what doesn't fit.

Documentation: the docs site lives in site/ (Starlight). Run it with cd site && npm ci && npm run dev. Its reference pages are generated from the code (uv run python scripts/gen_site_reference.py), and a test fails when they are out of date.

Documents with images, charts/graphs, and tabular data are explicitly supported — not plain-text-only.

Quick Start

Every stage of the pipeline is a component chosen by id. This example runs fully offline (demo embedder and LLM); swap ids to use real models — e.g. "embedder": "bge_m3" or "openai_small", "llm": "anthropic" with ANTHROPIC_API_KEY set.

import asyncio

from mosaic import RAGPipeline

config = {
    "ingestion": [{"id": "markdown", "path": "./docs/"}],
    "chunking": {"id": "recursive_hierarchical", "chunk_size": 512},
    "embedder": {"id": "demo", "dim": 384},
    "vector_store": {"id": "chroma", "collection": "my_kb"},
    "sparse_index": "bm25",
    "retrieval": {"id": "hybrid", "top_k": 20},
    "llm": "demo",
}


async def main() -> None:
    pipeline = RAGPipeline.from_dict(config)
    report = await pipeline.ingest()
    print(f"{report.documents} documents, {report.chunks} chunks, {len(report.failed)} failed")
    response = await pipeline.aquery("What is Python used for?")
    print(response.answer)


asyncio.run(main())

The same config can live in a YAML file (RAGPipeline.from_config("mosaic.yaml")); see mosaic.yaml for every stage. Mosaic checks that your choices work together before building anything — no API keys needed to validate:

from mosaic.core.validation import validate_config

result = validate_config(
    {"embedder": "openai_small", "vector_store": "chroma", "retrieval": "hybrid"}
)
print(result.compatible)  # False
print(result.errors[0])   # hybrid retrieval needs a sparse source: select a sparse_index, ...
assert not result.compatible

Not sure what to pick? Describe what you need and the Advisor recommends a complete pipeline, with a reason for every choice. It is deterministic and offline, and its recommendations always validate:

from mosaic.advisor import Brief, available_secrets, recommend
from mosaic.core.validation import validate_config

rec = recommend(Brief(
    use_case="support",          # what people will ask about
    sources=["markdown"],         # where the answers are kept
    local_only=True,              # data must stay on your own machines
    exact_terms=True,             # questions include product codes or error messages
    available_secrets=available_secrets(),  # names of API keys that are set, never values
))
for decision in rec.decisions:
    print(f"{decision.stage_label}: {decision.label} - {decision.reasons[0]}")
assert validate_config(rec.config).compatible

You can also wire components yourself; every argument is a port (mosaic.core.interfaces):

import asyncio

from mosaic import RAGPipeline
from mosaic.chunking import RecursiveHierarchicalChunker
from mosaic.ingestion import MarkdownLoader
from mosaic.parsing import RoutingParser
from mosaic.retrieval import HybridRetriever
from mosaic.storage import BM25Index, ChromaVectorStore
from mosaic.testing.demo import DemoEmbedder, DemoLLM  # or BGEEmbedder, AnthropicLLM, ...

embedder = DemoEmbedder(dim=384)
vector_store = ChromaVectorStore(collection="my_kb_explicit")
bm25 = BM25Index()

pipeline = RAGPipeline(
    ingestors=[MarkdownLoader(path="./docs/")],
    parser=RoutingParser(),
    chunker=RecursiveHierarchicalChunker(chunk_size=512),
    embedder=embedder,
    indexes=[vector_store, bm25],  # every index is written on ingest
    retriever=HybridRetriever(embedder, vector_store, sparse_index=bm25),
    llm=DemoLLM(),
)


async def main() -> None:
    await pipeline.ingest()
    response = await pipeline.aquery("Who created Python?")
    print(response.answer, response.sources)


asyncio.run(main())

Use with coding agents

Coding agents (Claude Code, Cursor, Copilot and others) can design, check, run and measure Mosaic pipelines through an MCP server or a command line with JSON output, instead of writing pipeline code from memory. AGENTS.md (generated from the component registry) explains the workflow.

pip install "mosaic-rag[agent]"

# MCP server (stdio); tools may only touch files under --root
claude mcp add mosaic -- mosaic-mcp --root .

# or the command line (add --json to any command)
mosaic recommend --brief '{"use_case": "support", "local_only": true}' --profile ./docs -o mosaic.yaml
mosaic validate mosaic.yaml            # issues carry codes, config paths and fixes
mosaic fix mosaic.yaml --write         # apply every unambiguous fix
mosaic scaffold mosaic.yaml --kind fastapi -o app/
mosaic ingest mosaic.yaml --path ./docs --demo
mosaic query mosaic.yaml "How do I return an item?" --demo
mosaic eval mosaic.yaml cases.yaml --path ./docs --demo

For other MCP clients, register the command mosaic-mcp with the arguments --root <project>. Tools: list_components, describe_component, recommend_pipeline, validate_config, apply_fixes, review_config, explain_config, profile_corpus, next_experiments, ingest, query, evaluate, compare_configs, scaffold_app.

Architecture

• Platform: pipeline builder, compatibility validation, ingest/query API, web app (platform/web/)
• SDK: building blocks + RAGPipeline orchestrator; Ports & Adapters via mosaic.core.interfaces
• Advisor: optional recommendations from a plain-language brief (and a corpus profile), fit review, retrieval metrics and config comparison — separate from compatibility validation (mosaic.advisor)
• Copilot (Phase 2+): natural-language interface for discussing and editing pipeline config

The core seams (mosaic.core):

Adding a component touches no file in mosaic.core: implement the port, declare it with @register(stage, id, config=..., capabilities=..., ui=...) in a module under the stage's package (or an installed plugin exposing the mosaic.components entry point), and it shows up in the catalog, the validator, the builder and the contract test suites (tests/contracts/).

The roadmap is in docs/superpowers/plans/2026-10-28-roadmap.md. See Mosaic_SDK_Document.md for the full specification and docs/superpowers/specs/2026-10-16-core-seams-design.md for the core seams design.

Platform UI

A React + TypeScript app in platform/web/. It opens on Start, a guided first run for people who have documents but no terminal: upload files, answer a few plain questions (where the answers are kept, how many documents and which languages are filled in from the files), build the recommended pipeline in demo mode or for real, and ask a first question.

Four workspaces follow: Build (compose a pipeline from components, live compatibility check, settings, YAML import/export, apply in demo or real mode), Data (upload files, run ingest, review reports), Ask (answers with evidence) and Evaluate (test questions with the file that answers each, scored by how often that file comes back; the Advisor's suggested changes compared side by side, best first; smoke checks), plus a Components reference. On Build, Help me choose asks the same questions and recommends a pipeline with a reason for each choice; nothing changes until you click Use this pipeline. The API's OpenAPI document is the contract (platform/web/src/api/openapi.json, regenerate with uv run python scripts/export_openapi.py).

Uploads are stored per tenant (X-Tenant-Id) under MOSAIC_UPLOAD_DIR (default .mosaic/uploads), with a size limit per file (MOSAIC_UPLOAD_MAX_MB, default 20) and a file limit per tenant (MOSAIC_UPLOAD_MAX_FILES, default 500). Queries sent with a tenant only find that tenant's documents, but the header is not authenticated yet: anyone can name any tenant, and a request without one searches every tenant's documents. Sign-in is planned (Stage 4 of the roadmap).

uv sync --extra all --extra dev

# Production: build once, the API serves it at /
(cd platform/web && npm ci && npm run build)
uv run uvicorn mosaic.api.main:app --port 8000        # http://localhost:8000

# Development: API and hot-reloading app side by side
uv run uvicorn mosaic.api.main:app --reload --port 8000
(cd platform/web && npm run dev)                      # http://localhost:5173, proxies the API

# Optionally start with a pipeline loaded
MOSAIC_CONFIG=mosaic.dev.yaml uv run uvicorn mosaic.api.main:app --port 8000

Checks: npm run check (types, lint, unit tests) and npm run e2e (browser journeys, including the guided first run, plus accessibility audits; needs a build). Without a build the API still runs and / explains how to build.

API docs: http://localhost:8000/docs

Installation

Mosaic is published on PyPI as mosaic-rag (the name mosaic is taken); you still import it as mosaic.

pip install "mosaic-rag[embedding,storage,api]"
# or full install:
pip install "mosaic-rag[all]"

Open-source embedders (BGE, E5, sentence-transformers) install with the embedding extra; managed APIs (OpenAI, Cohere) use the same extra with your API keys.