Qwen3-ASR-1.7B模型蒸馏教程：小模型继承大模型能力-深圳市維司達科技有限公司

Qwen3-ASR-1.7B模型蒸馏教程：小模型继承大模型能力

最近阿里开源的Qwen3-ASR-1.7B语音识别模型确实让人眼前一亮，支持52种语言和方言，识别准确率还特别高。但问题来了，1.7B的参数量对很多实际应用场景来说还是有点大，特别是想在手机、智能音箱这些设备上跑起来，对模型大小和推理速度都有严格要求。

这时候模型蒸馏技术就派上用场了。简单来说，就是让一个小模型去“学习”大模型的能力，就像学生跟着老师学一样。今天我就来手把手教你，怎么用知识蒸馏技术，让一个小模型也能拥有Qwen3-ASR-1.7B的强大语音识别能力。

1. 准备工作：环境搭建与数据准备

1.1 环境配置

首先得把环境准备好。我建议用Python 3.9以上版本，然后安装必要的依赖库。

# 创建虚拟环境 python -m venv asr_distill_env source asr_distill_env/bin/activate # Linux/Mac # 或者 asr_distill_env\Scripts\activate # Windows # 安装核心依赖 pip install torch torchaudio transformers datasets pip install accelerate peft bitsandbytes pip install scikit-learn soundfile

如果你有GPU的话，最好装一下CUDA版本的PyTorch，这样训练会快很多。没有GPU也没关系，CPU也能跑，就是慢一点。

1.2 下载Qwen3-ASR-1.7B模型

蒸馏的第一步是要有个好老师，我们先下载Qwen3-ASR-1.7B这个大模型。

from transformers import AutoModelForSpeechSeq2Seq, AutoProcessor # 下载教师模型（Qwen3-ASR-1.7B） teacher_model_name = "Qwen/Qwen3-ASR-1.7B" print(f"正在下载教师模型: {teacher_model_name}") teacher_model = AutoModelForSpeechSeq2Seq.from_pretrained( teacher_model_name, torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32, low_cpu_mem_usage=True, use_safetensors=True ) processor = AutoProcessor.from_pretrained(teacher_model_name) # 把模型放到GPU上（如果有的话） device = "cuda" if torch.cuda.is_available() else "cpu" teacher_model.to(device) teacher_model.eval() # 设置为评估模式 print(f"教师模型已加载到 {device}")

1.3 准备学生模型

学生模型我们选个小一点的，比如Whisper-tiny或者自己设计的小模型。这里我用Whisper-tiny作为例子，因为它结构相对简单，参数量只有39M，很适合蒸馏。

from transformers import WhisperForConditionalGeneration # 下载学生模型（Whisper-tiny） student_model_name = "openai/whisper-tiny" print(f"正在下载学生模型: {student_model_name}") student_model = WhisperForConditionalGeneration.from_pretrained( student_model_name, torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32 ) student_model.to(device) print(f"学生模型已加载到 {device}")

1.4 准备训练数据

蒸馏需要一些语音数据作为训练样本。你可以用公开的语音数据集，比如LibriSpeech、Common Voice，或者用自己的业务数据。

from datasets import load_dataset import soundfile as sf # 加载一个公开的语音数据集（这里用Common Voice中文版作为例子） def load_audio_dataset(split="train", language="zh-CN", max_samples=1000): dataset = load_dataset("mozilla-foundation/common_voice_16_1", language, split=split) # 只取前max_samples个样本，避免数据太多 dataset = dataset.select(range(min(len(dataset), max_samples))) # 预处理函数 def preprocess_function(examples): # 读取音频文件 audio_arrays = [] for audio_path in examples["audio"]: audio, sr = sf.read(audio_path["path"]) audio_arrays.append(audio) # 用教师模型的processor处理音频 inputs = processor( audio_arrays, sampling_rate=16000, text=examples["sentence"], padding=True, truncation=True, max_length=480000, # 30秒音频 return_tensors="pt" ) return inputs # 应用预处理 dataset = dataset.map( preprocess_function, batched=True, batch_size=8, remove_columns=dataset.column_names ) return dataset # 加载训练集和验证集 train_dataset = load_audio_dataset("train", max_samples=500) val_dataset = load_audio_dataset("validation", max_samples=100) print(f"训练集大小: {len(train_dataset)}") print(f"验证集大小: {len(val_dataset)}")

