AI语音识别与实时转写完全教程

1. ASR语音识别技术概述

自动语音识别（Automatic Speech Recognition, ASR）是将语音信号转换为文本的技术。现代ASR系统的核心架构经历了从传统GMM-HMM到端到端深度学习的演进。

技术演进路线

传统方法（2012年前）：GMM-HMM（高斯混合模型-隐马尔可夫模型）是经典方案。声学特征（MFPL/MFCC）经过GMM建模后，通过HMM进行时序对齐，配合N-gram语言模型解码。精度有限，需要大量人工调参。

深度学习方法（2012-2018）：DNN替代GMM作为声学模型，CTC（Connectionist Temporal Classification）损失函数解决了输入输出长度不对齐的问题。Listen, Attend and Spell（LAS）引入了注意力机制的端到端架构。

预训练大模型时代（2022至今）：Whisper的出现标志着ASR进入大模型时代。通过68万小时多语言弱监督数据训练，单一模型即可处理多语言、多任务（识别、翻译、时间戳、语种检测）。Conformer架构（CNN+Transformer结合）成为工业界主流。

核心技术指标

• WER（Word Error Rate）：词错误率，最核心的评估指标，计算方式为 (插入+删除+替换) / 总词数
• CER（Character Error Rate）：字符错误率，中文场景更常用
• RTF（Real Time Factor）：实时率，处理时间/音频时长，RTF < 1表示能实时处理
• 延迟（Latency）：从说话结束到文本输出的时间，流式场景要求 < 500ms

2. 主流模型对比（Whisper / Paraformer / WeNet）

Whisper

OpenAI开源的通用语音识别模型，基于Encoder-Decoder Transformer架构。

优势：多语言支持好（99种语言）、零样本能力强、生态丰富 劣势：非流式设计（需完整音频段）、中文精度不如专用模型、大模型推理慢

Paraformer

阿里达摩院开源的非自回归端到端模型，中文识别精度极高。

优势：中文CER低、推理速度快（比Whisper large快10倍+）、支持流式 劣势：多语言支持有限、社区生态不如Whisper

WeNet

出门问问开源的生产级ASR工具包，主打易部署和流式支持。

优势：原生流式支持、C++ Runtime部署简单、支持多种解码方式 劣势：需自行训练或使用预训练模型、开箱即用体验不如Whisper

选型建议

3. Whisper模型部署与使用

基础安装与使用

pip install openai-whisper
# 或使用更快的实现
pip install faster-whisper

基本用法

import whisper

# 加载模型（首次运行会自动下载）
model = whisper.load_model("medium")

# 转录音频文件
result = model.transcribe("audio.mp3", language="zh")
print(result["text"])

# 带时间戳的转录
for segment in result["segments"]:
    start = segment["start"]
    end = segment["end"]
    text = segment["text"]
    print(f"[{start:.1f}s - {end:.1f}s] {text}")

faster-whisper：高性能推理

faster-whisper基于CTranslate2，推理速度比原版快4倍，显存占用更少：

from faster_whisper import WhisperModel

# 使用float16量化，GPU推理
model = WhisperModel("large-v3", device="cuda", compute_type="float16")

segments, info = model.transcribe(
    "audio.mp3",
    language="zh",
    beam_size=5,
    vad_filter=True,           # 启用VAD过滤静音段
    vad_parameters=dict(
        min_silence_duration_ms=500,
        speech_pad_ms=200,
    ),
)

print(f"检测到语言: {info.language}, 置信度: {info.language_probability:.2f}")

for segment in segments:
    print(f"[{segment.start:.2f}s -> {segment.end:.2f}s] {segment.text}")

Whisper的Prompt机制

Whisper支持传入prompt来引导解码，提高专业术语识别率：

# 上下文提示：提供相关的专业术语
initial_prompt = """
以下是一段关于人工智能的技术讨论，可能涉及以下术语：
大语言模型、Transformer、注意力机制、梯度下降、反向传播、
CUDA、PyTorch、TensorFlow、LoRA、量化、蒸馏
"""

result = model.transcribe(
    "tech_meeting.mp3",
    language="zh",
    initial_prompt=initial_prompt,
    condition_on_previous_text=True,  # 利用前文上下文
)

批量处理与GPU优化

import torch
from faster_whisper import WhisperModel
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor

model = WhisperModel("large-v3", device="cuda", compute_type="float16")

def transcribe_file(audio_path: str) -> dict:
    segments, info = model.transcribe(audio_path, language="zh", beam_size=5)
    text = "".join(seg.text for seg in segments)
    return {"file": audio_path, "text": text, "language": info.language}

# 批量处理
audio_files = list(Path("./audio_files").glob("*.mp3"))
results = [transcribe_file(str(f)) for f in audio_files]

# 利用batch推理进一步提速
from faster_whisper import BatchedInferencePipeline

batched_model = BatchedInferencePipeline(model=model)
segments, info = batched_model.transcribe(
    "long_audio.mp3",
    language="zh",
    batch_size=16,  # 批量大小
)

4. 实时语音流识别方案

实时流式识别是会议转写、实时字幕等场景的核心需求。实现方案取决于模型架构。

方案一：VAD + 分段识别（Whisper适用）

Whisper本身不支持流式，但可以通过VAD（Voice Activity Detection）将音频切成片段后逐段识别：

