📕 RNN 기반 자연어 처리 모델

RNN 기반 자연어 처리 모델입니다. 텔레그램 톡방에서 데이터를 추출한 뒤, 필터 과정을 거쳐 학습을 진행합니다.
다른 플랫폼은 충분한 언어 데이터가 없어서 시도하지 않았습니다.

깃허브

GitHub - 909ma/RNN-based-Natural-Language-Processing-Model: RNN 기반 자연어 처리 모델입니다. 텔레그램에서 추

---

소스코드는 네 가지가 있습니다. 단계별로 나누었을 때 세 단계로 구별할 수 있습니다.

• 1 텔레그램 json to csv.py

# JSON 파일을 읽어오는 함수
# 텔레그램에서 추출한 대화 데이터 파일을 user, text 형태의 csv파일로 추출
import json
import csv


def read_json_file(file_path):
    with open(file_path, 'r', encoding='utf-8') as json_file:
        data = json.load(json_file)
    return data


def extract_actor_text_to_csv(json_data, output_csv_file):
    # CSV 파일을 쓰기 모드로 엽니다.
    with open(output_csv_file, 'w', newline='', encoding='utf-8') as csv_file:
        writer = csv.writer(csv_file)

        # CSV 파일의 헤더를 쓰기
        writer.writerow(["Actor", "Text"])

        prev_actor = None
        prev_text = []

        for message in json_data.get("messages", []):
            actor = message.get("from", "") if message.get(
                "from") else message.get("actor", "")
            text = message.get("text", "")

            try:
                # 'text' 필드가 문자열이 아닌 경우 처리
                if not isinstance(text, str):
                    text = str(text)  # 문자열로 변환

                # 텍스트가 없거나 'link'를 포함하면 무시
                if not text or any('link' in item.get('type', '') for item in message.get("text_entities", [])):
                    continue

                # 줄 바꿈 문자를 공백으로 대체
                text = text.replace("\n", " ")

                if actor == prev_actor:
                    prev_text.append(text)  # 리스트에 텍스트 추가
                else:
                    if prev_actor is not None:
                        # 리스트를 문자열로 합쳐서 쓰기
                        writer.writerow([prev_actor, ' '.join(prev_text)])
                    prev_actor = actor
                    prev_text = [text]  # 리스트로 초기화

            except Exception as e:
                print(f"오류 발생! 메시지: {message}")
                print(f"에러 메시지: {str(e)}")

        # 마지막 줄을 쓰기
        if prev_actor is not None:
            # 리스트를 문자열로 합쳐서 쓰기
            writer.writerow([prev_actor, ' '.join(prev_text)])


# JSON 파일 경로
json_file_path = 'result_test.json'  # 실제 JSON 파일 경로로 변경하세요.

# CSV 파일 경로
output_csv_file_path = 'output.csv'  # 저장할 CSV 파일 경로로 변경하세요.

# JSON 파일 읽기
json_data = read_json_file(json_file_path)

# actor와 text 필드 추출 및 CSV로 저장
extract_actor_text_to_csv(json_data, output_csv_file_path)

print(f"데이터 추출 및 저장이 완료되었습니다. {output_csv_file_path} 파일을 확인하세요.")

• 한 문장을 여러 번의 줄바꿈을 사용하여 메세지를 보낸 경우 한 문장으로 합쳐서 csv파일로 기록합니다.
• 빈 메세지이거나 하이퍼링크를 공유한 경우는 추출이 되지 않도록 필터를 걸었습니다.

 

---

• 1 텔레그램 json to csv(유저 필터).py

# JSON 파일을 읽어오는 함수
# 텔레그램에서 추출한 대화 데이터 파일을 user, text 형태의 csv파일로 추출
# 특정인의 발언만 기록하는 기능 개선
import json
import csv


def read_json_file(file_path):
    with open(file_path, 'r', encoding='utf-8') as json_file:
        data = json.load(json_file)
    return data


