自动语音识别大语言模型的训练、评估与部署

1

!pip install packageName

1

(base) C:\Users\username>

1

conda create --name myenv

1
2

conda activate myenv
conda deactivate

1
2

pip install packageName>=0.0.1
conda install packageName

1

python xxx.py

1

nvidia-smi

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

import torch

print(f' CUDA availability on PyTorch is {torch.cuda.is_available()}')
print(f' Current PyTorch version is {torch.__version__}')
print(f' Current CUDA version is {torch.version.cuda}')
print(f' cuDNN version is {torch.backends.cudnn.version()}')
print(f' The number of available GPU devices is {torch.cuda.device_count()}')

# Use CUDA on the device
device = torch.device("cuda")

1
2

from huggingface_hub import notebook_login
notebook_login()

1
2
3
4

from transformers import WhisperFeatureExtractor, WhisperTokenizer
WhisperFeatureExtractor.from_pretrained("model_id")
WhisperTokenizer.from_pretrained("model_id")
WhisperProcessor.from_pretrained("model_id")

1

language="lang", task="transcribe"  # or "translation"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

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")

        label_features = [{"input_ids": feature["labels"]} for feature in features]
        labels_batch = self.processor.tokenizer.pad(label_features, return_tensors="pt")

        labels = labels_batch["input_ids"].masked_fill(labels_batch.attention_mask.ne(1), -100)

        if (labels[:, 0] == self.processor.tokenizer.bos_token_id).all().cpu().item():
            labels = labels[:, 1:]

        batch["labels"] = labels

        return batch

1
2

import evaluate
metric = evaluate.load("wer")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

def compute_metrics(pred):
    pred_ids = pred.predictions
    label_ids = pred.label_ids

    label_ids[label_ids == -100] = tokenizer.pad_token_id

    pred_str = tokenizer.batch_decode(pred_ids, skip_special_tokens=True)
    label_str = tokenizer.batch_decode(label_ids, skip_special_tokens=True)

    wer = 100 * metric.compute(predictions=pred_str, references=label_str)

    return {"wer": wer}

1
2
3
4
5

from transformers import WhisperForConditionalGeneration
model = WhisperForConditionalGeneration.from_pretrained("model_id")

model.config.forced_decoder_ids = None
model.config.suppress_tokens = []

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