import numpy as np
import sounddevice as sd
from faster_whisper import WhisperModel
from silero_vad import load_silero_vad, get_speech_timestamps

model = WhisperModel("small", device="cuda", compute_type="float16")
vad_model = load_silero_vad()

class StreamingTranscriber:
    def __init__(self, sample_rate=16000, chunk_duration=0.5):
        self.sample_rate = sample_rate
        self.chunk_size = int(sample_rate * chunk_duration)
        self.audio_buffer = np.array([], dtype=np.float32)
        self.silence_counter = 0
        self.min_speech_duration = 1.0    # 最少1秒才识别
        self.max_speech_duration = 30.0   # 最多30秒强制切段
        self.silence_threshold = 3        # 连续3个chunk无语音则结束

    def process_chunk(self, audio_chunk: np.ndarray) -> str | None:
        """处理一个音频chunk，返回识别文本或None"""
        self.audio_buffer = np.concatenate([self.audio_buffer, audio_chunk])

        # VAD检测当前chunk是否有语音
        speech_prob = vad_model(torch.from_numpy(audio_chunk), self.sample_rate).item()
        has_speech = speech_prob > 0.5

        if has_speech:
            self.silence_counter = 0
        else:
            self.silence_counter += 1

        buffer_duration = len(self.audio_buffer) / self.sample_rate

        # 触发识别条件：检测到静音且有足够音频，或超过最大时长
        should_transcribe = (
            (self.silence_counter >= self.silence_threshold and buffer_duration >= self.min_speech_duration)
            or buffer_duration >= self.max_speech_duration
        )

        if should_transcribe and buffer_duration >= self.min_speech_duration:
            segments, _ = model.transcribe(
                self.audio_buffer,
                language="zh",
                beam_size=3,
                vad_filter=True,
            )
            text = "".join(seg.text for seg in segments)
            self.audio_buffer = np.array([], dtype=np.float32)
            self.silence_counter = 0
            return text

        return None

# 实时录音+识别
def run_realtime():
    transcriber = StreamingTranscriber()
    print("开始说话... (Ctrl+C停止)")

    def audio_callback(indata, frames, time_info, status):
        audio_chunk = indata[:, 0].astype(np.float32)
        result = transcriber.process_chunk(audio_chunk)
        if result:
            print(f"识别结果: {result}")

    with sd.InputStream(
        samplerate=16000,
        channels=1,
        dtype='float32',
        blocksize=8000,  # 0.5秒
        callback=audio_callback,
    ):
        import time
        while True:
            time.sleep(0.1)

if __name__ == "__main__":
    run_realtime()

方案二：Paraformer流式模式

Paraformer原生支持流式识别，延迟更低：

from funasr import AutoModel

# 加载流式模型
model = AutoModel(
    model="paraformer-zh-streaming",
    vad_model="fsmn-vad",
    punc_model="ct-punc",
)

# 流式推理
import numpy as np
import sounddevice as sd

cache = {}
chunk_size = [0, 10, 5]  # [look-back, chunk, look-ahead]，单位60ms

def stream_callback(indata, frames, time_info, status):
    audio_chunk = indata[:, 0]

    result = model.generate(
        input=audio_chunk,
        cache=cache,
        chunk_size=chunk_size,
        is_final=False,
        language="zh",
    )

    if result and result[0]["text"]:
        print(f"实时: {result[0]['text']}", end="\r")

with sd.InputStream(samplerate=16000, channels=1, dtype='float32',
                     blocksize=960, callback=stream_callback):  # 60ms
    import time
    while True:
        time.sleep(0.1)

方案三：WebSocket流式服务

搭建WebSocket服务，客户端实时发送音频流，服务端返回识别结果：

# 服务端
import asyncio
import websockets
import numpy as np
from faster_whisper import WhisperModel

model = WhisperModel("small", device="cuda", compute_type="float16")

async def handle_client(websocket):
    audio_buffer = bytearray()

    async for message in websocket:
        if isinstance(message, bytes):
            # 接收音频数据
            audio_buffer.extend(message)

            # 每收到约1秒音频就进行识别
            if len(audio_buffer) >= 16000 * 2:  # 16kHz * 2bytes * 1s
                audio_np = np.frombuffer(bytes(audio_buffer), dtype=np.int16).astype(np.float32) / 32768.0
                audio_buffer.clear()

                segments, _ = model.transcribe(audio_np, language="zh", beam_size=3)
                text = "".join(seg.text for seg in segments)

                if text.strip():
                    await websocket.send(json.dumps({
                        "type": "transcript",
                        "text": text,
                        "is_final": False,
                    }))

async def main():
    async with websockets.serve(handle_client, "0.0.0.0", 8765):
        print("WebSocket ASR服务启动在 ws://0.0.0.0:8765")
        await asyncio.Future()

asyncio.run(main())

// 浏览器客户端
class StreamingASR {
  constructor(wsUrl) {
    this.ws = new WebSocket(wsUrl);
    this.ws.binaryType = 'arraybuffer';
    this.mediaRecorder = null;
    this.onTranscript = () => {};
  }

