Spaces:
Sleeping
Sleeping
from transformers import pipeline | |
import torch | |
import random | |
from masking_methods import MaskingProcessor | |
class SamplingProcessorWithPipeline: | |
def __init__(self, model_name='bert-base-uncased'): | |
self.unmasker = pipeline('fill-mask', model=model_name) | |
self.tokenizer = self.unmasker.tokenizer | |
def fill_masked_sentence(self, masked_sentence, sampling_technique, temperature=1.0): | |
""" | |
Fills each mask in the masked sentence using the specified sampling technique. | |
Args: | |
masked_sentence (str): Sentence with [MASK] tokens. | |
sampling_technique (str): Sampling technique to use (e.g., "inverse_transform", "exponential_minimum", "temperature", "greedy"). | |
temperature (float): Temperature parameter for sampling methods. | |
Returns: | |
str: Sentence with the masks filled. | |
""" | |
while '[MASK]' in masked_sentence: | |
# Get predictions for the first [MASK] | |
predictions = self.unmasker(masked_sentence) | |
print(f' predictions : {predictions}') | |
print(f' type of predictions : {type(predictions)}') | |
# Ensure predictions is a list of dictionaries | |
if not isinstance(predictions, list) or not all(isinstance(pred, dict) for pred in predictions): | |
raise ValueError("Unexpected structure in predictions from the pipeline.") | |
# Extract logits (scores) from the predictions | |
logits = torch.tensor([pred['score'] for pred in predictions], dtype=torch.float32) | |
if sampling_technique == "inverse_transform": | |
probs = torch.softmax(logits / temperature, dim=-1) | |
cumulative_probs = torch.cumsum(probs, dim=-1) | |
random_prob = random.random() | |
sampled_index = torch.where(cumulative_probs >= random_prob)[0][0].item() | |
elif sampling_technique == "exponential_minimum": | |
probs = torch.softmax(logits / temperature, dim=-1) | |
exp_probs = torch.exp(-torch.log(probs)) | |
random_probs = torch.rand_like(exp_probs) | |
sampled_index = torch.argmax(random_probs * exp_probs).item() | |
elif sampling_technique == "temperature": | |
logits = torch.clamp(logits, min=-1e8, max=1e8) | |
probs = torch.softmax(logits / temperature, dim=-1) | |
if torch.any(torch.isnan(probs)) or torch.any(torch.isinf(probs)): | |
raise ValueError("The computed probabilities contain NaN or inf values.") | |
probs = torch.max(probs, torch.tensor(1e-8, device=logits.device)) | |
probs = probs / torch.sum(probs) | |
probs = probs.flatten() | |
if probs.size(0) > 1: | |
sampled_index = torch.multinomial(probs, 1).item() | |
else: | |
sampled_index = torch.argmax(probs).item() | |
elif sampling_technique == 'greedy': | |
sampled_index = torch.argmax(logits).item() | |
else: | |
raise ValueError(f"Unknown sampling technique: {sampling_technique}") | |
# Replace the first [MASK] with the selected word | |
sampled_token = predictions[sampled_index]['token_str'] | |
masked_sentence = masked_sentence.replace('[MASK]', sampled_token, 1) | |
return masked_sentence | |
# Example usage | |
if __name__ == "__main__": | |
from transformers import BertTokenizer | |
# Define sentences and result_dict | |
sentences = [ | |
"The quick brown fox jumps over the lazy dog.", | |
"A quick brown dog outpaces a lazy fox.", | |
"Quick brown dog leaps over lazy the fox." | |
] | |
result_dict = { | |
"The quick brown fox jumps over the lazy dog.": {'quick brown': [(0, 1)], 'fox': [(2, 2)], 'lazy': [(4, 4)], 'dog': [(5, 5)]}, | |
"A quick brown dog outpaces a lazy fox.": {'quick brown': [(0, 1)], 'fox': [(5, 5)], 'lazy': [(4, 4)], 'dog': [(2, 2)]}, | |
"Quick brown dog leaps over lazy the fox.": {'quick brown': [(0, 1)], 'fox': [(5, 5)], 'lazy': [(4, 4)], 'dog': [(2, 2)]} | |
} | |
masking_processor = MaskingProcessor() | |
masking_results = masking_processor.process_sentences(sentences, result_dict, method="random", remove_stopwords=False) | |
# Use SamplingProcessor | |
sampling_processor = SamplingProcessorWithPipeline() | |
# Iterate through masking results to apply sampling | |
for sentence, result in masking_results.items(): | |
print(f"Original Sentence (Random): {sentence}") | |
print(f"Masked Sentence (Random): {result['masked_sentence']}") | |
masked_sentence = result["masked_sentence"] | |
# Apply different sampling techniques | |
for technique in ["inverse_transform", "exponential_minimum", "temperature", "greedy"]: | |
print(f"Sampling Technique: {technique}") | |
filled_sentence = sampling_processor.fill_masked_sentence( | |
masked_sentence=masked_sentence, | |
sampling_technique=technique, | |
temperature=1.0 # Adjust temperature as needed | |
) | |
print(f"Filled Sentence: {filled_sentence}\n") | |
print('--------------------------------') | |