一、为什么边缘会议设备更需要轻量级 ASR

在企业会议系统中，语音识别模型通常部署在中心机房，由多块 GPU 统一提供推理服务。但在安全敏感部门、隔离网络、临时指挥场所和无法稳定连接中心服务器的环境中，会议录音往往需要直接在会议室边缘节点完成处理。

这类部署方式有几个明显特点：会议音频不离开本地设备；模型权重提前下载并离线加载；识别接口只在内网或本机开放；系统不依赖外部 API，也不在运行时访问模型仓库；会议结束后，可以立即生成转写文本，再交给上层模块整理纪要。
Qwen3-ASR 系列包含 1.7B 和 0.6B 两种规模。根据Modelscope资料显示，两者都支持语言识别和语音识别，覆盖30种语言和22种中文方言，也支持流式与离线统一推理以及长音频处理；其中0.6B版本更侧重准确率与推理效率之间的平衡。

因此，在算力、功耗和设备空间受限的边缘节点上，Qwen3-ASR-0.6B比更大参数量模型更容易落地。它可以部署在会议一体机、小型GPU服务器或机房边缘计算节点中，承担会议转写的前置工作。

本文将从环境准备开始，把Qwen3-ASR-0.6B封装成一个本地ASR服务，再将识别结果送入会悟的会议处理链路。

需要提前说明：本地运行、数据不出域和离线部署，是安全敏感场景的重要技术条件，但不等于系统自动满足涉密合规要求。正式用于涉密环境时，还需要结合网络分区、设备认证、身份鉴别、访问控制、日志审计、存储加密和单位内部保密制度进行整体设计。

二、先设计完整链路，而不是只启动一个模型

边缘会议转写不能只考虑模型推理，还要考虑音频采集、文件格式、长音频处理、并发控制、异常恢复和结果回传。

本文采用的链路如下：

会议录音设备
  ↓
边缘节点接收音频
  ↓
ffmpeg统一采样率和声道
  ↓
长音频检测与分段
  ↓
Qwen3-ASR-0.6B本地转写
  ↓
语言、文本、耗时、RTF等结构化结果
  ↓
熙瑾会悟：会议纪要、议题、决策、待办和归档

Qwen3-ASR-0.6B位于链路前半段，解决“录音里说了什么”的问题。后半段则继续解决段落整理、会议摘要、决策提取和行动项归纳。

边缘部署的关键不是把所有模块都放进一个进程，而是让各模块之间保持清晰边界。ASR服务发生异常时，不应导致会议业务系统、文件存储和纪要生成模块同时退出。

三、创建隔离环境，避免污染已有 Python 服务

官方项目推荐使用独立的Python 3.12环境，并通过qwen-asr软件包加载模型。该软件包提供Transformers和vLLM两种后端；Transformers后端安装简单，适合先验证单机链路，vLLM后端则更适合高吞吐和流式识别。

创建环境：

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

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

检查环境：

python -V
pip show qwen-asr
python -c "import torch; print(torch.__version__)"
python -c "import torch; print('CUDA:', torch.cuda.is_available())"

预期结果：

Python 3.12.x
CUDA: True

踩坑提醒：不要直接安装进业务系统的 base 环境

ASR项目通常会修改torch、transformers、tokenizers、vllm等核心依赖。如果直接安装到已有业务环境，可能出现以下问题：

ImportError: undefined symbol
transformers版本不兼容
torch与CUDA版本不匹配
原有服务启动后显存占用异常
uvicorn实际调用了另一个Python环境

安装完成后，应确认实际使用的是当前环境中的解释器：

which python
which pip
which uvicorn

head -n 1 "$(which uvicorn)"

如果uvicorn的首行仍然指向其他Python路径，不要直接使用它，可以改成：

python -m uvicorn app:app --host 127.0.0.1 --port 7861

这样能确保启动命令使用当前环境的Python解释器。

四、提前下载模型，杜绝运行时访问外网

如果代码里直接填写：