  async start() {
    const stream = await navigator.mediaDevices.getUserMedia({
      audio: { sampleRate: 16000, channelCount: 1 }
    });

    const audioContext = new AudioContext({ sampleRate: 16000 });
    const source = audioContext.createMediaStreamSource(stream);
    const processor = audioContext.createScriptProcessor(4096, 1, 1);

    processor.onaudioprocess = (e) => {
      const audioData = e.inputBuffer.getChannelData(0);
      // Float32 -> Int16
      const int16Data = new Int16Array(audioData.length);
      for (let i = 0; i < audioData.length; i++) {
        int16Data[i] = Math.max(-32768, Math.min(32767, audioData[i] * 32768));
      }
      this.ws.send(int16Data.buffer);
    };

    source.connect(processor);
    processor.connect(audioContext.destination);

    this.ws.onmessage = (event) => {
      const data = JSON.parse(event.data);
      if (data.type === 'transcript') {
        this.onTranscript(data.text, data.is_final);
      }
    };
  }

  stop() {
    this.mediaRecorder?.stop();
    this.ws.close();
  }
}

// 使用
const asr = new StreamingASR('ws://localhost:8765');
asr.onTranscript = (text, isFinal) => {
  document.getElementById('output').textContent += text;
};
asr.start();

5. 多语言与方言识别

语种检测

Whisper内置语种检测能力，可以在转录前自动识别语言：

import whisper

model = whisper.load_model("medium")
audio = whisper.load_audio("unknown_language.mp3")
audio = whisper.pad_or_trim(audio)

# 生成梅尔频谱图
mel = whisper.log_mel_spectrogram(audio).to(model.device)

# 检测语言
_, probs = model.detect_language(mel)
detected_lang = max(probs, key=probs.get)
print(f"检测到语言: {detected_lang}，置信度: {probs[detected_lang]:.2f}")

# 打印所有语言概率（Top 10）
sorted_langs = sorted(probs.items(), key=lambda x: x[1], reverse=True)[:10]
for lang, prob in sorted_langs:
    print(f"  {lang}: {prob:.3f}")

方言识别策略

中文方言识别是一个挑战。通用策略：

# 策略1：使用方言专用模型
# FunASR提供了粤语等方言模型
from funasr import AutoModel

# 粤语模型
cantonese_model = AutoModel(model="paraformer-zh")

# 策略2：通过prompt引导Whisper识别方言
result = model.transcribe(
    "cantonese_audio.mp3",
    language="zh",           # 仍标记为中文
    initial_prompt="以下是粤语对话，",  # prompt引导
)

# 策略3：多模型投票
def ensemble_transcribe(audio_path, models):
    results = []
    for m in models:
        result = m.transcribe(audio_path, language="zh")
        results.append(result["text"])

    # 取最长的结果（通常方言识别中，正确识别的结果包含更多细节）
    return max(results, key=len)

多语言混合场景

处理中英混合（Code-Switching）场景：

# Whisper对中英混合天然支持较好
result = model.transcribe(
    "code_switching.mp3",
    language=None,       # 不指定语言，让模型自动判断
    task="transcribe",   # "transcribe"保持原语言，"translate"翻译为英文
    condition_on_previous_text=True,
)

# 逐段检测语言
for segment in result["segments"]:
    # whisper每个segment没有直接的语言字段
    # 可以通过检测文本中的字符来判断
    text = segment["text"]
    has_chinese = any('\u4e00' <= c <= '\u9fff' for c in text)
    has_english = any(c.isascii() and c.isalpha() for c in text)
    if has_chinese and has_english:
        lang_tag = "中英混合"
    elif has_chinese:
        lang_tag = "中文"
    else:
        lang_tag = "英文"
    print(f"[{segment['start']:.1f}s] ({lang_tag}) {text}")

6. 说话人分离（Speaker Diarization）

说话人分离是将音频中的不同说话人区分开来的技术，在会议转写中至关重要。

基于pyannote的说话人分离

# pip install pyannote.audio
from pyannote.audio import Pipeline

# 加载预训练的说话人分离Pipeline（需要HuggingFace Token）
pipeline = Pipeline.from_pretrained(
    "pyannote/speaker-diarization-3.1",
    use_auth_token="YOUR_HF_TOKEN",
)

# 执行分离
diarization = pipeline("meeting.wav")

# 打印结果
for turn, _, speaker in diarization.itertracks(yield_label=True):
    print(f"{turn.start:.1f}s - {turn.end:.1f}s: {speaker}")

ASR + 说话人分离联合

将识别结果与说话人信息合并：

from faster_whisper import WhisperModel
from pyannote.audio import Pipeline
import numpy as np

def transcribe_with_speakers(audio_path: str) -> list[dict]:
    # 1. ASR识别
    asr_model = WhisperModel("medium", device="cuda", compute_type="float16")
    segments_iter, _ = asr_model.transcribe(audio_path, language="zh")
    asr_segments = [(seg.start, seg.end, seg.text) for seg in segments_iter]

    # 2. 说话人分离
    diarization_pipeline = Pipeline.from_pretrained("pyannote/speaker-diarization-3.1")
    diarization = diarization_pipeline(audio_path)

    # 3. 对齐：将ASR片段分配给说话人
    results = []
    for seg_start, seg_end, text in asr_segments:
        # 找到与当前ASR片段重叠最大的说话人
        best_speaker = "UNKNOWN"
        best_overlap = 0

        for turn, _, speaker in diarization.itertracks(yield_label=True):
            overlap_start = max(seg_start, turn.start)
            overlap_end = min(seg_end, turn.end)
            overlap = max(0, overlap_end - overlap_start)

            if overlap > best_overlap:
                best_overlap = overlap
                best_speaker = speaker

        results.append({
            "start": seg_start,
            "end": seg_end,
            "speaker": best_speaker,
            "text": text,
        })

    return results

