Two-Tower Models: cómo los sistemas de recomendación encuentran lo que te gusta entre millones de opciones

import tensorflow as tf

# Crear una capa de embedding para 10,000 peliculas
# Cada pelicula se representa como un vector de 128 dimensiones
embedding_layer = tf.keras.layers.Embedding(
    input_dim=10000,   # numero de peliculas
    output_dim=128     # dimensiones del embedding
)

# Obtener el embedding de la pelicula con ID 42
movie_id = tf.constant([42])
movie_embedding = embedding_layer(movie_id)
print(f"Forma del embedding: {movie_embedding.shape}")  # (1, 128)

import tensorflow as tf
from tensorflow import keras

# Torre de usuario
user_input = keras.Input(shape=(1,), name="user_id")
user_embedding = keras.layers.Embedding(10000, 128)(user_input)
user_embedding = keras.layers.Flatten()(user_embedding)
user_vector = keras.layers.Dense(128, activation="relu")(user_embedding)
user_vector = keras.layers.Dense(64, activation="relu")(user_vector)
user_output = keras.layers.Dense(128)(user_vector)

# Torre de item (pelicula)
item_input = keras.Input(shape=(1,), name="item_id")
item_embedding = keras.layers.Embedding(50000, 128)(item_input)
item_embedding = keras.layers.Flatten()(item_embedding)
item_vector = keras.layers.Dense(128, activation="relu")(item_embedding)
item_vector = keras.layers.Dense(64, activation="relu")(item_vector)
item_output = keras.layers.Dense(128)(item_vector)

# Similitud: dot product entre ambas torres
similarity = keras.layers.Dot(axes=1, normalize=True)([user_output, item_output])

# Modelo completo
model = keras.Model(inputs=[user_input, item_input], outputs=similarity)
model.compile(optimizer="adam", loss="binary_crossentropy")
model.summary()

import numpy as np

a = np.array([0.9, 0.1, 0.8, 0.3])
b = np.array([0.8, 0.2, 0.7, 0.4])

# Dot Product
dot = np.dot(a, b)
print(f"Dot Product: {dot:.4f}")  # 1.33

# Cosine Similarity
cos_sim = dot / (np.linalg.norm(a) * np.linalg.norm(b))
print(f"Cosine Similarity: {cos_sim:.4f}")  # 0.9887

# Distancia Euclidiana
euclidean = np.linalg.norm(a - b)
print(f"Distancia Euclidiana: {euclidean:.4f}")  # 0.2000

import numpy as np

def random_hash_function(dim, num_planes=8):
    """Genera planos aleatorios para LSH"""
    planes = np.random.randn(num_planes, dim)

    def hash_vector(v):
        projections = planes @ v
        # Cada proyeccion se convierte en 0 o 1
        return tuple((projections > 0).astype(int))

    return hash_vector

# Crear funcion hash para vectores de 128 dimensiones
hasher = random_hash_function(dim=128)

# Vectores similares obtienen el mismo hash (con alta probabilidad)
v1 = np.random.randn(128)
v2 = v1 + np.random.randn(128) * 0.1  # muy similar a v1
v3 = np.random.randn(128)              # completamente diferente

print(f"Hash v1: {hasher(v1)}")
print(f"Hash v2: {hasher(v2)}")  # probablemente igual a v1
print(f"Hash v3: {hasher(v3)}")  # probablemente diferente

import faiss
import numpy as np

# Generar embeddings simulados (1 millon de peliculas, 128 dimensiones)
dimension = 128
num_items = 1_000_000
item_embeddings = np.random.randn(num_items, dimension).astype("float32")

# Crear indice FAISS (IndexFlatIP = Inner Product, ideal para Two-Tower)
index = faiss.IndexFlatIP(dimension)
# Para millones de vectores, usar un indice aproximado:
# index = faiss.IndexIVFFlat(faiss.IndexFlatIP(dimension), dimension, 1000)

# Agregar todos los embeddings al indice
faiss.normalize_L2(item_embeddings)  # normalizar para cosine similarity
index.add(item_embeddings)

print(f"Vectores indexados: {index.ntotal}")  # 1,000,000

# Buscar las 10 peliculas mas similares a un usuario
user_embedding = np.random.randn(1, dimension).astype("float32")
faiss.normalize_L2(user_embedding)

distances, indices = index.search(user_embedding, k=10)
print(f"Top 10 peliculas mas similares: {indices[0]}")
print(f"Scores de similitud: {distances[0]}")

import tensorflow as tf
import tensorflow_recommenders as tfrs
import numpy as np

# Embeddings pre-calculados del modelo Two-Tower
item_embeddings = np.random.randn(100000, 128).astype("float32")
item_ids = [f"movie_{i}" for i in range(100000)]

# Crear indice ScaNN optimizado para busqueda rapida
scann_index = tfrs.layers.factorized_top_k.ScaNN(
    model=tf.keras.Sequential([
        tf.keras.layers.InputLayer(input_shape=(128,)),
    ]),
    k=10,
    num_leaves=1000,
    num_leaves_to_search=100,
)

# Indexar todos los embeddings
scann_index.index_from_dataset(
    tf.data.Dataset.from_tensor_slices(item_embeddings).batch(1000)
)

# Buscar recomendaciones para un usuario
user_embedding = np.random.randn(1, 128).astype("float32")
scores, movie_indices = scann_index(tf.constant(user_embedding))
print(f"Top 10 recomendaciones: {movie_indices.numpy()[0]}")