MODEL_PATH = "Qwen/Qwen3-ASR-0.6B"

模型加载时可能尝试连接模型仓库。普通开发环境这样做很方便，但不适合隔离网络和安全敏感部署。

官方仓库提供了ModelScope和Hugging Face两种下载方式。国内环境可以提前通过ModelScope下载到本地目录。

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

边缘节点正式上线前，可以断开外网再做一次加载测试：

export HF_HUB_OFFLINE=1
export TRANSFORMERS_OFFLINE=1

踩坑提醒：本地目录不完整会触发二次下载

模型目录只复制权重文件、不复制配置文件时，加载过程仍可能尝试联网寻找缺失内容。

离线交付时不要只检查模型文件大小，还应检查配置、分词器和处理器相关文件是否完整。可以先在联网机器上成功加载一次，再将整个模型目录打包。

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

对于安全敏感部门，模型文件、安装包和容器镜像都应建立版本清单和哈希校验记录，避免现场使用来源不明或被替换的文件。

五、第一次推理：先用短音频验证基础链路

准备一段10秒到30秒的会议音频：

/data/test/meeting_30s.wav

新建test_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_device_and_dtype() -> 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_device_and_dtype()

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

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

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

for result in results:
    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_asr.py

输出可能类似：

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

language: Chinese
text: 本次会议主要确认边缘节点部署计划，并讨论后续接口联调和系统测试安排。

官方qwen-asr接口支持本地文件路径、URL、Base64数据以及numpy数组+采样率等形式输入；language=None可以自动检测语言，也可以显式传入Chinese或English跳过语言识别。

踩坑提醒：0.6B不等于一定可以用CPU实时运行

0.6B表示模型参数规模较小，但不代表任何低功耗CPU都能达到实时会议转写。

实际速度还会受到以下因素影响：

CPU或GPU型号
推理精度
音频时长
max_new_tokens
并发数量
是否安装FlashAttention
是否使用Transformers或vLLM后端
磁盘读取速度

CPU模式更适合验证流程，不应在没有实测RTF的情况下直接承诺实时效果。

踩坑提醒：不要一开始就把batch设得很大

边缘节点显存通常有限，建议先使用：

max_inference_batch_size=1
max_new_tokens=512

链路稳定后再逐步提高。官方说明中也明确指出，减小max_inference_batch_size有助于避免OOM，而长音频需要适当增加max_new_tokens。

六、统一音频格式，减少“模型没问题但结果异常”

会议录音可能来自USB麦克风、录音笔、会议主机、浏览器或第三方录播设备。常见格式包括：

wav
mp3
m4a
aac
webm
opus
双声道PCM
48kHz录音

正式进入ASR前，建议统一转成16kHz、单声道WAV：

sudo apt-get update
sudo apt-get install -y ffmpeg

ffmpeg -y \
  -i input.m4a \
  -ar 16000 \
  -ac 1 \
  -c:a pcm_s16le \
  output.wav

批量处理：

mkdir -p normalized

find raw_audio -maxdepth 1 -type f -print0 |
while IFS= read -r -d '' file; do
    name="$(basename "$file")"
    base="${name%.*}"

    ffmpeg -hide_banner -loglevel error -y \
      -i "$file" \
      -ar 16000 \
      -ac 1 \
      -c:a pcm_s16le \
      "normalized/${base}.wav" \
    || echo "转换失败：$file" >&2
done

检查音频参数：

ffprobe -v error \
  -select_streams a:0 \
  -show_entries stream=codec_name,sample_rate,channels,duration \
  -of default=noprint_wrappers=1 \
  output.wav

预期结果：

codec_name=pcm_s16le
sample_rate=16000
channels=1
duration=125.430000

踩坑提醒：只改文件后缀没有任何作用

将meeting.m4a直接重命名为meeting.wav，并不会改变内部编码格式。模型可能读取失败，也可能识别结果异常。

格式转换必须通过ffmpeg或音频处理库完成。

