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:]