Metadata filter

Vector search that respects structure you already know. Prefix a query with a

README Try it Source

Showcase — Metadata filter

Vector search that respects structure you already know. Prefix a query with a topic and a colon — billing: how do I get my money back — and the store keeps only topic=billing documents, then ranks those by meaning. Filter narrows the candidates; embeddings rank what's left. Valid topics: account, billing, shipping, returns.

Run

bash bootstrap-secrets.sh              # reads ../../../../.env, writes secrets/
docker compose up --build              # default: PROVIDER=openai

Open http://localhost:3000. Run against Gemini with PROVIDER=gemini docker compose up --build.

What's where

backend/store.py — the shared VectorStore; search() takes a where metadata predicate applied before ranking.
backend/ai_openai.py / backend/ai_gemini.py — docs with topic metadata and the "topic: query" parsing.
frontend/app/page.tsx — query box + filtered, ranked results.

Stop

docker compose down

Run locally

Download the project as a ZIP and run it with Docker. Brings up a FastAPI backend + Next.js frontend on localhost:3000.

Download metadata-filter.zip

unzip metadata-filter.zip
cd metadata-filter
bash bootstrap-secrets.sh   # one-time: pulls API keys into ./secrets
docker compose up --build   # default provider: openai
# or:  PROVIDER=gemini docker compose up --build

Type some input, pick a provider, and run the same code shown in Source against the live API. Sign-in required.

Input

Provider

Output

The same modules the Run button hits. The whole project (frontend, Dockerfile, compose) is in the ZIP under README.

`backend/ai_openai.py`

"""Showcase 2 (OpenAI): metadata-filtered similarity search.

Pure vector search ignores structure you already know — which customer, which
date range, which category. Real stores let you filter by metadata AND rank by
similarity. Type "billing: how do I get my money back" and the store keeps only
billing docs, then ranks those by meaning. Filter narrows; embeddings rank.
"""
from openai import OpenAI

from store import VectorStore

_client = OpenAI()

_MODEL = "text-embedding-3-small"

_DOCS = [
    {"id": "a1", "text": "Reset your password from the login page's 'Forgot password' link.", "meta": {"topic": "account"}},
    {"id": "a2", "text": "Turn on two-factor authentication under Account > Security.", "meta": {"topic": "account"}},
    {"id": "b1", "text": "Refunds return to your original card within 5-7 business days.", "meta": {"topic": "billing"}},
    {"id": "b2", "text": "Update your card or billing address under Account > Billing.", "meta": {"topic": "billing"}},
    {"id": "b3", "text": "We accept Visa, Mastercard, and American Express.", "meta": {"topic": "billing"}},
    {"id": "s1", "text": "Standard shipping takes 3-5 business days; express takes 1-2.", "meta": {"topic": "shipping"}},
    {"id": "s2", "text": "Track your package from the link in the shipping confirmation email.", "meta": {"topic": "shipping"}},
    {"id": "r1", "text": "Return unopened items within 30 days for a full refund.", "meta": {"topic": "returns"}},
    {"id": "r2", "text": "Opened items are eligible for store credit within 14 days.", "meta": {"topic": "returns"}},
]

_TOPICS = {"account", "billing", "shipping", "returns"}
_store: VectorStore | None = None


def _embed(texts: list[str]) -> list[list[float]]:
    return [item.embedding for item in _client.embeddings.create(model=_MODEL, input=texts).data]


def _get_store() -> VectorStore:
    global _store
    if _store is None:
        _store = VectorStore(_embed)
        _store.add(_DOCS)
    return _store


def run(text: str) -> str:
    # Optional "topic: query" prefix selects a metadata filter.
    prefix, sep, rest = text.partition(":")
    if sep and prefix.strip().lower() in _TOPICS and rest.strip():
        where, query = {"topic": prefix.strip().lower()}, rest.strip()
    else:
        where, query = None, text.strip()

    hits = _get_store().search(query, k=4, where=where)
    if not hits:
        return f"No documents match filter {where}. Valid topics: {', '.join(sorted(_TOPICS))}."
    header = f"filter={where or 'none'}  query={query!r}\n\n"
    rows = [f"{score:.3f}  [{item['meta']['topic']}]  {item['text']}" for score, item in hits]
    return header + "\n".join(rows)

`backend/ai_gemini.py`

"""Showcase 2 (Gemini): metadata-filtered similarity search.

Same docs, same filter-then-rank, Gemini's embedding model.
"""
import os

from google import genai

from store import VectorStore

_client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])

_MODEL = "gemini-embedding-001"

_DOCS = [
    {"id": "a1", "text": "Reset your password from the login page's 'Forgot password' link.", "meta": {"topic": "account"}},
    {"id": "a2", "text": "Turn on two-factor authentication under Account > Security.", "meta": {"topic": "account"}},
    {"id": "b1", "text": "Refunds return to your original card within 5-7 business days.", "meta": {"topic": "billing"}},
    {"id": "b2", "text": "Update your card or billing address under Account > Billing.", "meta": {"topic": "billing"}},
    {"id": "b3", "text": "We accept Visa, Mastercard, and American Express.", "meta": {"topic": "billing"}},
    {"id": "s1", "text": "Standard shipping takes 3-5 business days; express takes 1-2.", "meta": {"topic": "shipping"}},
    {"id": "s2", "text": "Track your package from the link in the shipping confirmation email.", "meta": {"topic": "shipping"}},
    {"id": "r1", "text": "Return unopened items within 30 days for a full refund.", "meta": {"topic": "returns"}},
    {"id": "r2", "text": "Opened items are eligible for store credit within 14 days.", "meta": {"topic": "returns"}},
]

_TOPICS = {"account", "billing", "shipping", "returns"}
_store: VectorStore | None = None


def _embed(texts: list[str]) -> list[list[float]]:
    return [e.values for e in _client.models.embed_content(model=_MODEL, contents=texts).embeddings]


def _get_store() -> VectorStore:
    global _store
    if _store is None:
        _store = VectorStore(_embed)
        _store.add(_DOCS)
    return _store


def run(text: str) -> str:
    prefix, sep, rest = text.partition(":")
    if sep and prefix.strip().lower() in _TOPICS and rest.strip():
        where, query = {"topic": prefix.strip().lower()}, rest.strip()
    else:
        where, query = None, text.strip()

    hits = _get_store().search(query, k=4, where=where)
    if not hits:
        return f"No documents match filter {where}. Valid topics: {', '.join(sorted(_TOPICS))}."
    header = f"filter={where or 'none'}  query={query!r}\n\n"
    rows = [f"{score:.3f}  [{item['meta']['topic']}]  {item['text']}" for score, item in hits]
    return header + "\n".join(rows)