踩坑提醒：双声道可能包含两个不同音源

部分会议设备会把本地麦克风和远端会议音频分别写入左右声道。直接转成单声道可能发生混音。

先检查左右声道内容：

ffmpeg -y -i meeting.wav \
  -map_channel 0.0.0 left.wav

ffmpeg -y -i meeting.wav \
  -map_channel 0.0.1 right.wav

确认两个声道的业务含义后，再决定是否混合或分别识别。

七、封装 FastAPI，并用锁限制边缘节点并发

单次推理跑通后，可以封装为本地HTTP服务。

新建asr_server.py：

import asyncio
import hashlib
import os
import shutil
import subprocess
import tempfile
import time
import uuid
from pathlib import Path
from typing import Optional

import torch
from fastapi import FastAPI, File, Form, HTTPException, UploadFile
from pydantic import BaseModel
from qwen_asr import Qwen3ASRModel

MODEL_PATH = os.getenv(
    "ASR_MODEL_PATH",
    "/data/models/Qwen3-ASR-0.6B",
)
MAX_UPLOAD_BYTES = int(
    os.getenv("MAX_UPLOAD_BYTES", str(2 * 1024**3))
)
MAX_AUDIO_SECONDS = float(
    os.getenv("MAX_AUDIO_SECONDS", "14400")
)

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


def resolve_device_and_dtype() -> 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_device_and_dtype()

model = Qwen3ASRModel.from_pretrained(
    MODEL_PATH,
    dtype=DTYPE,
    device_map=DEVICE,
    max_inference_batch_size=int(
        os.getenv("MAX_INFERENCE_BATCH_SIZE", "1")
    ),
    max_new_tokens=int(
        os.getenv("MAX_NEW_TOKENS", "1024")
    ),
)

# 边缘节点默认串行推理，避免多个长会议同时挤爆显存。
inference_lock = asyncio.Lock()

app = FastAPI(
    title="Qwen3-ASR Edge Service",
    version="1.0.0",
)


class ASRResponse(BaseModel):
    request_id: str
    language: str
    text: str
    duration_sec: float
    process_sec: float
    rtf: float
    file_sha256: str


def get_audio_duration(audio_path: str) -> float:
    cmd = [
        "ffprobe",
        "-v",
        "error",
        "-show_entries",
        "format=duration",
        "-of",
        "default=noprint_wrappers=1:nokey=1",
        audio_path,
    ]

    try:
        output = subprocess.check_output(
            cmd,
            stderr=subprocess.STDOUT,
            timeout=30,
        )
        return float(output.decode().strip())
    except Exception as exc:
        raise ValueError(
            f"无法读取音频时长：{exc}"
        ) from exc


def calculate_sha256(file_path: str) -> str:
    digest = hashlib.sha256()

    with open(file_path, "rb") as file_obj:
        while chunk := file_obj.read(1024 * 1024):
            digest.update(chunk)

    return digest.hexdigest()


async def save_upload_limited(
    upload: UploadFile,
    destination: Path,
) -> int:
    written = 0

    with destination.open("wb") as target:
        while True:
            chunk = await upload.read(1024 * 1024)
            if not chunk:
                break

            written += len(chunk)
            if written > MAX_UPLOAD_BYTES:
                raise HTTPException(
                    status_code=413,
                    detail="音频文件超过允许大小",
                )

            target.write(chunk)

    return written


@app.get("/health")
def health_check():
    return {
        "status": "ok",
        "model": Path(MODEL_PATH).name,
        "device": DEVICE,
        "dtype": str(DTYPE),
    }