# 使用
results = transcribe_with_speakers("meeting.wav")
for r in results:
    print(f"[{r['start']:.1f}s-{r['end']:.1f}s] {r['speaker']}: {r['text']}")

合并连续同说话人片段

def merge_consecutive_speaker_segments(segments: list[dict], gap_threshold: float = 1.0) -> list[dict]:
    """合并同一说话人的连续片段"""
    if not segments:
        return []

    merged = [segments[0].copy()]
    for seg in segments[1:]:
        last = merged[-1]
        if seg["speaker"] == last["speaker"] and seg["start"] - last["end"] < gap_threshold:
            last["end"] = seg["end"]
            last["text"] += seg["text"]
        else:
            merged.append(seg.copy())

    return merged

7. 语音增强与降噪预处理

在真实环境中，音频往往包含噪声、回声、混响等干扰。预处理能显著提升识别精度。

使用noisereduce降噪

# pip install noisereduce
import noisereduce as nr
import soundfile as sf

def denoise_audio(input_path: str, output_path: str):
    # 读取音频
    audio, sr = sf.read(input_path)

    # 降噪
    reduced_noise = nr.reduce_noise(
        y=audio,
        sr=sr,
        stationary=False,     # 非稳态噪声（更通用）
        prop_decrease=0.75,   # 噪声衰减比例
        freq_mask_smooth_hz=500,
        time_mask_smooth_ms=50,
    )

    sf.write(output_path, reduced_noise, sr)
    print(f"降噪完成: {output_path}")

音频预处理Pipeline

import numpy as np
import soundfile as sf
from scipy import signal

class AudioPreprocessor:
    def __init__(self, target_sr=16000):
        self.target_sr = target_sr

    def process(self, audio_path: str) -> np.ndarray:
        """完整的音频预处理流水线"""
        # 1. 读取并重采样
        audio, sr = sf.read(audio_path)
        if len(audio.shape) > 1:
            audio = np.mean(audio, axis=1)  # 立体声转单声道
        if sr != self.target_sr:
            audio = signal.resample(audio, int(len(audio) * self.target_sr / sr))

        # 2. 音量归一化
        audio = self._normalize(audio)

        # 3. 高通滤波去除低频噪声
        audio = self._highpass_filter(audio, cutoff=80)

        # 4. 降噪
        audio = self._denoise(audio)

        # 5. 去除静音段（可选）
        audio = self._remove_silence(audio)

        return audio

    def _normalize(self, audio: np.ndarray, target_db: float = -20.0) -> np.ndarray:
        """响度归一化"""
        rms = np.sqrt(np.mean(audio ** 2))
        current_db = 20 * np.log10(rms + 1e-10)
        gain = 10 ** ((target_db - current_db) / 20)
        return np.clip(audio * gain, -1.0, 1.0)

    def _highpass_filter(self, audio: np.ndarray, cutoff: int = 80) -> np.ndarray:
        """高通滤波"""
        nyq = self.target_sr / 2
        b, a = signal.butter(4, cutoff / nyq, btype='high')
        return signal.filtfilt(b, a, audio)

    def _denoise(self, audio: np.ndarray) -> np.ndarray:
        """降噪"""
        import noisereduce as nr
        return nr.reduce_noise(y=audio, sr=self.target_sr, stationary=False, prop_decrease=0.7)

    def _remove_silence(self, audio: np.ndarray, threshold_db: float = -40) -> np.ndarray:
        """去除首尾静音"""
        frame_len = int(self.target_sr * 0.025)
        energy = np.array([
            np.mean(audio[i:i+frame_len] ** 2)
            for i in range(0, len(audio) - frame_len, frame_len)
        ])
        energy_db = 10 * np.log10(energy + 1e-10)

        # 找到有效语音区域
        mask = energy_db > threshold_db
        if not mask.any():
            return audio

        first_speech = np.argmax(mask) * frame_len
        last_speech = (len(mask) - 1 - np.argmax(mask[::-1])) * frame_len
        return audio[first_speech:last_speech + frame_len]

8. 标点恢复与文本后处理

ASR输出通常是无标点的连续文本，标点恢复对可读性至关重要。

基于模型的标点恢复

# FunASR内置标点模型
from funasr import AutoModel

punc_model = AutoModel(model="ct-punc")

def restore_punctuation(text: str) -> str:
    result = punc_model.generate(input=text)
    return result[0]["text"]

# 示例
raw_text = "今天天气很好我们去公园散步吧你觉得怎么样"
punctuated = restore_punctuation(raw_text)
print(punctuated)
# 输出: "今天天气很好，我们去公园散步吧，你觉得怎么样？"

文本后处理Pipeline

import re

