当前位置：首页 > news >正文

PyTorch + PaddlePaddle 语音识别

news 2025/8/8 5:26:58

PyTorch + PaddlePaddle 语音识别

2. 环境配置

2.1 安装依赖

# 安装PyTorch
pip install torch==2.0.0 torchaudio==2.0.0# 安装PaddlePaddle
pip install paddlepaddle==2.5.0 paddlespeech==1.4.0# 安装其他依赖
pip install numpy scipy librosa soundfile
pip install transformers datasets
pip install tensorboard matplotlib

2.2 验证安装

import torch
import paddle
import paddlespeech
import torchaudioprint(f"PyTorch version: {torch.__version__}")
print(f"PaddlePaddle version: {paddle.__version__}")
print(f"CUDA available (PyTorch): {torch.cuda.is_available()}")
print(f"CUDA available (Paddle): {paddle.device.is_compiled_with_cuda()}")

3. 基础理论

3.1 语音识别流程

音频输入 → 特征提取 → 声学模型 → 解码器 → 文本输出

3.2 关键技术

特征提取: MFCC, Mel-Spectrogram, Filter Bank
声学模型: CNN, RNN, Transformer
解码算法: CTC, Attention, Transducer

4. 数据预处理

4.1 音频特征提取类

import torch
import torchaudio
import numpy as np
from torch.nn.utils.rnn import pad_sequenceclass AudioFeatureExtractor:"""音频特征提取器"""def __init__(self, sample_rate=16000, n_mfcc=13, n_mels=80):self.sample_rate = sample_rateself.n_mfcc = n_mfccself.n_mels = n_mels# PyTorch transformsself.mfcc_transform = torchaudio.transforms.MFCC(sample_rate=sample_rate,n_mfcc=n_mfcc,melkwargs={'n_mels': n_mels})self.mel_transform = torchaudio.transforms.MelSpectrogram(sample_rate=sample_rate,n_mels=n_mels,n_fft=512,hop_length=160)def extract_mfcc(self, waveform):"""提取MFCC特征"""mfcc = self.mfcc_transform(waveform)# 添加一阶和二阶差分delta1 = torchaudio.functional.compute_deltas(mfcc)delta2 = torchaudio.functional.compute_deltas(delta1)features = torch.cat([mfcc, delta1, delta2], dim=1)return featuresdef extract_mel_spectrogram(self, waveform):"""提取Mel频谱特征"""mel_spec = self.mel_transform(waveform)# 转换为对数尺度mel_spec = torch.log(mel_spec + 1e-9)return mel_specdef normalize(self, features):"""特征归一化"""mean = features.mean(dim=-1, keepdim=True)std = features.std(dim=-1, keepdim=True)return (features - mean) / (std + 1e-5)

4.2 数据加载器

from torch.utils.data import Dataset, DataLoader
import pandas as pdclass SpeechDataset(Dataset):"""语音识别数据集"""def __init__(self, data_path, transcript_path, feature_extractor):self.data_path = data_pathself.transcripts = pd.read_csv(transcript_path)self.feature_extractor = feature_extractor# 字符到索引的映射self.char2idx = self._build_vocab()self.idx2char = {v: k for k, v in self.char2idx.items()}def _build_vocab(self):"""构建词汇表"""vocab = set()for text in self.transcripts['text']:vocab.update(list(text))char2idx = {'<pad>': 0, '<sos>': 1, '<eos>': 2, '<unk>': 3}for char in sorted(vocab):char2idx[char] = len(char2idx)return char2idxdef __len__(self):return len(self.transcripts)def __getitem__(self, idx):row = self.transcripts.iloc[idx]audio_path = f"{self.data_path}/{row['audio_file']}"# 加载音频waveform, sr = torchaudio.load(audio_path)# 重采样if sr != self.feature_extractor.sample_rate:resampler = torchaudio.transforms.Resample(sr, self.feature_extractor.sample_rate)waveform = resampler(waveform)# 提取特征features = self.feature_extractor.extract_mel_spectrogram(waveform)features = self.feature_extractor.normalize(features)# 文本编码text = row['text']encoded = [self.char2idx.get(c, self.char2idx['<unk>']) for c in text]encoded = [self.char2idx['<sos>']] + encoded + [self.char2idx['<eos>']]return features, torch.LongTensor(encoded)def collate_fn(batch):"""批处理函数"""features, texts = zip(*batch)# Paddingfeatures_padded = pad_sequence([f.transpose(0, 1) for f in features], batch_first=True, padding_value=0)texts_padded = pad_sequence(texts, batch_first=True, padding_value=0)# 创建掩码feature_lengths = torch.LongTensor([f.size(1) for f in features])text_lengths = torch.LongTensor([len(t) for t in texts])return features_padded, texts_padded, feature_lengths, text_lengths