@app.post("/v1/audio/transcriptions", response_model=ASRResponse)
async def transcribe(
    file: UploadFile = File(...),
    language: Optional[str] = Form(default="Chinese"),
):
    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:
        await save_upload_limited(file, input_path)

        duration_sec = get_audio_duration(str(input_path))
        if duration_sec <= 0:
            raise HTTPException(
                status_code=400,
                detail="音频时长无效",
            )

        if duration_sec > MAX_AUDIO_SECONDS:
            raise HTTPException(
                status_code=413,
                detail="音频时长超过接口限制，请先分段",
            )

        file_sha256 = calculate_sha256(str(input_path))
        started_at = time.perf_counter()

        async with inference_lock:
            results = await asyncio.to_thread(
                model.transcribe,
                audio=str(input_path),
                language=language or None,
            )

        process_sec = time.perf_counter() - started_at

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

        result = results[0]
        rtf = (
            process_sec / duration_sec
            if duration_sec > 0
            else 0.0
        )

        # 安全场景不记录原文件名和完整文本。
        print(
            f"[ASR] request_id={request_id} "
            f"duration={duration_sec:.3f}s "
            f"process={process_sec:.3f}s "
            f"rtf={rtf:.4f} "
            f"language={result.language} "
            f"sha256={file_sha256[:12]}"
        )

        return {
            "request_id": request_id,
            "language": result.language,
            "text": result.text,
            "duration_sec": round(duration_sec, 3),
            "process_sec": round(process_sec, 3),
            "rtf": round(rtf, 4),
            "file_sha256": file_sha256,
        }

    except HTTPException:
        raise
    except Exception as exc:
        print(
            f"[ASR-ERROR] request_id={request_id} "
            f"error={type(exc).__name__}: {exc}"
        )
        raise HTTPException(
            status_code=500,
            detail="语音识别失败",
        ) from exc
    finally:
        shutil.rmtree(temp_dir, ignore_errors=True)

启动服务：

export ASR_MODEL_PATH=/data/models/Qwen3-ASR-0.6B
export MAX_INFERENCE_BATCH_SIZE=1
export MAX_NEW_TOKENS=1024
export MAX_AUDIO_SECONDS=14400

python -m uvicorn asr_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_30s.wav" \
  -F "language=Chinese"

返回结果：

{
  "request_id": "1f3e982860bf4bd2",
  "language": "Chinese",
  "text": "本次会议主要确认边缘节点部署方案，并讨论后续接口联调计划。",
  "duration_sec": 30.216,
  "process_sec": 2.754,
  "rtf": 0.0911,
  "file_sha256": "1eb5635e9b3a..."
}

这里的RTF计算方式是：

RTF = 处理耗时 ÷ 音频时长

RTF小于1，表示处理速度快于音频播放速度。

踩坑提醒：不要启动多个Uvicorn Worker

下面的命令会让每个Worker都加载一份模型：

uvicorn asr_server:app \
  --host 0.0.0.0 \
  --port 7861 \
  --workers 4

结果通常不是吞吐变成四倍，而是显存被四份模型占满。

模型服务建议使用：

--workers 1

并通过应用内部队列、信号量或外部任务队列控制并发。

踩坑提醒：RTF不要写反

部分项目使用：

音频时长 ÷ 处理耗时

作为“实时倍速”，另一些项目使用：

处理耗时 ÷ 音频时长

作为RTF。

两种数值含义相反。日志、监控和汇报表必须统一口径，否则很容易把性能结果解释反。

八、长会议不要一次硬塞，分段后更容易恢复

官方模型支持长音频处理，但在边缘节点中，仍建议对长会议进行分段。这样可以降低峰值显存，限制单次失败范围，并支持从中断位置继续执行。

先按固定时长切分：

from pathlib import Path

from pydub import AudioSegment