class TextPostProcessor:
    def __init__(self):
        # 常见错误修正词典
        self.corrections = {
            "人工只能": "人工智能",
            "机器学刁": "机器学习",
            "深渡学习": "深度学习",
        }

    def process(self, text: str) -> str:
        text = self._fix_punctuation(text)
        text = self._fix_spacing(text)
        text = self._fix_numbers(text)
        text = self._apply_corrections(text)
        text = self._remove_repetitions(text)
        return text.strip()

    def _fix_punctuation(self, text: str) -> str:
        """修正标点问题"""
        # 移除重复标点
        text = re.sub(r'[。，、！？]{2,}', lambda m: m.group()[0], text)
        # 确保句末有标点
        if text and text[-1] not in '。！？…':
            text += '。'
        return text

    def _fix_spacing(self, text: str) -> str:
        """修正空格"""
        # 中文之间不应有空格
        text = re.sub(r'([\u4e00-\u9fff])\s+([\u4e00-\u9fff])', r'\1\2', text)
        # 中英文之间加空格
        text = re.sub(r'([\u4e00-\u9fff])([a-zA-Z])', r'\1 \2', text)
        text = re.sub(r'([a-zA-Z])([\u4e00-\u9fff])', r'\1 \2', text)
        return text

    def _fix_numbers(self, text: str) -> str:
        """数字格式化"""
        # "一 百 二 十 三" -> "一百二十三"
        text = re.sub(r'(?<=[一二三四五六七八九十百千万亿])\s+(?=[一二三四五六七八九十百千万亿])', '', text)
        return text

    def _apply_corrections(self, text: str) -> str:
        """应用错误修正"""
        for wrong, correct in self.corrections.items():
            text = text.replace(wrong, correct)
        return text

    def _remove_repetitions(self, text: str) -> str:
        """去除ASR常见的重复识别"""
        # 去除连续重复的短语
        text = re.sub(r'(.{2,10})\1{2,}', r'\1', text)
        return text

热词增强

针对特定领域的专业术语，可以通过热词机制提升识别率：

# FunASR支持热词
from funasr import AutoModel

model = AutoModel(model="paraformer-zh")

# 通过hotword参数传入热词
result = model.generate(
    input="audio.wav",
    hotword="大语言模型 Transformer 注意力机制 LoRA 微调 量化",
)

# Whisper的等效方式：通过initial_prompt
result = model.transcribe(
    "audio.wav",
    language="zh",
    initial_prompt="以下内容涉及：大语言模型、Transformer、注意力机制、LoRA微调、量化部署。",
)

9. 会议实时转写系统搭建

构建一个完整的会议实时转写系统，需要整合ASR、说话人分离、标点恢复等多个模块。

系统架构

麦克风 -> 音频采集 -> VAD分段 -> ASR识别 -> 标点恢复 -> 说话人分离 -> 前端展示
                                    |
                              WebSocket实时推送

核心实现

# meeting_transcriber.py
import asyncio
import json
import time
import numpy as np
from dataclasses import dataclass, asdict

@dataclass
class TranscriptSegment:
    speaker: str
    text: str
    start_time: float
    end_time: float
    is_final: bool

class MeetingTranscriber:
    def __init__(self):
        from faster_whisper import WhisperModel
        from funasr import AutoModel as FunASRModel
        from pyannote.audio import Pipeline as DiarizationPipeline

        self.asr_model = WhisperModel("medium", device="cuda", compute_type="float16")
        self.punc_model = FunASRModel(model="ct-punc")
        self.diarization = DiarizationPipeline.from_pretrained("pyannote/speaker-diarization-3.1")

        self.audio_buffer = np.array([], dtype=np.float32)
        self.sample_rate = 16000
        self.segment_duration = 5.0   # 每5秒处理一次
        self.listeners = []           # WebSocket监听器

    async def add_listener(self, ws):
        self.listeners.append(ws)

    async def remove_listener(self, ws):
        self.listeners.remove(ws)

    async def broadcast(self, segment: TranscriptSegment):
        data = json.dumps(asdict(segment), ensure_ascii=False)
        for ws in self.listeners[:]:
            try:
                await ws.send(data)
            except:
                self.listeners.remove(ws)

    def feed_audio(self, audio_chunk: np.ndarray):
        """喂入音频数据"""
        self.audio_buffer = np.concatenate([self.audio_buffer, audio_chunk])

        buffer_duration = len(self.audio_buffer) / self.sample_rate
        if buffer_duration >= self.segment_duration:
            asyncio.create_task(self._process_buffer())

    async def _process_buffer(self):
        """处理音频缓冲区"""
        if len(self.audio_buffer) < self.sample_rate:  # 至少1秒
            return

        audio_to_process = self.audio_buffer.copy()
        self.audio_buffer = np.array([], dtype=np.float32)

        # ASR识别
        segments, _ = self.asr_model.transcribe(
            audio_to_process,
            language="zh",
            beam_size=3,
            vad_filter=True,
        )

        raw_text = "".join(seg.text for seg in segments)
        if not raw_text.strip():
            return

        # 标点恢复
        punc_result = self.punc_model.generate(input=raw_text)
        text = punc_result[0]["text"]

        # 发送结果
        segment = TranscriptSegment(
            speaker="检测中...",
            text=text,
            start_time=time.time(),
            end_time=time.time(),
            is_final=True,
        )
        await self.broadcast(segment)

WebSocket服务

# server.py
import asyncio
import websockets
import numpy as np
from meeting_transcriber import MeetingTranscriber

transcriber = MeetingTranscriber()