5. 模型架构设计

5.1 PyTorch模型实现

import torch.nn as nn
import torch.nn.functional as Fclass ConformerBlock(nn.Module):"""Conformer块 - 结合CNN和Transformer的优势"""def __init__(self, dim, num_heads=8, conv_kernel_size=31, dropout=0.1):super().__init__()# Feed Forward Moduleself.ff1 = nn.Sequential(nn.LayerNorm(dim),nn.Linear(dim, dim * 4),nn.SiLU(),nn.Dropout(dropout),nn.Linear(dim * 4, dim),nn.Dropout(dropout))# Multi-Head Self Attentionself.attn = nn.MultiheadAttention(dim, num_heads, dropout=dropout)self.attn_norm = nn.LayerNorm(dim)# Convolution Moduleself.conv = nn.Sequential(nn.LayerNorm(dim),nn.Conv1d(dim, dim * 2, 1),nn.GLU(dim=1),nn.Conv1d(dim, dim, conv_kernel_size, padding=conv_kernel_size//2, groups=dim),nn.BatchNorm1d(dim),nn.SiLU(),nn.Conv1d(dim, dim, 1),nn.Dropout(dropout))# Feed Forward Moduleself.ff2 = nn.Sequential(nn.LayerNorm(dim),nn.Linear(dim, dim * 4),nn.SiLU(),nn.Dropout(dropout),nn.Linear(dim * 4, dim),nn.Dropout(dropout))self.final_norm = nn.LayerNorm(dim)def forward(self, x, mask=None):# First Feed Forwardx = x + 0.5 * self.ff1(x)# Multi-Head Self Attentionattn_out = self.attn_norm(x)attn_out, _ = self.attn(attn_out, attn_out, attn_out, attn_mask=mask)x = x + attn_out# Convolutionconv_out = x.transpose(1, 2)conv_out = self.conv(conv_out)x = x + conv_out.transpose(1, 2)# Second Feed Forwardx = x + 0.5 * self.ff2(x)return self.final_norm(x)class ConformerASR(nn.Module):"""基于Conformer的语音识别模型"""def __init__(self, input_dim, vocab_size, dim=256, num_blocks=12, num_heads=8):super().__init__()# 输入投影self.input_proj = nn.Linear(input_dim, dim)# 位置编码self.pos_encoding = PositionalEncoding(dim)# Conformer块self.conformer_blocks = nn.ModuleList([ConformerBlock(dim, num_heads) for _ in range(num_blocks)])# CTC输出层self.ctc_proj = nn.Linear(dim, vocab_size)# Attention解码器（可选）self.decoder = TransformerDecoder(dim, vocab_size, num_layers=6)def forward(self, x, x_lengths=None, targets=None, target_lengths=None):# 输入投影x = self.input_proj(x)x = self.pos_encoding(x)# 创建掩码if x_lengths is not None:max_len = x.size(1)mask = torch.arange(max_len, device=x.device).expand(len(x_lengths), max_len) >= x_lengths.unsqueeze(1)else:mask = None# Conformer编码for block in self.conformer_blocks:x = block(x, mask)# CTC输出ctc_out = self.ctc_proj(x)outputs = {'ctc_out': ctc_out}# 如果有目标，使用注意力解码器if targets is not None:decoder_out = self.decoder(x, targets, mask)outputs['decoder_out'] = decoder_outreturn outputsclass PositionalEncoding(nn.Module):"""位置编码"""def __init__(self, d_model, max_len=5000):super().__init__()pe = torch.zeros(max_len, d_model)position = torch.arange(0, max_len).unsqueeze(1).float()div_term = torch.exp(torch.arange(0, d_model, 2).float() *-(np.log(10000.0) / d_model))pe[:, 0::2] = torch.sin(position * div_term)pe[:, 1::2] = torch.cos(position * div_term)self.register_buffer('pe', pe.unsqueeze(0))def forward(self, x):return x + self.pe[:, :x.size(1)]

5.2 集成PaddlePaddle预训练模型

