Notes

import gensim
import torch
import torch.nn as nn
import torch.optim as optim
import pandas as pd
import numpy as np

class Tokenizer:

def __init__(self):
self.dictionary = {}
self.reverse_dictionary = {}

# Add the padding token
self.__add_to_dict('<pad>')

# Add characters and numbers to the dictionary
for i in range(10):
self.__add_to_dict(str(i))
for i in range(26):
self.__add_to_dict(chr(ord('a') + i))
self.__add_to_dict(chr(ord('A') + i))

# Add space and punctuation to the dictionary
self.__add_to_dict('.')
self.__add_to_dict(' ')
symbols = ['!', '"', '#', '$', '%', '&', "'", '(', ')', '*', '+', ',', '-', '/',
':', ';', '<', '=', '>', '?', '@', '[', '\' , ']', '^', '_', '`', '{', '|',
'}', '~','’', 'xa0', 'n']
for symbol in symbols:
self.__add_to_dict(symbol)


def __add_to_dict(self, character):
if character not in self.dictionary:
self.dictionary[character] = len(self.dictionary)
self.reverse_dictionary[self.dictionary[character]] = character

def tokenize(self, text):
return [self.dictionary[c] for c in text]

def character_to_token(self, character):
return self.dictionary[character]

def token_to_character(self, token):
return self.reverse_dictionary[token]

def size(self):
return len(self.dictionary)


# Define the environment

class CustomEnv:
def __init__(self, data_df,tokenizer):
self.data_df = data_df
self.tokenizer = tokenizer
self.num_episodes = len(data_df)
self.current_episode = 0

def reset(self):
self.current_episode += 1
if self.current_episode >= self.num_episodes:
self.current_episode = 0
episode = self.data_df.iloc[self.current_episode]
return episode['question']

def step(self, action):
episode = self.data_df.iloc[self.current_episode]
state = episode['question']
correct_response = episode['correct_answer']
incorrect_response = episode['incorrect_answer']

#tokenize the state(question)
state_token = self.tokenizer.tokenize(state)
state_numerical = [self.tokenizer.character_to_token(token) for token in state_tokens]


new_response = generate_response(state) # Use your language model here
state_tensor = torch.tensor(state_numerical,dtype = torch.float32)
# Define your reward function based on similarity metrics or other criteria
reward = 1.0 # Placeholder, replace with your reward calculation
done = True # Episodes terminate after a single step
return state_tensor, reward, done


# Define the policy network
class PolicyNetwork(nn.Module):
def __init__(self, state_dim, action_dim):
super(PolicyNetwork, self).__init__()
self.fc = nn.Sequential(
nn.Linear(state_dim, 64),
nn.ReLU(),
nn.Linear(64, action_dim),
nn.Softmax(dim=-1)
)

def forward(self, state):
return self.fc(state)





from pandas_ods_reader import read_ods


data = read_ods("/content/drive/MyDrive/sample_data/Fire_security.ods", columns=["question", "correct_response","incorrect_response"])


tokenizer = Tokenizer()

# Initialize the environment and policy network
env = CustomEnv(data_df = data, tokenizer = tokenizer)
policy = PolicyNetwork(tokenizer.size(), 2) # Adapt the input and output dimensions
optimizer = optim.Adam(policy.parameters(), lr=0.01)








# Training loop
num_episodes = 1000
for episode in range(num_episodes):
state = env.reset()
episode_states = []
episode_actions = []
episode_rewards = []

while True:
state_tensor = torch.tensor([state], dtype=torch.float32)
episodes_states.append(state_tensor)
action_probs = policy(state_tensor)
action = np.random.choice(2, p=action_probs.detach().numpy()[0])
next_state, reward, done = env.step(action)

# episode_states.append(state)
episode_actions.append(action)
episode_rewards.append(reward)

state = next_state

if done:
break

# Compute discounted returns
discounted_returns = []
running_add = 0
for r in reversed(episode_rewards):
running_add = r + 0.99 * running_add # Discount factor: 0.99
discounted_returns.insert(0, running_add)

# Calculate loss and perform policy gradient update
action_probs = policy(torch.cat(episode_states))
selected_action_probs = torch.gather(action_probs, 1, torch.tensor(episode_actions).unsqueeze(1))
loss = -torch.sum(torch.log(selected_action_probs) * torch.FloatTensor(discounted_returns))

optimizer.zero_grad()
loss.backward()
optimizer.step()

# Print episode information
if episode % 10 == 0:
print(f"Episode {episode}, Total Reward: {np.sum(episode_rewards)}")

# Use the trained policy for inference
with torch.no_grad():
test_state = env.reset()
while True:
test_state = torch.tensor([test_state], dtype=torch.float32)
action_probs = policy(test_state)
action = np.argmax(action_probs.detach().numpy()[0])
next_state, _, done = env.step(action)
test_state = next_state
if done:
break



# Model training and response generation



pip install pandas_ods_reader

import torch
from torch.utils.data import DataLoader, Dataset
from pandas_ods_reader import read_ods
import pandas as pd
import nltk
nltk.download('punkt')
max_sequence_length = 200

# Start, padding, and end tokens
START_TOKEN = '[START]'
PADDING_TOKEN = '[PAD]'
END_TOKEN = '[END]'

# Load your dataset into a DataFrame
data = read_ods("/content/drive/MyDrive/sample_data/Fire_security.ods", columns=["question", "answer"])

questions = data['question'].tolist()
answers = data['answer'].tolist()

# Build vocabulary
def build_vocabulary(data):
vocab = set()
for sentence in data:
tokens = nltk.word_tokenize(sentence.lower())
vocab.update(tokens)
vocab = list(vocab)
vocab.insert(0, START_TOKEN)
vocab.append(PADDING_TOKEN)
vocab.append(END_TOKEN)
return vocab


combined_vocabulary = build_vocabulary(questions + answers)
combined_to_index = {word: idx for idx, word in enumerate(combined_vocabulary)}
index_to_combined = {idx: word for idx, word in enumerate(combined_vocabulary)}
PAD_INDEX = combined_to_index[PADDING_TOKEN]
# Convert sentences to indices

def sentences_to_indices(sentences, vocabulary, max_length):
return [
[vocabulary[START_TOKEN]] +
[vocabulary.get(token, combined_to_index[PADDING_TOKEN]) for token in
nltk.word_tokenize(sentence.lower())][:max_length-2] +
[vocabulary[END_TOKEN]]
for sentence in sentences
]




questions_indices = sentences_to_indices(questions, combined_to_index, max_sequence_length)
answers_indices = sentences_to_indices(answers, combined_to_index, max_sequence_length)

# Define your Dataset class to handle question-answer pairs
class QADataset(Dataset):
def __init__(self, questions_indices, answers_indices):
self.questions_indices = questions_indices
self.answers_indices = answers_indices

def __len__(self):
return len(self.questions_indices)

def __getitem__(self, idx):
return {
'input_question': self.questions_indices[idx],
'target_answer': self.answers_indices[idx]
}
def collate_fn(batch):
combined_sequences = []
for item in batch:
combined_sequence = item['input_question'] + [combined_to_index[PADDING_TOKEN]] + item['target_answer']
combined_sequences.append(combined_sequence)

max_sequence_length = max(len(seq) for seq in combined_sequences)
combined_sequences = [seq + [combined_to_index[PADDING_TOKEN]] * (max_sequence_length - len(seq)) for seq in combined_sequences]

combined_sequence_with_end = [
seq[:max_sequence_length - 1] +[combined_to_index[END_TOKEN]]
for seq in combined_sequences
]
return {
'combined_sequence': torch.tensor(combined_sequence_with_end)
}

# Create your Dataset and DataLoader
train_dataset = QADataset(questions_indices, answers_indices)
batch_size = 30 # Set your desired batch size
train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=False, collate_fn=collate_fn)

print("Vocabulary : ", questions_indices)

print("Index Pair : Word", combined_to_index)


# Sample input text
input_text = "What should I do if a fire breaks out at Home ?"

# Tokenization and adding special tokens using your vocabulary
input_tokens = [START_TOKEN] + input_text.lower().split() + [END_TOKEN]
# Converting tokens to indices using your vocabulary
input_indices = [combined_to_index[token] for token in input_tokens]

print(input_tokens)

print(input_indices)

import torch.nn.functional as F
import math
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, Dataset
from tqdm import tqdm


class PositionalEncoding(nn.Module):
def __init__(self, d_model, max_sequence_length):
super(PositionalEncoding, self).__init__()
self.dropout = nn.Dropout(0.1)
pe = torch.zeros(max_sequence_length, d_model)
position = torch.arange(0, max_sequence_length, dtype=torch.float).unsqueeze(1)
div_term = torch.exp(torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model))
pe[:, 0::2] = torch.sin(position * div_term)
pe[:, 1::2] = torch.cos(position * div_term)
self.register_buffer('pe', pe)

def forward(self, x):
x = x + self.pe[:x.size(1), :]
return self.dropout(x)

class Transformer(nn.Module):
def __init__(self, d_model, nhead, num_encoder_layers, num_decoder_layers, ffn_hidden, max_sequence_length, vocab_size):
super(Transformer, self).__init__()

self.embedding = nn.Embedding(vocab_size, d_model)
self.positional_encoding = PositionalEncoding(d_model, max_sequence_length)

self.transformer = nn.Transformer(d_model=d_model, nhead=nhead, num_encoder_layers=num_encoder_layers, num_decoder_layers=num_decoder_layers, dim_feedforward=ffn_hidden)

self.fc = nn.Linear(d_model, vocab_size)

def forward(self, src, tgt):
src = self.embedding(src)
tgt = self.embedding(tgt)

src = self.positional_encoding(src)
tgt = self.positional_encoding(tgt)

output = self.transformer(src, tgt)

return self.fc(output)

# Define the parameters
d_model = 512
nhead = 8
num_encoder_layers = 2
num_decoder_layers = 2
ffn_hidden = 2048
max_sequence_length = 200
vocab_size = len(combined_vocabulary) # Use the combined vocabulary

# Instantiate the model
model = Transformer(
d_model,
nhead,
num_encoder_layers,
num_decoder_layers,
ffn_hidden,
max_sequence_length,
vocab_size
)

device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model.to(device)

# Print the model architecture
print(model)

# Define loss function and optimizer
criterion = nn.CrossEntropyLoss(ignore_index=PAD_INDEX)
optimizer = optim.Adam(model.parameters(), lr=0.001)

# Training loop
num_epochs = 1
for epoch in range(num_epochs):
model.train()
total_loss = 0.0

for batch in tqdm(train_loader, desc=f"Epoch {epoch + 1}"):
combined_sequence = batch['combined_sequence'].to(device) # Use 'combined_sequence' key

optimizer.zero_grad()
output = model(combined_sequence, combined_sequence) # Use the same sequence for both input and target
output_dim = output.shape[-1]

# Flatten the output and target sequences for loss calculation
output = output.contiguous().view(-1, output_dim)
target_sequence = combined_sequence.view(-1) # Flatten the target

loss = criterion(output, target_sequence)
loss.backward()
optimizer.step()

total_loss += loss.item()

avg_loss = total_loss / len(train_loader)
print(f"Epoch {epoch + 1}, Loss: {avg_loss:.4f}")



torch.save(model.state_dict(), "/content/drive/MyDrive/Tikva_model/own_model/model.pth")

import pickle
pickle.dump(model, open('/content/drive/MyDrive/Tikva_model/own_model/model.pkl', 'wb'))

import pickle

model = pickle.load(open('/content/drive/MyDrive/Tikva_model/own_model/model.pkl', 'rb'))

import torch
def generate_response():
# Sample input text
input_text = "How can I prevent fires at home ?"
# Tokenization and adding special tokens using your vocabulary
input_tokens = [START_TOKEN] + input_text.lower().split() + [END_TOKEN]
# Converting tokens to indices using your vocabulary
input_indices = [combined_to_index[token] for token in input_tokens]
# Padding and reshaping
max_input_length = max_sequence_length # Choose an appropriate length
padded_input = input_indices + [combined_to_index[PADDING_TOKEN]] * (max_input_length - len(input_indices))
input_tensor = torch.tensor(padded_input).unsqueeze(0) # Add a batch dimension
# Move the tensor to the appropriate device (GPU if available)
input_tensor = input_tensor.to(device)

# Pass through the model
with torch.no_grad():
model.eval() # Set the model to evaluation mode
output = model(input_tensor, input_tensor) # Pass input_tensor as both src and tgt
predicted_indices = output.argmax(dim=-1) # Get the indices with the highest probability

# Convert predicted indices back to tokens using your vocabulary
predicted_tokens = [index_to_combined[idx.item()] for idx in predicted_indices[0]]
# Remove special tokens and padding tokens
predicted_tokens = [token for token in predicted_tokens if token not in [START_TOKEN, END_TOKEN, PADDING_TOKEN]]
# Convert predicted tokens to a readable sentence
predicted_sentence = " ".join(predicted_tokens)
print("Predicted sentence:", predicted_sentence)

generate_response()