一、为什么要迁移一个已经稳定运行的 ASR 模型

在会议系统中更换语音识别模型，通常不是简单地修改一个模型路径。

ASR 模块的输出会继续进入说话人处理、时间轴对齐、会议纪要生成、全文检索和资料归档。底层模型一旦发生变化，语言参数、时间戳结构、长音频处理方式和异常返回格式都可能随之改变。

Whisper Turbo 是 Whisper large-v3 的优化版本，官方定位是在尽量控制准确率损失的情况下提高转写速度；faster-whisper 也支持加载 large-v3-turbo，并通过 CTranslate2 提供本地推理能力。

这次迁移并不是因为原模型无法使用，而是会议场景出现了新的要求：

中文会议和方言语音占比逐渐提高。

离线文件和实时会议希望采用同一模型体系。

边缘服务器需要更轻量的模型规模。

模型、依赖和推理服务需要完整离线交付。

业务接口不能因为更换模型而大范围修改。

Qwen3-ASR 系列包含 0.6B 和 1.7B 两种规模，支持语言识别、语音转写、长音频以及流式和离线推理；官方仓库当前列出的覆盖范围为 52 种语言和方言。流式能力通过 vLLM 后端提供，时间戳则可以结合独立的 Qwen3-ForcedAligner-0.6B 生成。

因此，本次改造的目标不是证明某一个模型在所有场景下都更好，而是完成一套可测试、可灰度、可回滚的 ASR 引擎迁移。

二、迁移时只替换引擎，不重写整条业务链路

最稳妥的改造方式，是保留现有的音频预处理、VAD、任务队列、接口地址和返回结构，只在内部增加一层 ASR 适配器。

整体结构如下：

会议录音
  ↓
音频格式检查
  ↓
采样率与声道统一
  ↓
VAD和长音频切分
  ↓
ASR统一适配层
  ├── Whisper large-v3 Turbo
  └── Qwen3-ASR-0.6B
  ↓
统一识别结果
  ↓
会议纪要、待办、检索和归档

迁移期间，业务系统仍然调用原来的接口：

POST /v1/audio/transcriptions

返回字段保持一致：

{
  "engine": "qwen3-asr",
  "model": "Qwen3-ASR-0.6B",
  "language": "Chinese",
  "text": "本次会议主要讨论语音识别模型迁移和接口兼容问题。",
  "duration_sec": 126.42,
  "process_sec": 10.83,
  "rtf": 0.0857
}

这样做有两个好处。

第一，新旧模型可以并行存在，便于对同一批会议录音进行双跑测试。

第二，新模型出现问题时，只需要修改配置并重启 ASR 服务，不需要重新发布会议业务系统。

三、先固定旧版 Whisper 的真实配置

在迁移前，需要先确认旧服务到底是怎样运行的。

原有 faster-whisper 代码可能类似：

from faster_whisper import WhisperModel

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

segments, info = model.transcribe(
    "/data/audio/meeting.wav",
    language="zh",
    beam_size=5,
    vad_filter=True,
    condition_on_previous_text=False,
)

text_parts = []

for segment in segments:
    text_parts.append(segment.text.strip())

full_text = "".join(text_parts)

print("language:", info.language)
print("text:", full_text)

真正需要记录的不只是模型名称，还包括以下参数：

engine: faster-whisper

model:
  name: large-v3-turbo
  device: cuda
  compute_type: float16

decode:
  language: zh
  beam_size: 5
  vad_filter: true
  condition_on_previous_text: false
  word_timestamps: false

faster-whisper 的推理结果会受到 beam size、VAD、量化精度和上下文参数影响。进行新旧模型比较时，不能只看模型名字，还要尽量保持输入音频和外围处理条件一致。

同时记录旧服务的软件版本：

python -V

pip show faster-whisper
pip show ctranslate2
pip show torch

nvidia-smi

导出依赖：

pip freeze > whisper-requirements-lock.txt

这份配置既是测试基线，也是后续回滚时需要保留的运行依据。

四、为 Qwen3-ASR 创建独立环境

不建议直接在原来的 Whisper 环境中安装新模型。