import paddle
from paddlespeech.cli.asr import ASRExecutorclass HybridASRModel:"""混合ASR模型 - 结合PyTorch和PaddlePaddle"""def __init__(self, pytorch_model, paddle_model_name='conformer_wenetspeech'):self.pytorch_model = pytorch_model# 初始化PaddlePaddle ASRself.paddle_asr = ASRExecutor()self.paddle_asr.model_name = paddle_model_namedef pytorch_inference(self, audio_features):"""使用PyTorch模型推理"""self.pytorch_model.eval()with torch.no_grad():outputs = self.pytorch_model(audio_features)predictions = torch.argmax(outputs['ctc_out'], dim=-1)return predictionsdef paddle_inference(self, audio_path):"""使用PaddlePaddle模型推理"""result = self.paddle_asr(audio_file=audio_path)return resultdef ensemble_inference(self, audio_path, audio_features, weights=[0.5, 0.5]):"""集成推理"""# PyTorch预测pytorch_pred = self.pytorch_inference(audio_features)pytorch_text = self.decode_predictions(pytorch_pred)# PaddlePaddle预测paddle_text = self.paddle_inference(audio_path)# 结合结果（这里简化处理，实际可以使用更复杂的集成策略）if weights[0] > weights[1]:return pytorch_textelse:return paddle_textdef decode_predictions(self, predictions, idx2char):"""解码预测结果"""texts = []for pred in predictions:chars = [idx2char[idx.item()] for idx in pred if idx != 0]text = ''.join(chars)texts.append(text)return texts

6. 完整实现案例

6.1 训练脚本

import torch
import torch.nn as nn
from torch.optim import Adam
from torch.optim.lr_scheduler import CosineAnnealingLR
from torch.nn import CTCLoss
import tensorboardclass ASRTrainer:"""ASR模型训练器"""def __init__(self, model, train_loader, val_loader, config):self.model = modelself.train_loader = train_loaderself.val_loader = val_loaderself.config = config# 优化器self.optimizer = Adam(model.parameters(), lr=config['lr'], betas=(0.9, 0.98), eps=1e-9)# 学习率调度器self.scheduler = CosineAnnealingLR(self.optimizer, T_max=config['epochs'])# 损失函数self.ctc_loss = CTCLoss(blank=0, reduction='mean', zero_infinity=True)# TensorBoardself.writer = tensorboard.SummaryWriter(config['log_dir'])# 设备self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')self.model.to(self.device)def train_epoch(self, epoch):"""训练一个epoch"""self.model.train()total_loss = 0for batch_idx, (features, targets, feat_lens, target_lens) in enumerate(self.train_loader):# 移动到设备features = features.to(self.device)targets = targets.to(self.device)feat_lens = feat_lens.to(self.device)target_lens = target_lens.to(self.device)# 前向传播outputs = self.model(features, feat_lens)log_probs = F.log_softmax(outputs['ctc_out'], dim=-1)# 计算CTC损失log_probs = log_probs.transpose(0, 1)  # (T, N, C)loss = self.ctc_loss(log_probs, targets, feat_lens, target_lens)# 反向传播self.optimizer.zero_grad()loss.backward()# 梯度裁剪torch.nn.utils.clip_grad_norm_(self.model.parameters(), 5.0)self.optimizer.step()total_loss += loss.item()# 记录if batch_idx % 10 == 0:print(f'Epoch {epoch}, Batch {batch_idx}/{len(self.train_loader)}, 'f'Loss: {loss.item():.4f}')self.writer.add_scalar('train/batch_loss', loss.item(), epoch * len(self.train_loader) + batch_idx)avg_loss = total_loss / len(self.train_loader)self.writer.add_scalar('train/epoch_loss', avg_loss, epoch)return avg_lossdef validate(self, epoch):"""验证"""self.model.eval()total_loss = 0total_cer = 0with torch.no_grad():for features, targets, feat_lens, target_lens in self.val_loader:features = features.to(self.device)targets = targets.to(self.device)feat_lens = feat_lens.to(self.device)target_lens = target_lens.to(self.device)outputs = self.model(features, feat_lens)log_probs = F.log_softmax(outputs['ctc_out'], dim=-1)log_probs = log_probs.transpose(0, 1)loss = self.ctc_loss(log_probs, targets, feat_lens, target_lens)total_loss += loss.item()# 计算CERpredictions = torch.argmax(outputs['ctc_out'], dim=-1)cer = self.calculate_cer(predictions, targets)total_cer += ceravg_loss = total_loss / len(self.val_loader)avg_cer = total_cer / len(self.val_loader)self.writer.add_scalar('val/loss', avg_loss, epoch)self.writer.add_scalar('val/cer', avg_cer, epoch)return avg_loss, avg_cerdef calculate_cer(self, predictions, targets):"""计算字符错误率"""# 简化的CER计算total_chars = 0total_errors = 0for pred, target in zip(predictions, targets):# 移除padding和重复pred = self.remove_duplicates_and_blank(pred)target = target[target != 0]# 计算编辑距离errors = self.edit_distance(pred, target)total_errors += errorstotal_chars += len(target)return total_errors / max(total_chars, 1)def remove_duplicates_and_blank(self, sequence):"""移除重复和空白标记"""result = []prev = Nonefor token in sequence:if token != 0 and token != prev:result.append(token)prev = tokenreturn torch.tensor(result)def edit_distance(self, seq1, seq2):"""计算编辑距离"""m, n = len(seq1), len(seq2)dp = [[0] * (n + 1) for _ in range(m + 1)]for i in range(m + 1):dp[i][0] = ifor j in range(n + 1):dp[0][j] = jfor i in range(1, m + 1):for j in range(1, n + 1):if seq1[i-1] == seq2[j-1]:dp[i][j] = dp[i-1][j-1]else:dp[i][j] = 1 + min(dp[i-1][j], dp[i][j-1], dp[i-1][j-1])return dp[m][n]def train(self):"""完整训练流程"""best_cer = float('inf')for epoch in range(self.config['epochs']):print(f'\n--- Epoch {epoch + 1}/{self.config["epochs"]} ---')# 训练train_loss = self.train_epoch(epoch)print(f'Training Loss: {train_loss:.4f}')# 验证val_loss, val_cer = self.validate(epoch)print(f'Validation Loss: {val_loss:.4f}, CER: {val_cer:.4f}')# 调整学习率self.scheduler.step()# 保存最佳模型if val_cer < best_cer:best_cer = val_certorch.save({'epoch': epoch,'model_state_dict': self.model.state_dict(),'optimizer_state_dict': self.optimizer.state_dict(),'cer': val_cer,}, f'{self.config["save_dir"]}/best_model.pt')print(f'Saved best model with CER: {val_cer:.4f}')self.writer.close()print(f'\nTraining completed. Best CER: {best_cer:.4f}')

