File size: 3,890 Bytes
4d89080 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 | import tensorflow as tf # type: ignore
from keras_preprocessing.text import Tokenizer # type: ignore
from keras_preprocessing.sequence import pad_sequences # type: ignore
import numpy as np # type: ignore
import os
os.environ['TF_CPP_MIN_LOG_LEVEL'] = '3'
# Optional: suppress ABSEIL warnings if TensorFlow still leaks something
try:
import absl.logging # type: ignore
absl.logging.set_verbosity(absl.logging.ERROR)
except ImportError:
pass
def load_dialogues(datapath="C:/Users/90372/ai/data/", num_words=10000, max_len=40): # noqa
lines_file = os.path.join(datapath, "movie_lines.txt")
convos_file = os.path.join(datapath, "movie_conversations.txt")
id2line = {}
with open(lines_file, encoding="utf-8", errors='ignore') as f:
for line in f:
parts = line.strip().split(" +++$+++ ")
if len(parts) == 5:
line_id, text = parts[0], parts[4]
id2line[line_id] = text
inputs, responses = [], []
with open(convos_file, encoding="utf-8", errors='ignore') as f:
for line in f:
parts = line.strip().split(" +++$+++ ")
if len(parts) == 4:
line_ids = eval(parts[3])
for i in range(len(line_ids) - 1):
input_line = id2line.get(line_ids[i], "")
response_line = id2line.get(line_ids[i + 1], "")
if input_line and response_line:
inputs.append(input_line)
responses.append(response_line)
catchAllError = "I'm sorry, I cant seem to remember anything right now."
tokenizer = Tokenizer(num_words=num_words, oov_token=catchAllError)
tokenizer.fit_on_texts(inputs + responses)
input_sequences = tokenizer.texts_to_sequences(inputs)
response_sequences = tokenizer.texts_to_sequences(responses)
inputs_padded = pad_sequences(input_sequences, maxlen=max_len, padding="post", truncating="post") # noqa
responses_padded = pad_sequences(response_sequences, maxlen=max_len, padding="post", truncating="post") # noqa
return inputs_padded, responses_padded, tokenizer
def build_seq2seq_model(vocab_size, max_len):
inputs = tf.keras.Input(shape=(max_len,))
embedding = tf.keras.layers.Embedding(vocab_size, 256)(inputs)
lstm = tf.keras.layers.LSTM(256, return_sequences=True)(embedding)
attention = tf.keras.layers.Attention()([lstm, lstm])
decoder_lstm = tf.keras.layers.LSTM(256)(attention)
outputs = tf.keras.layers.Dense(vocab_size)(decoder_lstm) # Final output layer
model = tf.keras.Model(inputs, outputs)
loss = tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True) # Removed label_smoothing
model.compile(optimizer=tf.keras.optimizers.Adam(learning_rate=1e-4), loss=loss) # Compile model
return model
def train_conversational_model():
inputs, responses, tokenizer = load_dialogues()
target_sequences = responses[:, 1:] # Shift the target by one timestep
target_sequences = np.pad(target_sequences, ((0, 0), (0, 1)), mode='constant') # Pad the target sequences
model = build_seq2seq_model(vocab_size=10000, max_len=inputs.shape[1])
model.fit(inputs, target_sequences, epochs=5, batch_size=32) # Train with shifted target sequences
return model, tokenizer
def generate_response(model, tokenizer, prompt, max_len=40):
sequence = tokenizer.texts_to_sequences([prompt])
sequence = pad_sequences(sequence, maxlen=max_len)
pred = model.predict(sequence)
next_word_id = np.argmax(pred[0])
return tokenizer.index_word.get(next_word_id, "<UNK>")
if __name__ == "__main__":
model, tokenizer = train_conversational_model()
while True:
prompt = input("User: ")
if prompt.lower() == "exit":
break
print("AI:", generate_response(model, tokenizer, prompt))
|