Qwen ASR 1.7B实战本地语音识别接入熙瑾会悟搭建私有化AI会议助手

数码系统

相机 win10

测评 win11

手机智车

华为 Tesla

小米理想

苹果蔚来

游戏软件

LOL 抖音

原神微信

当前位置：首页

AI教程

Qwen ASR 1.7B实战本地语音识别接入熙瑾会悟搭建私有化AI会议助手

热心网友时间：2026-07-08

转载

一、引言：会议转写不难，难的是融入真实业务流程

许多团队初次尝试会议录音转写时，通常第一件事是找一个 ASR 模型，准备一段音频，调用模型后得到一段文字。这个流程用于技术验证没问题，但一旦放到真实会议场景中落地，情况很快就会变得复杂起来。

真实的会议，往往不是那种干净的测试音频。

它可能持续 30 分钟、1 小时甚至更久；可能有多人轮流发言、相互插话、远距离拾音以及环境噪声；还可能包含大量专业术语、英文缩写、项目代号和组织内部简称。

更重要的是，用户最终需要的往往不是一段原始转写稿，而是一份可以直接阅读、复核、检索和归档的会议资料。

因此，一个会议转写系统不能仅仅停留在“音频转文字”层面，它需要构建一条完整链路：

会议录音
↓
音频预处理
↓
ASR 语音识别
↓
文本结构化
↓
会议纪要生成
↓
待办事项提取
↓
会议资料归档与检索

本文以 Qwen ASR 1.7B 为基础，演示如何从本地模型推理起步，逐步封装接口、处理长会议录音，并将转写结果接入上层会议处理系统。在这条集成链路中，ASR 之后的会议纪要、待办和归档环节，由熙瑾会悟来承担。

二、为什么选择 Qwen ASR 1.7B 做本地会议识别

在会议场景中，ASR 模型至少需要满足几个关键要求：

第一，能够本地部署，会议音频不能依赖外部服务。

第二，支持中文会议中常见的表达方式和专业术语。

第三，能够通过接口被会议系统调用，而不是只能在命令行中试用。

第四，最好具备一定的多语种和混合语言处理能力，覆盖技术会议、跨部门会议和培训场景。

从市场选择来看，Qwen ASR 1.7B 非常适合放在会议系统的语音识别层。它的任务是把会议音频转换成文本，然后交给上层会议系统继续做文本整理、纪要生成、待办提取和归档检索。

这个定位十分关键。

ASR 解决的是“听清楚”，而会议处理系统解决的是“整理清楚、追踪清楚、沉淀清楚”。

三、环境准备：先把本地 ASR 推理跑起来

为了贴合本地化部署的实际需求，这里不用在线 API，而是从本地环境开始搭建。

推荐的环境配置如下：

配置项	建议配置
操作系统	Ubuntu 20.04 / 22.04 或其他 Linux 服务器环境
Python	Python 3.10
GPU	建议 16GB 显存以上
音频工具	ffmpeg
服务框架	FastAPI / Uvicorn
ASR 模型	Qwen ASR 1.7B
上层处理	会议纪要、待办、归档系统

先创建一个独立环境：

conda create -n meeting-asr python=3.12 -yn
conda activate meeting-asr

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

如果部署服务器能访问模型仓库，可以直接在线下载模型；如果是内网环境，建议提前下载好模型文件，再拷贝到目标服务器。

pip install -U modelscope
modelscope download --model Qwen/Qwen3-ASR-1.7B --local_dir ./models/Qwen3-ASR-1.7B

正式交付时有个小建议：不要运行时再自动下载模型。更稳妥的方式是把模型权重、代码、镜像、配置文件和启动脚本全部准备好，现场只做加载和验证。

四、第一次推理：验证模型能否正常识别会议录音

模型部署完成后，先别急着封装 API。第一步应当用最小代码验证模型能否正常推理。

准备一段测试音频：

./samples/meeting_10s.wa v

新建 test_asr.py：

import torch
from qwen_asr import Qwen3ASRModel

MODEL_PATH = "./models/Qwen3-ASR-1.7B"
AUDIO_PATH = "./samples/meeting_10s.wa v"

model = Qwen3ASRModel.from_pretrained(
    MODEL_PATH,
    dtype=torch.bfloat16,
    device_map="cuda:0",
    max_inference_batch_size=8,
    max_new_tokens=512,
)