6.2 主程序

def main():"""主程序"""# 配置config = {'data_path': './data/speech','transcript_path': './data/transcripts.csv','batch_size': 32,'epochs': 100,'lr': 1e-3,'log_dir': './logs','save_dir': './models','input_dim': 80,'vocab_size': 5000,'model_dim': 256,'num_blocks': 12,'num_heads': 8}# 初始化特征提取器feature_extractor = AudioFeatureExtractor(sample_rate=16000, n_mels=80)# 创建数据集train_dataset = SpeechDataset(config['data_path'], config['transcript_path'],feature_extractor)# 划分训练集和验证集train_size = int(0.9 * len(train_dataset))val_size = len(train_dataset) - train_sizetrain_dataset, val_dataset = torch.utils.data.random_split(train_dataset, [train_size, val_size])# 创建数据加载器train_loader = DataLoader(train_dataset, batch_size=config['batch_size'],shuffle=True,collate_fn=collate_fn,num_workers=4)val_loader = DataLoader(val_dataset,batch_size=config['batch_size'],shuffle=False,collate_fn=collate_fn,num_workers=4)# 创建模型model = ConformerASR(input_dim=config['input_dim'],vocab_size=config['vocab_size'],dim=config['model_dim'],num_blocks=config['num_blocks'],num_heads=config['num_heads'])# 创建训练器trainer = ASRTrainer(model, train_loader, val_loader, config)# 开始训练trainer.train()# 创建混合模型hybrid_model = HybridASRModel(model)# 测试推理test_audio = './test.wav'waveform, sr = torchaudio.load(test_audio)features = feature_extractor.extract_mel_spectrogram(waveform)features = features.unsqueeze(0)  # 添加批次维度# PyTorch推理pytorch_result = hybrid_model.pytorch_inference(features)print(f"PyTorch Result: {pytorch_result}")# PaddlePaddle推理paddle_result = hybrid_model.paddle_inference(test_audio)print(f"PaddlePaddle Result: {paddle_result}")# 集成推理ensemble_result = hybrid_model.ensemble_inference(test_audio, features)print(f"Ensemble Result: {ensemble_result}")if __name__ == "__main__":main()