def extract_actor_text_to_csv(json_data, output_csv_file, target_name):
    with open(output_csv_file, 'w', newline='', encoding='utf-8') as csv_file:
        writer = csv.writer(csv_file)
        writer.writerow(["Actor", "Text"])
        for message in json_data.get("messages", []):
            actor = message.get("from", "") if message.get(
                "from") else message.get("actor", "")
            text = message.get("text", "")

            try:
                # 'text' 필드가 문자열이 아닌 경우 처리
                if not isinstance(text, str):
                    text = str(text)  # 문자열로 변환

                # 텍스트가 없거나 'link'를 포함하면 무시
                if not text or any('link' in item.get('type', '') for item in message.get("text_entities", [])):
                    continue

                # 줄 바꿈 문자를 공백으로 대체
                text = text.replace("\n", " ")

                if actor == target_name:
                    # 추출 대상 사용자와 일치하면 CSV 파일에 쓰기
                    writer.writerow([actor, text])

            except Exception as e:
                print(f"오류 발생! 메시지: {message}")
                print(f"에러 메시지: {str(e)}")
                # break # 오류가 많이 나면 주석을 푸세요


# JSON 파일 경로
json_file_path = 'result_test.json'  # 실제 JSON 파일 경로로 변경하세요.

# 추출하고자 하는 특정 대상
target_name = '홍길동'  # 실제 대상 이름으로 변경하세요.

# CSV 파일 경로
output_csv_file_path = f'output_{target_name}.csv'

json_data = read_json_file(json_file_path)
extract_actor_text_to_csv(json_data, output_csv_file_path, target_name)
print(f"데이터 추출 및 저장이 완료되었습니다. {output_csv_file_path} 파일을 확인하세요.")

• 원하는 사람만의 대화 기록을 저장합니다.
• 원문을 가공하지 않고 그대로 저장하되, 텍스트가 없는 내용이거나 하이퍼링크 공유인 경우는 저장하지 않도록 설정하였습니다.

 

---

• 2 LSTM 기반 학습 및 모델, 변수 저장.py

from tensorflow.keras.preprocessing.text import tokenizer_from_json
from tensorflow.keras.preprocessing.sequence import pad_sequences
from tensorflow.keras.layers import Embedding, Dense, LSTM
from tensorflow.keras.preprocessing.text import Tokenizer
from tensorflow.keras.utils import to_categorical
from tensorflow.keras.models import Sequential
from string import punctuation
import tensorflow as tf
import pandas as pd
import numpy as np
import pickle


# GPU 사용 가능 여부 확인
physical_devices = tf.config.list_physical_devices('GPU')
if len(physical_devices) > 0:
    tf.config.experimental.set_memory_growth(physical_devices[0], True)
    print("GPU가 사용 가능합니다.")
else:
    print("GPU를 찾을 수 없습니다. CPU를 사용합니다.")

df = pd.read_csv('output.csv')
Text = []
# 헤드라인의 값들을 리스트로 저장
Text.extend(list(df.Text.values))
print('총 샘플의 개수 : {}'.format(len(Text)))

Text = [word for word in Text if word != "ㅋ"]
print('노이즈값 제거 후 샘플의 개수 : {}'.format(len(Text)))

tokenizer = Tokenizer()
tokenizer.fit_on_texts(Text)
vocab_size = len(tokenizer.word_index) + 1
print('단어 집합의 크기 : %d' % vocab_size)

sequences = list()

for sentence in Text:
    encoded = tokenizer.texts_to_sequences([sentence])[0]
    for i in range(1, len(encoded)):
        sequence = encoded[:i+1]
        sequences.append(sequence)

max_len = max(len(l) for l in sequences)
print('샘플의 최대 길이 : {}'.format(max_len))

sequences = pad_sequences(sequences, maxlen=max_len, padding='pre')
sequences = np.array(sequences)
X = sequences[:, :-1]
y = sequences[:, -1]
y = to_categorical(y, num_classes=vocab_size)