training_args = Seq2SeqTrainingArguments(
    output_dir="kawadlc/whisperv1",          # 仓库名
    per_device_train_batch_size=16,          # 每 GPU 训练批次大小
    gradient_accumulation_steps=1,           # 批次大小减半时翻倍
    learning_rate=1e-5,                      # 控制过拟合/欠拟合的关键参数
    weight_decay=1e-2,                       # 正则化机制
    warmup_steps=200,                        # 提升早期性能
    max_steps=3000,                          # 总优化步数
    gradient_checkpointing=True,             # 节省内存
    evaluation_strategy="steps",             # 评估策略（其他："epoch"）
    fp16=True,                               # 半精度浮点
    per_device_eval_batch_size=8,            # 每 GPU 评估批次大小
    predict_with_generate=True,              # 执行生成
    generation_max_length=200,               # 自回归生成最大 token 数
    eval_steps=500,                          # 每次评估间隔步数
    report_to=["tensorboard"],               # 保存训练日志到 TensorBoard
    load_best_model_at_end=True,             # 结束时加载最优模型
    metric_for_best_model="wer",             # 最优模型判定指标
    greater_is_better=False,                 # WER 越低越好
    push_to_hub=False,                       # 推送到 Hub（可选）
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

from transformers import Seq2SeqTrainer

trainer = Seq2SeqTrainer(
    args=training_args,
    model=model,
    train_dataset=common_voice["train"],
    eval_dataset=common_voice["test"],
    data_collator=data_collator,
    compute_metrics=compute_metrics,
    tokenizer=processor.feature_extractor,
)

1
2
3
4
5

import gc
import torch

gc.collect()
torch.cuda.empty_cache()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

common_voice = DatasetDict()

common_voice["train"] = load_dataset("common_voice", "ja", split="train+validated", use_auth_token=True)
common_voice["validation"] = load_dataset("common_voice", "ja", split="validation", use_auth_token=True)
common_voice["test"] = load_dataset("common_voice", "ja", split="test", use_auth_token=True)

# 创建 DatasetDict，选择训练和评估的样本量
common_voice = DatasetDict({
    "train": common_voice['train'].select(range(3500)),
    "validation": common_voice['validation'].select(range(500)),
    "test": common_voice['test'].select(range(100)),
})

# 移除训练中不需要的列
common_voice = common_voice.remove_columns(["age", "client_id", "down_votes", "gender", "path", "up_votes"])

1

common_voice = common_voice.cast_column("audio", Audio(sampling_rate=16000))

1
2
3
4
5

# 转为小写并忽略撇号
text = [s.lower() for s in text]
punctuation_without_apostrophe = string.punctuation.replace("'", "")
translator = str.maketrans('', '', punctuation_without_apostrophe)
text = [s.translate(translator) for s in text]

1
2
3
4

# 移除特殊标记
def remove_tags(text):
    clean = re.compile('<.*?>')
    return re.sub(clean, '', text)

 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

model = WhisperForConditionalGeneration.from_pretrained(
    'openai/whisper-large-v2',
    load_in_8bit=True,
    device_map="auto"
)

from peft import LoraConfig, PeftModel, LoraModel, LoraConfig, get_peft_model, prepare_model_for_int8_training

model = prepare_model_for_int8_training(model)

def make_inputs_require_grad(module, input, output):
    output.requires_grad_(True)

model.model.encoder.conv1.register_forward_hook(make_inputs_require_grad)

config = LoraConfig(
    r=32,
    lora_alpha=64,
    target_modules=["q_proj", "v_proj"],
    lora_dropout=0.05,
    bias="none"
)

model = get_peft_model(model, config)
model.print_trainable_parameters()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

training_args = Seq2SeqTrainingArguments(
    output_dir="jackdu/whisper-peft",
    per_device_train_batch_size=8,
    gradient_accumulation_steps=1,
    learning_rate=1e-4,
    weight_decay=0.01,
    warmup_steps=0,
    num_train_epochs=3,
    evaluation_strategy="steps",
    fp16=True,
    per_device_eval_batch_size=8,
    generation_max_length=150,
    logging_steps=100,
    remove_unused_columns=False,  # PEFT 必需
    label_names=["labels"],       # PEFT 必需
)

1
2
3
4
5
6
7
8
9

plt.figure(figsize=(10, 6))
plt.plot(training_epoch, training_loss, label="Training Loss")
plt.plot(evaluation_epoch, evaluation_loss, label="Evaluation Loss")
plt.xlabel("Training Epochs")
plt.ylabel("Loss")
plt.title("Loss Curves for Whisper Fine-Tuning")
plt.legend()
plt.grid(True)
plt.show()

1

pip install -qq https://github.com/pyannote/pyannote-audio/archive/refs/heads/develop.zip

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

login(read_token)
device = "cuda:0" if torch.cuda.is_available() else "cpu"

from pyannote.audio import Pipeline, Audio
sd_pipeline = Pipeline.from_pretrained("pyannote/speaker-diarization", use_auth_token=True)
wav_files = glob.glob(os.path.join(audio_dirpath, '*.wav'))

pipe = pipeline(
    "automatic-speech-recognition",
    model="openai/whisper-large-v2",
    chunk_length_s=30,
    device=device,
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

results = []

for audio_file in wav_files:
    diarization = sd_pipeline(audio_file, min_speakers=min_speakers, max_speakers=max_speakers)
    audio = Audio(sample_rate=16000, mono='random')
    for segment, _, speaker in diarization.itertracks(yield_label=True):
        waveform, sample_rate = audio.crop(audio_file, segment)
        text = pipe(
            {"raw": waveform.squeeze().numpy(), "sampling_rate": sample_rate},
            batch_size=8,
            generate_kwargs={"language": "<|zh|>", "task": "transcribe"}
        )["text"]
        results.append({
            'start': segment.start,
            'stop': segment.end,
            'speaker': speaker,
            'text': text
        })

1
2

conda install pytorch==2.0.0 torchaudio==2.0.0 pytorch-cuda=11.8 -c pytorch -c nvidia
pip install git+https://github.com/m-bain/whisperx.git

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

diarize_model = whisperx.DiarizationPipeline(
    model_name="pyannote/speaker-diarization",
    use_auth_token='hf_token',
    device=device
)

model = whisperx.load_model(
    whisper_arch=model,
    device=device,
    compute_type=compute_type,
    language=language_abbr
)

audio = whisperx.load_audio(matching_file_path)
diarize_segments = diarize_model(matching_file_path, min_speakers=6, max_speakers=6)
result = model.transcribe(audio, batch_size=batch_size)
result = whisperx.assign_word_speakers(diarize_segments, result)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

from transformers import Wav2Vec2ForCTC, AutoProcessor

model_id = "facebook/mms-1b-all"
target_lang = "cmn-script_simplified"
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
processor = AutoProcessor.from_pretrained(model_id)
model = Wav2Vec2ForCTC.from_pretrained(model_id)

processor.tokenizer.set_target_lang(target_lang)
model.load_adapter(target_lang)
model = model.to(device)

1
2

pip install pytest-runner
pip install paddlespeech

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

from paddlespeech.cli.asr.infer import ASRExecutor
asr = ASRExecutor()

for audio_file in wav_files:
    result = asr(
        model='conformer_wenetspeech',
        lang='zh',
        sample_rate=16000,
        audio_file=audio_file,
        device=paddle.get_device()
    )
    transcript.append(result)

1

pip install speechbrain

1
2
3
4
5
6
7
8

from speechbrain.pretrained import EncoderDecoderASR

asr_model = EncoderDecoderASR.from_hparams(
    source="speechbrain/asr-transformer-aishell",
    savedir="pretrained_models/asr-transformer-aishell",
    run_opts={"device": "cuda"}
)
result = asr_model.transcribe_file(audio_file)

1

pip install espnet_model_zoo

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

from espnet2.bin.asr_inference import Speech2Text

speech2text = Speech2Text.from_pretrained(
    model_id,
    maxlenratio=0.0,
    minlenratio=0.0,
    beam_size=20,
    ctc_weight=0.3,
    lm_weight=0.5,
    penalty=0.0,
    nbest=1
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

pip install azure-storage-blob

from azure.storage.blob import BlobServiceClient, BlobClient, ContainerClient

def upload_zip_to_azure_blob(account_name, account_key, container_name, local_zip_path, zip_blob_name):
    connection_string = f"DefaultEndpointsProtocol=https;AccountName={account_name};AccountKey={account_key};EndpointSuffix=core.windows.net"
    blob_service_client = BlobServiceClient.from_connection_string(connection_string)

    container_client = blob_service_client.get_container_client(container_name)
    if not container_client.exists():
        container_client.create_container()

    zip_blob_client = container_client.get_blob_client(zip_blob_name)
    with open(local_zip_path, "rb") as zip_file:
        zip_blob_client.upload_blob(zip_file)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

pip install azure-cognitiveservices-speech

from azure.cognitiveservices.speech import SpeechConfig, SpeechRecognizer, AudioConfig

predictions = []

for root, _, files in os.walk(wav_base_path):
    for file_name in files:
        if file_name.endswith(".wav"):
            audio_file_path = os.path.join(root, file_name)
            audio_config = AudioConfig(filename=audio_file_path)
            speech_recognizer = SpeechRecognizer(speech_config=speech_config, audio_config=audio_config)
            result = speech_recognizer.recognize_once()
            predictions.append(result.text)