2. 知识蒸馏的核心原理

在开始写代码之前，我先用大白话解释一下知识蒸馏到底是怎么回事。

想象一下，你是个学生，要学一门很难的课。有两种学习方法：

只看标准答案（硬标签） - 只知道对错，不知道为什么
跟着老师学（软标签） - 老师不仅告诉你答案，还告诉你解题思路，哪些地方容易错

知识蒸馏就是第二种方法。大模型（老师）不仅给出最终的文字转录结果，还给出每个词的概率分布。小模型（学生）不仅要学会给出正确结果，还要学会老师的“思考方式”。

具体来说，蒸馏过程关注三个损失：

硬标签损失：学生模型的输出和真实标签的差异
软标签损失：学生模型的输出和老师模型输出的差异
隐藏层损失：学生模型中间层的表示和老师模型中间层表示的差异

3. 实现知识蒸馏训练

3.1 定义蒸馏损失函数

这是蒸馏最核心的部分，我们要定义学生模型怎么向老师模型学习。

import torch import torch.nn as nn import torch.nn.functional as F class DistillationLoss(nn.Module): def __init__(self, temperature=2.0, alpha=0.5, beta=0.3): """ 初始化蒸馏损失函数 参数: - temperature: 温度参数，控制软标签的平滑程度 - alpha: 硬标签损失的权重 - beta: 软标签损失的权重 """ super().__init__() self.temperature = temperature self.alpha = alpha self.beta = beta self.ce_loss = nn.CrossEntropyLoss() self.kl_loss = nn.KLDivLoss(reduction="batchmean") def forward(self, student_logits, teacher_logits, labels): """ 计算蒸馏损失 参数: - student_logits: 学生模型的输出logits - teacher_logits: 教师模型的输出logits - labels: 真实标签 """ # 硬标签损失（学生 vs 真实标签） hard_loss = self.ce_loss(student_logits.view(-1, student_logits.size(-1)), labels.view(-1)) # 软标签损失（学生 vs 教师） # 应用温度缩放 student_probs = F.log_softmax(student_logits / self.temperature, dim=-1) teacher_probs = F.softmax(teacher_logits / self.temperature, dim=-1) soft_loss = self.kl_loss(student_probs, teacher_probs) * (self.temperature ** 2) # 总损失 total_loss = self.alpha * hard_loss + self.beta * soft_loss return total_loss, hard_loss, soft_loss

3.2 实现特征蒸馏

除了输出层的蒸馏，我们还可以让学生模型学习老师模型的中间特征表示。

class FeatureDistillationLoss(nn.Module): def __init__(self, layer_mapping=None): """ 特征蒸馏损失 参数: - layer_mapping: 教师和学生模型层的对应关系 """ super().__init__() self.mse_loss = nn.MSELoss() self.layer_mapping = layer_mapping or {} def compute_feature_loss(self, student_features, teacher_features): """计算特征层之间的损失""" losses = [] for student_layer, teacher_layer in self.layer_mapping.items(): if student_layer in student_features and teacher_layer in teacher_features: # 对齐特征维度（如果需要） s_feat = student_features[student_layer] t_feat = teacher_features[teacher_layer] # 如果维度不匹配，进行适配 if s_feat.size() != t_feat.size(): # 简单的适配方法：平均池化或线性投影 if s_feat.size(-1) != t_feat.size(-1): adapter = nn.Linear(s_feat.size(-1), t_feat.size(-1)).to(s_feat.device) s_feat = adapter(s_feat) loss = self.mse_loss(s_feat, t_feat) losses.append(loss) return torch.stack(losses).mean() if losses else torch.tensor(0.0)

3.3 完整的蒸馏训练循环

现在我们把所有部分组合起来，实现完整的训练过程。