def split_audio_with_overlap(
    audio_path: str,
    output_root: str = "./segments",
    chunk_ms: int = 180_000,
    overlap_ms: int = 3_000,
) -> list[dict]:
    source = Path(audio_path)
    output_dir = Path(output_root) / source.stem
    output_dir.mkdir(parents=True, exist_ok=True)

    audio = AudioSegment.from_file(source)
    segments: list[dict] = []

    start_ms = 0
    index = 0

    while start_ms < len(audio):
        end_ms = min(start_ms + chunk_ms, len(audio))
        segment_audio = audio[start_ms:end_ms]

        segment_path = (
            output_dir
            / f"{source.stem}_{index:04d}.wav"
        )

        segment_audio.export(
            segment_path,
            format="wav",
            parameters=[
                "-ar",
                "16000",
                "-ac",
                "1",
            ],
        )

        segments.append(
            {
                "index": index,
                "path": str(segment_path),
                "start_ms": start_ms,
                "end_ms": end_ms,
            }
        )

        if end_ms >= len(audio):
            break

        start_ms = end_ms - overlap_ms
        index += 1

    return segments

调用接口：

import json
from pathlib import Path

import requests

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


def transcribe_segment(
    segment: dict,
    language: str = "Chinese",
) -> dict:
    with open(segment["path"], "rb") as audio_file:
        response = requests.post(
            ASR_API,
            files={
                "file": (
                    Path(segment["path"]).name,
                    audio_file,
                    "audio/wav",
                )
            },
            data={"language": language},
            timeout=600,
        )

    response.raise_for_status()
    result = response.json()

    return {
        **segment,
        "language": result["language"],
        "text": result["text"],
        "process_sec": result["process_sec"],
        "rtf": result["rtf"],
    }


def transcribe_long_meeting(
    audio_path: str,
    checkpoint_path: str,
) -> list[dict]:
    checkpoint = Path(checkpoint_path)

    if checkpoint.exists():
        completed = json.loads(
            checkpoint.read_text(encoding="utf-8")
        )
    else:
        completed = []

    completed_indexes = {
        item["index"] for item in completed
    }

    segments = split_audio_with_overlap(audio_path)

    for segment in segments:
        if segment["index"] in completed_indexes:
            print(
                f"跳过已完成片段：{segment['index']}"
            )
            continue

        result = transcribe_segment(segment)
        completed.append(result)

        checkpoint.write_text(
            json.dumps(
                completed,
                ensure_ascii=False,
                indent=2,
            ),
            encoding="utf-8",
        )

        print(
            f"完成片段 {result['index']}，"
            f"RTF={result['rtf']}"
        )

    return sorted(
        completed,
        key=lambda item: item["index"],
    )

执行：

results = transcribe_long_meeting(
    "/data/meetings/project-review.wav",
    "/data/outputs/project-review.checkpoint.json",
)

这样即使第18段处理失败，程序重新运行时也可以从第18段继续，而不需要重新识别整场会议。

踩坑提醒：重叠分段会产生重复文字

相邻片段保留3秒重叠，可以减少句子被切断的问题，但也会让同一句话在两段结果里重复出现。

可以增加一个简单的文本合并函数：

from difflib import SequenceMatcher


def merge_with_overlap(
    previous: str,
    current: str,
    max_compare_chars: int = 120,
    min_match_chars: int = 8,
) -> str:
    if not previous:
        return current

    left = previous[-max_compare_chars:]
    right = current[:max_compare_chars]

    matcher = SequenceMatcher(
        None,
        left,
        right,
        autojunk=False,
    )
    match = matcher.find_longest_match(
        0,
        len(left),
        0,
        len(right),
    )

    if match.size < min_match_chars:
        return previous.rstrip() + "\n" + current.lstrip()

    cut_position = match.b + match.size

    return (
        previous.rstrip()
        + current[cut_position:].lstrip()
    )


def merge_all_segments(results: list[dict]) -> str:
    merged = ""

    for item in sorted(
        results,
        key=lambda value: value["index"],
    ):
        merged = merge_with_overlap(
            merged,
            item["text"],
        )

    return merged

这种字符级去重只能解决明显重复。涉及方言、同音词或标点变化时，还需要更稳定的语义合并策略。

九、接入熙瑾会悟：把听写稿变成会议结果

ASR接口返回的内容本质上仍然是一份听写稿。真实会议流程还需要继续提取会议主题、主要议题、关键决策、待办事项、责任人和风险点。

