commit 14724cd7122d09ad2bf1e2a7cc1ace7f81862c2d
parent c82e284c00e85275181f14259dda076fd844b902
Author: Sondre Wold <sondrewold1@gmail.com>
Date: Thu, 25 Sep 2025 09:15:58 +0200
Refactor and clean up the code
Diffstat:
| D | debug/analyze_embeddings.py | | | 114 | ------------------------------------------------------------------------------- |
| D | debug/plot_sims.py | | | 182 | ------------------------------------------------------------------------------- |
| M | experiment.py | | | 509 | ++++--------------------------------------------------------------------------- |
| M | get_epsilon.py | | | 71 | ++++++++++++++++------------------------------------------------------- |
| A | models.py | | | 429 | +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ |
| M | random_experiment.py | | | 15 | ++++++++------- |
| M | statistical_testing.py | | | 58 | +++++++++++++++++++++++++++++++++++++++++++++------------- |
7 files changed, 522 insertions(+), 856 deletions(-)
diff --git a/debug/analyze_embeddings.py b/debug/analyze_embeddings.py
@@ -1,114 +0,0 @@
-import argparse
-import pathlib
-import pickle
-
-from numpy import mean
-from experiment import Point
-
-import pandas as pd
-
-from typing import Any
-
-import torch
-from get_epsilon import get_dataset_and_model_name
-
-
-def parse_args() -> argparse.Namespace:
- parser = argparse.ArgumentParser(description="Gather epsilon values")
- parser.add_argument("-points_path", type=pathlib.Path)
- parser.add_argument("-result_path", type=pathlib.Path)
- return parser.parse_args()
-
-
-def main() -> None:
- args = parse_args()
- points_folder: str = args.points_path
- raw_files = list(pathlib.Path(points_folder).glob("*.pkl"))
-
- final_result: dict[str, list[Any]] = {
- "dataset": [],
- "model": [],
- "a_norm": [],
- "b_norm": [],
- "ac_norm": [],
- "bc_norm": [],
- "a_mean": [],
- "b_mean": [],
- "ac_mean": [],
- "bc_mean": [],
- "distance": [],
- }
-
- 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)
- a_norms: list[float] = []
- b_norms: list[float] = []
- ac_norms: list[float] = []
- bc_norms: list[float] = []
- a_means: list[torch.Tensor] = []
- b_means: list[torch.Tensor] = []
- ac_means: list[torch.Tensor] = []
- bc_means: list[torch.Tensor] = []
- for point in points:
- a_norms.append(
- torch.linalg.norm(torch.Tensor(point.phrase_a_embedded)).item()
- )
- b_norms.append(
- torch.linalg.norm(torch.Tensor(point.phrase_b_embedded)).item()
- )
- ac_norms.append(
- torch.linalg.norm(torch.Tensor(point.context_ac_embedded)).item()
- )
- bc_norms.append(
- torch.linalg.norm(torch.Tensor(point.context_bc_embedded)).item()
- )
- a_means.append(torch.Tensor(point.phrase_a_embedded))
- b_means.append(torch.Tensor(point.phrase_b_embedded))
- ac_means.append(torch.Tensor(point.context_ac_embedded))
- bc_means.append(torch.Tensor(point.context_bc_embedded))
-
- final_result["dataset"].append(dataset_name)
- final_result["model"].append(model_name)
- final_result["a_norm"].append(mean(a_norms))
- final_result["b_norm"].append(mean(b_norms))
- final_result["ac_norm"].append(mean(ac_norms))
- final_result["bc_norm"].append(mean(bc_norms))
-
- final_result["a_mean"].append(
- torch.mean(torch.stack(a_means), dim=-1).norm().item()
- )
- final_result["b_mean"].append(
- torch.mean(torch.stack(b_means), dim=-1).norm().item()
- )
- final_result["ac_mean"].append(
- torch.mean(torch.stack(ac_means), dim=-1).norm().item()
- )
- final_result["bc_mean"].append(
- torch.mean(torch.stack(bc_means), dim=-1).norm().item()
- )
-
- final_result["distance"].append(
- (
- (
- torch.mean(torch.stack(ac_means), dim=-1)
- + torch.mean(torch.stack(bc_means), dim=-1)
- )
- / 2
- - (
- torch.mean(torch.stack(a_means), dim=-1)
- + torch.mean(torch.stack(b_means), dim=-1)
- )
- / 2
- )
- .norm()
- .item()
- )
-
- df = pd.DataFrame.from_dict(final_result)
- print(df)
-
-
-if __name__ == "__main__":
- main()
diff --git a/debug/plot_sims.py b/debug/plot_sims.py
@@ -1,182 +0,0 @@
-import sys
-
-parent_dir = ".."
-sys.path.append(parent_dir)
-import argparse
-import pathlib
-import pickle
-
-import numpy
-
-from numpy import mean
-from experiment import Point
-
-import matplotlib.pyplot as plt
-
-import pandas as pd
-
-from typing import Any
-
-import torch
-from get_epsilon import get_dataset_and_model_name
-
-
-def parse_args() -> argparse.Namespace:
- parser = argparse.ArgumentParser(description="Gather epsilon values")
- parser.add_argument("-points_path", type=pathlib.Path)
- parser.add_argument("-random_path", type=pathlib.Path)
- parser.add_argument("-result_path", type=pathlib.Path)
- return parser.parse_args()
-
-
-def annotate_means(ax, data, positions):
- for i, (d, pos) in enumerate(zip(data, positions)):
- mean_val = numpy.mean(d)
- ax.text(
- pos,
- mean_val + 0.02 * (ax.get_ylim()[1] - ax.get_ylim()[0]), # Check this LLM spagetthi
- f"{mean_val:.3f}",
- ha="center",
- va="bottom",
- fontsize=8,
- bbox=dict(facecolor="white", alpha=0.7, edgecolor="none")
- )
-
-
-def main() -> None:
- args = parse_args()
- points_folder: str = args.points_path
- raw_files = list(pathlib.Path(points_folder).glob("*.pkl"))
-
- final_result: dict[str, list[Any]] = {
- "dataset": [],
- "model": [],
- "phrase_sim": [],
- "context_sim": [],
- }
-
- chosen_dataset = "ncip-non"
-
-
- 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)
- if dataset_name != chosen_dataset:
- continue
-
- word_sims = [
- (
- point.phrase_similarity
- if isinstance(point.phrase_similarity, float)
- else point.phrase_similarity.item()
- )
- for point in points
- ]
- context_sims = [
- (
- point.contextualized_similarity
- if isinstance(point.contextualized_similarity, float)
- else point.contextualized_similarity.item()
- )
- for point in points
- ]
-
- print(model_name, dataset_name)
- #final_result["dataset"].append(dataset_name)
- final_result["model"].append(model_name)
- final_result["phrase_sim"].append(word_sims)
- final_result["context_sim"].append(context_sims)
-
-
-
-
-
- # Sort the "model" list alphabetically
- sorted_models = sorted(final_result["model"])
-
- # Create a mapping from model name to its sorted index
- model_order = {model: idx for idx, model in enumerate(sorted_models)}
-
- # Sort all other lists in the dict based on the order of "model"
- final_result = {
- key: sorted(value, key=lambda x: model_order[final_result["model"][value.index(x)]])
- for key, value in final_result.items()
- }
- fig, axs = plt.subplots(nrows=2, ncols=2, figsize=(10, 4))
- axs[0, 0].violinplot(
- final_result["phrase_sim"],
- [y + 1 for y in range(len(final_result["phrase_sim"]))],
- )
- axs[0, 0].set_xticks(
- [y + 1 for y in range(len(final_result["phrase_sim"]))],
- labels=final_result["model"],
- )
- annotate_means(
- axs[0, 0],
- final_result["phrase_sim"],
- [y + 1 for y in range(len(final_result["phrase_sim"]))]
- )
- axs[1, 0].violinplot(
- final_result["context_sim"],
- [y + 1 for y in range(len(final_result["context_sim"]))],
- )
- axs[1, 0].set_xticks(
- [y + 1 for y in range(len(final_result["context_sim"]))],
- labels=final_result["model"],
- )
- annotate_means(
- axs[1, 0],
- final_result["context_sim"],
- [y + 1 for y in range(len(final_result["context_sim"]))]
- )
-
- axs[0, 0].set_title(f"{chosen_dataset} phrase similarity")
- axs[0, 1].set_title(f"Random phrase similarity")
- axs[1, 0].set_title(f"{chosen_dataset} context similarity")
- axs[1, 1].set_title(f"Random context similarity")
-
- if args.random_path is not None:
- with open(args.random_path, "rb") as f:
- random_result = pickle.load(f)
- sorted_models = sorted(random_result["model"])
-
- # Create a mapping from model name to its sorted index
- model_order = {model: idx for idx, model in enumerate(sorted_models)}
-
- # Sort all other lists in the dict based on the order of "model"
- random_result = {
- key: sorted(value, key=lambda x: model_order[random_result["model"][value.index(x)]])
- for key, value in random_result.items()
- }
-
- axs[0, 1].violinplot(
- random_result["phrase_sim"],
- [y + 1 for y in range(len(random_result["phrase_sim"]))],
- )
- axs[0, 1].set_xticks(
- [y + 1 for y in range(len(random_result["phrase_sim"]))],
- labels=random_result["model"],
- )
- annotate_means(
- axs[0, 1],
- random_result["phrase_sim"],
- [y + 1 for y in range(len(random_result["phrase_sim"]))]
- )
- axs[1, 1].violinplot(
- random_result["context_sim"],
- [y + 1 for y in range(len(random_result["context_sim"]))],
- )
- axs[1, 1].set_xticks(
- [y + 1 for y in range(len(random_result["context_sim"]))],
- labels=random_result["model"],
- )
- annotate_means(
- axs[1, 1],
- random_result["context_sim"],
- [y + 1 for y in range(len(random_result["context_sim"]))]
- )
- plt.show()
-
-if __name__ == "__main__":
- main()
diff --git a/experiment.py b/experiment.py
@@ -14,9 +14,7 @@ import tqdm
from transformers import AutoTokenizer, AutoModel
from sentence_transformers import SentenceTransformer
-
-
-DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
+from models import ModelFactory, Model
def parse_args() -> argparse.Namespace:
@@ -27,11 +25,7 @@ def parse_args() -> argparse.Namespace:
parser.add_argument(
"-experiment_type", type=str, choices=["sick", "ncip", "ncip-non"]
)
- parser.add_argument(
- "-model",
- type=str,
- choices=list(ModelFactory._factory.keys())
- )
+ parser.add_argument("-model", type=str, choices=list(ModelFactory._factory.keys()))
return parser.parse_args()
@@ -55,435 +49,18 @@ class Point(NamedTuple):
comp_class: str
-class ResultObject(NamedTuple):
- model_name: str
- dataset_name: str
- points: list[Point]
-
-
class SentencePairs(NamedTuple):
sentence_A: list[str]
sentence_B: list[str]
classes: list[str]
-def normalize_v2():
- """
- For a context c and a substring x (to be replaced), replace x with n candidates, embed them, calculate the mean the std.dev.
- Subtract the mean and divide by the std.dev (gaussian assumption) (mu(ac) - (mean(c with x)) / std.dev
-
- Sample x from a list of random words.
- """
- pass
-
-
-class Model(ABC):
-
- @abstractmethod
- def embed(self, string: str) -> torch.Tensor:
- pass
-
- def normalize(
- self,
- composed: torch.Tensor,
- original_phrase_removed: torch.Tensor,
- lambda_value: float,
- ):
- return composed - (original_phrase_removed * lambda_value)
-
- @abstractmethod
- def calculate_similarity(self, a: torch.Tensor, b: torch.Tensor) -> float:
- pass
-
-
-class ModelFactory:
-
- _factory: dict[str, type[Model]] = {}
-
- def __init__(self, name: str):
- self._name = name
-
- def __call__(self, model: type[Model]) -> type[Model]:
- self._factory[self._name] = model
- return model
-
- @classmethod
- def get(cls: type[ModelFactory], name: str, *args: Any, **kwargs: Any) -> Model:
- return cls._factory[name](*args, **kwargs)
-
-
-# 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"
- 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.lower(), return_tensors="pt")["input_ids"]
- ).last_hidden_state[:, 0, :]
-
- 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-cls")
-class ModernBERT_cls(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 self.encoder(
- self.tokenizer(string.lower(), return_tensors="pt")["input_ids"]
- ).last_hidden_state[:, 0, :]
-
- 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-mean")
-class ModernBERT_mean(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 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):
-
- 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 self.encoder(
- self.tokenizer(string.lower(), return_tensors="pt")["input_ids"]
- ).last_hidden_state[:, 0, :]
-
- 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-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):
-
- 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:
- 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("BERT-clssep")
-class BERT_clssep(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:
- 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("S-BERT")
-class S_BERT(Model):
-
- def __init__(self):
- self.encoder = SentenceTransformer("all-MiniLM-L6-v2", 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()
-
-
-@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).item()
-
-
-@ModelFactory("E5")
-class E5(Model):
-
- def __init__(self):
- self.encoder = SentenceTransformer(
- "intfloat/multilingual-e5-large-instruct", 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).item()
-
-
-@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).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.
Subtracts the longest common prefix and suffix and takes whats left. Not the smartest algorithm, perhaps.
"""
- if a == b: # Should not happen for SICK or NCIP, but you never know.
+ if a == b: # Should not happen, but you never know....
return None
try:
@@ -535,7 +112,6 @@ class Experiment:
self.model: Model = model
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
@@ -545,12 +121,13 @@ class Experiment:
self.phrase_b_length: list[int] = []
def run_experiment(self) -> None:
- for i, (sentence_a, sentence_b, cls) in enumerate(tqdm.tqdm(zip(*self.sentence_pairs))):
+ for i, (sentence_a, sentence_b, cls) in enumerate(
+ tqdm.tqdm(zip(*self.sentence_pairs))
+ ):
diffs: list[str] | None = find_diff(sentence_a, sentence_b)
if not diffs:
- self.diff_errors += 1
- print(f"Error at index {i}, sentence: {sentence_a}")
- continue
+ # We need exact matching of the pairs, so abort mission here.
+ raise Exception(f"Error at index {i}, sentence: {sentence_a}")
phrase_a: str
phrase_b: str
@@ -580,23 +157,6 @@ class Experiment:
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, "")
- 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
- )
- b_in_context_embedded = self.model.normalize(
- b_in_context_embedded, b_removed_embedded, self.normalization_value
- )
-
context_similarity: float = self.model.calculate_similarity(
a_in_context_embedded, b_in_context_embedded
)
@@ -613,7 +173,7 @@ class Experiment:
phrase_b_embedded=phrase_b_embedded,
context_ac_embedded=a_in_context_embedded,
context_bc_embedded=b_in_context_embedded,
- comp_class=cls
+ comp_class=cls,
)
)
@@ -622,21 +182,10 @@ 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()
- sep = "\t"
- if args.experiment_type in ["ncip", "ncip-non"]:
- sep = ","
-
+ args: argparse.Namespace = parse_args()
+ sep: str = ","
df: pandas.DataFrame = pandas.read_csv(args.data_path, sep=sep)
if args.experiment_type == "ncip":
@@ -650,39 +199,33 @@ def main() -> None:
synonym_sentence: list[str] = df[
"synonym both"
].values.tolist() # Both the modifier and the noun are swapped with synonyms from word net.
-
+
sentence_A: list[str] = original_sentence
sentence_B: list[str] = synonym_sentence
- # Validate the find_diff function somewhat
- for i, (sA, sB, goldA) in enumerate(zip(sentence_A, sentence_B, original_idiom)):
+ # Validate the find_diff function somewhat.
+ # The find_diff between sentA and sentB should return the origin MWE.
+ for i, (sA, sB, goldA) in enumerate(
+ zip(sentence_A, sentence_B, original_idiom)
+ ):
res = find_diff(sA, sB)
if res is not None:
diffa, diffb = res
if diffa != goldA:
- print(i, sA, sB, goldA, diffa, diffb)
- exit()
+ raise Exception(i, sA, sB, goldA, diffa, diffb)
elif args.experiment_type == "ncip-non":
# Here we measure the difference between two non-idiomatic phrases in context.
sentence_A = df["synonym both"].values.tolist()
sentence_B = df["both synonym alt1_1"].values.tolist()
-
- elif args.experiment_type == "sick":
- lexA = df.loc[df["sentence_A_expRule"] == "S2_lex"]
- lexB = df.loc[df["sentence_B_expRule"] == "S2_lex"]
- lex = pandas.concat([lexA, lexB])
- sentence_A = lex["sentence_A"].values.tolist()
- sentence_B = lex["sentence_B"].values.tolist()
-
else:
raise ValueError("Undefined dataset type...")
- clss: list[str] = df[
- "class"
- ].values.tolist() # The comp. classes
+ clss: list[str] = df["class"].values.tolist() # The comp. classes
- sentence_pairs = SentencePairs(sentence_A=sentence_A, sentence_B=sentence_B, classes=clss)
+ sentence_pairs = SentencePairs(
+ sentence_A=sentence_A, sentence_B=sentence_B, classes=clss
+ )
models = (
[args.model] if args.model is not None else list(ModelFactory._factory.keys())
@@ -691,8 +234,7 @@ def main() -> None:
for model_type in models:
model = ModelFactory.get(model_type)
result_path = (
- args.result_path
- / f"{args.experiment_type}_{model_type}_lambda{int(args.normalization_value * 10)}"
+ args.result_path / f"{args.experiment_type}_{model_type}"
) # shitty logging of lambda parameter, need to fix
experiment = Experiment(
sentence_pairs=sentence_pairs,
@@ -702,9 +244,7 @@ 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"Embedding errors: {experiment.embed_errors}")
print(f"Saving experiment to: {str(result_path)}")
sent_a_mean = numpy.mean(experiment.sentence_a_length)
@@ -726,7 +266,6 @@ 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
@@ -9,7 +9,6 @@ import matplotlib.pyplot as plt
from typing import Any
import json
-
import pandas
import argparse
@@ -24,12 +23,6 @@ def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Gather epsilon values")
parser.add_argument("-points_path", type=pathlib.Path)
parser.add_argument("-result_path", type=pathlib.Path)
- parser.add_argument(
- "-percentile",
- type=float,
- default=None,
- help="The percentile for the epsilon value",
- )
return parser.parse_args()
@@ -40,23 +33,6 @@ 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]]:
@@ -90,14 +66,16 @@ def main() -> None:
"original": [],
"divisive": [],
"additive": [],
- "simscores": [],
"classes": [],
}
+
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)
+ dataset_name, model_name = get_dataset_and_model_name(
+ raw_file.with_suffix("").name
+ )
+ print(f"Getting epsilons for: {model_name}")
n: int = len(points)
word_sims: numpy.ndarray = numpy.array(
[point.phrase_similarity for point in points]
@@ -106,37 +84,20 @@ def main() -> None:
[point.contextualized_similarity for point in points]
)
classes = [point.comp_class 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]
+ 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
- final_original: Any = float(
- numpy.percentile(tmp_result["original"], args.percentile)
- )
- final_division: Any = float(
- numpy.percentile(tmp_result["divisive"], args.percentile)
- )
- final_additive: Any = float(
- numpy.percentile(tmp_result["additive"], args.percentile)
- )
- else:
- final_original = tmp_result["original"]
- final_division = tmp_result["divisive"]
- final_additive = tmp_result["additive"]
-
- final_result["original"].append(final_original)
- final_result["divisive"].append(final_division)
- final_result["additive"].append(final_additive)
- final_result["classes"] = classes
-
- # df: pandas.DataFrame = pandas.DataFrame.from_dict(final_result)
- # df.to_csv(args.result_path, sep=",", index=False)
+ final_result["original"].append(tmp_result["original"])
+ final_result["divisive"].append(tmp_result["divisive"])
+ final_result["additive"].append(tmp_result["additive"])
+ final_result["classes"].append(classes)
+
with open(args.result_path, "w") as f:
json.dump(final_result, f)
diff --git a/models.py b/models.py
@@ -0,0 +1,429 @@
+from __future__ import annotations
+from typing import NamedTuple, Any
+from abc import ABC, abstractmethod
+from abc import ABC, abstractmethod
+from transformers import AutoTokenizer, AutoModel
+from sentence_transformers import SentenceTransformer
+import torch
+
+DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
+
+
+class Model(ABC):
+
+ @abstractmethod
+ def embed(self, string: str) -> torch.Tensor:
+ pass
+
+ def normalize(
+ self,
+ composed: torch.Tensor,
+ original_phrase_removed: torch.Tensor,
+ lambda_value: float,
+ ):
+ return composed - (original_phrase_removed * lambda_value)
+
+ @abstractmethod
+ def calculate_similarity(self, a: torch.Tensor, b: torch.Tensor) -> float:
+ pass
+
+
+class ModelFactory:
+
+ _factory: dict[str, type[Model]] = {}
+
+ def __init__(self, name: str):
+ self._name = name
+
+ def __call__(self, model: type[Model]) -> type[Model]:
+ self._factory[self._name] = model
+ return model
+
+ @classmethod
+ def get(cls: type[ModelFactory], name: str, *args: Any, **kwargs: Any) -> Model:
+ return cls._factory[name](*args, **kwargs)
+
+
+# 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"
+ 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.lower(), return_tensors="pt")["input_ids"]
+ ).last_hidden_state[:, 0, :]
+
+ 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-cls")
+class ModernBERT_cls(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 self.encoder(
+ self.tokenizer(string.lower(), return_tensors="pt")["input_ids"]
+ ).last_hidden_state[:, 0, :]
+
+ 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-mean")
+class ModernBERT_mean(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 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):
+
+ 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 self.encoder(
+ self.tokenizer(string.lower(), return_tensors="pt")["input_ids"]
+ ).last_hidden_state[:, 0, :]
+
+ 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-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):
+
+ 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:
+ 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("BERT-clssep")
+class BERT_clssep(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:
+ 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("S-BERT")
+class S_BERT(Model):
+
+ def __init__(self):
+ self.encoder = SentenceTransformer("all-MiniLM-L6-v2", 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()
+
+
+@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).item()
+
+
+@ModelFactory("E5")
+class E5(Model):
+
+ def __init__(self):
+ self.encoder = SentenceTransformer(
+ "intfloat/multilingual-e5-large-instruct", 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).item()
+
+
+@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).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()
diff --git a/random_experiment.py b/random_experiment.py
@@ -23,10 +23,7 @@ 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("-n", type=int, default=100, help="The number of samples to calculate for")
- parser.add_argument(
- "-experiment_type", type=str, choices=["sick", "ncip", "ncip-non"]
- )
+ parser.add_argument("-n", type=int, default=1000, help="The number of samples to calculate for")
parser.add_argument(
"-model",
type=str,
@@ -41,14 +38,17 @@ def main() -> None:
df: pandas.DataFrame = pandas.read_csv(args.data_path, sep=sep)
print(f"Similarity between random items, n_samples={args.n}")
+
sentences: list[str] = df["original sentence"].values.tolist()
- phrases: list[str] = df["compound"].values.tolist()
+ phrases: list[str] = df["compound noun"].values.tolist()
models = [args.model] if args.model is not None else ModelFactory._factory.keys()
result: dict[str, list] = {
"model": [],
"phrase_sim": [],
- "context_sim": []
+ "context_sim": [],
+ "phrase_sim_avg": [],
+ "context_sim_avg": []
}
for model_type in models:
@@ -74,10 +74,11 @@ def main() -> None:
phrase_sim = model.calculate_similarity(phrase_a_embedded, phrase_b_embedded)
phrase_sims.append(phrase_sim)
- print(f"{context_sim}")
result["model"].append(model_type)
result["phrase_sim"].append(phrase_sims)
result["context_sim"].append(context_sims)
+ result["phrase_sim_avg"].append(numpy.mean(phrase_sims))
+ result["context_sim_avg"].append(numpy.mean(context_sims))
print(f"Model: {model}")
print("Random context sims", numpy.mean(context_sims))
diff --git a/statistical_testing.py b/statistical_testing.py
@@ -26,12 +26,30 @@ 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")
+ parser.add_argument(
+ "-epsilon_type",
+ type=str,
+ choices=["original", "divisive", "additive"],
+ default="original",
+ help="Use the score method based on the He et al paper",
+ )
+ parser.add_argument(
+ "-class_type",
+ type=str,
+ nargs="*",
+ choices=["C", "NC", "PC", "NCPC"],
+ help="The compositionality class to look at",
+ )
return parser.parse_args()
-class ResultPoint():
- def __init__(self, model_name: str = "", idiomatic_result: list[float] = [], non_idiomatic_result: list[float] = []):
+class ResultPoint:
+ def __init__(
+ self,
+ model_name: str = "",
+ idiomatic_result: list[float] = [],
+ non_idiomatic_result: list[float] = [],
+ ):
self.model_name: str = model_name
self.idiomatic_result: list[float] = idiomatic_result
self.non_idiomatic_result: list[float] = non_idiomatic_result
@@ -62,27 +80,41 @@ 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"
+ key = args.epsilon_type
if model_name not in model_results:
model_results[model_name] = ResultPoint(model_name=model_name)
+
+ result_list: list[float] = []
+ # Filter the type of samples we look for (C, NC, PC)
+ for item, c in zip(loaded_dict[key][i], loaded_dict["classes"][i]):
+ if c in args.class_type:
+ result_list.append(item)
+
if is_idiomatic:
- model_results[model_name].idiomatic_result = loaded_dict[key][i]
+ model_results[model_name].idiomatic_result = result_list
else:
- model_results[model_name].non_idiomatic_result = loaded_dict[key][i]
+ model_results[model_name].non_idiomatic_result = result_list
for model, res in model_results.items():
idiom_variance = numpy.var(res.idiomatic_result)
non_idiom_variance = numpy.var(res.non_idiomatic_result)
idiom_is_normal = stats.normaltest(res.idiomatic_result).pvalue > 0.05
non_idiom_is_normal = stats.normaltest(res.non_idiomatic_result).pvalue > 0.05
- coxon = stats.wilcoxon(res.idiomatic_result, res.non_idiomatic_result) # Dependence, not normally distributed, unequal sample size
- U1 = coxon.statistic
- U2 = len(res.idiomatic_result) * len(res.non_idiomatic_result) - U1 # as per the documentaiton notes on the scipy function
- cl_effect_size = U1 / (len(res.idiomatic_result) * len(res.non_idiomatic_result)) # as per Wikipedia on Mann Whitney U
- r_effect_size = ((2 * U1) / (len(res.idiomatic_result) * len(res.non_idiomatic_result))) - 1 # as per Wikipedia on Mann Whitney U
+ coxon = stats.wilcoxon(
+ res.idiomatic_result, res.non_idiomatic_result, alternative="greater"
+ )
+ T = coxon.statistic
+ # As per the Wikipedia page of the Wilxocon signed-rank test. T divided by the sum of ranks.
+ rank_sums = sum([i for i in range(1, len(res.idiomatic_result) + 1)])
+ r_effect_size = T / rank_sums
is_signficant: bool = coxon.pvalue < 0.05
- hodges_lehman = hodges_lehmann_2sample(res.idiomatic_result, res.non_idiomatic_result) # To gauge the practical difference between the two groups, as suggested by the Wikipedia page on Mann Whitney U
- print(f"######### \n Model name: {model} \n Idiom Variance (n={len(res.idiomatic_result)}): {idiom_variance}, Idiom is normally distributed: {idiom_is_normal}, Non-Idiom Variance (n={len(res.non_idiomatic_result)}): {non_idiom_variance}, Non-idiom is normal distributed: {non_idiom_is_normal} \n Hodges-Lehman Estimator: {hodges_lehman} \n Two-sided Wilcoxon signed-rank test: p_value={coxon.pvalue}, is_significant: {is_signficant} \n Effect size (Common Language F-test): {cl_effect_size} \n Effect size (Rank-biserial correlation): {r_effect_size}")
+ hodges_lehman = hodges_lehmann_2sample(
+ res.idiomatic_result, res.non_idiomatic_result
+ ) # To gauge the practical difference between the two groups, as suggested by the Wikipedia page on Mann Whitney U
+
+ print(
+ f"######### \n Model name: {model} \n Idiom Variance (n={len(res.idiomatic_result)}): {idiom_variance}, Idiom is normally distributed: {idiom_is_normal}, Non-Idiom Variance (n={len(res.non_idiomatic_result)}): {non_idiom_variance}, Non-idiom is normal distributed: {non_idiom_is_normal} \n Hodges-Lehman Estimator: {hodges_lehman} \n Two-sided Wilcoxon signed-rank test: p_value={coxon.pvalue}, is_significant: {is_signficant} \n Effect size (Rank-biserial correlation): {r_effect_size}"
+ )
if __name__ == "__main__":