from tqdm import tqdm from torch.utils.data import DataLoader def train_distillation( teacher_model, student_model, train_dataset, val_dataset, processor, num_epochs=10, batch_size=4, learning_rate=5e-5, temperature=2.0 ): """ 执行知识蒸馏训练 参数: - teacher_model: 教师模型（Qwen3-ASR-1.7B） - student_model: 学生模型（Whisper-tiny） - train_dataset: 训练数据集 - val_dataset: 验证数据集 - processor: 音频处理器 - num_epochs: 训练轮数 - batch_size: 批次大小 - learning_rate: 学习率 - temperature: 蒸馏温度 """ # 创建数据加载器 train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True) val_loader = DataLoader(val_dataset, batch_size=batch_size) # 定义损失函数 distillation_loss = DistillationLoss(temperature=temperature) feature_loss = FeatureDistillationLoss() # 定义优化器 optimizer = torch.optim.AdamW(student_model.parameters(), lr=learning_rate) # 学习率调度器 scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=num_epochs) # 训练循环 best_val_loss = float('inf') training_history = [] for epoch in range(num_epochs): print(f"\n{'='*50}") print(f"Epoch {epoch+1}/{num_epochs}") print(f"{'='*50}") # 训练阶段 student_model.train() train_loss = 0.0 progress_bar = tqdm(train_loader, desc="Training") for batch in progress_bar: # 获取批次数据 input_features = batch["input_features"].to(device) attention_mask = batch.get("attention_mask", None) if attention_mask is not None: attention_mask = attention_mask.to(device) labels = batch["labels"].to(device) # 前向传播 - 教师模型（不计算梯度） with torch.no_grad(): teacher_outputs = teacher_model( input_features=input_features, attention_mask=attention_mask, labels=labels ) teacher_logits = teacher_outputs.logits # 前向传播 - 学生模型 student_outputs = student_model( input_features=input_features, attention_mask=attention_mask, labels=labels, output_hidden_states=True # 输出隐藏状态用于特征蒸馏 ) student_logits = student_outputs.logits # 计算损失 # 1. 输出层蒸馏损失 distill_loss, hard_loss, soft_loss = distillation_loss( student_logits, teacher_logits, labels ) # 2. 特征蒸馏损失（可选） # 这里需要根据具体模型结构定义层映射关系 # feat_loss = feature_loss.compute_feature_loss( # student_outputs.hidden_states, # teacher_outputs.hidden_states # ) # 总损失 total_loss = distill_loss # + 0.1 * feat_loss # 可以加上特征损失 # 反向传播 optimizer.zero_grad() total_loss.backward() torch.nn.utils.clip_grad_norm_(student_model.parameters(), max_norm=1.0) optimizer.step() # 更新进度条 train_loss += total_loss.item() progress_bar.set_postfix({ "loss": total_loss.item(), "hard": hard_loss.item(), "soft": soft_loss.item() }) avg_train_loss = train_loss / len(train_loader) # 验证阶段 student_model.eval() val_loss = 0.0 with torch.no_grad(): for batch in tqdm(val_loader, desc="Validation"): input_features = batch["input_features"].to(device) attention_mask = batch.get("attention_mask", None) if attention_mask is not None: attention_mask = attention_mask.to(device) labels = batch["labels"].to(device) # 学生模型推理 student_outputs = student_model( input_features=input_features, attention_mask=attention_mask, labels=labels ) # 计算验证损失 loss = student_outputs.loss val_loss += loss.item() avg_val_loss = val_loss / len(val_loader) # 打印epoch结果 print(f"\nEpoch {epoch+1} 结果:") print(f"训练损失: {avg_train_loss:.4f}") print(f"验证损失: {avg_val_loss:.4f}") # 保存最佳模型 if avg_val_loss < best_val_loss: best_val_loss = avg_val_loss torch.save({ 'epoch': epoch, 'model_state_dict': student_model.state_dict(), 'optimizer_state_dict': optimizer.state_dict(), 'loss': best_val_loss, }, 'best_student_model.pth') print(f"保存最佳模型，验证损失: {best_val_loss:.4f}") # 更新学习率 scheduler.step() # 记录训练历史 training_history.append({ 'epoch': epoch + 1, 'train_loss': avg_train_loss, 'val_loss': avg_val_loss, 'learning_rate': scheduler.get_last_lr()[0] }) print(f"\n{'='*50}") print(f"训练完成！最佳验证损失: {best_val_loss:.4f}") print(f"{'='*50}") return student_model, training_history

