|
import pandas as pd
|
|
import numpy as np
|
|
import matplotlib.pyplot as plt
|
|
from sklearn.model_selection import train_test_split
|
|
from keras.preprocessing.sequence import pad_sequences
|
|
from keras.utils import to_categorical
|
|
from keras.models import Sequential, load_model
|
|
from keras.layers import LSTM, Embedding, Dense, TimeDistributed, Bidirectional
|
|
import keras
|
|
import os
|
|
import banglanltk as bn
|
|
|
|
|
|
data = pd.read_excel("b-nersuzi.xlsx", sheet_name="b-ner")
|
|
|
|
|
|
data = data.fillna(method='ffill')
|
|
|
|
|
|
agg_func = lambda s: [(w, t) for w, t in zip(s["Word"].values.tolist(), s['Tag'].values.tolist())]
|
|
agg_data = data.groupby(['Sentence #']).apply(agg_func).reset_index().rename(columns={0:'Sentence_POS_Tag_Pair'})
|
|
|
|
|
|
def preprocess_data(data):
|
|
data['Sentence'] = data['Sentence_POS_Tag_Pair'].apply(lambda sentence: " ".join(map(str, [s[0] for s in sentence])))
|
|
data['Tag'] = data['Sentence_POS_Tag_Pair'].apply(lambda sentence: " ".join(map(str, [s[1] for s in sentence])))
|
|
data['tokenised_sentences'] = data['Sentence'].apply(bn.word_tokenize)
|
|
data['tag_list'] = data['Tag'].apply(lambda x: x.split())
|
|
return data
|
|
|
|
|
|
agg_data = preprocess_data(agg_data)
|
|
|
|
|
|
tokenized_sentences = agg_data['tokenised_sentences'].tolist()
|
|
tags_list = agg_data['tag_list'].tolist()
|
|
|
|
|
|
words = set(word for sent in tokenized_sentences for word in sent)
|
|
word_to_idx = {word: i + 1 for i, word in enumerate(words)}
|
|
num_words = len(words) + 1
|
|
tags = set(tag for tag_list in tags_list for tag in tag_list)
|
|
tag_to_idx = {tag: i for i, tag in enumerate(tags)}
|
|
num_tags = len(tags)
|
|
|
|
|
|
max_len = max(len(sent) for sent in tokenized_sentences)
|
|
encoded_sentences = [[word_to_idx[word] for word in sent] for sent in tokenized_sentences]
|
|
encoded_sentences = pad_sequences(encoded_sentences, maxlen=max_len, padding='post')
|
|
encoded_tags = [[tag_to_idx[tag] for tag in tag_list] for tag_list in tags_list]
|
|
encoded_tags = pad_sequences(encoded_tags, maxlen=max_len, padding='post')
|
|
encoded_tags = [to_categorical(tag, num_classes=num_tags) for tag in encoded_tags]
|
|
|
|
|
|
X_train, X_test, y_train, y_test = train_test_split(encoded_sentences, encoded_tags, test_size=0.2, random_state=42)
|
|
|
|
|
|
model_path = "best_model.h5"
|
|
if os.path.exists(model_path):
|
|
model = load_model(model_path)
|
|
else:
|
|
model = Sequential()
|
|
model.add(Embedding(input_dim=num_words, output_dim=50, input_length=max_len))
|
|
model.add(Bidirectional(LSTM(units=100, return_sequences=True)))
|
|
model.add(TimeDistributed(Dense(units=num_tags, activation='softmax')))
|
|
|
|
model.compile(optimizer='adam', loss='categorical_crossentropy', metrics=['accuracy'])
|
|
|
|
|
|
class SaveModelCallback(keras.callbacks.Callback):
|
|
def on_epoch_end(self, epoch, logs={}):
|
|
if logs.get('val_accuracy') >= 0.99:
|
|
self.model.save("best_model.h5")
|
|
print("\nValidation accuracy reached 99% or above. Model saved.")
|
|
|
|
|
|
history = model.fit(X_train, np.array(y_train), batch_size=32, epochs=7, validation_split=0.1, callbacks=[SaveModelCallback()])
|
|
|
|
|
|
loss, accuracy = model.evaluate(X_test, np.array(y_test))
|
|
print("Test Loss:", loss)
|
|
print("Test Accuracy:", accuracy)
|
|
|
|
|
|
def predict_entities(input_sentence):
|
|
tokenized_input = bn.word_tokenize(input_sentence)
|
|
encoded_input = [word_to_idx[word] if word in word_to_idx else 0 for word in tokenized_input]
|
|
padded_input = pad_sequences([encoded_input], maxlen=max_len, padding='post')
|
|
predictions = model.predict(padded_input)
|
|
predicted_tags = np.argmax(predictions, axis=-1)
|
|
reverse_tag_map = {v: k for k, v in tag_to_idx.items()}
|
|
predicted_tags = [reverse_tag_map[idx] for idx in predicted_tags[0][:len(tokenized_input)]]
|
|
tagged_sentence = [(word, tag) for word, tag in zip(tokenized_input, predicted_tags)]
|
|
return tagged_sentence
|
|
|
|
|
|
user_input = input("Enter a Bengali sentence: ")
|
|
predicted_tags = predict_entities(user_input)
|
|
for word, tag in predicted_tags:
|
|
print(f"{word}: {tag}") |