提交数据时，不建议只传一整段文本。更稳妥的方式是同时保留片段时间范围、识别语言、模型版本和处理记录：

from datetime import datetime

import requests

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


def submit_meeting_result(
    meeting_id: str,
    transcript: str,
    segments: list[dict],
) -> dict:
    payload = {
        "meeting_id": meeting_id,
        "meeting_title": "边缘节点部署评审会",
        "meeting_time": datetime.now().isoformat(),
        "source": "qwen3-asr-edge",
        "asr_model": "Qwen3-ASR-0.6B",
        "language": "Chinese",
        "transcript": transcript,
        "segments": [
            {
                "index": item["index"],
                "start_ms": item["start_ms"],
                "end_ms": item["end_ms"],
                "text": item["text"],
            }
            for item in segments
        ],
        "output_format": {
            "summary": True,
            "topics": True,
            "decisions": True,
            "todos": True,
            "risks": True,
        },
    }

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

    return response.json()

返回结构可以设计为：

{
  "summary": "会议确认采用边缘节点运行本地ASR，并完成音频预处理、接口隔离和离线模型管理。",
  "topics": [
    "边缘节点部署方式",
    "模型离线加载",
    "长会议分段策略",
    "会议数据安全"
  ],
  "decisions": [
    "ASR接口仅在本机和业务内网开放",
    "模型文件采用离线交付和哈希校验",
    "长会议采用分段转写和断点续跑"
  ],
  "todos": [
    {
      "owner": "运维组",
      "task": "完成边缘节点镜像制作",
      "deadline": "本周五"
    },
    {
      "owner": "测试组",
      "task": "验证两小时会议的转写稳定性",
      "deadline": "下周二"
    }
  ],
  "risks": [
    "边缘设备显存有限，需要限制并发",
    "不同录音设备的声道结构需要提前确认"
  ]
}

这一层完成后，系统才从“本地ASR服务”变成可进入日常流程的会议处理系统。

十、需要时间戳时，别忘了额外的 Forced Aligner

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",
    dtype=torch.bfloat16,
    device_map="cuda:0",
    max_inference_batch_size=1,
    max_new_tokens=512,
    forced_aligner=(
        "/data/models/Qwen3-ForcedAligner-0.6B"
    ),
    forced_aligner_kwargs={
        "dtype": torch.bfloat16,
        "device_map": "cuda:0",
    },
)

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

for result in results:
    print(result.language)
    print(result.text)

    for timestamp in result.time_stamps:
        print(timestamp)

踩坑提醒：加入时间戳模型后，显存不会保持原样

ASR模型和Forced Aligner会同时占用资源。原本只够运行0.6B ASR的边缘设备，增加对齐模型后可能发生OOM。

时间戳不是所有会议都必须实时生成。资源有限时，可以采用两阶段处理：

第一阶段：Qwen3-ASR-0.6B生成转写文本
第二阶段：会议结束后按需执行时间戳对齐

这样可以把有限算力优先用于会议转写。

十一、流式识别要换到 vLLM 后端

官方说明中，Qwen3-ASR的流式推理目前只在vLLM后端提供，而且流式模式不支持批量推理，也不返回时间戳。

安装：

pip install -U "qwen-asr[vllm]"

启动本地服务：

export CUDA_VISIBLE_DEVICES=0

qwen-asr-serve \
  /data/models/Qwen3-ASR-0.6B \
  --gpu-memory-utilization 0.65 \
  --host 127.0.0.1 \
  --port 8000

官方qwen-asr-serve是对vllm serve的封装，可以通过OpenAI风格的聊天接口提交音频请求。

踩坑提醒：不要把“离线识别”和“流式识别”当成同一种服务

离线识别通常接收一个完整文件，重点是准确率、长音频和批处理。

流式识别持续接收PCM音频块，重点是首字延迟、增量结果稳定性和断线恢复。

两种服务的接口、状态管理和容量规划都不同，不能只把上传接口改成WebSocket就认为完成了流式改造。

