Implementación pgvector — NutriAI (Hybrid Search)
# nutriai/vector_store.py — Implementación pgvector production-ready
import asyncpg
from typing import List, Dict, Optional
import numpy as np
class NutriAIVectorStore:
async def init(self, connection_string: str):
self.pool = await asyncpg.create_pool(connection_string)
async with self.pool.acquire() as conn:
await conn.execute("CREATE EXTENSION IF NOT EXISTS vector")
# Tabla con metadatos para filtrado pre-búsqueda
await conn.execute("""
CREATE TABLE IF NOT EXISTS fichas_nutricionales (
id TEXT PRIMARY KEY, contenido TEXT,
categoria TEXT, -- proteinas|carbohidratos|vitaminas|lipidos
aprobado_clinico BOOL DEFAULT FALSE,
fuente TEXT, -- oms|harvard|ncbi|intra
embedding vector(1536)
)""")
# Índice HNSW — mejor balance latencia/recall para 450K vecs
await conn.execute("""
CREATE INDEX IF NOT EXISTS idx_fichas_hnsw
ON fichas_nutricionales
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64)""")
# Índice FTS para búsqueda híbrida
await conn.execute("""
CREATE INDEX IF NOT EXISTS idx_fichas_fts
ON fichas_nutricionales
USING gin(to_tsvector('spanish', contenido))""")
async def hybrid_search(
self, query_embedding: List[float], query_text: str,
categoria: Optional[str] = None,
solo_aprobados: bool = False,
top_k: int = 10, vector_weight: float = 0.6
) -> List[Dict]:
"""Búsqueda híbrida: cosine similarity + BM25 full-text + filtros clínicos."""
filters = []
if categoria: filters.append(f"categoria = '{categoria}'")
if solo_aprobados: filters.append("aprobado_clinico = TRUE")
where_clause = ("WHERE " + " AND ".join(filters)) if filters else ""
async with self.pool.acquire() as conn:
# SET ef_search en la sesión para controlar recall/latencia
await conn.execute("SET hnsw.ef_search = 128")
rows = await conn.fetch(f"""
WITH vector_scores AS (
SELECT id, contenido, categoria, aprobado_clinico,
1 - (embedding <=> $1::vector) AS v_score
FROM fichas_nutricionales {where_clause}
ORDER BY embedding <=> $1::vector LIMIT $3 * 3
),
text_scores AS (
SELECT id,
ts_rank(to_tsvector('spanish', contenido),
plainto_tsquery('spanish', $2)) AS t_score
FROM fichas_nutricionales {where_clause}
WHERE to_tsvector('spanish', contenido) @@
plainto_tsquery('spanish', $2)
)
SELECT v.id, v.contenido, v.categoria, v.aprobado_clinico,
(v.v_score * $4 + COALESCE(t.t_score, 0) * (1 - $4)) AS score
FROM vector_scores v LEFT JOIN text_scores t ON v.id = t.id
ORDER BY score DESC LIMIT $3
""", query_embedding, query_text, top_k, vector_weight)
return [dict(r) for r in rows]