이 포스트는 '구름'에서 진행하는 KDT 프로그램인 'AI 자연어처리 전문가 양성 과정'의 프로젝트 과정에서 수행했던 Whisper의 한국어 음성 통화 ASR 작업용 파인튜닝 과정 및 코드를 기록한 포스트입니다.
'Fine-Tune Whisper For Multilingual ASR with 🤗 Transformers'를 참고하였으며, 코드와 함께 적어 둔 주석이나 과정 등 또한 해당 포스트의 글을 해석하고 필요한 경우 일부 수정한 것입니다.
OpenAI의 Whisper는 발표 직후 상당한 화제가 된 End-to-End ASR 모델입니다. 별다른 파인튜닝 없이도 상당한 수준의 정확도를 보이며 실시간 번역과 함께 발화자 표시, 타임라인 표시 등 다양한 기능을 지원하고 있어 특히 해외 영상 자막 제작 등에서 활발하게 사용되고 있습니다.
'구름'의 'AI 자연어처리 전문가 양성 과정'의 마무리 과제로서 프로젝트를 진행하면서 음성 데이터를 실시간으로 전사 처리하여 텍스트 데이터로 변환하는 ASR 모델의 구현이 필요했고, Whisper의 모델을 파인튜닝하여 프로젝트에 사용했습니다. 그 과정에서 한국어로 된 자료가 많지 않다고 느껴 부족하나마 해당 내용을 정리하였습니다. 전체 코드와 내용은 colab 파일을 통해서도 확인할 수 있습니다.
!pip install datasets>=2.6.1
!pip install git+https://github.com/huggingface/transformers
!pip install evaluate>=0.30
!pip install jiwer
!pip install accelerate -U
!pip install transformers[torch]
음성 인식 Pipeline은 다음과 같이 세 가지 구성으로 나눌 수 있다.
1. 오디오 raw data에 대하여 pre-process를 진행할 feature extractor
2. sequence-to-sequence 매핑을 수행할 model
3. model의 output을 text 포맷으로 후처리할 tokenizer
HuggingFace의 Transformers는 Whisper의 모델과 함께 이와 연관된 WhisperFeatureExtractor와 WhisperTokenizer를 함께 제공하고 있다.
발화는 시간에 따라 변화하는 1차원 배열로 표현된다. 각 시계열별로 입력된 배열의 값은 그 순간의 신호가 갖는 진폭이다. 그 진폭의 정보만으로, 우리는 음성 데이터의 주파수 스펙트럼을 재구성하고 모든 음향적 특징을 복원할 수 있다.
발화는 연속적이므로, 진폭의 값은 무한하다. 그리고 이는 유한한 배열을 기대하는 컴퓨터 장치에 문제를 야기한다. 따라서, 우리는 고정된 시간 간격에 따라 신호에서 값을 샘플링하여 우리의 음성 신호를 분할한다. 오디오를 샘플링하는 간격은 sampling rate로 알려져 있으며, 이는 일반적으로 '샘플/초' 또는 헤르츠(Hz) 단위로 측정된다. 높은 sample rate로 샘플링하는 것은 연속적인 발화 신호에 대하여 더 나은 접근을 가능케 하지만, 초당 더 많은 값을 저장하도록 요구하기도 한다.
우리가 가진 오디오 데이터의 sampling rate를 우리가 사용할 모델이 기대하는 sampling rate와 일치시키는 것은 매우 중요한데, 서로 다른 sampling rate의 오디오 신호는 매우 다른 분포를 보이기 때문이다. 오디오 신호는 적절한 sampling rate로 처리되어야만 한다. 그렇지 않으면 예상치 못한 결과를 야기할 수 있다. 예를 들어, 16kHz로 샘플링된 오디오 샘플을 8kHz의 sampling rate로 들으면 해당 오디오의 속도가 절반으로 들릴 것이다. 동일한 방식으로, 잘못된 sampling rate의 오디오 데이터를 ASR 모델에 넘길 경우 모델이 불안정해질 수 있다. Whisper Feature Extractor는 16kHz의 오디오 데이터를 요구하며, 우리는 우리가 가진 오디오 데이터를 이에 맞추어야 한다.
Whisper Feature Extractor는 두 가지 동작을 수행한다. 먼저 오디오 샘플에 대하여 padding 및 trucating을 진행하여 모든 샘플들의 길이를 30초로 맞춘다. 30초보다 짧은 샘플은 0으로 채워 30초로 맞추고(음성 신호에서 0은 신호 없음 혹은 침묵을 의미한다), 30초보다 긴 샘플들은 30초로 잘라낸다. 이처럼 인풋 단계에서 각 배치의 모든 요소들이 최대 길이로 채워지거나 잘라지기 때문에, 우리는 Whisper 모델에 데이터를 전달할 때 Attention mask를 필요로 하지 않는다. Whisper 모델은 이 부분에서 독특한데, 대부분의 오디오 모델은 sequence가 패딩된 위치와 self-attention 메커니즘이 어느 부분을 무시해야 하는지에 대한 상세 정보를 제공하는 attention mask를 필요로 한다. 그러나 Whisper는 이러한 attention mask 없이 발화 신호로부터 무시해야 할 인풋을 직접 추론하도록 학습되었다.
Whisper Feature Extractor의 두 번째 동작은 피댕된 오디오 배열을 log-Mel 스펙트럼으로 변환하는 것이다. 이 스펙트럼은 푸리에 변환과 같이 주파수 신호를 시각적으로 나타낸 것이다.Whisper 모델의 인풋으로 주어져야 하는 것이 바로 이 log-Mel 스펙트럼이다.
Transformer의 Whisper Feature Extractor는 패딩과 스펙트로그램 변환을 단 한 줄의 코드로 수행할 수 있다.
from transformers import WhisperFeatureExtractor
# 파인튜닝을 진행하고자 하는 모델의 feature extractor를 로드
feature_extractor = WhisperFeatureExtractor.from_pretrained("openai/whisper-base")
Whisper 모델의 아웃풋은 vocabulary item들의 단어 중 예측된 텍스트를 나타내는 index값이다. Tokenizer는 이러한 텍스트 토큰 시퀀스와 실제 텍스트 문자열을 매핑해준다.
from transformers import WhisperTokenizer
# 파인튜닝을 진행하고자 하는 모델의 tokenizer를 로드
tokenizer = WhisperTokenizer.from_pretrained("openai/whisper-base", language="Korean", task="transcribe")
WhisperTokenizer는 인코딩 과정에서 'Special token'들을 부여한다. 여기에는 문장의 시작과 끝을 나타내는 토큰(전사의 시작과 끝을 나타내는 토큰을 포함한다), 언어를 나타내는 토큰, 태스크(전사, 번역 등)를 나타내는 토큰 등이 있다.
input_str = "저는 서울중앙지검 지능범죄수사팀 최인호 검사입니다."
labels = tokenizer(input_str).input_ids
decoded_with_special = tokenizer.decode(labels, skip_special_tokens=False)
decoded_str = tokenizer.decode(labels, skip_special_tokens=True)
print(f"Input: {input_str}")
print(f"Decoded w/ special: {decoded_with_special}")
print(f"Decoded w/out special: {decoded_str}")
print(f"Are equal: {input_str == decoded_str}")
Input: 저는 서울중앙지검 지능범죄수사팀 최인호 검사입니다.
Decoded w/ special: <|startoftranscript|><|ko|><|transcribe|><|notimestamps|>저는 서울중앙지검 지능범죄수사팀 최인호 검사입니다.<|endoftext|>
Decoded w/out special: 저는 서울중앙지검 지능범죄수사팀 최인호 검사입니다.
Are equal: True
Feature Extactor와 Tokenizer를 간편히 사용하기 위해, 두 모듈을 WhisperProcessor 클래스 하나로 묶을 수 있다. Processor는 Feature Extactor와 Tokenizer를 상속하며, 오디오 인풋의 입력과 모델의 예측에 사용할 수 있다. 이로써 우리는 훈련 과정에서 Processor와 Model이라는 두 개의 객체만 추적하면 된다.
from transformers import WhisperProcessor
processor = WhisperProcessor.from_pretrained("openai/whisper-base", language="Korean", task="transcribe")
보유한 음성 데이터와 이에 매치되는 텍스트 전사 데이터를 모델에 입력할 수 있는 형태로 변환해야 한다. 앞서 언급했듯 WhisperFeatureExtractor는 16kHz의 오디오 데이터를 입력으로 받으므로 이에 맞추어 오디오 데이터를 리샘플링해야 한다.
본 프로젝트에서는 AI-HUB의 '저음질 전화 음성인식 데이터'의 일부를 활용한다. 이 데이터셋은 각 문장별로 음성 데이터(wav 파일)와 해당 음성 데이터의 전사 데이터(text 파일)이 같은 파일명으로 매치되어 있으며, labeled data의 경우 해당 통화 내용의 전체 텍스트가 정리된 json 파일이 1개씩 포함되어 있다. 자세한 사항은 링크를 통해 확인하자.
이 데이터에 대하여 수행해야 할 전처리를 단계별로 구분하면 다음과 같다.
# 1. 오디오 파일 경로 취합
import glob
path = "오디오 파일들이 포함된 경로를 입력한다. - 예) /content/drive/MyDrive/NLP_Project_data/raw_data/*"
raw_data_list = glob.glob(path)
raw_data_list = sorted(raw_data_list)
print(f"file_list : {raw_data_list[:10]}")
print(len(raw_data_list))
# 2. 텍스트 파일 경로 취합
path = "텍스트 파일들이 포함된 경로를 입력한다. - 예) /content/drive/MyDrive/NLP_Project_data/labeled_data/*"
labeled_data_list = glob.glob(path)
# 레이블 데이터에는 json 데이터가 폴더별로 하나씩 있으므로 txt 파일만을 골라낸다.
labeled_data_list = sorted([file for file in labeled_data_list if file.endswith(".txt")])
print(f"file_list : {labeled_data_list[:10]}")
print(len(labeled_data_list))
transcript_list = []
for labeled_data in tqdm(labeled_data_list):
with open(labeled_data, 'r', encoding='UTF8') as f:
line = f.readline()
transcript_list.append(line)
df = pd.DataFrame(data=transcript_list, columns = ["transcript"])
# 텍스트 데이터로 만든 데이터프레임에 음성 파일 경로 컬럼을 추가
df["raw_data"] = raw_data_list
# Null data 유무 확인
df.isnull().values.sum()
취합한 데이터프레임을 Transformers의 DatasetDict로 만들어 두면 해당 데이터를 언제 어디서든 자유롭게 활용할 수 있다. 물론 컴퓨팅 환경이 적합하다면 오프라인이 훨씬 빠르겠지만, 그렇지 않다면 고려해볼 수 있는 사항이다.
from datasets import Dataset, DatasetDict
from datasets import Audio
# 오디오 파일 경로를 dict의 "audio" 키의 value로 넣고 이를 데이터셋으로 변환
# 이때, Whisper가 요구하는 사양대로 Sampling rate는 16,000으로 설정한다.
ds = Dataset.from_dict({"audio": [path for path in df["raw_data"]],
"transcripts": [transcript for transcript in df["transcript"]]}).cast_column("audio", Audio(sampling_rate=16000))
# 데이터셋을 훈련 데이터와 테스트 데이터, 밸리데이션 데이터로 분할
train_testvalid = ds.train_test_split(test_size=0.2)
test_valid = train_testvalid["test"].train_test_split(test_size=0.5)
datasets = DatasetDict({
"train": train_testvalid["train"],
"test": test_valid["test"],
"valid": test_valid["train"]})
# 작성한 데이터셋을 허깅페이스에 업로드
datasets.push_to_hub("업로드할 허깅페이스 주소 입력")
데이터셋을 업로드했다면 다운로드하여 사용한다.
from datasets import load_dataset
# 데이터셋을 업로드할 때 접근을 제한하거나 비공개로 설정한 경우 허깅페이스 로그인이 필요하다.
low_call_voices = load_dataset("다운받을 데이터셋의 주소 입력")
데이터셋에 대하여 다음의 작업을 수행할 함수를 선언한다.
1. 오디오 데이터를 로드하고 리샘플링을 실시
2. feature extractor를 통해 1차원 오디오 배열을 log-Mel spectrogram으로 변환
3. tokenizer를 이용해 전사 데이터를 label ids로 변환
def prepare_dataset(batch):
# 오디오 파일을 16kHz로 로드
audio = batch["audio"]
# input audio array로부터 log-Mel spectrogram 변환
batch["input_features"] = feature_extractor(audio["array"], sampling_rate=audio["sampling_rate"]).input_features[0]
# target text를 label ids로 변환
batch["labels"] = tokenizer(batch["transcripts"]).input_ids
return batch
# 데이터 전처리 함수를 데이터셋 전체에 적용
low_call_voices = datasets.map(prepare_dataset, remove_columns=datasets.column_names["train"], num_proc=None)
# 전처리 작업이 오래 걸릴 수 있으므로, colab을 사용하여 파인튜닝을 진행한다면 전처리가 완료된 데이터셋을 Hub에 저장하는 것을 추천한다.
low_call_voices.push_to_hub("업로드할 허깅페이스 주소 입력")
이 부분부터 런타임 유형을 GPU로 설정하는 것을 추천한다.
# Hub로부터 전처리가 완료된 데이터셋을 로드
from datasets import load_dataset
low_call_voices_prepreocessed = load_dataset("다운받을 데이터셋의 주소 입력")
low_call_voices_prepreocessed
이제 데이터가 준비되었으므로 training pipeline을 수행할 준비가 되었다. Hugging face의 'Trainer'를 사용하면 다음의 단계에 따라 간단히 수행할 수 있다.
Sequence-to-sequence 발화 모델을 위한 Data collator는 input_feature와 label을 독립적으로 다룬다는 점에서 독특한 성격을 보인다. input_feature는 feature extractor로, label은 tokenizer로 다루어야 한다.
input_feature는 30초 길이로 패딩되고 고정된 차원의 log_mel spectrogram으로 변환되었으므로, 우리가 할 일은 이를 PyTorch tensor로 변환하는 것뿐이다. 이를 위해 .pad 메서드의 return_tensor=pt 인자를 사용한다. 이때 이미 패딩이 완료되었으므로 여기서 추가적인 패딩 작업이 이루어지지는 않으며, 그저 input_feature를 PyTorch tensor로 변환하기만 할 것이다.
반면, label은 아직 패딩 작업이 이루어지지 않았다. 따라서 먼저 tokenizer의 .pad 메서드를 이용해 패딩 작업을 실시할 것이다. 패딩 토큰들은 -100으로 치환되며, 따라서 이 토큰들은 모델이 loss를 계산할 때는 이용되지 않을 것이다. 그리고 이후 training 작업 동안 우리는 label sequence의 시작 부분에 있는 transcript 토큰을 잘라낼 것이다.
앞서 선언했던 WhisperPrecessor를 이용해 feature extractor와 tokenizer 모두를 사용할 수 있다.
import torch
from dataclasses import dataclass
from typing import Any, Dict, List, Union
@dataclass
class DataCollatorSpeechSeq2SeqWithPadding:
processor: Any
def __call__(self, features: List[Dict[str, Union[List[int], torch.Tensor]]]) -> Dict[str, torch.Tensor]:
# 인풋 데이터와 라벨 데이터의 길이가 다르며, 따라서 서로 다른 패딩 방법이 적용되어야 한다. 그러므로 두 데이터를 분리해야 한다.
# 먼저 오디오 인풋 데이터를 간단히 토치 텐서로 반환하는 작업을 수행한다.
input_features = [{"input_features": feature["input_features"]} for feature in features]
batch = self.processor.feature_extractor.pad(input_features, return_tensors="pt")
# Tokenize된 레이블 시퀀스를 가져온다.
label_features = [{"input_ids": feature["labels"]} for feature in features]
# 레이블 시퀀스에 대해 최대 길이만큼 패딩 작업을 실시한다.
labels_batch = self.processor.tokenizer.pad(label_features, return_tensors="pt")
# 패딩 토큰을 -100으로 치환하여 loss 계산 과정에서 무시되도록 한다.
labels = labels_batch["input_ids"].masked_fill(labels_batch.attention_mask.ne(1), -100)
# 이전 토크나이즈 과정에서 bos 토큰이 추가되었다면 bos 토큰을 잘라낸다.
# 해당 토큰은 이후 언제든 추가할 수 있다.
if (labels[:, 0] == self.processor.tokenizer.bos_token_id).all().cpu().item():
labels = labels[:, 1:]
batch["labels"] = labels
return batch
# 훈련시킬 모델의 processor, tokenizer, feature extractor 로드
from transformers import WhisperTokenizer, WhisperFeatureExtractor
from transformers import WhisperProcessor
processor = WhisperProcessor.from_pretrained("openai/whisper-base", language="Korean", task="transcribe")
tokenizer = WhisperTokenizer.from_pretrained("openai/whisper-base", language="Korean", task="transcribe")
feature_extractor = WhisperFeatureExtractor.from_pretrained("openai/whisper-base")
# 데이터 콜레이터 초기화
data_collator = DataCollatorSpeechSeq2SeqWithPadding(processor=processor)
검증 데이터셋에 사용할 evaluation metrics를 정의한다. 영어의 경우 WER을 사용하지만, 한국어 데이터이므로 CER을 사용하는 것이 더 적절할 것이다.
import evaluate
metric = evaluate.load('cer')
compute_metrics라는 이름의 이 함수는 가장 먼저 label_ids에 있는 -100을 pad_token으로 치환한다(data collator가 비용함수를 계산할 때 pad token을 무시하도록 하기 위해 설정한). 그 뒤 예측치와 label ids를 문자열로 변환하고 예측치와 정답을 비교하여 CER을 계산해낸다.
def compute_metrics(pred):
pred_ids = pred.predictions
label_ids = pred.label_ids
# pad_token을 -100으로 치환
label_ids[label_ids == -100] = tokenizer.pad_token_id
# metrics 계산 시 special token들을 빼고 계산하도록 설정
pred_str = tokenizer.batch_decode(pred_ids, skip_special_tokens=True)
label_str = tokenizer.batch_decode(label_ids, skip_special_tokens=True)
cer = 100 * metric.compute(predictions=pred_str, references=label_str)
return {"cer": cer}
pre-trained Whisper base 모델의 checkpoint를 로드한다.
from transformers import WhisperForConditionalGeneration
model = WhisperForConditionalGeneration.from_pretrained("openai/whisper-base")
또한 문장의 생성 중 완전하게 억제되는 토큰들도 있다(suppress_tokens). 이 토큰들은 로그 확률을 -inf로 설정하며, 따라서 샘플링되지 않는다. 우리는 이 토큰들을 비어 있는 리스트로 치환할 것이다. 즉, 어떤 토큰도 억제되지 않는다.
model.config.forced_decoder_ids = None
model.config.suppress_tokens = []
마지막 단계로서, 트레이닝을 위한 모든 파라미터들을 정의해야 한다. 각각의 파라미터들은 다음과 같은 의미를 갖는다.
from transformers import Seq2SeqTrainingArguments
training_args = Seq2SeqTrainingArguments(
output_dir="repo_name", # 원하는 리포지토리 이름을 임력한다.
per_device_train_batch_size=16,
gradient_accumulation_steps=1, # 배치 크기가 2배 감소할 때마다 2배씩 증가
learning_rate=1e-5,
warmup_steps=500,
max_steps=4000, # epoch 대신 설정
gradient_checkpointing=True,
fp16=True,
evaluation_strategy="steps",
per_device_eval_batch_size=8,
predict_with_generate=True,
generation_max_length=225,
save_steps=1000,
eval_steps=1000,
logging_steps=25,
report_to=["tensorboard"],
load_best_model_at_end=True,
metric_for_best_model="cer", # 한국어의 경우 'wer'보다는 'cer'이 더 적합할 것
greater_is_better=False,
push_to_hub=True,
)
from transformers import Seq2SeqTrainer
trainer = Seq2SeqTrainer(
args=training_args,
model=model,
train_dataset=low_call_voices_prepreocessed["train"],
eval_dataset=low_call_voices_prepreocessed["valid"], # or "test"
data_collator=data_collator,
compute_metrics=compute_metrics,
tokenizer=processor.feature_extractor,
)
trainer.train()
kwargs = {
"dataset_tags": "사용한 데이터셋의 주소 입력",
"dataset": "사용한 데이터셋의 이름 입력", # a 'pretty' name for the training dataset
"dataset_args": "config: ko, split: valid",
"language": "ko",
"model_name": "모델 이름 입력", # a 'pretty' name for your model
"finetuned_from": "openai/whisper-base",
"tasks": "automatic-speech-recognition",
"tags": "hf-asr-leaderboard",
}
trainer.push_to_hub(**kwargs)
processor.push_to_hub("repo_name")
tokenizer.push_to_hub("repo_name")
# 파인 튜닝한 모델을 로드
from transformers import WhisperForConditionalGeneration, WhisperProcessor, WhisperFeatureExtractor, WhisperTokenizer
model = WhisperForConditionalGeneration.from_pretrained("허브에 업로드한 모델 주소 입력")
feature_extractor = WhisperFeatureExtractor.from_pretrained("허브에 업로드한 모델 주소 입력")
tokenizer = WhisperTokenizer.from_pretrained("허브에 업로드한 모델 주소 입력", language="Korean", task="transcribe")
processor = WhisperProcessor.from_pretrained("허브에 업로드한 모델 주소 입력")
from transformers import Seq2SeqTrainingArguments
training_args = Seq2SeqTrainingArguments(
output_dir="repo_name", # 원하는 리포지토리 이름을 임력한다.
per_device_train_batch_size=16,
gradient_accumulation_steps=1, # 배치 크기가 2배 감소할 때마다 2배씩 증가
learning_rate=1e-5,
warmup_steps=500,
max_steps=4000,
gradient_checkpointing=True,
fp16=True,
evaluation_strategy="steps",
per_device_eval_batch_size=8,
predict_with_generate=True,
generation_max_length=225,
save_steps=1000,
eval_steps=1000,
logging_steps=25,
report_to=["tensorboard"],
load_best_model_at_end=True,
metric_for_best_model="cer", # 한국어의 경우 'wer'보다는 'cer'이 더 적합할 것
greater_is_better=False,
push_to_hub=False,
)
from transformers import Seq2SeqTrainer
trainer = Seq2SeqTrainer(
args=training_args,
model=model,
train_dataset=low_call_voices_prepreocessed["train"],
eval_dataset=low_call_voices_prepreocessed["test"], # for evaluation(not validation)
data_collator=data_collator,
compute_metrics=compute_metrics,
tokenizer=processor.feature_extractor,
)
Evaluation 진행
trainer.evaluate()
여기까지 한국어 '저음질 전화망 음성인식 데이터'를 이용한 Whisper의 파인 튜닝 과정을 기록해 보았습니다. 참고하였던 'Fine-Tune Whisper For Multilingual ASR with 🤗 Transformers' 포스트와 비교하면 이론적인 내용도 상당히 많이 빠져 있고, 아무래도 번역 실력도 좋지 않다 보니 잘못된 내용이 있을 수도 있습니다.
코드의 경우에도 제가 작업한 코드에서 쓸모없는 부분을 정리한 상태입니다만, 작업이 작업인지라 최종적으로 코드를 실행해보지는 않은 상태라 오류가 있을 수도 있습니다.
즉, 여러모로 부족한 부분이 많은 포스트입니다. 따라서 Whisper의 파인튜닝 작업을 하고자 하시는 분들은 상기 참고 포스트와 함께 OpenAI의 공식 문서나 HuggingFace의 관련 문서도 살펴보시기를 강력히 권장드립니다.
다만 DatasetDict를 만드는 과정이나 훈련이 끝난 모델을 업로드하는 과정 등 제가 나름대로의 방법을 찾아 정리한 부분도 있습니다. 부족하나마 이를 함께 참고하시면 원하시는 도메인에 적합한 형태로 Whisper를 파인튜닝하는 데 도움이 되지 않을까 합니다.
서두에 밝혔듯, 자연어 처리 교육 과정을 밟고 있는 응애 개발자 지망생입니다. 잘못된 부분이나 부족한 부분은 지도해주시면 감사히 배우고 보완할 수 있도록 하겠습니다.
안녕하세요, 혹시 사용하신 컴퓨팅 자원(GPU)이나 학습 시간 등에 대해 구체적으로 알 수 있을까요?
너무 오래 걸리면 시간을 효율적으로 분배해야할 것 같아서 문의드려요.
안녕하세요 현재 진행중인 프로젝트 내에서 동일한 자료를 토대로 model whisper를 finetuning 작업을 진행하려 하고 있습니다. 혹시 ai hub의 ‘저음질 전화 음성인식 데이터’ 내의 \n \o 와 같은 텍스트의 전처리 과정을 거치고 fine tuning 을 진행하신 것인지 궁금합니다!