epsilon-compositionality

Code for our paper Measuring Idiomaticity in Text Embedding Models with 𝜀-compositionality
git clone https://esimon.eu/repos/epsilon-compositionality.git
Log | Files | Refs | Submodules | README

commit 9c81da9755d77a1bcb0783dd5f3e84b36105aa8d
parent 60f7046273badb01b582c8d03da76a81cb68b8b2
Author: Sondre Wold <sondrewold1@gmail.com>
Date:   Wed, 24 Sep 2025 12:41:34 +0200

updates, reproduce current table

Diffstat:
Mexperiment.py | 253++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++-----------
Mget_epsilon.py | 25+++++++++++++++++++++++++
Mrandom_experiment.py | 16+++++++++-------
Mreproduce.sh | 14+++++---------
Mstatistical_testing.py | 6++++--
5 files changed, 262 insertions(+), 52 deletions(-)

diff --git a/experiment.py b/experiment.py @@ -54,6 +54,12 @@ class Point(NamedTuple): context_bc_embedded: torch.Tensor +class ResultObject(NamedTuple): + model_name: str + dataset_name: str + points: list[Point] + + class SentencePairs(NamedTuple): sentence_A: list[str] sentence_B: list[str] @@ -104,8 +110,116 @@ class ModelFactory: return cls._factory[name](*args, **kwargs) -@ModelFactory("BERT-pooler") -class BERT_pooler(Model): +# Generated +def mean_pool_last_four_layers(string: str, tokenizer, encoder) -> torch.Tensor: + inputs = tokenizer(string.lower(), return_tensors="pt") + + # Get all hidden states (set output_hidden_states=True) + outputs = encoder(**inputs, output_hidden_states=True) + + # Get the last four hidden states (layers) + # hidden_states is a tuple of length num_layers+1 (includes embedding layer) + # We take layers [-4:] to get the last four layers (excluding embedding layer) + last_four_layers = outputs.hidden_states[-4:] + + # Stack the last four layers and average them + stacked_layers = torch.stack(last_four_layers, dim=0) # Shape: (4, batch_size, seq_len, hidden_size) + averaged_layers = torch.mean(stacked_layers, dim=0) # Shape: (batch_size, seq_len, hidden_size) + + attention_mask = inputs["attention_mask"] + + # Expand attention mask to match token embeddings dimensions + attention_mask_expanded = ( + attention_mask.unsqueeze(-1).expand(averaged_layers.size()).float() + ) + + # Apply mask to token embeddings (set padding tokens to 0) + masked_embeddings = averaged_layers * attention_mask_expanded + + # Sum embeddings and divide by number of non-padding tokens + summed_embeddings = torch.sum(masked_embeddings, dim=1) # Shape: (batch_size, hidden_size) + summed_mask = torch.clamp(attention_mask_expanded.sum(dim=1), min=1e-9) # Avoid division by zero + + # Create sentence representation by mean pooling + mean_pooled = summed_embeddings / summed_mask + + return mean_pooled + +# Generated +def mean_pool(string: str, tokenizer, encoder) -> torch.Tensor: + inputs = tokenizer(string.lower(), return_tensors="pt") + outputs = encoder(**inputs) + + token_embeddings = outputs.last_hidden_state + attention_mask = inputs["attention_mask"] + + # Expand attention mask to match token embeddings dimensions + attention_mask_expanded = ( + attention_mask.unsqueeze(-1).expand(token_embeddings.size()).float() + ) + + # Apply mask to token embeddings (set padding tokens to 0) + masked_embeddings = token_embeddings * attention_mask_expanded + + # Sum embeddings and divide by number of non-padding tokens + summed_embeddings = torch.sum( + masked_embeddings, dim=1 + ) # Shape: (batch_size, hidden_size) + summed_mask = torch.clamp( + attention_mask_expanded.sum(dim=1), min=1e-9 + ) # Avoid division by zero + + # Create sentence representation by mean pooling + mean_pooled = summed_embeddings / summed_mask + + return mean_pooled + + +# Generated +def cls_sep_pool(string: str, tokenizer, encoder) -> torch.Tensor: + inputs = tokenizer(string.lower(), return_tensors="pt") + outputs = encoder(**inputs) + token_embeddings = outputs.last_hidden_state + attention_mask = inputs["attention_mask"] + input_ids = inputs["input_ids"] + + batch_size, seq_len, hidden_size = token_embeddings.shape + + # Initialize result tensor + cls_sep_embeddings = torch.zeros(batch_size, hidden_size) + + for batch_idx in range(batch_size): + # Find the sequence length (excluding padding) + valid_length = attention_mask[batch_idx].sum().item() + + # CLS token is always at position 0 + cls_embedding = token_embeddings[batch_idx, 0] + + # Find SEP token position by looking for the SEP token ID + # SEP token is the last non-padding token before padding starts + sep_token_id = tokenizer.sep_token_id + sep_position = None + + # Search for SEP token in the valid sequence + for pos in range(valid_length): + if input_ids[batch_idx, pos] == sep_token_id: + sep_position = pos + break + + # If no SEP token found, use the last valid token + if sep_position is None: + raise Exception("No SEP token found") + + sep_embedding = token_embeddings[batch_idx, sep_position] + + # Sum CLS and SEP embeddings + cls_sep_embeddings[batch_idx] = cls_embedding + sep_embedding + + return cls_sep_embeddings + + +@ModelFactory("BERT-cls") +class BERT_cls(Model): def __init__(self) -> None: self.model_name: str = "bert-base-uncased" @@ -117,7 +231,7 @@ class BERT_pooler(Model): def embed(self, string: str) -> torch.Tensor: return self.encoder( self.tokenizer(string.lower(), return_tensors="pt")["input_ids"] - ).pooler_output + ).last_hidden_state[:, 0, :] def calculate_similarity(self, a: torch.Tensor, b: torch.Tensor) -> float: return torch.nn.functional.cosine_similarity( @@ -125,11 +239,11 @@ class BERT_pooler(Model): ).item() # dist(\mu(a), \mu(b)) -@ModelFactory("BERT-cls") -class BERT_cls(Model): +@ModelFactory("ModernBERT-cls") +class ModernBERT_cls(Model): def __init__(self) -> None: - self.model_name: str = "bert-base-uncased" + self.model_name: str = "answerdotai/ModernBERT-base" self.tokenizer = AutoTokenizer.from_pretrained(self.model_name) self.encoder = AutoModel.from_pretrained(self.model_name) self.encoder.eval() @@ -146,8 +260,8 @@ class BERT_cls(Model): ).item() # dist(\mu(a), \mu(b)) -@ModelFactory("ModernBERT-cls") -class ModernBERT_cls(Model): +@ModelFactory("ModernBERT-mean") +class ModernBERT_mean(Model): def __init__(self) -> None: self.model_name: str = "answerdotai/ModernBERT-base" @@ -157,15 +271,33 @@ class ModernBERT_cls(Model): @torch.no_grad() def embed(self, string: str) -> torch.Tensor: - return self.encoder( - self.tokenizer(string.lower(), return_tensors="pt")["input_ids"] - ).last_hidden_state[:, 0, :] + return mean_pool_last_four_layers(string=string, tokenizer=self.tokenizer, encoder=self.encoder) def calculate_similarity(self, a: torch.Tensor, b: torch.Tensor) -> float: return torch.nn.functional.cosine_similarity( a, b ).item() # dist(\mu(a), \mu(b)) + +@ModelFactory("ModernBERT-clssep") +class ModernBERT_clssep(Model): + + def __init__(self) -> None: + self.model_name: str = "answerdotai/ModernBERT-base" + self.tokenizer = AutoTokenizer.from_pretrained(self.model_name) + self.encoder = AutoModel.from_pretrained(self.model_name) + self.encoder.eval() + + @torch.no_grad() + def embed(self, string: str) -> torch.Tensor: + return cls_sep_pool(string=string, tokenizer=self.tokenizer, encoder=self.encoder) + + def calculate_similarity(self, a: torch.Tensor, b: torch.Tensor) -> float: + return torch.nn.functional.cosine_similarity( + a, b + ).item() # dist(\mu(a), \mu(b)) + + @ModelFactory("DeBERTa-cls") class DeBERTa_cls(Model): @@ -187,6 +319,44 @@ class DeBERTa_cls(Model): ).item() # dist(\mu(a), \mu(b)) +@ModelFactory("DeBERTa-mean") +class DeBERTa_mean(Model): + + def __init__(self) -> None: + self.model_name: str = "microsoft/deberta-v3-base" + self.tokenizer = AutoTokenizer.from_pretrained(self.model_name) + self.encoder = AutoModel.from_pretrained(self.model_name) + self.encoder.eval() + + @torch.no_grad() + def embed(self, string: str) -> torch.Tensor: + return mean_pool_last_four_layers(string=string, tokenizer=self.tokenizer, encoder=self.encoder) + + def calculate_similarity(self, a: torch.Tensor, b: torch.Tensor) -> float: + return torch.nn.functional.cosine_similarity( + a, b + ).item() # dist(\mu(a), \mu(b)) + + +@ModelFactory("DeBERTa-clssep") +class DeBERTa_clssep(Model): + + def __init__(self) -> None: + self.model_name: str = "microsoft/deberta-v3-base" + self.tokenizer = AutoTokenizer.from_pretrained(self.model_name) + self.encoder = AutoModel.from_pretrained(self.model_name) + self.encoder.eval() + + @torch.no_grad() + def embed(self, string: str) -> torch.Tensor: + return cls_sep_pool(string=string, tokenizer=self.tokenizer, encoder=self.encoder) + + def calculate_similarity(self, a: torch.Tensor, b: torch.Tensor) -> float: + return torch.nn.functional.cosine_similarity( + a, b + ).item() # dist(\mu(a), \mu(b)) + + @ModelFactory("BERT-mean") class BERT_mean(Model): @@ -199,32 +369,26 @@ class BERT_mean(Model): @torch.no_grad() def embed(self, string: str) -> torch.Tensor: - inputs = self.tokenizer(string.lower(), return_tensors="pt") - outputs = self.encoder(**inputs) + return mean_pool_last_four_layers(string=string, tokenizer=self.tokenizer, encoder=self.encoder) - token_embeddings = outputs.last_hidden_state - attention_mask = inputs["attention_mask"] - - # Expand attention mask to match token embeddings dimensions - attention_mask_expanded = ( - attention_mask.unsqueeze(-1).expand(token_embeddings.size()).float() - ) + def calculate_similarity(self, a: torch.Tensor, b: torch.Tensor) -> float: + return torch.nn.functional.cosine_similarity( + a, b + ).item() # dist(\mu(a), \mu(b)) - # Apply mask to token embeddings (set padding tokens to 0) - masked_embeddings = token_embeddings * attention_mask_expanded - # Sum embeddings and divide by number of non-padding tokens - summed_embeddings = torch.sum( - masked_embeddings, dim=1 - ) # Shape: (batch_size, hidden_size) - summed_mask = torch.clamp( - attention_mask_expanded.sum(dim=1), min=1e-9 - ) # Avoid division by zero +@ModelFactory("BERT-clssep") +class BERT_clssep(Model): - # Create sentence representation by mean pooling - mean_pooled = summed_embeddings / summed_mask + def __init__(self) -> None: + self.model_name: str = "bert-base-uncased" + self.tokenizer = AutoTokenizer.from_pretrained(self.model_name) + self.encoder = AutoModel.from_pretrained(self.model_name) + self.encoder.eval() - return mean_pooled + @torch.no_grad() + def embed(self, string: str) -> torch.Tensor: + return cls_sep_pool(string=string, tokenizer=self.tokenizer, encoder=self.encoder) def calculate_similarity(self, a: torch.Tensor, b: torch.Tensor) -> float: return torch.nn.functional.cosine_similarity( @@ -298,6 +462,20 @@ class QWEN(Model): return self.encoder.similarity(a, b).item() +@ModelFactory("UAE") +class UAE(Model): + + def __init__(self): + self.encoder = SentenceTransformer("WhereIsAI/UAE-Large-V1", device=DEVICE) + + def embed(self, string: str) -> torch.Tensor: + return self.encoder.encode(string) + + def calculate_similarity(self, a: torch.Tensor, b: torch.Tensor) -> float: + return self.encoder.similarity(a, b).item() + + + def find_diff(a: str, b: str) -> list[str] | None: """ Finds the longest continuous substring where A differs from B and B differs from A. @@ -306,8 +484,6 @@ def find_diff(a: str, b: str) -> list[str] | None: if a == b: # Should not happen for SICK or NCIP, but you never know. return None - # print(f"A={a}\nB={b}\n##############") - try: words_a = a.split() words_b = b.split() @@ -443,6 +619,14 @@ class Experiment: with open(self.result_path.with_suffix(".pkl"), "wb") as file: pickle.dump(self.points, file) + def save_results_2(self, dataset): + pathlib.Path(self.result_path).parent.mkdir(parents=True, exist_ok=True) + result = ResultObject(model_name=self.model.name, dataset_name=dataset, points=self.points) + with open(self.result_path.with_suffix(".pkl"), "wb") as file: + pickle.dump(self.points, result) + + + def main() -> None: args = parse_args() @@ -535,6 +719,7 @@ def main() -> None: print(descriptive_statistics) experiment.save_results() + #experiment.save_results_2(dataset=args.experiment_type) if __name__ == "__main__": diff --git a/get_epsilon.py b/get_epsilon.py @@ -1,6 +1,7 @@ import sys import pickle import numpy +import torch from torch import histogram from experiment import Point import matplotlib.pyplot as plt @@ -8,6 +9,7 @@ import matplotlib.pyplot as plt from typing import Any import json + import pandas import argparse @@ -38,6 +40,23 @@ def get_dataset_and_model_name(filename: str) -> tuple[str, str]: return dataset_name, model_name +def get_simscore( + ac_emb: list[torch.Tensor], bc_emb: list[torch.Tensor] + ) -> list[float]: + res: list[float] = [] + for ac, bc in zip(ac_emb, bc_emb): + ac = ac.view(1, -1) + bc = bc.view(1, -1) + try: + x = float(torch.nn.functional.cosine_similarity(ac, bc)) + except: + print(ac.shape) + print(bc.shape) + exit() + res.append(x) + return res + + def get_epsilons( word_sims: numpy.ndarray, context_sims: numpy.ndarray ) -> dict[str, list[float]]: @@ -71,11 +90,13 @@ def main() -> None: "original": [], "divisive": [], "additive": [], + "simscores": [], } for raw_file in raw_files: with open(raw_file, "rb") as file: points: list[Point] = pickle.load(file) dataset_name, model_name = get_dataset_and_model_name(raw_file.name) + print(model_name) n: int = len(points) word_sims: numpy.ndarray = numpy.array( [point.phrase_similarity for point in points] @@ -83,9 +104,13 @@ def main() -> None: context_sims: numpy.ndarray = numpy.array( [point.contextualized_similarity for point in points] ) + ac_embedded: list[torch.Tensor] = [torch.Tensor(point.context_ac_embedded) for point in points] + bc_embedded: list[torch.Tensor] = [torch.Tensor(point.context_bc_embedded) for point in points] tmp_result: dict[str, list[float]] = get_epsilons(word_sims, context_sims) + simscores: list[float] = get_simscore(ac_embedded, bc_embedded) final_result["dataset"].append(dataset_name) final_result["model"].append(model_name) + final_result["simscores"].append(simscores) if ( args.percentile is not None ): # Save the percentile epsilons to disk as a csv diff --git a/random_experiment.py b/random_experiment.py @@ -30,7 +30,7 @@ def parse_args() -> argparse.Namespace: parser.add_argument( "-model", type=str, - choices=["BERT-cls", "BERT-mean", "S-BERT", "S-GPT", "E5", "QWEN"], + choices=ModelFactory._factory.keys() ) return parser.parse_args() @@ -42,11 +42,10 @@ def main() -> None: print(f"Similarity between random items, n_samples={args.n}") sentences: list[str] = df["original sentence"].values.tolist() - phrases: list[str] = df["synonym both"].values.tolist() + phrases: list[str] = df["compound"].values.tolist() - choices = ["BERT-cls", "BERT-mean", "S-BERT", "E5", "QWEN"] - models = [args.model] if args.model is not None else choices - result = { + models = [args.model] if args.model is not None else ModelFactory._factory.keys() + result: dict[str, list] = { "model": [], "phrase_sim": [], "context_sim": [] @@ -70,9 +69,12 @@ def main() -> None: context_a_embedded = model.embed(context_a) context_b_embedded = model.embed(context_b) + context_sim = model.calculate_similarity(context_a_embedded, context_b_embedded) + context_sims.append(context_sim) + phrase_sim = model.calculate_similarity(phrase_a_embedded, phrase_b_embedded) + phrase_sims.append(phrase_sim) - context_sims.append(model.calculate_similarity(context_a_embedded, context_b_embedded)) - phrase_sims.append(model.calculate_similarity(phrase_a_embedded, phrase_b_embedded)) + print(f"{context_sim}") result["model"].append(model_type) result["phrase_sim"].append(phrase_sims) result["context_sim"].append(context_sims) diff --git a/reproduce.sh b/reproduce.sh @@ -1,14 +1,10 @@ -results_folder=./reproduction_model_results -epsilon_file=./distributions.json -final_result_file=./result.txt - +results_folder=./reproduction_model_results_24 +epsilon_file=./distributions_24.json +final_result_file=./result_24.txt echo "Running all the embedding models on NCIMP" -for model in "BERT-cls" "BERT-mean" "S-BERT" "S-GPT" "E5" "ModernBERT-cls" "DeBERTa-cls" "QWEN" -do - python experiment.py -data_path ncimp/ncip_clean.csv -experiment_type ncip -normalization_value 0.0 -result_path $results_folder -model $model - python experiment.py -data_path ncimp/ncip_clean.csv -experiment_type ncip-non -normalization_value 0.0 -result_path $results_folder -model $model -done +python experiment.py -data_path ncimp/ncip_clean.csv -experiment_type ncip -normalization_value 0.0 -result_path $results_folder +python experiment.py -data_path ncimp/ncip_clean.csv -experiment_type ncip-non -normalization_value 0.0 -result_path $results_folder echo "Calculating epsilon values..." python get_epsilon.py -points_path $results_folder -result_path $epsilon_file diff --git a/statistical_testing.py b/statistical_testing.py @@ -26,6 +26,7 @@ def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Calculate the Welsch T-Test") parser.add_argument("-epsilons_path", type=pathlib.Path) parser.add_argument("-result_path", type=pathlib.Path) + parser.add_argument("-simscore", action="store_true", help="Use the score method based on the He et al paper") return parser.parse_args() @@ -61,12 +62,13 @@ def main() -> None: model_name = loaded_dict["model"][i] print(f"{model_name}, {dataset_name}, {len(loaded_dict["original"][i])}") is_idiomatic = True if dataset_name == "ncip" else False + key = "simscore" if args.simscore else "original" if model_name not in model_results: model_results[model_name] = ResultPoint(model_name=model_name) if is_idiomatic: - model_results[model_name].idiomatic_result = loaded_dict["original"][i] + model_results[model_name].idiomatic_result = loaded_dict[key][i] else: - model_results[model_name].non_idiomatic_result = loaded_dict["original"][i] + model_results[model_name].non_idiomatic_result = loaded_dict[key][i] for model, res in model_results.items(): idiom_variance = numpy.var(res.idiomatic_result)