Skip to main content

Semantic Search for Multi-Tenant Products

Vector search in multi-tenant products has one rule that trumps all others: a user in tenant A must never see results from tenant B, even if tenant B has a more semantically similar document.

RushDB's semantic search enforces this at the storage layer. Every project is an isolated namespace. When you call db.ai.search(), the search is always scoped to your project. There is no global vector index, no shared ANN pool, no cross-project leakage.

This tutorial shows you exactly how the scoping works, how to add structured filters on top of semantic ranking, and how to build a multi-tenant search endpoint that is correct by construction.

How project-scoped search works​

The prefilter step narrows candidates to records in the current project (via a Cypher MATCH/WHERE clause) before any similarity computation runs. Exact cosine similarity is then applied to the prefiltered set.

This means:

  • Tenant isolation is guaranteed by the query engine, not by application-layer filtering
  • Adding a where clause narrows the candidate set further but does not change the isolation guarantee
  • Results always carry a __score (0–1) indicating cosine similarity

Step 1: Ingest multi-tenant content​

In a real multi-tenant product, each tenant has its own RushDB project and its own API key. For this tutorial, the "tenant isolation" is the project itself.

from rushdb import RushDB

db = RushDB("TENANT_PROJECT_API_KEY", base_url="https://api.rushdb.com/api/v1")

db.records.import_json({
    "label": "ARTICLE",
    "data": [
        {
            "title": "Reducing Infrastructure Costs with Spot Instances",
            "body": "A practical guide to lowering cloud bills using preemptible compute.",
            "category": "infrastructure",
            "authorId": "u-101",
            "publishedAt": "2025-01-15"
        },
        {
            "title": "Query Planner Internals",
            "body": "How modern databases choose execution plans when they get it wrong.",
            "category": "databases",
            "authorId": "u-202",
            "publishedAt": "2025-02-10"
        },
        {
            "title": "Incident Response Automation",
            "body": "Automating runbooks, page routing, and post-incident review with LLMs.",
            "category": "operations",
            "authorId": "u-101",
            "publishedAt": "2025-03-01"
        }
    ]
})

Step 2: Create an embedding index​

Semantic search requires an embedding index on the property you want to search. Create it once per label/property combination.

index = db.ai.indexes.create({
    "label": "ARTICLE",
    "propertyName": "body",
    "sourceType": "managed"
})

print("Index status:", index.data["status"])

Step 3: Poll until the index is ready​

Backfilling vectors takes time proportional to record count. Poll the stats endpoint until indexedRecords equals totalRecords.

import time

def wait_for_index(index_id: str, interval: float = 3.0) -> None:
    while True:
        stats = db.ai.indexes.stats(index_id)
        total = stats.data["totalRecords"]
        indexed = stats.data["indexedRecords"]
        print(f"Indexed {indexed} / {total}")
        if indexed >= total:
            break
        time.sleep(interval)

wait_for_index(index.data["id"])

No where filter — returns the top matches across all ARTICLE records in this project, ranked by cosine similarity.

results = db.ai.search({
    "query": "how to reduce cloud spending",
    "propertyName": "body",
    "labels": ["ARTICLE"],
    "limit": 5
})

for r in results.data:
    print(f"{r.score:.3f}  {r['title']}")

Step 5: Scope results with structured filters​

Add a where clause to narrow candidates before cosine ranking. This is the correct pattern for per-user or per-category search in a product.

# Scoped to one author
author_results = db.ai.search({
    "query": "automation and reliability",
    "propertyName": "body",
    "labels": ["ARTICLE"],
    "where": {"authorId": "u-101"},
    "limit": 5
})

# Scoped to categories
category_results = db.ai.search({
    "query": "database performance",
    "propertyName": "body",
    "labels": ["ARTICLE"],
    "where": {"category": {"$in": ["databases", "infrastructure"]}},
    "limit": 10
})

# Date-bounded
recent_results = db.ai.search({
    "query": "incident response",
    "propertyName": "body",
    "labels": ["ARTICLE"],
    "where": {"publishedAt": {"$gte": {"$year": 2025, "$month": 2, "$day": 1}}},
    "limit": 10
})

Step 6: Paginate over semantic results​

Use skip to page through ranked results.

def search_page(query: str, page: int, page_size: int = 10):
    return db.ai.search({
        "query": query,
        "propertyName": "body",
        "labels": ["ARTICLE"],
        "skip": page * page_size,
        "limit": page_size
    })

page0 = search_page("cloud infrastructure", 0)
page1 = search_page("cloud infrastructure", 1)

Step 7: Using external vectors (BYOV)​

If you manage your own embeddings, create an external index and push vectors inline at write time.

# Create external index
db.ai.indexes.create({
    "label": "ARTICLE",
    "propertyName": "body",
    "sourceType": "external",
    "dimensions": 1536,
    "similarityFunction": "cosine"
})

# Write record with inline vector
db.records.create(
    "ARTICLE",
    {
        "title": "Distributed Tracing at Scale",
        "body": "How to instrument services for trace collection.",
        "category": "observability",
    },
    vectors=[
        {
            "propertyName": "body",
            "vector": [/* 1536-dimension float array */]
        }
    ]
)

# Search with query vector
results = db.ai.search({
    "queryVector": [/* same 1536-dim array */],
    "propertyName": "body",
    "labels": ["ARTICLE"],
    "sourceType": "external",
    "limit": 5
})

The multi-tenant API endpoint pattern​

Here is a minimal handler that exposes project-scoped semantic search for a product:

# POST /api/search (e.g. Flask / FastAPI handler)
from rushdb import RushDB

def search_handler(tenant_api_key: str, query: str, category: str = None, page: int = 0, page_size: int = 10):
    db = RushDB(tenant_api_key)

    where = {}
    if category:
        where["category"] = category

    results = db.ai.search({
        "query": query,
        "propertyName": "body",
        "labels": ["ARTICLE"],
        "where": where if where else None,
        "skip": page * page_size,
        "limit": page_size
    })

    return {
        "results": [
            {"id": r.id, "title": r.get("title"), "score": r.score, "category": r.get("category")}
            for r in results.data
        ],
        "total": results.total
    }

Because RushDB(tenantApiKey) is project-scoped, there is no application-layer filtering needed. The isolation guarantee comes from the storage layer.

Production caveat​

Project-scoped isolation is enforced at the API key level. If you accidentally reuse the same API key across tenants (for example by sharing a single project for all tenants), no isolation exists. Each tenant must have a separate project with a separate API key. The recommended architecture is one RushDB project per tenant.

Next steps​