텍스트 데이터와 토큰화

from pathlib import Path

폴더 = Path('data/koen')
assert 폴더.is_dir(), f'{폴더} 폴더가 존재하지 않습니다.'

파일들 = list(폴더.glob('**/*.csv'))
datasets = {}
for 파일 in 파일들:
    split = 파일.stem
    datasets[split] = pd.read_csv(파일)
    print(f'{split:<10}: {len(datasets[split]):,}')

datasets['train'].sample(5)

분류코드 = datasets['train']['source']
분류코드도수 = 분류코드.value_counts()
pd.DataFrame({'도수': 분류코드도수, '비율': 분류코드도수 / sum(분류코드도수)}).round(3)

분류필터 = datasets['train']['source'] == 71265 # 구어체 텍스트
datasets['train'][분류필터].sample(5)

분류필터 = datasets['train']['source'] == 563 # 특허 텍스트
datasets['train'][분류필터].sample(5)

with open('한국어문장.txt', 'w') as 파일:
    for 색인, 행 in datasets['train'].iterrows():
        파일.write(행['ko'] + '\n')

# 결과 확인
with open('한국어문장.txt', 'r') as 파일:
    for _ in range(5):
        print(파일.readline().strip())

with open('영어문장.txt', 'w') as 파일:
    for 색인, 행 in datasets['train'].iterrows():
        파일.write(행['en'] + '\n')

# 결과 확인
with open('영어문장.txt', 'r') as 파일:
    for _ in range(5):
        print(파일.readline().strip())

import sentencepiece as spm

어휘수 = 30000
문장수 = 100000
spm.SentencePieceTrainer.train(
    # 유의! 각 파일경로는 공백없이 쉼표로 구분
    input='한국어문장.txt,영어문장.txt',
    model_prefix='spm',
    vocab_size=어휘수,
    # 무작위로 선택된 문장으로 모델 훈련
    input_sentence_size=문장수, # 훈련에 사용할 문장 수
    shuffle_input_sentence=True, # 입력 문장 순서 섞기
    # model_type='bpe' # Byte Pair Encoding (BPE): 1990년대
    # 특수토큰 지정
    pad_id=0, unk_id=1, bos_id=2, eos_id=3
)

예문 = [
    '한국어 문장을 형태소 단위로 분리합니다.',
    'tokenize an English sentence into subwords.',
]

tokenizer = spm.SentencePieceProcessor(model_file='spm.model')
for text in 예문:
    형태소목록 = tokenizer.encode(text, out_type=str)
    정수시퀀스 = tokenizer.encode(text, out_type=int)
    display(pd.DataFrame({
        '형태소': 형태소목록,
        '정수': 정수시퀀스
    }).T)

    텍스트복원 = tokenizer.decode(정수시퀀스)
    print(텍스트복원)

# 정수ID
pd.DataFrame({
    '토큰': [tokenizer.id_to_piece(i) for i in range(15)]
}).T

class TextDataset:
    def __init__(self, texts, labels=None, transform=None):
        self.texts = texts
        self.labels = labels
        self.transform = transform

    def __len__(self):
        return len(self.texts)
    
    def __getitem__(self, 색인):
        sample = self.texts[색인]
        if self.transform:
            sample = self.transform(sample)
            
        if self.labels is not None:
            label = self.labels[색인]
            return sample, label
        return sample

전처리 = lambda text: tokenizer.encode(text, out_type=int)

datasets = {}
for 파일경로 in 파일들:
    frame = pd.read_csv(파일경로)
    split = 파일경로.stem
    datasets[split] = TextDataset(frame['ko'], transform=전처리)

sample = datasets['train'][0]
print(sample)

def 전처리(text, 최대길이):
    정수시퀀스 = tokenizer.encode(text, out_type=int)    
    return 정수시퀀스[:최대길이]

