experiment creating domain specific embedding models using scikit-learn only

README.md

sklearn-embedding-transformer

sklearn-embedding-transformer is a lightweight Python library designed to create and use feature embeddings with scikit-learn. It allows you to handcraft features and generate embeddings using a Multi-Layer Perceptron (MLP), making it easy to integrate with scikit-learn pipelines and workflows. The library does not have any additional dependencies beyond scikit-learn.

Features

• Customizable Embeddings: Generate embeddings with adjustable hidden layers and embedding sizes.
• PCA for Dimensionality Reduction: Optionally apply PCA to reduce the dimensionality of the embeddings.
• Similarity Search: Perform similarity searches using cosine similarity to find the most similar items based on their embeddings.
• Easy Integration: Use seamlessly in scikit-learn pipelines and workflows.
• No Extra Dependencies: Built with scikit-learn only.

Installation

You can install the library via pip (TODO: publish it first...):

pip install sklearn-embedding-transformer

Usage

Creating and Training Embeddings

from sklearn_embedding_transformer import MLPEmbeddingTransformer

# Define and train the transformer
transformer = MLPEmbeddingTransformer(
    hidden_layer_sizes=(200, 100, 50),
    hidden_layer_index=1,
    embedding_size=64,
    max_iter=1000
)

# Example data: List of dictionaries with categorical features
data = [
    {'feature1': 'A', 'feature2': 'B'},
    {'feature1': 'A', 'feature2': 'C'},
    {'feature1': 'B', 'feature2': 'B'},
    {'feature1': 'B', 'feature2': 'C'},
    {'feature1': 'C', 'feature2': 'B'},
]
# Target labels - choose this in a way that creates meaningful embeddings for your specific domain
y = [0, 1, 0, 1, 0]

# Fit the transformer on your data
transformer.fit(data, y)

# Transform your data to get embeddings
embeddings = transformer.transform(data)

# Save and load the transformer
transformer.save('mlp_embedding_transformer.pkl')
loaded_transformer = MLPEmbeddingTransformer.load('mlp_embedding_transformer.pkl')
loaded_embeddings = loaded_transformer.transform(data)

Evaluating Classifiers

from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score

def evaluate_classifiers(X_train, X_test, y_train, y_test):
    classifiers = {
        'Logistic Regression': LogisticRegression(max_iter=1000),
        'MLP Classifier': MLPClassifier(max_iter=1000, random_state=42),
        'SVM': SVC(),
        'Random Forest': RandomForestClassifier(random_state=42)
    }

    results = {}
    for name, clf in classifiers.items():
        clf.fit(X_train, y_train)
        y_pred = clf.predict(X_test)
        accuracy = accuracy_score(y_test, y_pred)
        results[name] = accuracy
    return results

# Example usage
X_train_emb, X_test_emb, y_train_emb, y_test_emb = train_test_split(embeddings, y, test_size=0.2, random_state=42)
results_embeddings = evaluate_classifiers(X_train_emb, X_test_emb, y_train_emb, y_test_emb)
print("Evaluating classifiers using embeddings:")
for name, accuracy in results_embeddings.items():
    print(f"{name} accuracy: {accuracy:.2f}")

X_train_orig, X_test_orig, y_train_orig, y_test_orig = train_test_split(data, y, test_size=0.2, random_state=42)
vectorizer = DictVectorizer(sparse=False)
X_train_orig_vect = vectorizer.fit_transform(X_train_orig)
X_test_orig_vect = vectorizer.transform(X_test_orig)
results_original = evaluate_classifiers(X_train_orig_vect, X_test_orig_vect, y_train_orig, y_test_orig)
print("Evaluating classifiers using original features:")
for name, accuracy in results_original.items():
    print(f"{name} accuracy: {accuracy:.2f}")

Performing Similarity Search

from sklearn.metrics.pairwise import cosine_similarity
import numpy as np

def find_most_similar(embeddings, query_embedding, top_n=5):
    similarities = cosine_similarity(embeddings, query_embedding.reshape(1, -1))
    most_similar_indices = np.argsort(similarities.flatten())[-top_n:]
    return most_similar_indices, similarities[most_similar_indices]

# Example usage
query_index = 10
query_embedding = embeddings[query_index]
top_n = 5
most_similar_indices, similarities = find_most_similar(embeddings, query_embedding, top_n)
print(f"Query index: {query_index}")
print(f"Top {top_n} most similar indices:")
for idx, similarity in zip(most_similar_indices, similarities.flatten()):
    print(f"Index: {idx}, Similarity: {similarity:.4f}")

Use Cases

• Feature Engineering: Convert handcrafted features into embeddings for use in machine learning models.
• Recommendation Systems: Use embeddings to find items similar to a given query.
• Information Retrieval: Retrieve similar documents or records from a large dataset based on their embeddings.
• Anomaly Detection: Identify items that are most different from a given reference.

How to Choose the Classification Task for Training

The classification task used to train the embeddings should align with your specific use case:

• Media Classification: For example, classifying media types like radio, movie, or podcast can help in building embeddings that are tailored to these categories.
• Artist to Genre Mapping: For mapping artist names to genres, embeddings can help in finding the closest artists by genre similarity.

Similarity Search with Embeddings

You can compare embeddings using metrics like cosine similarity. This helps in identifying how similar two items are based on their vector representations. This can be useful for recommendation systems, similarity search, and more.

MLP_embeddings.py

import joblib
import numpy as np
from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.datasets import make_classification
from sklearn.decomposition import PCA
from sklearn.ensemble import RandomForestClassifier
from sklearn.feature_extraction import DictVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import accuracy_score
from sklearn.model_selection import train_test_split
from sklearn.neural_network import MLPClassifier
from sklearn.pipeline import Pipeline
from sklearn.svm import SVC