3.4 开始训练

现在我们可以开始蒸馏训练了。

# 设置训练参数 training_config = { "num_epochs": 15, "batch_size": 8, # 根据GPU内存调整 "learning_rate": 3e-5, "temperature": 2.0, "warmup_steps": 100, } # 开始训练 print("开始知识蒸馏训练...") print(f"教师模型: Qwen3-ASR-1.7B") print(f"学生模型: Whisper-tiny") print(f"训练参数: {training_config}") trained_student, history = train_distillation( teacher_model=teacher_model, student_model=student_model, train_dataset=train_dataset, val_dataset=val_dataset, processor=processor, num_epochs=training_config["num_epochs"], batch_size=training_config["batch_size"], learning_rate=training_config["learning_rate"], temperature=training_config["temperature"] )

4. 评估蒸馏效果

训练完成后，我们需要评估一下蒸馏的效果，看看小模型到底学到了多少。

4.1 计算词错误率（WER）

词错误率是语音识别最常用的评估指标。

from evaluate import load import numpy as np def evaluate_wer(model, dataset, processor, num_samples=50): """ 评估模型的词错误率 参数: - model: 要评估的模型 - dataset: 评估数据集 - processor: 音频处理器 - num_samples: 评估样本数量 """ wer_metric = load("wer") model.eval() predictions = [] references = [] # 随机选择一些样本进行评估 indices = np.random.choice(len(dataset), min(num_samples, len(dataset)), replace=False) with torch.no_grad(): for idx in tqdm(indices, desc="评估中"): sample = dataset[idx] # 获取音频特征 input_features = sample["input_features"].unsqueeze(0).to(device) attention_mask = sample.get("attention_mask", None) if attention_mask is not None: attention_mask = attention_mask.unsqueeze(0).to(device) # 模型推理 generated_ids = model.generate( input_features=input_features, attention_mask=attention_mask, max_length=448, num_beams=5, temperature=0.0 ) # 解码文本 transcription = processor.batch_decode(generated_ids, skip_special_tokens=True)[0] # 获取真实文本（需要根据数据集结构调整） # 这里假设数据集中有"text"字段 if "text" in sample: reference_text = sample["text"] else: # 如果没有text字段，尝试从labels解码 labels = sample["labels"] reference_text = processor.batch_decode(labels.unsqueeze(0), skip_special_tokens=True)[0] predictions.append(transcription) references.append(reference_text) # 计算WER wer = wer_metric.compute(predictions=predictions, references=references) # 打印一些例子 print("\n评估结果示例:") for i in range(min(3, len(predictions))): print(f"\n样本 {i+1}:") print(f"真实文本: {references[i]}") print(f"预测文本: {predictions[i]}") return wer, predictions, references # 评估教师模型 print("评估教师模型 (Qwen3-ASR-1.7B)...") teacher_wer, _, _ = evaluate_wer(teacher_model, val_dataset, processor, num_samples=30) print(f"教师模型 WER: {teacher_wer:.4f}") # 评估原始学生模型 print("\n评估原始学生模型 (Whisper-tiny)...") original_student = WhisperForConditionalGeneration.from_pretrained("openai/whisper-tiny").to(device) original_student_wer, _, _ = evaluate_wer(original_student, val_dataset, processor, num_samples=30) print(f"原始学生模型 WER: {original_student_wer:.4f}") # 评估蒸馏后的学生模型 print("\n评估蒸馏后的学生模型...") distilled_student_wer, _, _ = evaluate_wer(trained_student, val_dataset, processor, num_samples=30) print(f"蒸馏后学生模型 WER: {distilled_student_wer:.4f}") # 对比结果 print("\n" + "="*50) print("模型性能对比:") print("="*50) print(f"教师模型 (Qwen3-ASR-1.7B) WER: {teacher_wer:.4f}") print(f"原始学生模型 (Whisper-tiny) WER: {original_student_wer:.4f}") print(f"蒸馏后学生模型 WER: {distilled_student_wer:.4f}") print(f"WER提升: {(original_student_wer - distilled_student_wer):.4f} (相对提升 {((original_student_wer - distilled_student_wer)/original_student_wer*100):.1f}%)")