async def handle_client(websocket):
    await transcriber.add_listener(websocket)
    print(f"客户端已连接，当前监听数: {len(transcriber.listeners)}")

    try:
        async for message in websocket:
            if isinstance(message, bytes):
                # 接收音频数据（Int16格式）
                audio = np.frombuffer(message, dtype=np.int16).astype(np.float32) / 32768.0
                transcriber.feed_audio(audio)
    except websockets.exceptions.ConnectionClosed:
        pass
    finally:
        await transcriber.remove_listener(websocket)
        print(f"客户端已断开，当前监听数: {len(transcriber.listeners)}")

async def main():
    async with websockets.serve(handle_client, "0.0.0.0", 8765):
        print("会议转写服务启动在 ws://0.0.0.0:8765")
        await asyncio.Future()

asyncio.run(main())

前端展示页面

<!-- meeting_ui.html -->
<!DOCTYPE html>
<html>
<head>
    <title>实时会议转写</title>
    <style>
        body { font-family: system-ui; max-width: 800px; margin: 0 auto; padding: 20px; }
        .transcript-area { height: 60vh; overflow-y: auto; border: 1px solid #ddd; border-radius: 8px; padding: 16px; }
        .segment { margin-bottom: 12px; padding: 8px 12px; border-radius: 8px; background: #f5f5f5; }
        .segment .speaker { font-weight: bold; color: #2563eb; font-size: 0.85em; }
        .segment .text { margin-top: 4px; line-height: 1.6; }
        .segment .time { font-size: 0.75em; color: #999; }
        .controls { margin-top: 16px; display: flex; gap: 8px; }
        .btn { padding: 8px 24px; border: none; border-radius: 6px; cursor: pointer; font-size: 14px; }
        .btn-primary { background: #2563eb; color: white; }
        .btn-danger { background: #dc2626; color: white; }
        .status { display: inline-block; width: 8px; height: 8px; border-radius: 50%; margin-right: 8px; }
        .status.active { background: #22c55e; }
        .status.inactive { background: #9ca3af; }
    </style>
</head>
<body>
    <h1>🎙️ 实时会议转写</h1>
    <div class="controls">
        <button class="btn btn-primary" onclick="startRecording()">开始转写</button>
        <button class="btn btn-danger" onclick="stopRecording()">停止</button>
        <span><span class="status" id="status"></span><span id="statusText">未连接</span></span>
    </div>
    <div class="transcript-area" id="transcript"></div>

    <script>
    let ws, mediaRecorder, audioContext, processor;
    const transcript = document.getElementById('transcript');

    async function startRecording() {
        const stream = await navigator.mediaDevices.getUserMedia({ audio: { sampleRate: 16000, channelCount: 1 } });
        audioContext = new AudioContext({ sampleRate: 16000 });
        const source = audioContext.createMediaStreamSource(stream);
        processor = audioContext.createScriptProcessor(4096, 1, 1);

        ws = new WebSocket('ws://localhost:8765');
        ws.onopen = () => {
            document.getElementById('status').className = 'status active';
            document.getElementById('statusText').textContent = '转写中...';
        };
        ws.onmessage = (e) => {
            const data = JSON.parse(e.data);
            const div = document.createElement('div');
            div.className = 'segment';
            div.innerHTML = `
                <div class="speaker">${data.speaker} · <span class="time">${new Date(data.start_time * 1000).toLocaleTimeString()}</span></div>
                <div class="text">${data.text}</div>
            `;
            transcript.appendChild(div);
            transcript.scrollTop = transcript.scrollHeight;
        };

        processor.onaudioprocess = (e) => {
            const audio = e.inputBuffer.getChannelData(0);
            const int16 = new Int16Array(audio.length);
            for (let i = 0; i < audio.length; i++) int16[i] = audio[i] * 32768;
            if (ws.readyState === WebSocket.OPEN) ws.send(int16.buffer);
        };

        source.connect(processor);
        processor.connect(audioContext.destination);
    }

    function stopRecording() {
        processor?.disconnect();
        audioContext?.close();
        ws?.close();
        document.getElementById('status').className = 'status inactive';
        document.getElementById('statusText').textContent = '已停止';
    }
    </script>
</body>
</html>

10. 实战案例：多语言实时字幕系统

构建一个支持中英日韩多语言的实时字幕系统，适用于直播、视频会议等场景。

字幕系统架构

# subtitle_system.py
import asyncio
import time
from enum import Enum
from dataclasses import dataclass
from typing import Callable

class Language(Enum):
    ZH = "zh"
    EN = "en"
    JA = "ja"
    KO = "ko"
    AUTO = "auto"

@dataclass
class SubtitleLine:
    text: str
    language: Language
    start_ms: int
    end_ms: int
    confidence: float
    speaker_id: str | None = None

class MultiLanguageSubtitleSystem:
    def __init__(self):
        from faster_whisper import WhisperModel
        self.model = WhisperModel("medium", device="cuda", compute_type="float16")
        self.subtitles: list[SubtitleLine] = []
        self.callbacks: list[Callable] = []
        self.target_language: Language | None = None  # None表示不翻译

    def set_target_language(self, lang: Language | None):
        """设置目标语言（None=原语言，指定语言=翻译）"""
        self.target_language = lang

    def on_subtitle(self, callback: Callable[[SubtitleLine], None]):
        self.callbacks.append(callback)

    async def process_audio(self, audio: np.ndarray, timestamp_ms: int):
        """处理一段音频，生成字幕"""
        # 自动检测语言
        segments, info = self.model.transcribe(
            audio,
            language=None,         # 自动检测
            task="transcribe",     # 不翻译
            beam_size=3,
            vad_filter=True,
        )

        detected_lang = Language(info.language) if info.language in [l.value for l in Language] else Language.AUTO

        for seg in segments:
            subtitle = SubtitleLine(
                text=seg.text,
                language=detected_lang,
                start_ms=timestamp_ms + int(seg.start * 1000),
                end_ms=timestamp_ms + int(seg.end * 1000),
                confidence=seg.avg_logprob,
            )

            # 如果需要翻译
            if self.target_language and self.target_language != detected_lang:
                subtitle.text = await self._translate(subtitle.text, detected_lang, self.target_language)
                subtitle.language = self.target_language

            self.subtitles.append(subtitle)

            for cb in self.callbacks:
                cb(subtitle)

    async def _translate(self, text: str, from_lang: Language, to_lang: Language) -> str:
        """翻译字幕（可对接翻译API或本地模型）"""
        # 示例：使用Whisper的translate任务（仅支持翻译为英文）
        if to_lang == Language.EN:
            segments, _ = self.model.transcribe(
                text,
                language=from_lang.value,
                task="translate",
            )
            return "".join(seg.text for seg in segments)

        # 其他语言可对接翻译API
        # 这里返回原文作为fallback
        return text

    def export_srt(self, output_path: str):
        """导出SRT字幕文件"""
        with open(output_path, 'w', encoding='utf-8') as f:
            for i, sub in enumerate(self.subtitles, 1):
                start = self._ms_to_srt_time(sub.start_ms)
                end = self._ms_to_srt_time(sub.end_ms)
                f.write(f"{i}\n{start} --> {end}\n{sub.text}\n\n")

    @staticmethod
    def _ms_to_srt_time(ms: int) -> str:
        h = ms // 3600000
        m = (ms % 3600000) // 60000
        s = (ms % 60000) // 1000
        ms_rem = ms % 1000
        return f"{h:02d}:{m:02d}:{s:02d},{ms_rem:03d}"

    def export_vtt(self, output_path: str):
        """导出WebVTT字幕文件（Web标准）"""
        with open(output_path, 'w', encoding='utf-8') as f:
            f.write("WEBVTT\n\n")
            for i, sub in enumerate(self.subtitles, 1):
                start = self._ms_to_vtt_time(sub.start_ms)
                end = self._ms_to_vtt_time(sub.end_ms)
                f.write(f"{i}\n{start} --> {end}\n{sub.text}\n\n")

    @staticmethod
    def _ms_to_vtt_time(ms: int) -> str:
        h = ms // 3600000
        m = (ms % 3600000) // 60000
        s = (ms % 60000) // 1000
        ms_rem = ms % 1000
        return f"{h:02d}:{m:02d}:{s:02d}.{ms_rem:03d}"

OBS直播字幕集成

# obs_subtitle_overlay.py
import asyncio
import websockets
import json

class OBSSubtitleOverlay:
    """通过WebSocket向OBS的浏览器源推送字幕"""

    def __init__(self, obs_ws_url="ws://localhost:4455"):
        self.obs_ws_url = obs_ws_url
        self.subtitle_ws = None

    async def start_server(self, port=8766):
        """启动字幕WebSocket服务器，供OBS浏览器源连接"""
        async def handler(websocket):
            self.subtitle_ws = websocket
            try:
                async for _ in websocket:
                    pass  # 只用于发送，不接收
            finally:
                self.subtitle_ws = None

        async with websockets.serve(handler, "0.0.0.0", port):
            print(f"字幕服务器启动在 ws://0.0.0.0:{port}")
            await asyncio.Future()

    async def push_subtitle(self, text: str, duration_ms: int = 5000):
        if self.subtitle_ws:
            await self.subtitle_ws.send(json.dumps({
                "text": text,
                "duration": duration_ms,
            }))

# OBS浏览器源HTML模板
OBS_BROWSER_HTML = """
<!DOCTYPE html>
<html>
<style>
    body {
        margin: 0;
        background: transparent;
        display: flex;
        justify-content: center;
        align-items: flex-end;
        height: 100vh;
        padding-bottom: 40px;
        font-family: 'Noto Sans SC', sans-serif;
    }
    #subtitle {
        background: rgba(0, 0, 0, 0.75);
        color: white;
        padding: 8px 24px;
        border-radius: 6px;
        font-size: 24px;
        max-width: 80%;
        text-align: center;
        opacity: 0;
        transition: opacity 0.3s;
    }
    #subtitle.visible { opacity: 1; }
</style>
<body>
    <div id="subtitle"></div>
    <script>
        const el = document.getElementById('subtitle');
        const ws = new WebSocket('ws://localhost:8766');
        ws.onmessage = (e) => {
            const { text, duration } = JSON.parse(e.data);
            el.textContent = text;
            el.classList.add('visible');
            setTimeout(() => el.classList.remove('visible'), duration || 5000);
        };
    </script>
</body>
</html>
"""

11. 评估指标与优化策略

评估指标详解

import jiwer
from collections import Counter

def calculate_wer(reference: str, hypothesis: str) -> dict:
    """计算WER及相关指标"""
    measures = jiwer.compute_measures(reference, hypothesis)
    return {
        "wer": measures["wer"],
        "insertions": measures["insertions"],
        "deletions": measures["deletions"],
        "substitutions": measures["substitutions"],
        "hits": measures["hits"],
    }

def calculate_cer(reference: str, hypothesis: str) -> float:
    """计算字符错误率（中文适用）"""
    # 使用编辑距离
    ref_chars = list(reference.replace(" ", ""))
    hyp_chars = list(hypothesis.replace(" ", ""))

    # 动态规划计算编辑距离
    n, m = len(ref_chars), len(hyp_chars)
    dp = [[0] * (m + 1) for _ in range(n + 1)]

    for i in range(n + 1):
        dp[i][0] = i
    for j in range(m + 1):
        dp[0][j] = j

    for i in range(1, n + 1):
        for j in range(1, m + 1):
            if ref_chars[i-1] == hyp_chars[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[n][m] / max(n, 1)

def batch_evaluate(references: list[str], hypotheses: list[str]) -> dict:
    """批量评估"""
    total_wer = 0
    total_cer = 0

    for ref, hyp in zip(references, hypotheses):
        total_wer += calculate_wer(ref, hyp)["wer"]
        total_cer += calculate_cer(ref, hyp)

    n = len(references)
    return {
        "avg_wer": total_wer / n,
        "avg_cer": total_cer / n,
        "num_samples": n,
    }

优化策略

1. 模型选择优化

# 根据场景选择最优模型
def select_model(scenario: str):
    configs = {
        "realtime_low_latency": {
            "model": "small",
            "beam_size": 1,
            "compute_type": "int8",
            "vad_filter": True,
        },
        "offline_high_accuracy": {
            "model": "large-v3",
            "beam_size": 5,
            "compute_type": "float16",
            "vad_filter": True,
        },
        "batch_processing": {
            "model": "large-v3",
            "beam_size": 5,
            "compute_type": "float16",
            "batch_size": 16,
        },
        "edge_device": {
            "model": "tiny",
            "beam_size": 1,
            "compute_type": "int8",
        },
    }
    return configs.get(scenario, configs["offline_high_accuracy"])

2. 推理加速

# CTranslate2量化加速
from ctranslate2 import WhisperModel as CT2WhisperModel

# 转换模型格式（一次性）
converter = ctranslate2.converters.OpenAIConverter()
converter.convert(
    "large-v3",
    output_dir="whisper-large-v3-ct2",
    quantization="int8_float16",  # INT8权重 + FP16计算
)

# 使用量化模型推理
model = CT2WhisperModel("whisper-large-v3-ct2", device="cuda")
result = model.transcribe("audio.wav", beam_size=5)

3. VAD预过滤

from silero_vad import load_silero_vad, get_speech_timestamps
import torch

def filter_speech(audio: np.ndarray, sr: int = 16000) -> np.ndarray:
    """只保留有语音的部分，跳过静音"""
    model = load_silero_vad()
    audio_tensor = torch.from_numpy(audio)

    speech_timestamps = get_speech_timestamps(
        audio_tensor,
        model,
        sampling_rate=sr,
        min_speech_duration_ms=250,
        min_silence_duration_ms=100,
    )

    if not speech_timestamps:
        return audio

    # 拼接所有语音段
    speech_chunks = []
    for ts in speech_timestamps:
        speech_chunks.append(audio[ts['start']:ts['end']])

    return np.concatenate(speech_chunks)

4. 长音频分段策略

def transcribe_long_audio(audio_path: str, model, segment_duration: int = 30):
    """长音频分段识别，使用重叠窗口避免截断处丢字"""
    import soundfile as sf

    audio, sr = sf.read(audio_path)
    if len(audio.shape) > 1:
        audio = np.mean(audio, axis=1)

    segment_samples = segment_duration * sr
    overlap_samples = int(2.0 * sr)  # 2秒重叠
    step_samples = segment_samples - overlap_samples

    all_segments = []
    position = 0

    while position < len(audio):
        end = min(position + segment_samples, len(audio))
        chunk = audio[position:end]

        segments, _ = model.transcribe(chunk, language="zh", beam_size=5)

        for seg in segments:
            abs_start = position / sr + seg.start
            abs_end = position / sr + seg.end

            # 跳过重叠区域中已识别的部分
            if position > 0 and seg.start < 2.0:
                continue

            all_segments.append({
                "start": abs_start,
                "end": abs_end,
                "text": seg.text,
            })

        position += step_samples

    return all_segments

5. 生产环境部署建议

# docker-compose.yml
version: '3.8'
services:
  asr-server:
    build: .
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]
    ports:
      - "8765:8765"
    environment:
      - MODEL_SIZE=medium
      - COMPUTE_TYPE=float16
      - BEAM_SIZE=3
      - MAX_CONCURRENT=10
    volumes:
      - ./models:/root/.cache
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8765/health"]
      interval: 30s
      timeout: 10s
      retries: 3

以上涵盖了从基础ASR技术到完整会议转写系统搭建的全流程。在实际项目中，需要根据具体场景（实时性要求、语言需求、部署环境）选择合适的模型和架构方案。核心优化方向是：模型选择匹配场景需求、VAD预过滤减少无效计算、量化部署降低资源消耗、合理的分段策略平衡延迟与精度。