Whisper 服务通常依赖 faster-whisper 和 CTranslate2，而 Qwen3-ASR 会引入 PyTorch、Transformers，以及可选的 vLLM 和 FlashAttention。把两套依赖安装在同一个环境里，容易出现版本覆盖和二进制库冲突。

创建独立环境：

conda create -n qwen3-asr-migrate python=3.12 -y
conda activate qwen3-asr-migrate

python -m pip install --upgrade pip

pip install -U qwen-asr
pip install fastapi uvicorn python-multipart
pip install pydantic requests soundfile pydub

检查环境：

which python
which pip

python -V

python -c "import torch; print(torch.__version__)"
python -c "import torch; print('CUDA:', torch.cuda.is_available())"

pip show qwen-asr

启动服务时建议使用：

python -m uvicorn qwen_asr_server:app \
  --host 127.0.0.1 \
  --port 7862 \
  --workers 1

使用python -m uvicorn，可以避免终端虽然激活了新环境，但实际调用的仍然是其他环境中的uvicorn。

五、提前下载模型，并完成断网加载测试

联网环境下载模型：

pip install -U modelscope

mkdir -p /data/models

modelscope download \
  --model Qwen/Qwen3-ASR-0.6B \
  --local_dir /data/models/Qwen3-ASR-0.6B

检查模型目录：

find /data/models/Qwen3-ASR-0.6B \
  -maxdepth 2 \
  -type f |
head -30

du -sh /data/models/Qwen3-ASR-0.6B

生成离线压缩包：

tar -I 'zstd -T0' \
  -cf Qwen3-ASR-0.6B.tar.zst \
  -C /data/models \
  Qwen3-ASR-0.6B

sha256sum Qwen3-ASR-0.6B.tar.zst \
  > Qwen3-ASR-0.6B.tar.zst.sha256

目标服务器先校验，再解压：

sha256sum -c Qwen3-ASR-0.6B.tar.zst.sha256

mkdir -p /data/models

tar -I zstd \
  -xf Qwen3-ASR-0.6B.tar.zst \
  -C /data/models

设置离线模式：

export HF_HUB_OFFLINE=1
export TRANSFORMERS_OFFLINE=1

Qwen3-ASR 官方仓库提供 ModelScope 和 Hugging Face 两种模型获取方式，也支持从本地目录加载模型。

正式迁移前应在无法访问外网的环境中启动一次。只有断网后模型仍能完整加载，才能确认权重、配置、处理器和其他必要文件已经全部交付。

六、先完成最小推理验证

新建test_qwen3_asr.py：

import os

import torch
from qwen_asr import Qwen3ASRModel

MODEL_PATH = os.getenv(
    "ASR_MODEL_PATH",
    "/data/models/Qwen3-ASR-0.6B",
)

AUDIO_PATH = os.getenv(
    "ASR_TEST_AUDIO",
    "/data/test/meeting_30s.wav",
)


def resolve_runtime() -> tuple[str, torch.dtype]:
    if not torch.cuda.is_available():
        return "cpu", torch.float32

    if torch.cuda.is_bf16_supported():
        return "cuda:0", torch.bfloat16

    return "cuda:0", torch.float16


device, dtype = resolve_runtime()

print(f"model={MODEL_PATH}")
print(f"device={device}")
print(f"dtype={dtype}")

model = Qwen3ASRModel.from_pretrained(
    MODEL_PATH,
    device_map=device,
    dtype=dtype,
    max_inference_batch_size=1,
    max_new_tokens=512,
)

results = model.transcribe(
    audio=AUDIO_PATH,
    language="Chinese",
)

if not results:
    raise RuntimeError("模型未返回识别结果")

result = results[0]

print("language:", result.language)
print("text:", result.text)

执行：

export ASR_MODEL_PATH=/data/models/Qwen3-ASR-0.6B
export ASR_TEST_AUDIO=/data/test/meeting_30s.wav

python test_qwen3_asr.py

输出示例：

model=/data/models/Qwen3-ASR-0.6B
device=cuda:0
dtype=torch.bfloat16

language: Chinese
text: 本次会议主要讨论语音识别模型迁移和后续接口测试安排。