4.2 模型大小和推理速度对比

除了准确率，我们还要关心模型的大小和推理速度。

import time import os def compare_model_size_and_speed(teacher_model, student_model, original_student, processor): """ 比较模型大小和推理速度 """ results = {} # 计算模型大小 def get_model_size(model, model_name): # 保存模型到临时文件 temp_path = f"temp_{model_name}.pth" torch.save(model.state_dict(), temp_path) # 获取文件大小 size_bytes = os.path.getsize(temp_path) size_mb = size_bytes / (1024 * 1024) # 删除临时文件 os.remove(temp_path) return size_mb # 模型大小 results["teacher_size_mb"] = get_model_size(teacher_model, "teacher") results["original_student_size_mb"] = get_model_size(original_student, "original_student") results["distilled_student_size_mb"] = get_model_size(student_model, "distilled_student") # 推理速度测试 def measure_inference_time(model, model_name, num_runs=10): # 创建一个测试输入 test_input = torch.randn(1, 80, 3000).to(device) # 模拟3秒音频的mel特征 # 预热 for _ in range(3): _ = model.generate(input_features=test_input, max_length=100) # 测量推理时间 start_time = time.time() for _ in range(num_runs): with torch.no_grad(): _ = model.generate(input_features=test_input, max_length=100) end_time = time.time() avg_time = (end_time - start_time) / num_runs return avg_time # 推理时间 results["teacher_inference_time"] = measure_inference_time(teacher_model, "teacher") results["original_student_inference_time"] = measure_inference_time(original_student, "original_student") results["distilled_student_inference_time"] = measure_inference_time(student_model, "distilled_student") return results # 运行对比测试 print("进行模型大小和推理速度对比...") comparison_results = compare_model_size_and_speed( teacher_model, trained_student, original_student, processor ) print("\n" + "="*50) print("模型大小和推理速度对比:") print("="*50) print(f"教师模型 (Qwen3-ASR-1.7B):") print(f" - 模型大小: {comparison_results['teacher_size_mb']:.1f} MB") print(f" - 平均推理时间: {comparison_results['teacher_inference_time']*1000:.1f} ms") print(f"\n原始学生模型 (Whisper-tiny):") print(f" - 模型大小: {comparison_results['original_student_size_mb']:.1f} MB") print(f" - 平均推理时间: {comparison_results['original_student_inference_time']*1000:.1f} ms") print(f" - 相对于教师模型: 大小减少 {((comparison_results['teacher_size_mb'] - comparison_results['original_student_size_mb'])/comparison_results['teacher_size_mb']*100):.1f}%") print(f" - 推理速度提升: {comparison_results['teacher_inference_time']/comparison_results['original_student_inference_time']:.1f}倍") print(f"\n蒸馏后学生模型:") print(f" - 模型大小: {comparison_results['distilled_student_size_mb']:.1f} MB") print(f" - 平均推理时间: {comparison_results['distilled_student_inference_time']*1000:.1f} ms") print(f" - 相对于教师模型: 大小减少 {((comparison_results['teacher_size_mb'] - comparison_results['distilled_student_size_mb'])/comparison_results['teacher_size_mb']*100):.1f}%") print(f" - 推理速度提升: {comparison_results['teacher_inference_time']/comparison_results['distilled_student_inference_time']:.1f}倍")

5. 实际应用示例

训练好的小模型怎么用呢？我写几个实际的使用例子。

5.1 语音转文字API