datasets = {}
for split, 샘플수 in [('train', 10000), ('validation', 1000), ('test', 1000)]:
    frame = pd.read_csv(폴더 / f'{split}.csv')
    frame.dropna(inplace=True) # 결측값 제거
    # 각 클래스에서 샘플링
    구어체 = frame[frame['source'] == 71265].sample(샘플수, random_state=0)
    특허 = frame[frame['source'] == 563].sample(샘플수, random_state=0)

    samples = pd.concat([구어체, 특허], ignore_index=True)
    
    datasets[split] = TextDataset(
        texts=samples['ko'].tolist(),
        # 텍스트 분류를 위해 클래스 레이블도 함께 제공해야 합니다.
        labels=samples['source'].map({71265: 0, 563: 1}).tolist(),
        transform=lambda text: 전처리(text, 최대길이=300)
    )
    print(f'{split:<10}: {len(datasets[split]):,}')
    # 유형 도수 확인
    print(samples['source'].value_counts().to_string())

sample, label = datasets['train'][10001]
print(f'label: {label}\nsample: {sample}')
print('복원된 텍스트:', tokenizer.decode(sample))

문서길이 = [len(sample) for sample, _ in datasets['train']]
pd.Series(문서길이).describe().astype(int)

def 시퀀스패딩(sample, 최대길이, 패딩값=0):
    assert len(sample) > 0, '샘플이 비어 있습니다.'
    sample = sample[:최대길이]
    return sample + [패딩값] * (최대길이 - len(sample))

최대길이 = 20
정규화된_샘플 = []
for i in range(3):
    sample, _ = datasets['train'][i]
    정규화된_샘플.append(시퀀스패딩(sample, 최대길이))

pd.DataFrame(정규화된_샘플)

def 배치병합(batch):
    samples, labels = zip(*batch)
    문서길이 = [len(sample) for sample in samples]
    최대길이 = max(문서길이) # 배치 내에서 가장 긴 샘플의 길이를 최대길이로 설정
    정규화된_샘플 = [시퀀스패딩(sample, 최대길이) for sample in samples]
    return np.array(정규화된_샘플), np.array(labels)

# 첫 번째 배치
sample_batch = [datasets['train'][i] for i in range(3)]
Xs, ys = 배치병합(sample_batch)
print(Xs.shape, ys.shape)
# 다른 배치
sample_batch = [datasets['train'][i] for i in range(5, 8)]
Xs, ys = 배치병합(sample_batch)
print(Xs.shape, ys.shape)

dataloaders = {}
for split in datasets.keys():
    dataloaders[split] = torch.utils.data.DataLoader(
        datasets[split], shuffle=(split=='train'),
        batch_size=32, num_workers=2, collate_fn=배치병합
    )

배치시퀀스길이 = []
for X_batch, y_batch in dataloaders['train']:
    # print(X_batch.shape, y_batch.shape)
    배치시퀀스길이.append(X_batch.shape[1])

pd.Series(배치시퀀스길이).describe().astype(int)

import keras
from keras import layers

keras.backend.clear_session()

어휘수 = tokenizer.get_piece_size()
임베딩차원 = 64
print(f'어휘수: {어휘수}')
model = keras.Sequential([
    layers.Input(shape=(None,)), # 가변 길이 시퀀스 입력
    # 임베딩층: 정수 시퀀스를 밀집 벡터로 변환
    layers.Embedding(input_dim=어휘수, output_dim=임베딩차원),
    # 은닉층
    layers.GlobalAveragePooling1D(), # 시퀀스 차원 평균
    layers.Dense(64, activation='relu'),
    # 출력층
    layers.Dense(1, activation='sigmoid')
])
model.summary()

model.compile(
    loss=keras.losses.BinaryCrossentropy(),
    optimizer=keras.optimizers.RMSprop(),
    metrics=['accuracy']
)

history = model.fit(
    dataloaders['train'], validation_data=dataloaders['validation'],
    epochs=10,
    callbacks=[
        keras.callbacks.ModelCheckpoint('text_classifier.keras', save_best_only=True)
    ]
)

임베딩행렬 = model.layers[0].get_weights()[0]
print(f'임베딩행렬: {임베딩행렬.shape}')

단어 = '사랑'
정수ID = tokenizer.piece_to_id(단어)
print(f'단어: {단어}, 정수ID: {정수ID}')

단어임베딩 = 임베딩행렬[정수ID]
print(f'임베딩 벡터: {단어임베딩[:10].round(3)} ...')