音频-文本多模态

import librosa
import numpy as np

def compute_mfcc(audio_path: str, n_mfcc: int = 13, sr: int = 16000) -> np.ndarray:
    """从音频文件提取 MFCC 特征
    MFCC 通过梅尔尺度滤波和 DCT 压缩频谱信息
    保留最具判别性的低频倒谱系数"""
    y, _ = librosa.load(audio_path, sr=sr)
    mfcc = librosa.feature.mfcc(y=y, sr=sr, n_mfcc=n_mfcc)
    # 标准化：零均值单位方差
    mfcc = (mfcc - mfcc.mean(axis=1, keepdims=True)) / (mfcc.std(axis=1, keepdims=True) + 1e-8)
    return mfcc  # shape: (n_mfcc, time_steps)

import torchaudio
import torch

def compute_mel_spectrogram(waveform: torch.Tensor, sr: int = 16000) -> torch.Tensor:
    """将波形张量转换为梅尔频谱图
    torchaudio 提供 GPU 加速的频谱图计算"""
    transform = torchaudio.transforms.MelSpectrogram(
        sample_rate=sr,
        n_fft=400,          # 25ms at 16kHz
        hop_length=160,     # 10ms frame shift
        n_mels=80,
        f_min=0,
        f_max=8000
    )
    mel_spec = transform(waveform)  # shape: (1, n_mels, time)
    # 转换为对数尺度
    mel_spec = torch.log(mel_spec + 1e-6)
    return mel_spec

from transformers import WhisperProcessor, WhisperForConditionalGeneration
import torch

def transcribe_with_whisper(audio_path: str, model_name: str = "openai/whisper-medium") -> str:
    """使用 Whisper 模型进行语音识别
    Whisper 将音频编码为序列后自回归生成文本"""
    processor = WhisperProcessor.from_pretrained(model_name)
    model = WhisperForConditionalGeneration.from_pretrained(model_name)

    audio, sr = librosa.load(audio_path, sr=16000)
    inputs = processor(audio, sampling_rate=16000, return_tensors="pt")

    with torch.no_grad():
        generated = model.generate(
            inputs.input_features,
            max_length=448,
            num_beams=5
        )

    return processor.batch_decode(generated, skip_special_tokens=True)[0]

from transformers import Wav2Vec2ForCTC, Wav2Vec2Processor
import torch

def transcribe_with_wav2vec2(audio_path: str, lang: str = "zh") -> str:
    """使用 Wav2Vec2 进行 CTC 语音识别
    CTC 损失允许输入输出长度不一致，无需帧级对齐"""
    model_name = f"facebook/wav2vec2-large-xlsr-53-{lang}"
    processor = Wav2Vec2Processor.from_pretrained(model_name)
    model = Wav2Vec2ForCTC.from_pretrained(model_name)

    audio, sr = librosa.load(audio_path, sr=16000)
    inputs = processor(audio, sampling_rate=16000, return_tensors="pt")

    with torch.no_grad():
        logits = model(inputs.input_values).logits

    predicted_ids = torch.argmax(logits, dim=-1)
    return processor.batch_decode(predicted_ids)[0]

import torch
import torch.nn as nn
from torchvision import models

class AudioClassifier(nn.Module):
    """基于预训练 ResNet 的音频分类器
    将梅尔频谱图视为图像输入"""
    def __init__(self, num_classes: int = 527, pretrained: bool = True):
        super().__init__()
        self.backbone = models.resnet18(weights="DEFAULT" if pretrained else None)
        # 将第一层改为单通道输入
        self.backbone.conv1 = nn.Conv2d(
            1, 64, kernel_size=7, stride=2, padding=3, bias=False
        )
        # 替换分类头
        self.backbone.fc = nn.Linear(self.backbone.fc.in_features, num_classes)

    def forward(self, mel_spec: torch.Tensor) -> torch.Tensor:
        # mel_spec: (batch, 1, n_mels, time)
        return self.backbone(mel_spec)

from transformers import ASTForAudioClassification, AutoFeatureExtractor

def classify_audio(audio_path: str, model_name: str = "MIT/ast-finetuned-audioset-10-10-0.4593"):
    """使用 Audio Spectrogram Transformer 进行分类
    AST 将音频分类提升到 Transformer 架构"""
    model = ASTForAudioClassification.from_pretrained(model_name)
    extractor = AutoFeatureExtractor.from_pretrained(model_name)

    audio, sr = librosa.load(audio_path, sr=16000)
    inputs = extractor(audio, sampling_rate=sr, return_tensors="pt")

    with torch.no_grad():
        outputs = model(**inputs)
        probs = torch.softmax(outputs.logits, dim=-1)

    # 返回 Top-3 预测
    top3 = torch.topk(probs, k=3, dim=-1)
    labels = [model.config.id2label[i.item()] for i in top3.indices[0]]
    return list(zip(labels, top3.values[0].tolist()))

