Filtro de metadata
Búsqueda vectorial que respeta la estructura que ya conoces. Antepón a un query
Showcase — Filtro de metadata
Búsqueda vectorial que respeta la estructura que ya conoces. Antepón a un query
un topic y dos puntos — billing: how do I get my money back — y el store se
queda solo con los documentos topic=billing, luego los rankea por significado.
El filtro reduce los candidatos; los embeddings rankean lo que queda. Topics
válidos: account, billing, shipping, returns.
Córrelo
bash bootstrap-secrets.sh # reads ../../../../.env, writes secrets/
docker compose up --build # default: PROVIDER=openai
Abre http://localhost:3000. Córrelo contra Gemini con
PROVIDER=gemini docker compose up --build.
Qué hay aquí
backend/store.py— elVectorStorecompartido;search()toma un predicado de metadatawhereque se aplica antes de rankear.backend/ai_openai.py/backend/ai_gemini.py— docs con metadata detopicy el parseo de"topic: query".frontend/app/page.tsx— caja de query + resultados filtrados y rankeados.
Detenlo
docker compose down
Ejecútalo en tu máquina
Descarga el proyecto como ZIP y córrelo con Docker. Levanta un backend FastAPI y un frontend Next.js en localhost:3000.
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
Escribe algo, elige un proveedor y ejecuta el mismo código de Código contra la API real. Requiere iniciar sesión.
Los mismos módulos que ejecuta el botón Run. El proyecto completo (frontend, Dockerfile, compose) está en el ZIP, pestaña 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)
Archivos del proyecto
.gitignoreREADME.es.mdREADME.mdbackend/Dockerfilebackend/ai_gemini.pybackend/ai_openai.pybackend/main.pybackend/requirements.txtbackend/store.pybootstrap-secrets.shdocker-compose.ymlfrontend/Dockerfilefrontend/app/layout.tsxfrontend/app/page.tsxfrontend/next.config.tsfrontend/package.jsonfrontend/tsconfig.json