commit 686525fcec3aacf31d60ec9a1c312d14cf3a7bbd
parent 6f26c1c43b327708c744a8c8b4b30ee896b6f3a4
Author: Sondre Wold <sondrewold1@gmail.com>
Date: Sat, 12 Jul 2025 16:42:49 +0200
Experiment class
Diffstat:
| A | experiment.py | | | 251 | +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ |
1 file changed, 251 insertions(+), 0 deletions(-)
diff --git a/experiment.py b/experiment.py
@@ -0,0 +1,251 @@
+from __future__ import annotations
+from typing import Callable, NamedTuple, Any
+
+import sys
+import pandas
+import pathlib
+import argparse
+
+from abc import ABC, abstractmethod
+import torch
+import pickle
+import tqdm
+
+from transformers import AutoTokenizer, AutoModel
+from sentence_transformers import SentenceTransformer
+
+
+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"])
+ return parser.parse_args()
+
+
+class Point(NamedTuple):
+ phrase_1: str
+ phrase_2: str
+ context_1: str
+ context_2: str
+ phrase_similarity: float
+ contextualized_similarity: float
+
+
+class SentencePairs(NamedTuple):
+ sentence_A: list[str]
+ sentence_B: list[str]
+
+
+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)
+
+
+@ModelFactory("BERT_cls")
+class BERT_cls(Model):
+
+ def __init__(self):
+ self.model_name = "bert-base-uncased"
+ self.tokenizer = AutoTokenizer.from_pretrained(self.model_name)
+ self.encoder = AutoModel.from_pretrained(self.model_name)
+
+ def embed(self, string: str) -> torch.Tensor:
+ return self.encoder(
+ self.tokenizer(string, 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:
+ 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")
+
+ 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)
+
+
+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.
+ """
+ if a == b: # Should not happen for SICK, but you never know.
+ return None
+
+ words_a = a.split()
+ words_b = b.split()
+
+ prefix_len = 0
+ min_len = min(len(words_a), len(words_b))
+ for i in range(min_len):
+ if words_a[i] == words_b[i]:
+ prefix_len += 1
+ else:
+ break
+
+ suffix_len = 0
+ for i in range(1, min_len - prefix_len + 1):
+ if words_a[-i] == words_b[-i]:
+ suffix_len += 1
+ else:
+ break
+
+ diff_a = words_a[
+ prefix_len : len(words_a) - suffix_len if suffix_len > 0 else len(words_a)
+ ]
+ diff_b = words_b[
+ prefix_len : len(words_b) - suffix_len if suffix_len > 0 else len(words_b)
+ ]
+
+ if not diff_a or not diff_b:
+ return None
+ else:
+ diff_str_a = " ".join(diff_a) if diff_a else ""
+ diff_str_b = " ".join(diff_b) if diff_b else ""
+ if diff_str_b == b or diff_str_a == a:
+ return None
+ return [diff_str_a, diff_str_b]
+
+
+class Experiment:
+ def __init__(self, sentence_pairs: SentencePairs, model: Model, result_path: pathlib.Path, normalization_value: float):
+ self.model: Model = model
+ self.sentence_pairs = sentence_pairs
+ self.normalization_value: float = normalization_value
+ self.diff_errors: int = 0
+ self.points: list[Point] = []
+ self.result_path: pathlib.Path = result_path
+
+
+ def run_experiment(self):
+ for sentence_a, sentence_b in tqdm.tqdm(zip(*self.sentence_pairs)):
+ diffs: list[str] | None = find_diff(sentence_a, sentence_b)
+ if not diffs:
+ self.diff_errors += 1
+ continue
+
+ phrase_a: str
+ phrase_b: str
+ phrase_a, phrase_b = diffs
+
+ 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)
+
+ 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)
+
+ 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)
+
+ 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)
+
+ self.points.append(
+ Point(
+ phrase_1=phrase_a,
+ phrase_2=phrase_b,
+ context_1=sentence_a,
+ context_2=b_in_a,
+ phrase_similarity=phrase_similarity,
+ contextualized_similarity=context_similarity
+ )
+ )
+
+ def save_results(self):
+ pathlib.Path(self.result_path).mkdir(parents=True, exist_ok=True)
+ with open(self.result_path.with_suffix(".pkl"), "wb") as file:
+ pickle.dump(self.points, file)
+
+
+def main() -> None:
+ args = parse_args()
+ sep = "\t"
+ if args.experiment_type == "ncip":
+ sep = ","
+
+ df: pandas.DataFrame = pandas.read_csv(args.data_path, sep=sep)
+
+ if args.experiment_type == "ncip":
+ original_sentence: list[str] = df["original sentence"].values.tolist()
+ synonym_sentence: list[str] = df["synonym both"].values.tolist()
+ synonym_sentence2: list[str] = df["both synonym alt2_1"].values.tolist()
+ synonym_sentence3: list[str] = df["both synonym alt3_1"].values.tolist()
+ synonym_sentence4: list[str] = df["both synonym alt4_1"].values.tolist()
+
+ sentence_A = original_sentence * 4
+ sentence_B = synonym_sentence + synonym_sentence2 + synonym_sentence3 + synonym_sentence4
+ sentence_pairs = SentencePairs(sentence_A=sentence_A, sentence_B=sentence_B)
+ 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: list[str] = lex["sentence_A"].values.tolist()
+ sentence_B: list[str] = lex["sentence_B"].values.tolist()
+ sentence_pairs = SentencePairs(sentence_A=sentence_A, sentence_B=sentence_B)
+
+ else:
+ assert ValueError("Undefined experiment type...")
+
+
+ for model_type in ModelFactory._factory.keys():
+ model = ModelFactory.get(model_type)
+ result_path = args.result_path / f"{args.experiment_type}_{model_type}_lambda{int(args.normalization_value * 10)}" # shitty logging of lambda parameter, need to fix
+ experiment = Experiment(
+ sentence_pairs=sentence_pairs,
+ model=model,
+ result_path = result_path,
+ normalization_value=args.normalization_value,
+ )
+ print(f"Running experiment for model: {model_type}.")
+ experiment.run_experiment()
+ print(f"Saving experiment to: {str(result_path)}")
+ experiment.save_results()
+
+
+if __name__ == "__main__":
+ main()
+
