Spaces:
Sleeping
Sleeping
import torch | |
from transformers import ( | |
AutoTokenizer, | |
AutoModelForSeq2SeqLM, | |
AutoModelForTokenClassification, | |
pipeline) | |
from transformers import Pipeline | |
import re | |
model = AutoModelForSeq2SeqLM.from_pretrained("emirhangazi77/Turkish-T5") | |
tokenizer = AutoTokenizer.from_pretrained("emirhangazi77/Turkish-T5") | |
ner_model = AutoModelForTokenClassification.from_pretrained("akdeniz27/bert-base-turkish-cased-ner") # pretrained ner model | |
ner_tokenizer = AutoTokenizer.from_pretrained("akdeniz27/bert-base-turkish-cased-ner") # pretrained ner tokenizer | |
ner = pipeline('ner', model=ner_model, tokenizer=ner_tokenizer, aggregation_strategy="first") # | |
device = torch.device('cpu') | |
class Diacritic_Pipe(Pipeline): | |
def __init__(self,ner,model,tokenizer): | |
super().__init__(model = model, tokenizer = tokenizer) | |
self.ner_pipe = ner | |
def generate_result(self,text): | |
prefix = "Correct diacritics for : " | |
postfix = " </s>" | |
text = prefix + text + postfix | |
self.tokenizer.truncation_side = "left" | |
batch = self.tokenizer(text, return_tensors='pt', max_length = 64, truncation = False).to(device) | |
result = self.model.generate(**batch, max_new_tokens = 128) | |
result = self.tokenizer.batch_decode(result) | |
return str(result[0]) | |
def ner_predict_mapping(self,text, threshold=0.3): | |
result = self.ner_pipe(text) | |
if len(result) == 0: | |
return [] | |
else: | |
special_words = [result["word"] for result in result if result["score"] > threshold] | |
special_words_ = [] | |
for word_ in special_words: | |
if word_.lower()[0] == "i": | |
word_ = word_.replace("I","İ") | |
if len(word_.split()) > 1: | |
special_words_.extend(word_.split()) | |
else: | |
special_words_.append(word_) | |
return special_words_ | |
def split_text_into_n_worded_chunks(self,text, n): | |
words = text.split() | |
chunks = [] | |
for i in range(0, len(words), n): | |
chunks.append(' '.join(words[i:i+n])) | |
last_chunk_words = len(words) % n | |
if last_chunk_words != 0: | |
chunks[-1] = ' '.join(words[-last_chunk_words:]) | |
return chunks | |
def chunk_2(self,text): | |
chunks = self.split_text_into_n_worded_chunks(text, 2) | |
processed_chunks = [re.sub(r'(["q(°\[\]{}&´])\s+', r'\1',self.generate_result(chunk)) for chunk in chunks] | |
result = ' '.join(processed_chunks) | |
return result.replace("<pad>","").replace("</s>","").replace(" "," ") | |
def chunk_1(self,text): | |
chunks = self.split_text_into_n_worded_chunks(text, 1) | |
processed_chunks = [self.generate_result(chunk).replace(" ","") for chunk in chunks] | |
result = ''.join(processed_chunks) | |
return result.replace("<pad>"," ").replace("</s>","") | |
def process_text(self,text): | |
words = self.ner_predict_mapping(text) | |
two_chunk = self.chunk_2(text) | |
one_chunk = self.chunk_1(text) | |
if len(one_chunk.split()) != len(two_chunk.split()): | |
for word in words: | |
one_chunk = one_chunk.replace(word.lower().replace('i̇',"i"),word) | |
return one_chunk | |
else: | |
for word in words: | |
two_chunk = two_chunk.replace(word.lower().replace('i̇',"i"),word) | |
return two_chunk | |
def _sanitize_parameters(self, **kwargs): | |
preprocess_kwargs = {} | |
if "maybe_arg" in kwargs: | |
preprocess_kwargs["maybe_arg"] = kwargs["maybe_arg"] | |
return preprocess_kwargs, {}, {} | |
def preprocess(self, inputs, maybe_arg=2): | |
return {"model_input": inputs} | |
def _forward(self, model_inputs): | |
#model_inputs == {"model_input": model_input} | |
outputs = self.process_text(model_inputs["model_input"]) | |
# Maybe {"logits": Tensor(...)} | |
return outputs | |
def postprocess(self, model_outputs): | |
return model_outputs | |
import gradio as gr | |
diacritics = Diacritic_Pipe(ner = ner , model = model , tokenizer = tokenizer) | |
def fn(query): | |
response = diacritics(query) | |
return str(response) | |
def my_chatbot(input, history): | |
history = history or [] | |
my_history = list(sum(history, ())) | |
my_history.append(input) | |
my_input = ' '.join(my_history) | |
output = fn(input) | |
history.append((input, output)) | |
return history, history | |
import gradio as gr | |
import logging | |
import sys | |
logging.basicConfig(stream=sys.stdout, level=logging.INFO) | |
logging.getLogger().addHandler(logging.StreamHandler(stream=sys.stdout)) | |
with gr.Blocks() as demo: | |
gr.Markdown("""<h1><center>Diacritics on Turkish</center></h1>""") | |
chatbot = gr.Chatbot() | |
state = gr.State() | |
txt = gr.Textbox(show_label=False, placeholder="Ask me a question and press enter.") | |
txt.submit(my_chatbot, inputs=[txt, state], outputs=[chatbot, state]) | |
demo.launch(share=True) | |