官方 Python 接口支持本地文件、URL、Base64 数据以及数组形式的音频输入；language=None可以执行语言识别，也可以显式指定语言。

关键提醒：新旧模型的语言参数不同

Whisper 常用：

language="zh"

Qwen3-ASR 使用：

language="Chinese"

需要在统一适配层中完成转换：

LANGUAGE_TO_QWEN = {
    None: None,
    "": None,
    "auto": None,
    "zh": "Chinese",
    "zh-cn": "Chinese",
    "cmn": "Chinese",
    "en": "English",
    "ja": "Japanese",
    "ko": "Korean",
    "yue": "Cantonese",
}


def normalize_qwen_language(
    language: str | None,
) -> str | None:
    if language is None:
        return None

    key = language.strip().lower()

    return LANGUAGE_TO_QWEN.get(
        key,
        language,
    )

如果直接把旧接口中的zh原样传给新模型，即使接口没有立即报错，也可能产生非预期行为。

七、建立统一的数据结构

先定义新旧引擎共同使用的返回结构：

from dataclasses import dataclass, field


@dataclass
class TranscriptSegment:
    start_sec: float
    end_sec: float
    text: str


@dataclass
class TranscriptionResult:
    engine: str
    model: str
    language: str
    text: str
    duration_sec: float
    process_sec: float
    rtf: float
    segments: list[TranscriptSegment] = field(
        default_factory=list
    )

定义统一接口：

from abc import ABC, abstractmethod


class ASREngine(ABC):

    @abstractmethod
    def transcribe(
        self,
        audio_path: str,
        language: str | None = None,
    ) -> TranscriptionResult:
        raise NotImplementedError

这样业务服务不需要知道底层使用的是 faster-whisper 还是 Qwen3-ASR。

八、实现 Whisper 适配器

import subprocess
import time

from faster_whisper import WhisperModel


def get_audio_duration(audio_path: str) -> float:
    output = subprocess.check_output(
        [
            "ffprobe",
            "-v",
            "error",
            "-show_entries",
            "format=duration",
            "-of",
            "default=noprint_wrappers=1:nokey=1",
            audio_path,
        ],
        timeout=30,
    )

    return float(output.decode().strip())


class WhisperEngine(ASREngine):

    def __init__(
        self,
        model_path: str,
        device: str = "cuda",
        compute_type: str = "float16",
    ) -> None:
        self.model_name = "large-v3-turbo"

        self.model = WhisperModel(
            model_path,
            device=device,
            compute_type=compute_type,
        )

    def transcribe(
        self,
        audio_path: str,
        language: str | None = None,
    ) -> TranscriptionResult:
        duration_sec = get_audio_duration(audio_path)
        started_at = time.perf_counter()

        segments_generator, info = self.model.transcribe(
            audio_path,
            language=language,
            beam_size=5,
            vad_filter=True,
            condition_on_previous_text=False,
        )

        segments = []
        text_parts = []

        for item in segments_generator:
            text = item.text.strip()

            if not text:
                continue

            segments.append(
                TranscriptSegment(
                    start_sec=float(item.start),
                    end_sec=float(item.end),
                    text=text,
                )
            )
            text_parts.append(text)

        process_sec = time.perf_counter() - started_at

        return TranscriptionResult(
            engine="faster-whisper",
            model=self.model_name,
            language=info.language or language or "unknown",
            text="".join(text_parts),
            duration_sec=duration_sec,
            process_sec=process_sec,
            rtf=process_sec / duration_sec,
            segments=segments,
        )

原来的 Whisper 实现暂时不要删除。迁移阶段仍然需要它参与双跑测试和故障回滚。

九、实现 Qwen3-ASR 适配器

import time

import torch
from qwen_asr import Qwen3ASRModel