7. 模型训练与评估

7.1 数据增强技术

class AudioAugmentation:"""音频数据增强"""def __init__(self, sample_rate=16000):self.sample_rate = sample_ratedef add_noise(self, waveform, noise_factor=0.005):"""添加高斯噪声"""noise = torch.randn_like(waveform) * noise_factorreturn waveform + noisedef time_stretch(self, waveform, rate=1.2):"""时间拉伸"""# 使用torchaudio的时间拉伸return torchaudio.functional.time_stretch(waveform, rate)def pitch_shift(self, waveform, n_steps=2):"""音高变换"""return torchaudio.functional.pitch_shift(waveform, self.sample_rate, n_steps)def speed_perturb(self, waveform, speed_factor=1.1):"""速度扰动"""# 改变播放速度old_length = waveform.size(-1)new_length = int(old_length / speed_factor)indices = torch.linspace(0, old_length - 1, new_length).long()return waveform[..., indices]def spec_augment(self, spectrogram, freq_mask=15, time_mask=35):"""SpecAugment - 频谱增强"""# 频率掩码freq_mask_param = freq_masknum_freq_mask = 2for _ in range(num_freq_mask):f = torch.randint(0, freq_mask_param, (1,)).item()f_start = torch.randint(0, spectrogram.size(1) - f, (1,)).item()spectrogram[:, f_start:f_start + f, :] = 0# 时间掩码time_mask_param = time_masknum_time_mask = 2for _ in range(num_time_mask):t = torch.randint(0, time_mask_param, (1,)).item()t_start = torch.randint(0, spectrogram.size(2) - t, (1,)).item()spectrogram[:, :, t_start:t_start + t] = 0return spectrogram

7.2 评估指标

class ASRMetrics:"""ASR评估指标"""@staticmethoddef word_error_rate(reference, hypothesis):"""计算词错误率(WER)"""ref_words = reference.split()hyp_words = hypothesis.split()# 动态规划计算编辑距离d = np.zeros((len(ref_words) + 1, len(hyp_words) + 1))for i in range(len(ref_words) + 1):d[i][0] = ifor j in range(len(hyp_words) + 1):d[0][j] = jfor i in range(1, len(ref_words) + 1):for j in range(1, len(hyp_words) + 1):if ref_words[i-1] == hyp_words[j-1]:d[i][j] = d[i-1][j-1]else:d[i][j] = min(d[i-1][j] + 1,    # 删除d[i][j-1] + 1,    # 插入d[i-1][j-1] + 1   # 替换)return d[len(ref_words)][len(hyp_words)] / len(ref_words)@staticmethoddef character_error_rate(reference, hypothesis):"""计算字符错误率(CER)"""ref_chars = list(reference)hyp_chars = list(hypothesis)# 使用Levenshtein距离distance = edit_distance(ref_chars, hyp_chars)return distance / len(ref_chars)

8. 推理与部署

8.1 模型优化

class ModelOptimizer:"""模型优化器"""@staticmethoddef quantize_model(model, backend='qnnpack'):"""模型量化"""model.eval()# 设置量化后端torch.backends.quantized.engine = backend# 准备量化model.qconfig = torch.quantization.get_default_qconfig(backend)model_prepared = torch.quantization.prepare(model)# 校准（需要运行一些数据）# calibrate_model(model_prepared, calibration_loader)# 转换为量化模型model_quantized = torch.quantization.convert(model_prepared)return model_quantized@staticmethoddef export_onnx(model, dummy_input, output_path):"""导出ONNX模型"""model.eval()torch.onnx.export(model,dummy_input,output_path,export_params=True,opset_version=11,do_constant_folding=True,input_names=['input'],output_names=['output'],dynamic_axes={'input': {0: 'batch_size', 1: 'sequence'},'output': {0: 'batch_size', 1: 'sequence'}})print(f"Model exported to {output_path}")@staticmethoddef torch_script_trace(model, example_input):"""TorchScript追踪"""model.eval()traced_model = torch.jit.trace(model, example_input)return traced_model

8.2 实时推理服务

