[패스트캠퍼스 수강 후기] 인공지능강의 100% 환급 챌린지 48회차 미션

Part4-58. 맥락을 파악하는 Attention 기법 - 07. (STEP 3) Attention 신경망 구현 및 학습 - 1

seq2seq에 attention을 적용하여 진행

konlpy

pip install konlpy로 설치
OKT(Open Korean Twitter) 형태소 분석기

Encoder

class Encoder(tf.keras.Model):
    def __init__(self):
        super(Encoder, self).__init__()
        self.emb = tf.keras.layers.Embedding(NUM_WORDS, 64) #onehot형태로 들어오면 Embedding 진행
        # return_state를 True로 해야 hidden 과 cell state를 출력해줌
        # attention을 위해 return_sequences을 True로 변경해야함(모든 히든스테이트를 얻기 위해)
        self.lstm = tf.keras.layers.LSTM(512, return_sequences=True, return_state=True) 

    def call(self, x, training=False, mask=None):
        x = self.emb(x)
        H, h, c = self.lstm(x)
        return H, h, c

Decoder

class Decoder(tf.keras.Model):
    def __init__(self):
        super(Decoder, self).__init__()
        self.emb = tf.keras.layers.Embedding(NUM_WORDS, 64)
        # return_sequences가 False인 경우 dense()의 연산이 마지막 하나만 출력하게됨 
        self.lstm = tf.keras.layers.LSTM(512, return_sequences=True, return_state=True) 
        self.att = tf.keras.layers.Attention()
        self.dense = tf.keras.layers.Dense(NUM_WORDS, activation='softmax')

    def call(self, inputs, training=False, mask=None):
        x, s0, c0, H = inputs #입력, hidden, cell 3개를 입력 받음 
        x = self.emb(x)
        S, h, c = self.lstm(x, initial_state=[s0, c0]) #S는 히든스테이트를 모두 모아놓은거 

        #앞선 시간에 대해서 작성 
        S_ = tf.concat([s0[:, tf.newaxis, :], S[:, :-1, :]], axis=1) #s0가 2차원이므로 3차원으로 확장 #S는 마지막하나 제외
        A = self.att([S_, H])
        y = tf.concat([S, A], axis=-1) #Attention value를 추가 

        return self.dense(y), h, c #test를 할때 hidden과 cell state가 필요 

Seq2seq

class Seq2seq(tf.keras.Model):
    def __init__(self, sos, eos):
        super(Seq2seq, self).__init__()
        self.enc = Encoder()
        self.dec = Decoder()
        self.sos = sos
        self.eos = eos

    def call(self, inputs, training=False, mask=None):
        if training is True:
            x, y = inputs #학습을 위해 입력, 출력 모두 알아야함 
            H, h, c = self.enc(x) #hidden, cel state 구현
            y, _, _ = self.dec((y, h, c, H))
            return y
        else:
            x = inputs
            H, h, c = self.enc(x)

            y = tf.convert_to_tensor(self.sos) #sos입력 
            y = tf.reshape(y, (1, 1))

            seq = tf.TensorArray(tf.int32, 64)

            for idx in tf.range(64): #내부적으로 tf함수로 되어있다면 효율화된 tf for으로 변환됨
                y, h, c = self.dec([y, h, c, H]) # y는 softmax를 통해 나온 값
                y = tf.cast(tf.argmax(y, axis=-1), dtype=tf.int32) #따라서 argmax로 가장 높은 값을 선택 
                y = tf.reshape(y, (1, 1)) #배치를 고려해야하기 때문에 (1,1)으로 reshape
                seq = seq.write(idx, y)

                if y == self.eos:
                    break

            return tf.reshape(seq.stack(), (1, 64))

학습, 테스트 루프 정의

# Implement training loop
@tf.function
def train_step(model, inputs, labels, loss_object, optimizer, train_loss, train_accuracy):
    output_labels = labels[:, 1:]
    shifted_labels = labels[:, :-1]
    with tf.GradientTape() as tape:
        predictions = model([inputs, shifted_labels], training=True) # shifted_labels 입력할때 같이 넣어줘야함
        loss = loss_object(output_labels, predictions)
    gradients = tape.gradient(loss, model.trainable_variables)

    optimizer.apply_gradients(zip(gradients, model.trainable_variables))
    train_loss(loss)
    train_accuracy(output_labels, predictions)

# Implement algorithm test
@tf.function
def test_step(model, inputs):
    return model(inputs, training=False)

데이터셋 준비