踩坑提醒：vLLM和依赖版本必须固定

边缘节点交付后，不建议无条件执行：

pip install -U vllm

更稳妥的方式是保存经过验证的依赖清单：

pip freeze > requirements-lock.txt

或将完整环境封装为容器镜像，并记录：

镜像名称
镜像摘要
CUDA版本
驱动版本
PyTorch版本
qwen-asr版本
vLLM版本
模型文件哈希

十二、容器化部署：默认只监听本机

新建Dockerfile：

FROM python:3.12-slim

ENV PYTHONDONTWRITEBYTECODE=1 \
    PYTHONUNBUFFERED=1 \
    HF_HUB_OFFLINE=1 \
    TRANSFORMERS_OFFLINE=1 \
    HF_HOME=/tmp/hf-cache

RUN apt-get update \
    && apt-get install -y --no-install-recommends \
       ffmpeg \
       ca-certificates \
    && rm -rf /var/lib/apt/lists/*

RUN useradd \
    --create-home \
    --uid 10001 \
    appuser

WORKDIR /app

COPY requirements.txt /app/requirements.txt

RUN pip install \
    --no-cache-dir \
    -r /app/requirements.txt

COPY asr_server.py /app/asr_server.py

USER appuser

EXPOSE 7861

CMD [
  "python",
  "-m",
  "uvicorn",
  "asr_server:app",
  "--host",
  "0.0.0.0",
  "--port",
  "7861",
  "--workers",
  "1"
]

requirements.txt：

qwen-asr
fastapi
uvicorn
python-multipart
pydantic
soundfile
pydub
requests

docker-compose.yml：

services:
  qwen3-asr-edge:
    build:
      context: .
    image: meeting/qwen3-asr-edge:0.6b
    container_name: qwen3-asr-edge
    restart: unless-stopped

    # 只绑定到宿主机回环地址。
    ports:
      - "127.0.0.1:7861:7861"

    volumes:
      - /data/models/Qwen3-ASR-0.6B:/models/asr:ro
      - /data/asr-tmp:/work/tmp

    environment:
      ASR_MODEL_PATH: /models/asr
      MAX_INFERENCE_BATCH_SIZE: "1"
      MAX_NEW_TOKENS: "1024"
      MAX_AUDIO_SECONDS: "14400"
      HF_HUB_OFFLINE: "1"
      TRANSFORMERS_OFFLINE: "1"
      TMPDIR: /work/tmp

    read_only: true

    tmpfs:
      - /tmp:size=512m,mode=1777

    cap_drop:
      - ALL

    security_opt:
      - no-new-privileges:true

    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities:
                - gpu

    healthcheck:
      test:
        - CMD
        - python
        - -c
        - >
          import urllib.request;
          urllib.request.urlopen(
          'http://127.0.0.1:7861/health',
          timeout=3)
      interval: 30s
      timeout: 5s
      retries: 3
      start_period: 120s

启动：

docker compose up -d --build

docker compose ps
docker logs -f qwen3-asr-edge

踩坑提醒：EXPOSE不等于端口已经安全

Dockerfile中的：

EXPOSE 7861

只是声明容器使用该端口，不会自动限制访问来源。

真正决定外部是否可访问的是Compose或docker run中的端口映射：

ports:
  - "127.0.0.1:7861:7861"

不要在没有网关和访问控制的情况下直接写成：

ports:
  - "7861:7861"

后者会把接口暴露到宿主机所有网卡。

十三、安全敏感部门边缘部署，重点不只是“模型在本地”

Qwen3-ASR-0.6B参数规模较小，模型和服务可以放到会议室边缘服务器运行。这种架构很适合不能将录音上传公网、网络隔离严格、会议数据需要留在本地的部门。

但真正的安全部署至少还应处理以下问题。

1. 接口边界

ASR服务只监听：

127.0.0.1

或专用业务网卡，不直接暴露到办公网和互联网。

跨服务器调用时，应通过内部网关、双向TLS、VPN或严格的访问控制列表限制来源。

2. 模型与依赖供应链

模型、Python包、系统镜像和容器镜像应提前下载并进行哈希校验。

正式运行环境不应允许服务自行访问模型仓库，也不应在启动时动态安装依赖。

3. 临时文件清理

上传的音频会先写入临时目录。接口无论成功还是失败，都必须执行清理。

还要考虑：

进程被强制终止
服务器突然断电
容器异常退出
磁盘空间不足

可以增加定期清理任务：

find /data/asr-tmp \
  -type f \
  -mmin +60 \
  -delete

4. 日志脱敏

不建议在日志中记录：

会议标题
参会人姓名
原始文件名
完整转写文本
录音存储路径
业务系统Token

更适合记录：

request_id
音频时长
处理耗时
RTF
模型版本
文件哈希前缀
错误类型

5. 数据生命周期

会议原始音频、转写文本、纪要和临时分段文件，应分别设置保存期限和删除策略。

不能因为ASR部署在本地，就默认所有中间文件都可以永久保留。

6. 合规边界

边缘部署解决的是技术上的“本地处理”和“减少数据外流”。

能否用于具体涉密等级，还取决于整机、操作系统、数据库、密码设备、网络环境、运维流程和管理制度。模型本身不能替代涉密信息系统测评、审批和认证。

十四、常见报错与排查顺序

1. CUDA out of memory

先降低：

export MAX_INFERENCE_BATCH_SIZE=1
export MAX_NEW_TOKENS=512

确认没有多Worker：

ps -ef | grep '[u]vicorn'
nvidia-smi

如果同时加载了Forced Aligner，先关闭时间戳功能再验证。

2. undefined symbol

通常是Torch、CUDA、FlashAttention或其他二进制扩展版本不一致。

检查：

python -c "import torch; print(torch.__version__)"
python -c "import torch; print(torch.version.cuda)"
pip show flash-attn
nvidia-smi

不要在旧环境中反复覆盖安装不同版本的Torch。重新创建干净环境通常更可靠。

3. FlashAttention安装失败

官方建议FlashAttention用于降低显存占用并加速长输入和大批量推理，但它要求硬件兼容，并且模型使用float16或bfloat16。

内存不足时可限制编译任务：

MAX_JOBS=4 \
pip install -U flash-attn \
  --no-build-isolation

边缘设备不支持时，可以先不安装，使用默认注意力实现验证功能，不要把FlashAttention作为服务启动的强制依赖。

4. 中文识别结果为空或异常短

按顺序检查：

音频是否真的有人声
采样率和声道是否正确
文件是否损坏
max_new_tokens是否过小
language参数是否设置错误
音频是否包含超长静音

先转成标准WAV再次测试：

ffmpeg -y \
  -i input_audio \
  -ar 16000 \
  -ac 1 \
  -c:a pcm_s16le \
  debug.wav

5. 接口偶尔卡死

检查是否同时有多个请求进入模型。

边缘服务应默认串行推理。需要排队时，可以在API前增加Redis队列，或者让业务系统只提交任务，不同步等待整场会议完成。

6. 磁盘逐渐被占满

检查：

du -sh /data/asr-tmp
find /data/asr-tmp -type f | head
docker system df

很多ASR服务不是显存先出问题，而是临时音频和分段文件没有清理，最终把系统盘写满。

十五、总结：0.6B的价值在于把ASR推到会议现场

Qwen3-ASR-0.6B的意义，不只是参数量比1.7B更小，而是让会议转写更容易从中心GPU集群下沉到边缘节点。

会悟则在ASR之后继续处理会议内容，把原始听写稿整理成摘要、议题、决策、待办和可检索资料。

对于安全敏感部门，这套架构最大的价值，是让原始会议音频可以留在会议室设备或内部边缘节点，不必依赖公网语音服务。同时，轻量模型也更便于在算力、空间和功耗受限的设备中运行。