import torch
from audiocraft.models import MusicGen
from audiocraft.data.audio import audio_write

def generate_music_from_text(prompt: str, duration: int = 30):
    """使用 MusicGen 从文本描述生成音乐
    MusicGen 结合了 EnCodec 音频压缩和条件 Transformer"""
    model = MusicGen.get_pretrained("medium")
    model.set_generation_params(
        duration=duration,
        top_k=250,
        top_p=0.0,
        temperature=1.0,
        cfg_coef=3.0  # classifier-free guidance 强度
    )

    with torch.no_grad():
        output = model.generate([prompt], progress=True)

    # 保存生成的音频
    audio_write("generated", output[0].cpu(), model.sample_rate,
                strategy="loudness", loudness_compressor=True)
    return output

from encodec import EncodecModel
import torch

class AudioTokenExtractor:
    """使用 EnCodec 将音频编码为离散 token 序列
    这是 MusicLM 等生成模型的关键预处理步骤"""
    def __init__(self, bandwidth: float = 6.0):
        self.model = EncodecModel.encodec_model_24khz()
        self.model.set_target_bandwidth(bandwidth)

    def encode(self, waveform: torch.Tensor) -> torch.Tensor:
        """将波形编码为离散 token（多 codebook 量化）"""
        with torch.no_grad():
            encoded = self.model.encode(waveform.unsqueeze(0).unsqueeze(0))
        # encoded[0][0] 包含 codebook indices
        tokens = encoded[0][0]  # shape: (n_codebooks, seq_len)
        return tokens

    def decode(self, tokens: torch.Tensor) -> torch.Tensor:
        """从离散 token 重建波形"""
        with torch.no_grad():
            decoded = self.model.decode([(tokens, None)])
        return decoded[0].squeeze()

import torch
import torch.nn.functional as F
from laion_clap import CLAP_Module

class CrossModalRetriever:
    """基于 CLAP 的音频-文本双向检索系统"""
    def __init__(self, model_path: str = "laion/clap-htsat-unfused"):
        self.model = CLAP_Module(enable_fusion=False)
        self.model.load_ckpt(model_path)
        self.audio_embeddings = {}
        self.text_embeddings = {}

    def encode_audio(self, audio_path: str, key: str):
        """编码音频并存储嵌入向量"""
        audio_data, _ = librosa.load(audio_path, sr=48000)
        emb = self.model.get_audio_embedding_from_data(
            x=audio_data, use_tensor=True
        )
        self.audio_embeddings[key] = emb
        return emb

    def search_by_text(self, query: str, top_k: int = 5) -> list:
        """用文本查询检索最相似的音频"""
        text_emb = self.model.get_text_embedding(query, use_tensor=True)
        audio_embs = torch.stack(list(self.audio_embeddings.values()))
        # 计算余弦相似度
        sims = F.cosine_similarity(text_emb, audio_embs, dim=-1)
        top_indices = torch.topk(sims, k=top_k).indices
        keys = list(self.audio_embeddings.keys())
        return [(keys[i], sims[i].item()) for i in top_indices]

import torch

def contrastive_loss(audio_emb: torch.Tensor, text_emb: torch.Tensor,
                     temperature: torch.Tensor = torch.tensor(0.07)) -> torch.Tensor:
    """计算 CLAP 风格的对比损失
    音频到文本和文本到音频两个方向的 InfoNCE 损失"""
    # 归一化嵌入
    audio_emb = F.normalize(audio_emb, dim=-1)
    text_emb = F.normalize(text_emb, dim=-1)

    # 相似度矩阵: (batch, batch)
    logits = audio_emb @ text_emb.T / temperature.exp()

    # 对角线为正样本
    labels = torch.arange(logits.size(0), device=logits.device)
    loss_a2t = F.cross_entropy(logits, labels)  # 音频->文本
    loss_t2a = F.cross_entropy(logits.T, labels)  # 文本->音频

    return (loss_a2t + loss_t2a) / 2  # 双向对称损失

import torch
import torch.nn as nn
from transformers import AutoModel

class AudioProjector(nn.Module):
    """将音频表征投影到 LLM 嵌入空间
    这是多模态大模型中连接音频和文本的关键组件"""
    def __init__(self, audio_dim: int = 1024, llm_dim: int = 4096):
        super().__init__()
        self.projector = nn.Sequential(
            nn.Linear(audio_dim, llm_dim),
            nn.GELU(),
            nn.Linear(llm_dim, llm_dim),
        )

    def forward(self, audio_features: torch.Tensor) -> torch.Tensor:
        """audio_features: (batch, audio_seq_len, audio_dim)
        返回: (batch, audio_seq_len, llm_dim)"""
        return self.projector(audio_features)