class Qwen3ASREngine(ASREngine):

    def __init__(
        self,
        model_path: str,
        device: str = "cuda:0",
    ) -> None:
        if device.startswith("cuda"):
            dtype = (
                torch.bfloat16
                if torch.cuda.is_bf16_supported()
                else torch.float16
            )
        else:
            dtype = torch.float32

        self.model_name = "Qwen3-ASR-0.6B"

        self.model = Qwen3ASRModel.from_pretrained(
            model_path,
            device_map=device,
            dtype=dtype,
            max_inference_batch_size=1,
            max_new_tokens=1024,
        )

    def transcribe(
        self,
        audio_path: str,
        language: str | None = None,
    ) -> TranscriptionResult:
        duration_sec = get_audio_duration(audio_path)
        qwen_language = normalize_qwen_language(language)

        started_at = time.perf_counter()

        results = self.model.transcribe(
            audio=audio_path,
            language=qwen_language,
        )

        process_sec = time.perf_counter() - started_at

        if not results:
            raise RuntimeError(
                "Qwen3-ASR未返回识别结果"
            )

        result = results[0]

        return TranscriptionResult(
            engine="qwen3-asr",
            model=self.model_name,
            language=result.language
            or qwen_language
            or "unknown",
            text=result.text.strip(),
            duration_sec=duration_sec,
            process_sec=process_sec,
            rtf=process_sec / duration_sec,
            segments=[],
        )

这里先将segments设置为空列表，因为普通Qwen3-ASR转写结果和Whisper的原生分段时间戳并不是同一种结构。

先完成文本迁移，再单独解决时间戳，是风险更低的做法。

十、通过配置选择实际使用的引擎

import os


def create_asr_engine() -> ASREngine:
    provider = os.getenv(
        "ASR_PROVIDER",
        "whisper",
    ).strip().lower()

    if provider == "whisper":
        return WhisperEngine(
            model_path=os.getenv(
                "WHISPER_MODEL_PATH",
                "/data/models/whisper-large-v3-turbo",
            ),
            device=os.getenv(
                "WHISPER_DEVICE",
                "cuda",
            ),
            compute_type=os.getenv(
                "WHISPER_COMPUTE_TYPE",
                "float16",
            ),
        )

    if provider == "qwen3":
        return Qwen3ASREngine(
            model_path=os.getenv(
                "QWEN3_ASR_MODEL_PATH",
                "/data/models/Qwen3-ASR-0.6B",
            ),
            device=os.getenv(
                "QWEN3_ASR_DEVICE",
                "cuda:0",
            ),
        )

    raise ValueError(
        f"不支持的ASR_PROVIDER：{provider}"
    )

切换到旧模型：

export ASR_PROVIDER=whisper

切换到新模型：

export ASR_PROVIDER=qwen3

业务代码只保留：

asr_engine = create_asr_engine()

result = asr_engine.transcribe(
    "/data/audio/meeting.wav",
    language="zh",
)

后续回滚时，不需要再次修改Python代码。

十一、保持原有 FastAPI 接口不变

import asyncio
import os
import shutil
import tempfile
import uuid
from pathlib import Path
from typing import Optional

from fastapi import FastAPI
from fastapi import File
from fastapi import Form
from fastapi import HTTPException
from fastapi import UploadFile
from pydantic import BaseModel

app = FastAPI(
    title="Meeting ASR Compatibility Service",
    version="2.0.0",
)

asr_engine = create_asr_engine()
inference_lock = asyncio.Lock()

ALLOWED_SUFFIXES = {
    ".wav",
    ".mp3",
    ".m4a",
    ".aac",
    ".flac",
    ".ogg",
    ".opus",
    ".webm",
}


class SegmentResponse(BaseModel):
    start_sec: float
    end_sec: float
    text: str


class ASRResponse(BaseModel):
    request_id: str
    engine: str
    model: str
    language: str
    text: str
    duration_sec: float
    process_sec: float
    rtf: float
    segments: list[SegmentResponse]


@app.get("/health")
def health_check():
    return {
        "status": "ok",
        "engine": type(asr_engine).__name__,
        "provider": os.getenv(
            "ASR_PROVIDER",
            "whisper",
        ),
    }