import asyncio
import websockets
import json
import base64class ASRInferenceServer:"""ASR实时推理服务器"""def __init__(self, model, feature_extractor, port=8765):self.model = modelself.feature_extractor = feature_extractorself.port = portself.model.eval()async def process_audio(self, audio_data):"""处理音频数据"""# 解码base64音频数据audio_bytes = base64.b64decode(audio_data)# 转换为tensorwaveform = torch.frombuffer(audio_bytes, dtype=torch.float32)waveform = waveform.unsqueeze(0)# 提取特征features = self.feature_extractor.extract_mel_spectrogram(waveform)features = features.unsqueeze(0)# 推理with torch.no_grad():outputs = self.model(features)predictions = torch.argmax(outputs['ctc_out'], dim=-1)# 解码text = self.decode_predictions(predictions[0])return textdef decode_predictions(self, predictions):"""解码预测结果"""# 简化的解码逻辑chars = []prev = Nonefor p in predictions:if p != 0 and p != prev:  # 移除空白和重复chars.append(chr(p + 96))  # 简化的字符映射prev = preturn ''.join(chars)async def handle_client(self, websocket, path):"""处理客户端连接"""try:async for message in websocket:data = json.loads(message)if data['type'] == 'audio':# 处理音频result = await self.process_audio(data['audio'])# 发送结果response = {'type': 'transcription','text': result,'timestamp': data.get('timestamp', 0)}await websocket.send(json.dumps(response))except websockets.exceptions.ConnectionClosed:print("Client disconnected")except Exception as e:print(f"Error: {e}")def start(self):"""启动服务器"""start_server = websockets.serve(self.handle_client, "localhost", self.port)print(f"ASR Server started on port {self.port}")asyncio.get_event_loop().run_until_complete(start_server)asyncio.get_event_loop().run_forever()

8.3 客户端示例

class ASRClient:"""ASR客户端"""def __init__(self, server_url="ws://localhost:8765"):self.server_url = server_urlasync def stream_audio(self, audio_file):"""流式发送音频"""async with websockets.connect(self.server_url) as websocket:# 读取音频文件waveform, sr = torchaudio.load(audio_file)# 分块发送chunk_size = sr  # 1秒的音频for i in range(0, waveform.size(1), chunk_size):chunk = waveform[:, i:i+chunk_size]# 转换为字节audio_bytes = chunk.numpy().tobytes()audio_base64 = base64.b64encode(audio_bytes).decode()# 发送数据message = {'type': 'audio','audio': audio_base64,'timestamp': i / sr}await websocket.send(json.dumps(message))# 接收结果response = await websocket.recv()result = json.loads(response)print(f"[{result['timestamp']}s] {result['text']}")# 模拟实时流await asyncio.sleep(1)

9. 性能优化技巧

9.1 内存优化

class MemoryEfficientTraining:"""内存高效训练"""@staticmethoddef gradient_accumulation(model, dataloader, optimizer, accumulation_steps=4):"""梯度累积"""model.train()optimizer.zero_grad()for i, batch in enumerate(dataloader):outputs = model(batch)loss = compute_loss(outputs, batch)loss = loss / accumulation_stepsloss.backward()if (i + 1) % accumulation_steps == 0:optimizer.step()optimizer.zero_grad()@staticmethoddef mixed_precision_training(model, dataloader, optimizer):"""混合精度训练"""from torch.cuda.amp import autocast, GradScalerscaler = GradScaler()for batch in dataloader:optimizer.zero_grad()with autocast():outputs = model(batch)loss = compute_loss(outputs, batch)scaler.scale(loss).backward()scaler.step(optimizer)scaler.update()

9.2 推理加速

class InferenceAcceleration:"""推理加速技术"""@staticmethoddef batch_inference(model, audio_list, batch_size=32):"""批量推理"""model.eval()results = []with torch.no_grad():for i in range(0, len(audio_list), batch_size):batch = audio_list[i:i+batch_size]# 处理批次features = extract_features_batch(batch)outputs = model(features)results.extend(decode_batch(outputs))return results@staticmethoddef streaming_inference(model, audio_stream, window_size=1600, hop_size=800):"""流式推理"""model.eval()buffer = []for chunk in audio_stream:buffer.extend(chunk)while len(buffer) >= window_size:# 处理窗口window = buffer[:window_size]features = extract_features(window)with torch.no_grad():output = model(features)text = decode(output)yield text# 滑动窗口buffer = buffer[hop_size:]