def build_multimodal_input(text_tokens: torch.Tensor,
                           audio_tokens: torch.Tensor,
                           projector: AudioProjector) -> torch.Tensor:
    """将音频和文本 token 拼接为统一输入序列"""
    # 投影音频 token 到 LLM 空间
    projected = projector(audio_tokens)
    # 拼接: [音频 token 序列] + [文本 token 序列]
    return torch.cat([projected, text_tokens], dim=1)

from transformers import Qwen2AudioForConditionalGeneration, AutoProcessor
import torch

def audio_chat(audio_path: str, instruction: str) -> str:
    """使用 Qwen2-Audio 进行音频对话
    模型可以理解音频内容、说话人意图、背景信息等"""
    model = Qwen2AudioForConditionalGeneration.from_pretrained(
        "Qwen/Qwen2-Audio-7B-Instruct", torch_dtype=torch.float16
    )
    processor = AutoProcessor.from_pretrained("Qwen/Qwen2-Audio-7B-Instruct")

    conversation = [
        {"role": "user", "content": [
            {"type": "audio", "audio_url": audio_path},
            {"type": "text", "text": instruction}
        ]}
    ]
    text_prompt = processor.apply_chat_template(conversation)
    audio_input, _ = librosa.load(audio_path, sr=16000)

    inputs = processor(
        text=text_prompt, audios=[audio_input], return_tensors="pt",
        padding=True
    ).to(model.device)

    generated = model.generate(**inputs, max_new_tokens=256)
    return processor.decode(generated[0], skip_special_tokens=True)

import torch
import numpy as np
from transformers import WhisperProcessor, WhisperForConditionalGeneration
import torchaudio

class ProductionASR:
    """生产级语音识别 pipeline
    包含 VAD 切分、批量推理、后处理"""
    def __init__(self, model_name: str = "openai/whisper-large-v3"):
        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        self.processor = WhisperProcessor.from_pretrained(model_name)
        self.model = WhisperForConditionalGeneration.from_pretrained(
            model_name, torch_dtype=torch.float16
        ).to(self.device)

        # VAD 用于切分长音频
        self.vad_model, _ = torch.hub.load(
            "snakers4/silero-vad", "silero_vad", force_reload=False
        )

    def transcribe_long(self, audio_path: str, chunk_duration: int = 30) -> str:
        """处理长音频：VAD 切分 → 逐段识别 → 拼接结果"""
        waveform, sr = torchaudio.load(audio_path)
        if sr != 16000:
            waveform = torchaudio.functional.resample(waveform, sr, 16000)

        # 简单分段（实际应使用 VAD）
        chunk_size = chunk_duration * 16000
        num_chunks = int(np.ceil(waveform.size(1) / chunk_size))
        results = []

        for i in range(num_chunks):
            start = i * chunk_size
            end = min((i + 1) * chunk_size, waveform.size(1))
            chunk = waveform[:, start:end]

            inputs = self.processor(
                chunk.squeeze().numpy(),
                sampling_rate=16000,
                return_tensors="pt"
            ).to(self.device)

            with torch.no_grad():
                ids = self.model.generate(
                    inputs.input_features.half(),
                    language="chinese",
                    task="transcribe"
                )
            results.append(self.processor.decode(ids[0], skip_special_tokens=True))

        return " ".join(results)

import jiwer
from typing import List

def evaluate_asr(predictions: List[str], references: List[str]) -> dict:
    """评估语音识别结果
    计算 WER（词错误率）、CER（字符错误率）等指标"""
    # Word Error Rate
    wer = jiwer.wer(references, predictions)
    # Character Error Rate（中文更重要）
    cer = jiwer.cer(references, predictions)

    # 详细分解
    transform = jiwer.Compose([
        jiwer.RemoveMultipleSpaces(),
        jiwer.Strip(),
        jiwer.SentencesToListOfWords(),
    ])
    measures = jiwer.compute_measures(references, predictions)

    return {
        "wer": round(wer * 100, 2),      # 越低越好
        "cer": round(cer * 100, 2),       # 越低越好
        "substitutions": measures["substitutions"],
        "deletions": measures["deletions"],
        "insertions": measures["insertions"],
        "hits": measures["hits"],
    }


def optimize_throughput(audio_paths: list, asr: ProductionASR,
                        batch_size: int = 4) -> list:
    """批量推理优化：利用 GPU 并行处理多段音频"""
    all_results = []
    for i in range(0, len(audio_paths), batch_size):
        batch = audio_paths[i:i + batch_size]
        # 实际实现中应在 batch 维度并行
        for path in batch:
            all_results.append(asr.transcribe_long(path))
    return all_results