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 e6d15a7c6d4173253ae34d4ac3c99297eb7a796a
parent aa5867344c6953d55c1bae2ac4fd2b4b8072ca5d
Author: Sondre Wold <sondrewold1@gmail.com>
Date:   Wed, 16 Jul 2025 13:38:58 +0200

added more models

Diffstat:
Mexperiment.py | 195+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++-----
1 file changed, 185 insertions(+), 10 deletions(-)

diff --git a/experiment.py b/experiment.py @@ -5,24 +5,37 @@ import pandas import pathlib import argparse +import numpy + from abc import ABC, abstractmethod import torch import pickle import tqdm +from gensim.models import KeyedVectors + from transformers import AutoTokenizer, AutoModel from sentence_transformers import SentenceTransformer +DEVICE = "cuda" if torch.cuda.is_available() else "cpu" + + def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Experiment") parser.add_argument("-data_path", type=pathlib.Path) parser.add_argument("-result_path", type=pathlib.Path) parser.add_argument("-normalization_value", type=float) parser.add_argument("-experiment_type", type=str, choices=["sick", "ncip"]) + parser.add_argument("-model", type=str, choices=["BERT_cls", "BERT_mean", "S_BERT", "S_GPT", "E5", "QWEN"]) return parser.parse_args() +class EmbedError(Exception): + """If the model is undefined for a subword of the MultiWordExpression, for example, then I should skip this""" + pass + + class Point(NamedTuple): phrase_1: str phrase_2: str @@ -72,17 +85,52 @@ class ModelFactory: return cls._factory[name](*args, **kwargs) +# @ModelFactory("Word2Vec") +class Word2vec(Model): + """ + It does not make conceptually sense to test Word2Vec, as sim(\mu(ac), \mu(bc)) will + be equal to sim(\mu(a), \mu(b)) by definition if \mu(ac) + is the sum of the embeddings of the subwords in a and c. + + But, maybe we can do something else with it, so here it is. + """ + + def __init__(self) -> None: + self.model_path: str = "./models/GoogleNews-vectors-negative300.bin.gz" + self.encoder = KeyedVectors.load_word2vec_format(self.model_path, binary=True) + + + def embed(self, string: str) -> torch.Tensor: + words: list[str] = string.split() + + # In He et.al (2025), they ignore OOV word word2vec. + present_words = [word for word in words if word in self.encoder] + + embeddings = [self.encoder[word] for word in present_words] + sentence_embedding = numpy.sum(embeddings, axis=0) + return torch.tensor(sentence_embedding) + + + def calculate_similarity(self, a: torch.Tensor, b: torch.Tensor) -> float: + return torch.nn.functional.cosine_similarity( + a, b, dim=0 + ).item() + + + @ModelFactory("BERT_cls") class BERT_cls(Model): def __init__(self): - self.model_name = "bert-base-uncased" + 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() + @torch.no_grad() def embed(self, string: str) -> torch.Tensor: return self.encoder( - self.tokenizer(string, return_tensors="pt")["input_ids"] + self.tokenizer(string.lower(), return_tensors="pt")["input_ids"] ).pooler_output # the phrase in a that is not in b def calculate_similarity(self, a: torch.Tensor, b: torch.Tensor) -> float: @@ -91,11 +139,49 @@ class BERT_cls(Model): ).item() # dist(\mu(a), \mu(b)) +@ModelFactory("BERT_mean") +class BERT_mean(Model): + + 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() + + @torch.no_grad() + def embed(self, string: str) -> torch.Tensor: + inputs = self.tokenizer(string.lower(), return_tensors="pt") + outputs = self.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 + + 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("S_BERT") class S_BERT(Model): def __init__(self): - self.encoder = SentenceTransformer("all-MiniLM-L6-v2") + self.encoder = SentenceTransformer("all-MiniLM-L6-v2", device=DEVICE) def embed(self, string: str) -> torch.Tensor: return self.encoder.encode([string]) @@ -104,6 +190,50 @@ class S_BERT(Model): return self.encoder.similarity(a, b) +@ModelFactory("S_GPT") +class S_GPT(Model): + + def __init__(self): + self.encoder = SentenceTransformer("Muennighoff/SGPT-125M-weightedmean-nli-bitfit", 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) + + +@ModelFactory("E5") +class E5(Model): + + def __init__(self): + self.encoder = SentenceTransformer("intfloat/multilingual-e5-large-instruct", 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) + + +@ModelFactory("QWEN") +class QWEN(Model): + + def __init__(self): + self.encoder = SentenceTransformer("Qwen/Qwen3-Embedding-0.6B", device=DEVICE) + self.task = "Given a sentence, encode its semantic properties" + + def get_detailed_instruct(self, query: str) -> str: + return f'Instruct: {self.task}\nQuery: {query}' + + def embed(self, string: str) -> torch.Tensor: + prepared_string = self.get_detailed_instruct(string) + return self.encoder.encode([prepared_string]) + + def calculate_similarity(self, a: torch.Tensor, b: torch.Tensor) -> float: + return self.encoder.similarity(a, b) + + 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. @@ -159,8 +289,13 @@ class Experiment: self.sentence_pairs = sentence_pairs self.normalization_value: float = normalization_value self.diff_errors: int = 0 + self.embed_errors: int = 0 self.points: list[Point] = [] self.result_path: pathlib.Path = result_path + self.sentence_a_length: list[int] = [] + self.sentence_b_length: list[int] = [] + self.phrase_a_length: list[int] = [] + self.phrase_b_length: list[int] = [] def run_experiment(self) -> None: for sentence_a, sentence_b in tqdm.tqdm(zip(*self.sentence_pairs)): @@ -173,22 +308,40 @@ class Experiment: phrase_b: str phrase_a, phrase_b = diffs + self.sentence_a_length.append(len(sentence_a.split())) + self.sentence_b_length.append(len(sentence_b.split())) + self.phrase_a_length.append(len(phrase_a.split())) + self.phrase_b_length.append(len(phrase_b.split())) + b_in_a: str = sentence_a.replace(phrase_a, phrase_b) - phrase_a_embedded: torch.Tensor = self.model.embed(phrase_a) - phrase_b_embedded: torch.Tensor = self.model.embed(phrase_b) + try: + phrase_a_embedded: torch.Tensor = self.model.embed(phrase_a) + phrase_b_embedded: torch.Tensor = self.model.embed(phrase_b) + except EmbedError as e: + self.embed_errors += 1 + continue phrase_similarity: float = self.model.calculate_similarity( phrase_a_embedded, phrase_b_embedded ) - a_in_context_embedded = self.model.embed(sentence_a) - b_in_context_embedded = self.model.embed(b_in_a) + + try: + a_in_context_embedded = self.model.embed(sentence_a) + b_in_context_embedded = self.model.embed(b_in_a) + except EmbedError as e: + self.embed_errors += 1 + continue if self.normalization_value > 0.0: a_removed: str = sentence_a.replace(phrase_a, "") b_removed: str = sentence_b.replace(phrase_b, "") - a_removed_embedded = self.model.embed(a_removed) - b_removed_embedded = self.model.embed(b_removed) + try: + a_removed_embedded = self.model.embed(a_removed) + b_removed_embedded = self.model.embed(b_removed) + except EmbedError as e: + self.embed_errors += 1 + continue a_in_context_embedded = self.model.normalize( a_in_context_embedded, a_removed_embedded, self.normalization_value @@ -248,7 +401,9 @@ def main() -> None: sentence_pairs = SentencePairs(sentence_A=sentence_A, sentence_B=sentence_B) - for model_type in ModelFactory._factory.keys(): + models = [args.model] if args.model is not None else list(ModelFactory._factory.keys()) + + for model_type in models: model = ModelFactory.get(model_type) result_path = ( args.result_path @@ -262,7 +417,27 @@ def main() -> None: ) print(f"Running experiment for model: {model_type}.") experiment.run_experiment() + print(f"Diff errors: {experiment.diff_errors}, Embedding errors: {experiment.embed_errors}") print(f"Saving experiment to: {str(result_path)}") + + sent_a_mean = numpy.mean(experiment.sentence_a_length) + sent_b_mean = numpy.mean(experiment.sentence_b_length) + phrase_a_mean = numpy.mean(experiment.phrase_a_length) + phrase_b_mean = numpy.mean(experiment.phrase_b_length) + + sent_a_std = numpy.std(experiment.sentence_a_length) + sent_b_std = numpy.std(experiment.sentence_b_length) + phrase_a_std = numpy.std(experiment.phrase_a_length) + phrase_b_std = numpy.std(experiment.phrase_b_length) + + descriptive_statistics = f""" + Sentence A: {sent_a_mean} +- {sent_a_std} \\ + Sentence B: {sent_b_mean} +- {sent_b_std} \\ + Phrase A: {phrase_a_mean} +- {phrase_a_std} \\ + Phrase B: {phrase_b_mean} +- {phrase_b_std} + """ + print(descriptive_statistics) + experiment.save_results()