@app.post(
    "/v1/audio/transcriptions",
    response_model=ASRResponse,
)
async def transcribe(
    file: UploadFile = File(...),
    language: Optional[str] = Form(default="zh"),
):
    request_id = uuid.uuid4().hex[:16]
    suffix = Path(file.filename or "").suffix.lower()

    if suffix not in ALLOWED_SUFFIXES:
        raise HTTPException(
            status_code=415,
            detail=f"不支持的格式：{suffix}",
        )

    temp_dir = Path(
        tempfile.mkdtemp(
            prefix=f"asr-{request_id}-"
        )
    )
    input_path = temp_dir / f"input{suffix}"

    try:
        with input_path.open("wb") as target:
            while chunk := await file.read(
                1024 * 1024
            ):
                target.write(chunk)

        async with inference_lock:
            result = await asyncio.to_thread(
                asr_engine.transcribe,
                str(input_path),
                language,
            )

        return {
            "request_id": request_id,
            "engine": result.engine,
            "model": result.model,
            "language": result.language,
            "text": result.text,
            "duration_sec": round(
                result.duration_sec,
                3,
            ),
            "process_sec": round(
                result.process_sec,
                3,
            ),
            "rtf": round(result.rtf, 4),
            "segments": [
                {
                    "start_sec": item.start_sec,
                    "end_sec": item.end_sec,
                    "text": item.text,
                }
                for item in result.segments
            ],
        }

    except Exception as exc:
        print(
            f"[ASR-ERROR] "
            f"request_id={request_id} "
            f"type={type(exc).__name__} "
            f"message={exc}"
        )

        raise HTTPException(
            status_code=500,
            detail="语音识别失败",
        ) from exc

    finally:
        shutil.rmtree(
            temp_dir,
            ignore_errors=True,
        )

启动：

export ASR_PROVIDER=qwen3
export QWEN3_ASR_MODEL_PATH=/data/models/Qwen3-ASR-0.6B

python -m uvicorn asr_compat_server:app \
  --host 127.0.0.1 \
  --port 7861 \
  --workers 1

测试：

curl -X POST \
  http://127.0.0.1:7861/v1/audio/transcriptions \
  -F "file=@/data/test/meeting.wav" \
  -F "language=zh"

关键提醒：模型进程不要随意增加 Worker

每个Uvicorn Worker都会独立加载模型。设置：

--workers 4

通常意味着显存中同时存在四份模型，而不是单纯获得四倍吞吐。

GPU模型服务建议保持：

--workers 1

再通过任务队列、信号量或请求锁控制并发。

十二、时间戳不能直接照搬 Whisper 的实现

Whisper和faster-whisper可以直接输出分段时间信息，而Qwen3-ASR需要结合Qwen3-ForcedAligner-0.6B生成词级或字符级时间戳。

官方说明中，Forced Aligner支持11种语言，并支持最长5分钟语音的文本对齐。流式模式当前不返回时间戳。

下载模型：

modelscope download \
  --model Qwen/Qwen3-ForcedAligner-0.6B \
  --local_dir /data/models/Qwen3-ForcedAligner-0.6B

加载方式：

import torch
from qwen_asr import Qwen3ASRModel

model = Qwen3ASRModel.from_pretrained(
    "/data/models/Qwen3-ASR-0.6B",
    device_map="cuda:0",
    dtype=torch.bfloat16,
    max_inference_batch_size=1,
    max_new_tokens=1024,
    forced_aligner=(
        "/data/models/"
        "Qwen3-ForcedAligner-0.6B"
    ),
    forced_aligner_kwargs={
        "device_map": "cuda:0",
        "dtype": torch.bfloat16,
    },
)

results = model.transcribe(
    audio="/data/test/meeting_3min.wav",
    language="Chinese",
    return_time_stamps=True,
)

result = results[0]

print(result.text)

for item in result.time_stamps:
    print(item)

关键提醒：不要为了保持旧接口，第一天就强行上线时间戳

额外加载Forced Aligner会增加显存占用，也会改变长音频处理流程。

更稳妥的迁移顺序是：

第一阶段：替换文本识别引擎
第二阶段：验证纪要、检索和长音频效果
第三阶段：接入Forced Aligner
第四阶段：恢复时间轴和点击回听能力

对齐模型单次处理长度有限，长会议需要先切成不超过限制的片段，再分别执行识别和时间对齐。

十三、用同一批会议录音进行双跑测试

不要只拿一段干净普通话测试音频决定是否上线。