embedding_dim = 10
hidden_units = 128

model = Sequential()
model.add(Embedding(vocab_size, embedding_dim))
model.add(LSTM(hidden_units))
model.add(Dense(vocab_size, activation='softmax'))
model.compile(loss='categorical_crossentropy',
              optimizer='adam', metrics=['accuracy'])

# 모델 학습
model.fit(X, y, epochs=1, verbose=2)

# 모델 저장
model.save('my_model.keras', overwrite=True)
print("모델 저장이 완료되었습니다.")

# max_len 저장
with open('max_len.pkl', 'wb') as f:
    pickle.dump(max_len, f)

# tokenizer 저장
with open('tokenizer.json', 'w', encoding='utf-8') as f:
    f.write(tokenizer.to_json())

   결과 파일은 세 가지로 나옵니다.

1. my_model.keras
2. max_len.pkl
3. tokenizer.json

  사용하는 데이터의 특성에 맞추어 노이즈값을 제거하세요. 기본 설정은 'ㅋ' 제외입니다.

  학습을 진행할 때에 진행하고자 하는 목표에 맞추어 max_len, sequences, epochs 값들을 바꾸시면 됩니다.

 

---

• 3 모델 및 변수 로드 후 자연어 생성.py

from tensorflow.keras.preprocessing.text import tokenizer_from_json
from tensorflow.keras.preprocessing.sequence import pad_sequences
from tensorflow.keras.preprocessing.text import Tokenizer
from tensorflow.keras.models import load_model
import numpy as np
import pandas as pd
import pickle


def sentence_generation(model, tokenizer, current_word, n):
    init_word = current_word
    sentence = ''

    for _ in range(n):
        encoded = tokenizer.texts_to_sequences([current_word])[0]
        encoded = pad_sequences([encoded], maxlen=max_len-1, padding='pre')

        result = model.predict(encoded, verbose=0)
        result = np.argmax(result, axis=1)

        for word, index in tokenizer.word_index.items():
            if index == result:
                break

        current_word = current_word + ' ' + word
        sentence = sentence + ' ' + word

    sentence = init_word + sentence
    return sentence


# max_len 로드
with open('max_len.pkl', 'rb') as f:
    max_len = pickle.load(f)

# tokenizer 로드
with open('tokenizer.json', 'r', encoding='utf-8') as f:
    tokenizer = tokenizer_from_json(f.read())

# 모델 로드
loaded_model = load_model('my_model.keras')

# 시작 단어와 생성할 단어 수 설정
seed_text = '코로나'
num_words_to_generate = 7

# 텍스트 생성
generated_text = sentence_generation(
    loaded_model, tokenizer, seed_text, num_words_to_generate)

print("="*100)
print(generated_text)
print("="*100)

  결과는 마지막에 출력됩니다.

 

---

결과1

  '그래서' 라는 말을 시작으로 6가지 길이의 단어를 생성했을 때 위와 같은 말을 생성하였습니다. 혹시 가지고 있는 데이터에서 위와 똑같은 문장이 있는지 확인을 했는 결과...

결과1 - 유사 문장 발견

  가장 유사한 문장을 하나 발견하였습니다. 그러나 완벽하게 똑같지는 않으나 유사도가 높은 면을 보면 어떤 과정으로 학습이 되는지 잘 나타내는 모습임을 알 수 있습니다.

 

---

결과2

  친구 이름을 넣고 문장을 생성하였습니다. 혹시 비슷한 문장을 말한 적이 있는지 확인했으나 발견하지 못했습니다.

 

---

결과3

  이번에는 '미드'를 넣고 문장을 생성했습니다. 원천 데이터 속에서 비슷한 문장을 찾아보았으나 똑같은 말을 한 적이 없었습니다.

 

---

참고 문헌

• 딥러닝을 이용한 자연어 처리 입문: https://wikidocs.net/45101

08-06 RNN을 이용한 텍스트 생성(Text Generation using RNN)