from fastapi import FastAPI, UploadFile, File import uvicorn from pydantic import BaseModel import io app = FastAPI(title="蒸馏语音识别API") class TranscriptionResponse(BaseModel): text: str confidence: float processing_time: float # 加载蒸馏后的模型 def load_distilled_model(model_path="best_student_model.pth"): """加载训练好的蒸馏模型""" # 先加载基础模型结构 model = WhisperForConditionalGeneration.from_pretrained("openai/whisper-tiny") # 加载蒸馏后的权重 checkpoint = torch.load(model_path, map_location=device) model.load_state_dict(checkpoint['model_state_dict']) model.to(device) model.eval() return model distilled_model = load_distilled_model() @app.post("/transcribe", response_model=TranscriptionResponse) async def transcribe_audio(file: UploadFile = File(...)): """ 语音转文字API 支持格式: wav, mp3, flac等 """ start_time = time.time() try: # 读取上传的音频文件 audio_bytes = await file.read() audio_io = io.BytesIO(audio_bytes) # 读取音频 audio, sr = sf.read(audio_io) # 重采样到16kHz（如果需要） if sr != 16000: import librosa audio = librosa.resample(audio, orig_sr=sr, target_sr=16000) # 预处理音频 inputs = processor( audio, sampling_rate=16000, return_tensors="pt", padding=True ) input_features = inputs.input_features.to(device) # 推理 with torch.no_grad(): generated_ids = distilled_model.generate( input_features=input_features, max_length=448, num_beams=5, temperature=0.0 ) # 解码文本 transcription = processor.batch_decode(generated_ids, skip_special_tokens=True)[0] processing_time = time.time() - start_time # 这里可以添加置信度计算（比如beam search的概率） confidence = 0.95 # 示例值，实际应该计算 return TranscriptionResponse( text=transcription, confidence=confidence, processing_time=processing_time ) except Exception as e: return {"error": str(e)} # 运行API if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000)

5.2 实时语音识别

import pyaudio import numpy as np import queue import threading class RealtimeASR: """实时语音识别类""" def __init__(self, model, processor, chunk_duration=1.0): """ 初始化实时语音识别 参数: - model: 语音识别模型 - processor: 音频处理器 - chunk_duration: 每次处理的音频时长（秒） """ self.model = model self.processor = processor self.chunk_duration = chunk_duration # 音频参数 self.sample_rate = 16000 self.chunk_size = int(self.sample_rate * chunk_duration) # 音频缓冲区 self.audio_buffer = queue.Queue() self.is_recording = False # 初始化PyAudio self.p = pyaudio.PyAudio() def start_recording(self): """开始录音""" self.is_recording = True # 打开音频流 self.stream = self.p.open( format=pyaudio.paFloat32, channels=1, rate=self.sample_rate, input=True, frames_per_buffer=self.chunk_size, stream_callback=self.audio_callback ) print("开始录音... (按Ctrl+C停止)") def audio_callback(self, in_data, frame_count, time_info, status): """音频回调函数""" if self.is_recording: # 将音频数据放入缓冲区 audio_data = np.frombuffer(in_data, dtype=np.float32) self.audio_buffer.put(audio_data) return (None, pyaudio.paContinue) def process_audio(self): """处理音频缓冲区中的数据""" while self.is_recording or not self.audio_buffer.empty(): try: # 从缓冲区获取音频数据 audio_chunk = self.audio_buffer.get(timeout=0.1) # 预处理音频 inputs = self.processor( audio_chunk, sampling_rate=self.sample_rate, return_tensors="pt" ) input_features = inputs.input_features.to(device) # 推理 with torch.no_grad(): generated_ids = self.model.generate( input_features=input_features, max_length=200, num_beams=3 ) # 解码文本 text = self.processor.batch_decode(generated_ids, skip_special_tokens=True)[0] # 打印识别结果 if text.strip(): # 只打印非空文本 print(f"识别结果: {text}") except queue.Empty: continue except Exception as e: print(f"处理错误: {e}") def stop_recording(self): """停止录音""" self.is_recording = False self.stream.stop_stream() self.stream.close() print("录音已停止") def run(self): """运行实时语音识别""" # 启动录音线程 record_thread = threading.Thread(target=self.start_recording) record_thread.start() # 启动处理线程 process_thread = threading.Thread(target=self.process_audio) process_thread.start() try: # 等待用户中断 while True: time.sleep(0.1) except KeyboardInterrupt: print("\n正在停止...") self.stop_recording() record_thread.join() process_thread.join() # 清理PyAudio self.p.terminate() # 使用示例 if __name__ == "__main__": # 加载蒸馏模型 distilled_model = load_distilled_model() # 创建实时识别器 realtime_asr = RealtimeASR(distilled_model, processor) # 运行实时识别 realtime_asr.run()