测试集至少应覆盖：

普通会议室录音
远距离拾音
多人讨论
中英文混合
地方口音或方言
专业术语
长时间静音
线上会议回放
双声道录音
一小时以上长会议

批量对比脚本：

import csv
import time
from pathlib import Path

import requests

WHISPER_API = (
    "http://127.0.0.1:7861"
    "/v1/audio/transcriptions"
)

QWEN_API = (
    "http://127.0.0.1:7862"
    "/v1/audio/transcriptions"
)

AUDIO_SUFFIXES = {
    ".wav",
    ".mp3",
    ".m4a",
    ".aac",
    ".flac",
}


def call_asr(
    api_url: str,
    audio_path: Path,
) -> dict:
    started_at = time.perf_counter()

    with audio_path.open("rb") as audio_file:
        response = requests.post(
            api_url,
            files={
                "file": (
                    audio_path.name,
                    audio_file,
                )
            },
            data={"language": "zh"},
            timeout=1800,
        )

    elapsed = time.perf_counter() - started_at
    response.raise_for_status()

    data = response.json()
    data["client_elapsed_sec"] = elapsed

    return data


def run_comparison(
    audio_dir: str,
    output_csv: str,
) -> None:
    rows = []

    for audio_path in sorted(
        Path(audio_dir).iterdir()
    ):
        if audio_path.suffix.lower() not in AUDIO_SUFFIXES:
            continue

        print(f"测试：{audio_path.name}")

        whisper_result = call_asr(
            WHISPER_API,
            audio_path,
        )

        qwen_result = call_asr(
            QWEN_API,
            audio_path,
        )

        rows.append(
            {
                "filename": audio_path.name,
                "duration_sec": qwen_result[
                    "duration_sec"
                ],
                "whisper_process_sec": (
                    whisper_result["process_sec"]
                ),
                "whisper_rtf": whisper_result[
                    "rtf"
                ],
                "whisper_text": whisper_result[
                    "text"
                ],
                "qwen_process_sec": qwen_result[
                    "process_sec"
                ],
                "qwen_rtf": qwen_result["rtf"],
                "qwen_text": qwen_result["text"],
            }
        )

    with open(
        output_csv,
        "w",
        encoding="utf-8-sig",
        newline="",
    ) as output_file:
        writer = csv.DictWriter(
            output_file,
            fieldnames=rows[0].keys(),
        )
        writer.writeheader()
        writer.writerows(rows)


run_comparison(
    "/data/asr-benchmark",
    "/data/results/asr-comparison.csv",
)

评估时建议同时记录以下结果：

字错误率或人工错字数
人名和机构名识别
专业术语识别
方言识别
中英文混合文本
断句可读性
重复与幻觉
音频时长
处理耗时
RTF
峰值显存
异常率

不要只看RTF，也不要只凭一两段主观听感判断准确率。

更可靠的方式，是让会议纪要使用人员对隐藏模型名称的结果进行盲评。

十四、通过影子模式完成灰度迁移

正式切换前，可以让旧模型继续向业务系统返回结果，同时在后台调用新模型。

from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=1)


def process_with_shadow(
    audio_path: str,
    language: str | None,
) -> TranscriptionResult:
    primary_result = whisper_engine.transcribe(
        audio_path,
        language,
    )

    executor.submit(
        run_shadow_qwen,
        audio_path,
        language,
    )

    return primary_result


def run_shadow_qwen(
    audio_path: str,
    language: str | None,
) -> None:
    try:
        result = qwen_engine.transcribe(
            audio_path,
            language,
        )

        save_shadow_result(result)

    except Exception as exc:
        print(
            "[SHADOW-ASR-ERROR] "
            f"{type(exc).__name__}: {exc}"
        )

影子模式下：

Whisper结果：继续进入正式业务流程
Qwen结果：只保存测试指标，不展示给用户

经过一段时间的真实会议测试后，再逐步调整流量：

阶段一：0%，仅影子运行
阶段二：内部测试账号使用新模型
阶段三：10%会议任务
阶段四：30%会议任务
阶段五：全部切换

这种方式比一次性替换模型更容易发现少数特殊音频中的问题。