chatbot_data : 수작업으로 만든데이터
카페에서 일어나는 질문과 답변형태로 진행
질문-답변순서로 작성되어 있으며 약 1000건

dataset_file = 'chatbot_data.csv' # acquired from 'http://www.aihub.or.kr' and modified
okt = Okt()

with open(dataset_file, 'r') as file:
    lines = file.readlines()
    #morphs 한줄한줄에 대한 형태로 분석 진행 
    #가져올때 tab을 하나씩 추가해줌(SOS역할)
    #newline캐릭터가 미리 붙어있게 되는데 (EOS로 사용)
    seq = [' '.join(okt.morphs(line)) for line in lines] 


questions = seq[::2]
answers = ['\t ' + lines for lines in seq[1::2]]

num_sample = len(questions)

perm = list(range(num_sample))
random.seed(0)
random.shuffle(perm)

train_q = list()
train_a = list()
test_q = list()
test_a = list()

for idx, qna in enumerate(zip(questions, answers)):
    q, a = qna
    if perm[idx] > num_sample//5:
        train_q.append(q)
        train_a.append(a)
    else:
        test_q.append(q)
        test_a.append(a)
#가져오는 문장에서 토큰화(각 단어를 숫자로 변형) 진행, 탭과 뉴라인은 토큰화 하지 않음
tokenizer = tf.keras.preprocessing.text.Tokenizer(num_words=NUM_WORDS,
                                                  filters='!"#$%&()*+,-./:;<=>?@[\\]^_`{|}~')

tokenizer.fit_on_texts(train_q + train_a)

#seq은 숫자로된 단어들의 나열이라 생각하면 됨
train_q_seq = tokenizer.texts_to_sequences(train_q)
train_a_seq = tokenizer.texts_to_sequences(train_a)

test_q_seq = tokenizer.texts_to_sequences(test_q)
test_a_seq = tokenizer.texts_to_sequences(test_a)

x_train = tf.keras.preprocessing.sequence.pad_sequences(train_q_seq,
                                                        value=0,
                                                        padding='pre',
                                                        maxlen=64)
#출력데이터는 앞쪽으로 정렬, maxlen을 65로 해준이유는 EOS가 붙어있기 때문에                                               
y_train = tf.keras.preprocessing.sequence.pad_sequences(train_a_seq,
                                                        value=0,
                                                        padding='post',
                                                        maxlen=65)


x_test = tf.keras.preprocessing.sequence.pad_sequences(test_q_seq,
                                                       value=0,
                                                       padding='pre',
                                                       maxlen=64)
y_test = tf.keras.preprocessing.sequence.pad_sequences(test_a_seq,
                                                       value=0,
                                                       padding='post',
                                                       maxlen=65)

train_ds = tf.data.Dataset.from_tensor_slices((x_train, y_train)).shuffle(10000).batch(32).prefetch(1024)
test_ds = tf.data.Dataset.from_tensor_slices((x_test, y_test)).batch(1).prefetch(1024)

학습환경 정의

# Create model
# sos는 탭, eos는 뉴라인
model = Seq2seq(sos=tokenizer.word_index['\t'],
                eos=tokenizer.word_index['\n'])

# Define loss and optimizer
loss_object = tf.keras.losses.SparseCategoricalCrossentropy()
optimizer = tf.keras.optimizers.Adam()

# Define performance metrics
train_loss = tf.keras.metrics.Mean(name='train_loss')
train_accuracy = tf.keras.metrics.SparseCategoricalAccuracy(name='train_accuracy')

학습루프 동작

for epoch in range(EPOCHS):
    for seqs, labels in train_ds:
        train_step(model, seqs, labels, loss_object, optimizer, train_loss, train_accuracy)

    template = 'Epoch {}, Loss: {}, Accuracy: {}'
    print(template.format(epoch + 1,
                          train_loss.result(),
                          train_accuracy.result() * 100))

    train_loss.reset_states()
    train_accuracy.reset_states()

테스트 루프

for test_seq, test_labels in test_ds:
    prediction = test_step(model, test_seq)
    test_text = tokenizer.sequences_to_texts(test_seq.numpy())
    gt_text = tokenizer.sequences_to_texts(test_labels.numpy())
    texts = tokenizer.sequences_to_texts(prediction.numpy())
    print('_')
    print('q: ', test_text)
    print('a: ', gt_text)
    print('p: ', texts)

seq2seq와 attention 비교

좌 attention, 우 seq2seq
attention이 추가되면서 성능이 향상된것을 볼 수 있음

딥러닝/인공지능 올인원 패키지 Online