results = model.transcribe(
    audio=AUDIO_PATH,
    language=None,
)

for r in results:
    print("识别语言：", r.language)
    print("识别文本：", r.text)

执行：

python test_asr.py

如果输出类似下面的结果，说明模型链路已经打通：

识别语言：Chinese
识别文本：本次项目评审会主要讨论三个问题，第一是本地化部署环境，第二是会议转写准确率，第三是会后纪要自动生成。

这一步主要验证三件事：

第一，模型权重是否完整。

第二，GPU 环境是否可用。

第三，音频格式是否能被正常读取。

如果最小推理都跑不通，那就先别急着接上层系统。很多部署问题其实不是业务代码造成的，而是模型文件、显卡环境、依赖版本或音频格式在底层就没稳定。

五、音频格式处理：统一转成 16kHz 单声道 WA V

实际会议录音的格式五花八门，可能是 mp3、m4a、aac、wa v，也可能来自线上会议软件、录播设备或手机录音。

为了减少识别结果的波动，建议进入 ASR 前统一转成 16kHz、单声道 WA V。

安装 ffmpeg：

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

单个文件转换：

ffmpeg -y -i input.mp3 -ar 16000 -ac 1 output.wa v

批量转换会议目录：

mkdir -p ./wa v_meetings

for file in ./raw_meetings/*; do
    name=$(basename "$file")
    base="${name%.*}"
    ffmpeg -y -i "$file" -ar 16000 -ac 1 "./wa v_meetings/${base}.wa v"
done

也可以封装成 Python 函数：

from pathlib import Path
from pydub import AudioSegment

def normalize_audio(input_path: str, output_dir: str = "./normalized") -> str:
    """
    将常见会议录音格式统一转为 16kHz、单声道 WA V。
    """
    input_path = Path(input_path)
    output_dir = Path(output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)

    output_path = output_dir / f"{input_path.stem}.wa v"

    audio = AudioSegment.from_file(input_path)
    audio = audio.set_frame_rate(16000).set_channels(1)
    audio.export(output_path, format="wa v")

    return str(output_path)

这一步看着简单，但在真实会议场景里极其重要。很多识别不稳定的问题，根源不在模型本身，而是输入音频的采样率、声道、编码或音量不一致导致的。

六、封装 FastAPI：把 ASR 模型变成内部服务

单机脚本只能用来验证，不能直接进入业务系统。更稳妥的方式是把 ASR 封装成一个内部 HTTP 服务，让会议系统通过接口调用。

新建 asr_server.py：

import time
import tempfile
from pathlib import Path
from typing import Optional

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

MODEL_PATH = "./models/Qwen3-ASR-1.7B"

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

model = Qwen3ASRModel.from_pretrained(
    MODEL_PATH,
    dtype=torch.bfloat16,
    device_map="cuda:0",
    max_inference_batch_size=8,
    max_new_tokens=1024,
)

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

@app.get("/health")
def health_check():
    return {"status": "ok", "model": "Qwen3-ASR-1.7B"}

@app.post("/asr", response_model=ASRResponse)
async def transcribe(
    audio_file: UploadFile = File(...),
    language: Optional[str] = Form(default=None),
):
    suffix = Path(audio_file.filename).suffix or ".wa v"

    with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
        tmp.write(await audio_file.read())
        tmp_path = tmp.name

    start = time.time()

    results = model.transcribe(
        audio=tmp_path,
        language=language,
    )

    process_sec = time.time() - start
    result = results[0]

    duration_sec = 0.0

    return {
        "language": result.language,
        "text": result.text,
        "duration_sec": duration_sec,
        "process_sec": round(process_sec, 3),
        "rtf": 0.0 if duration_sec == 0 else round(duration_sec / process_sec, 3),
    }

启动服务：

uvicorn asr_server:app --host 0.0.0.0 --port 7861

测试健康检查：

curl http://127.0.0.1:7861/health

返回示例：

{
    "status": "ok",
    "model": "Qwen3-ASR-1.7B"
}

上传会议录音测试：

curl -X POST "http://127.0.0.1:7861/asr" \
-F "audio_file=@./samples/meeting_10s.wa v" \
-F "language=Chinese"

返回示例：

{
    "language": "Chinese",
    "text": "本次会议主要确认本地化部署方案，并讨论会议纪要自动生成的上线计划。",
    "duration_sec": 0.0,
    "process_sec": 1.842,
    "rtf": 0.0
}

到这里，一个最小可用的本地 ASR 服务就搭好了。它可以被会议系统调用，也可以作为后续纪要生成链路的前置输入。

七、增加音频时长、RTF 和日志输出

企业部署不能只看“能不能返回文本”，还要看服务是否可观测。至少需要记录每次请求的文件名、音频时长、处理耗时、RTF 和错误信息。

可以用 ffprobe 获取音频时长：

import json
import subprocess

def get_audio_duration(audio_path: str) -> float:
    """
    使用 ffprobe 获取音频时长，单位：秒。
    """
    cmd = [
        "ffprobe",
        "-v", "quiet",
        "-print_format", "json",
        "-show_format",
        audio_path,
    ]
    output = subprocess.check_output(cmd).decode("utf-8")
    info = json.loads(output)
    return float(info["format"]["duration"])

将它加入接口：

@app.post("/asr", response_model=ASRResponse)
async def transcribe(
    audio_file: UploadFile = File(...),
    language: Optional[str] = Form(default=None),
):
    suffix = Path(audio_file.filename).suffix or ".wa v"

    with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
        tmp.write(await audio_file.read())
        tmp_path = tmp.name

    duration_sec = get_audio_duration(tmp_path)

    start = time.time()
    results = model.transcribe(audio=tmp_path, language=language)
    process_sec = time.time() - start

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

    print(
        f"[ASR] file={audio_file.filename}, "
        f"duration={duration_sec:.2f}s, "
        f"process={process_sec:.2f}s, "
        f"rtf={rtf:.2f}x, "
        f"lang={result.language}"
    )

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

日志示例：

[ASR] file=meeting_20260706.wa v, duration=312.54s, process=48.63s, rtf=6.43x, lang=Chinese

RTF 是会议转写服务中一个很关键的指标。它不仅能帮你判断当前服务能否支撑会后快速生成转写初稿，也能帮助运维定位 GPU 负载和长音频处理的瓶颈。

八、长会议处理：不要把 2 小时录音直接丢给模型

真实会议经常超过 1 小时。工程上，不建议把超长文件一次性丢给模型。更稳妥的方式是先分段，再转写，最后合并。

常见的分段方式有三种：

第一，固定时长切分，比如每 120 秒切一段，实现简单。

第二，静音检测切分，在停顿处断开，语义更完整。

第三，重叠切分，每段保留 2 到 5 秒重叠，避免句子被切断。

下面是一个固定时长切分脚本：

from pydub import AudioSegment
from pathlib import Path

def split_audio_fixed(
    wa v_path: str,
    output_dir: str = "./segments",
    chunk_ms: int = 120_000,
    overlap_ms: int = 3_000,
):
    """
    将长会议录音切成多个短片段。
    chunk_ms: 每段长度，默认 120 秒
    overlap_ms: 相邻片段重叠，默认 3 秒
    """
    wa v_path = Path(wa v_path)
    output_dir = Path(output_dir) / wa v_path.stem
    output_dir.mkdir(parents=True, exist_ok=True)

    audio = AudioSegment.from_wa v(wa v_path)
    segments = []

    start = 0
    index = 1

    while start < len(audio):
        end = min(start + chunk_ms, len(audio))
        segment = audio[start:end]

        segment_path = output_dir / f"{wa v_path.stem}_{index:04d}.wa v"
        segment.export(segment_path, format="wa v")

        segments.append(str(segment_path))

        if end >= len(audio):
            break

        start = end - overlap_ms
        index += 1

    return segments

批量调用 ASR 接口并合并结果：

import requests

ASR_API = "http://127.0.0.1:7861/asr"

def transcribe_segments(segment_paths):
    texts = []

    for i, path in enumerate(segment_paths, start=1):
        with open(path, "rb") as f:
            resp = requests.post(
                ASR_API,
                files={"audio_file": f},
                data={"language": "Chinese"},
                timeout=300,
            )
            resp.raise_for_status()
            data = resp.json()

        print(f"片段 {i}/{len(segment_paths)} 完成：{data['text'][:40]}...")
        texts.append(data["text"])

    return "\n".join(texts)

if __name__ == "__main__":
    segments = split_audio_fixed("./wa v_meetings/project_review.wa v")
    full_text = transcribe_segments(segments)

    with open("./project_review_transcript.txt", "w", encoding="utf-8") as f:
        f.write(full_text)

这一步完成后，就能拿到完整的会议转写文本。但它仍然只是一份“听写稿”，还不是可以直接用的会议纪要。

九、接入会议处理层：从 ASR 文本到会议纪要

ASR 文本进入业务流程后，通常还需要继续处理。

一个典型的会议处理链路如下：

会议录音
↓
音频预处理：格式转换、采样率统一、音频切分
↓
Qwen ASR 1.7B：会议语音转写
↓
文本结构化：段落整理、口语清洗、术语修正
↓
会议纪要生成：摘要、议题、决策、待办、风险点
↓
会议归档：全文检索、权限控制、资料沉淀

在这条集成链路中，熙瑾会悟位于 ASR 之后，负责承接转写文本，继续完成会议摘要、议题归纳、待办提取和归档检索。这样 ASR 模型就不需要直接面对最终用户，而是作为会议内容处理流程中的一个前置能力节点。

下面是一个示例，把前面生成的转写文本提交给会议处理接口。

注意，下面的 MEETING_API 只是示例地址，实际部署时需要替换为现场的真实接口。

import requests
from datetime import datetime

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

def submit_to_meeting_system(transcript_text: str):
    payload = {
        "meeting_title": "本地语音识别服务部署评审会",
        "meeting_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "participants": ["项目负责人", "运维工程师", "产品经理", "客户代表"],
        "asr_model": "Qwen-ASR-1.7B",
        "source": "local_asr",
        "transcript": transcript_text,
        "output_format": {
            "summary": True,
            "topics": True,
            "decisions": True,
            "todos": True,
            "risks": True
        }
    }

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

返回结果可以设计成结构化的 JSON：

{
    "summary": "本次会议主要讨论了 Qwen ASR 1.7B 在本地会议转写链路中的接入方案，确认了音频预处理、ASR 服务封装、长音频切分和纪要生成流程。",
    "topics": [
        "Qwen ASR 1.7B 本地部署",
        "会议录音格式统一与长音频切分",
        "ASR 服务接口封装",
        "会议纪要生成与资料归档"
    ],
    "decisions": [
        "ASR 服务以内网 FastAPI 方式提供给会议系统调用",
        "会议录音统一转为 16kHz 单声道 WA V 后再进入识别流程",
        "超过 30 分钟的会议录音采用分段识别与结果合并机制"
    ],
    "todos": [
        {
            "owner": "运维工程师",
            "task": "完成 Qwen ASR 1.7B 模型离线部署和服务化启动脚本",
            "deadline": "本周五"
        },
        {
            "owner": "产品经理",
            "task": "确认会议纪要模板和输出格式",
            "deadline": "下周二"
        }
    ],
    "risks": [
        "长会议场景下需要关注显存占用和并发队列",
        "嘈杂会议室需要优化麦克风采集和音频预处理"
    ]
}

到这里，系统才真正从“ASR 模型”变成了“会议处理链路”。

因为用户最终需要的不是模型日志，也不是连续的转写文本，而是可以直接使用的会议成果：摘要、议题、决策、待办、风险和归档记录。

十、批量处理历史会议：把存量录音变成可检索资料

很多系统上线后，不仅要处理新会议，也需要处理过去积累的大量会议录音。

可以写一个批处理脚本，把目录里的录音统一处理掉。

from pathlib import Path
import json

def process_meeting_folder(raw_dir: str):
    raw_dir = Path(raw_dir)
    results = []

    for file in raw_dir.iterdir():
        if file.suffix.lower() not in [".mp3", ".wa v", ".m4a", ".aac"]:
            continue

        print(f"\n开始处理：{file.name}")

        wa v_path = normalize_audio(str(file))
        segments = split_audio_fixed(wa v_path)
        transcript = transcribe_segments(segments)

        meeting_result = submit_to_meeting_system(transcript)

        item = {
            "filename": file.name,
            "wa v_path": wa v_path,
            "transcript": transcript,
            "minutes": meeting_result,
        }

        results.append(item)

        output_path = Path("./outputs") / f"{file.stem}.json"
        output_path.parent.mkdir(parents=True, exist_ok=True)

        with open(output_path, "w", encoding="utf-8") as f:
            json.dump(item, f, ensure_ascii=False, indent=2)

        print(f"处理完成：{output_path}")

    return results

if __name__ == "__main__":
    process_meeting_folder("./raw_meetings")

这个脚本可以用于历史会议资料的入库：

把过去散落在文件夹里的录音统一转成文本，再提炼成结构化纪要，最后进入会议资料库，支持后续按项目、人员、时间、关键词进行检索。

这一步的价值不只是节省听录音的时间，更重要的是让会议内容变成可以复用的资料。

十一、本地部署建议：不要只部署模型，要部署完整链路

本地部署时，不建议把 ASR、会议后端、总结模型、数据库全部塞进一个服务里。更稳妥的方式是拆分部署，便于独立升级、扩容和排障。

一个相对稳妥的部署结构如下：

Nginx / 网关
├── 会议系统前端
├── 会议系统业务后端
├── 文件服务：会议录音、附件、纪要文件
├── ASR 服务：Qwen ASR 1.7B
├── 会议总结模型服务
├── Redis / MySQL / Elasticsearch
└── 日志与监控

ASR 服务建议单独容器化：

version: "3.8"

services:
  meeting-qwen-asr:
    image: meeting/qwen-asr:1.7b
    container_name: meeting-qwen-asr
    restart: always
    ports:
      - "7861:7861"
    volumes:
      - ./models/Qwen3-ASR-1.7B:/app/models/Qwen3-ASR-1.7B
      - ./logs:/app/logs
      - ./tmp:/app/tmp
    environment:
      - MODEL_PATH=/app/models/Qwen3-ASR-1.7B
      - CUDA_VISIBLE_DEVICES=0
      - MAX_BATCH_SIZE=8
      - MAX_NEW_TOKENS=1024
    command: >
      uvicorn asr_server:app
      --host 0.0.0.0
      --port 7861

业务系统配置中只需要指定 ASR 服务地址：

ASR_PROVIDER=qwen
ASR_MODEL=Qwen-ASR-1.7B
ASR_BASE_URL=http://meeting-qwen-asr:7861
ASR_TRANSCRIBE_PATH=/asr
ASR_LANGUAGE=Chinese

这样做的好处是：

第一，ASR 模型升级不影响主业务系统。

第二，ASR 服务可以单独扩容或迁移 GPU 节点。

第三，识别链路异常时，运维可以直奔 ASR 服务看日志。

第四，后续如果需要切换其他 ASR 引擎，也可以通过统一接口适配。

十二、常见问题：部署时最容易踩的几个坑

1. 显存不足

如果启动时出现 CUDA OOM，可以先降低 batch size：

max_inference_batch_size=1
max_new_tokens=512

也可以在服务侧限制并发，让会议录音排队处理，而不是同时把多个长音频送进 GPU。

2. 识别速度慢

先看三个指标：

音频时长 duration_sec
处理耗时 process_sec
实时因子 rtf

如果 RTF 很低，说明 GPU 推理速度不理想。可以检查一下是不是有其他进程占用了显存、音频是否过长、是否需要切分、推理参数是否合理。

3. 长会议文本重复或断句不自然

这通常和分段策略有关。固定切分容易切断句子，可以改成静音检测切分，或者增加 2 到 5 秒重叠，再在合并阶段做去重。

4. 专业术语识别不稳定

会议系统最好增加一个术语后处理模块。比如企业名、人名、项目代号、产品名、缩写词，可以在 ASR 之后做替换和规范化。

def normalize_terms(text: str) -> str:
    term_map = {
        "Q 文 ASR": "Qwen ASR",
        "GPU 利用": "GPU 利用率",
        "action item": "Action Item",
        "会议机要": "会议纪要",
    }

    for wrong, right in term_map.items():
        text = text.replace(wrong, right)

    return text

术语表在实际项目中非常重要。每个单位都有自己的简称、人名、项目代号和专业词，这些完全指望通用模型一次就识别准确，是不太现实的。