十五、接回熙瑾会悟时，业务层不应感知模型差异

ASR服务最终只需要输出统一结构：

{
  "engine": "qwen3-asr",
  "model": "Qwen3-ASR-0.6B",
  "language": "Chinese",
  "text": "会议确认了模型迁移计划和后续测试范围。",
  "segments": [],
  "duration_sec": 315.28,
  "process_sec": 27.64,
  "rtf": 0.0877
}

熙瑾会悟继续使用其中的文本、语言、片段和模型版本信息，完成纪要生成、议题归纳、待办提取和会议资料归档。

提交接口：

from datetime import datetime

import requests

MEETING_API = (
    "http://127.0.0.1:18080"
    "/api/meeting/minutes/generate"
)


def submit_transcript(
    meeting_id: str,
    asr_result: TranscriptionResult,
) -> dict:
    payload = {
        "meeting_id": meeting_id,
        "meeting_time": datetime.now().isoformat(),
        "asr_engine": asr_result.engine,
        "asr_model": asr_result.model,
        "language": asr_result.language,
        "transcript": asr_result.text,
        "segments": [
            {
                "start_sec": segment.start_sec,
                "end_sec": segment.end_sec,
                "text": segment.text,
            }
            for segment in asr_result.segments
        ],
        "metrics": {
            "duration_sec": asr_result.duration_sec,
            "process_sec": asr_result.process_sec,
            "rtf": asr_result.rtf,
        },
    }

    response = requests.post(
        MEETING_API,
        json=payload,
        timeout=300,
    )
    response.raise_for_status()

    return response.json()

业务层不直接调用某个模型SDK，也不依赖模型专属语言代码。这样后续再次升级模型时，改动仍然集中在ASR适配层。

十六、保留一键回滚能力

切换配置：

export ASR_PROVIDER=qwen3

回滚配置：

export ASR_PROVIDER=whisper

重新启动：

systemctl restart meeting-asr

或者使用Docker Compose：

ASR_PROVIDER=whisper \
docker compose up -d \
  --force-recreate \
  meeting-asr

关键提醒：不要在新模型上线当天删除旧环境

至少应保留：

Whisper模型文件
旧版容器镜像
旧版依赖锁定文件
旧服务配置
回滚脚本
旧版测试结果

同时避免新旧服务复用同一个可写模型目录，以免升级或清理操作破坏旧模型。

迁移验收后，再根据内部版本管理策略决定旧模型的保留期限。

十七、迁移完成后的验收清单

上线前可以逐项检查：

[ ] 新旧模型使用同一批测试音频
[ ] 语言参数已经完成映射
[ ] API地址和字段保持兼容
[ ] RTF计算口径保持一致
[ ] 长音频能够完整处理
[ ] 临时音频能够自动清理
[ ] 单Worker和并发锁已经生效
[ ] 纪要生成链路能够正常接收结果
[ ] 时间戳缺失时前端不会报错
[ ] 模型可以断网启动
[ ] 镜像和模型文件已经记录哈希
[ ] 新模型异常时可以快速切回旧模型

只有这些项目全部通过，模型切换才算真正完成。

十八、总结：一次可靠的模型迁移，重点不在“换掉模型”

从 Whisper large-v3 Turbo 切换到 Qwen3-ASR-0.6B，真正需要解决的不是一段推理代码，而是新旧模型之间的工程差异。

其中最关键的部分包括：

统一语言参数
统一接口返回结构
保留旧引擎
建立双跑测试
分阶段处理时间戳
限制模型并发
准备灰度方案
保留一键回滚

Qwen3-ASR-0.6B为中文会议、方言覆盖、流式与离线统一以及边缘部署提供了新的选择，但是否正式替换原模型，仍然应由真实会议测试结果决定，而不是只参考公开指标。

在这次改造中，熙瑾会悟的会议业务链路并没有因为底层模型变化而重新设计。ASR适配层屏蔽了不同模型的接口差异，上层仍然接收统一的转写结果并完成纪要、待办和资料归档。

这也是生产系统更换AI模型时更值得复用的思路：模型可以持续升级，但业务接口、数据结构和回滚能力应尽量保持稳定。