class MLPEmbeddingTransformer(BaseEstimator, TransformerMixin):
    def __init__(self, hidden_layer_sizes=(10, 5),
                 hidden_layer_index=0,
                 embedding_size=None,
                 max_iter=1000, random_state=42):
        self.hidden_layer_sizes = hidden_layer_sizes
        self.hidden_layer_index = hidden_layer_index
        self.embedding_size = embedding_size
        self.max_iter = max_iter
        self.random_state = random_state
        self.vectorizer = DictVectorizer(sparse=False)
        self.mlp = MLPClassifier(hidden_layer_sizes=self.hidden_layer_sizes,
                                 max_iter=self.max_iter,
                                 random_state=self.random_state)
        self.pipeline = Pipeline([
            ('vectorizer', self.vectorizer),
            ('mlp', self.mlp),
        ])
        self.pca = None  # Initialize PCA later based on data

    def fit(self, X, y=None):
        self.pipeline.fit(X, y)
        hidden_activations = self._get_hidden_activations(X)
        if self.embedding_size:
            # Ensure embedding size does not exceed the number of features in hidden activations
            n_features = hidden_activations.shape[1]
            n_components = min(self.embedding_size, n_features)
            self.pca = PCA(n_components=n_components)
            self.pca.fit(hidden_activations)
        return self

    def transform(self, X):
        hidden_activations = self._get_hidden_activations(X)
        if self.embedding_size:
            hidden_activations = self.pca.transform(hidden_activations)
        return hidden_activations

    def _get_hidden_activations(self, X):
        X_transformed = self.vectorizer.transform(X)
        coefs = self.mlp.coefs_
        intercepts = self.mlp.intercepts_

        activation = X_transformed
        for i in range(self.hidden_layer_index + 1):
            activation = np.dot(activation, coefs[i]) + intercepts[i]
            activation = np.maximum(activation, 0)  # ReLU activation
        return activation

    def save(self, filepath):
        joblib.dump(self, filepath)

    @classmethod
    def load(cls, filepath):
        return joblib.load(filepath)


# Function to train and evaluate classifiers
def evaluate_classifiers(X_train, X_test, y_train, y_test):
    classifiers = {
        'Logistic Regression': LogisticRegression(max_iter=1000),
        'MLP Classifier': MLPClassifier(max_iter=1000, random_state=42),
        'SVM': SVC(),
        'Random Forest': RandomForestClassifier(random_state=42)
    }

    results = {}
    for name, clf in classifiers.items():
        clf.fit(X_train, y_train)
        y_pred = clf.predict(X_test)
        accuracy = accuracy_score(y_test, y_pred)
        results[name] = accuracy
    return results


# Example usage:
if __name__ == "__main__":
    # Generate a synthetic dataset
    X, y = make_classification(n_samples=500, n_features=45, n_informative=20, n_redundant=15, random_state=42)

    # Convert the dataset to a list of dictionaries
    data = [{f'feature{i}': X[j, i] for i in range(X.shape[1])} for j in range(X.shape[0])]

    # Create and train the MLP embedding transformer with embedding size specified
    transformer = MLPEmbeddingTransformer(hidden_layer_sizes=(200, 100, 50),
                                          hidden_layer_index=1,
                                          embedding_size=64,  # size of desired output vector
                                          max_iter=1000)
    transformer.fit(data, y)
    embeddings = transformer.transform(data)

    # Print the resulting embeddings shape
    print("Embeddings shape:", embeddings.shape)
    # Embeddings shape: (500, 64)

    # Save the transformer to a file
    transformer.save('mlp_embedding_transformer.pkl')

    # Load the transformer from the file
    loaded_transformer = MLPEmbeddingTransformer.load('mlp_embedding_transformer.pkl')
    loaded_embeddings = loaded_transformer.transform(data)

    # Verify that the loaded transformer produces the same embeddings
    print("Loaded transformer embeddings shape:", loaded_embeddings.shape)
    # Loaded transformer embeddings shape: (500, 64)

    # Split the data for training and testing
    X_train_emb, X_test_emb, y_train_emb, y_test_emb = train_test_split(embeddings, y,
                                                                        test_size=0.2, random_state=42)
    X_train_orig, X_test_orig, y_train_orig, y_test_orig = train_test_split(data, y,
                                                                            test_size=0.2, random_state=42)

    # Vectorize the original data
    vectorizer = DictVectorizer(sparse=False)
    X_train_orig_vect = vectorizer.fit_transform(X_train_orig)
    X_test_orig_vect = vectorizer.transform(X_test_orig)

    # Evaluate classifiers using embeddings
    print("Evaluating classifiers using embeddings:")
    results_embeddings = evaluate_classifiers(X_train_emb, X_test_emb, y_train_emb, y_test_emb)
    for name, accuracy in results_embeddings.items():
        print(f"{name} accuracy: {accuracy:.2f}")
    # Evaluating classifiers using embeddings:
    # Logistic Regression accuracy: 1.00
    # MLP Classifier accuracy: 0.99
    # SVM accuracy: 1.00
    # Random Forest accuracy: 0.91

    # Evaluate classifiers using original features
    print("Evaluating classifiers using original features:")
    results_original = evaluate_classifiers(X_train_orig_vect, X_test_orig_vect, y_train_orig, y_test_orig)
    for name, accuracy in results_original.items():
        print(f"{name} accuracy: {accuracy:.2f}")
    # Evaluating classifiers using original features:
    # Logistic Regression accuracy: 0.72
    # MLP Classifier accuracy: 0.88
    # SVM accuracy: 0.85
    # Random Forest accuracy: 0.80