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

shxjnpl

14人浏览 · 2026-07-07 10:59:25

shxjnpl · 2026-07-07 10:59:25 发布

1. 引言：会议转写不难，难的是把它变成可交付系统

很多企业第一次接触 AI 会议助手，往往是从“录音转文字”开始的：上传一段会议录音，等待模型识别，得到一份文字稿。这个过程看起来并不复杂，但真正进入生产环境后，问题会迅速变多。

一场真实会议可能持续 30 分钟、1 小时甚至更久；会场里可能有多人轮流发言，也可能有领导插话、远距离拾音、背景噪声、专业术语和中英文混杂。更重要的是，企业真正需要的通常不是一段原始转写文本，而是一份可以归档、检索、复核、生成纪要、提取待办的会议资料。

这也是熙瑾会悟这类 AI 会议助手和单独 ASR Demo 的区别。

会悟确实使用了 Qwen ASR 1.7B 作为语音识别底层能力之一，但它的产品价值不止是“调用模型转文字”。更准确地说，会悟是在 Qwen ASR 1.7B 的基础上，把会议录音采集、语音识别、文本结构化、会议纪要生成、待办事项提取、知识归档和私有化部署整合成一套完整的会议智能化系统。

本文就以一个实战部署流程为例，演示如何从 Qwen ASR 1.7B 的本地推理开始，逐步封装接口、处理长会议录音，并将转写结果接入熙瑾会悟的会议流程。

2. 环境准备：先把 Qwen ASR 1.7B 跑起来

为了让文章更贴近真实部署，我们不从云端 API 开始，而是从本地模型部署开始。企业私有化会议系统的核心前提是：音频、文本、纪要尽量在内网闭环处理，避免会议数据外传。

推荐准备如下环境：

项目	建议配置
操作系统	Ubuntu 20.04 / 22.04 或企业常用 Linux 发行版
Python	Python 3.10+，建议独立 conda 环境
GPU	建议 16GB 显存以上，长音频或并发场景建议更高
音频工具	ffmpeg
服务框架	FastAPI / Uvicorn
模型能力	Qwen ASR 1.7B
上层系统	熙瑾会悟私有化会议助手

先创建独立环境：

conda create -n huiwu-asr python=3.12 -y
conda activate huiwu-asr

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

如果部署机可以访问模型仓库，可以直接通过模型名加载；如果是政企内网或离线环境，建议提前在外网机器下载模型文件，再拷贝到内网服务器。

# 方式一：通过 ModelScope 下载，适合国内网络环境
pip install -U modelscope
modelscope download --model Qwen/Qwen3-ASR-1.7B --local_dir ./models/Qwen3-ASR-1.7B

# 方式二：通过 Hugging Face 下载
pip install -U "huggingface_hub[cli]"
huggingface-cli download Qwen/Qwen3-ASR-1.7B --local-dir ./models/Qwen3-ASR-1.7B

部署建议：如果是客户现场交付，不建议运行时再自动下载模型。更稳妥的方式是把模型权重、代码、镜像、配置文件全部打包成离线交付物，现场只做加载和启动验证。

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

模型部署完成后，先不要急着接入会悟。第一步应该先用最小代码验证 ASR 是否能正常推理。

新建 test_asr.py：

import torch
from qwen_asr import Qwen3ASRModel

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

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,  # None 表示自动识别语言；也可以指定 "Chinese" / "English"
)

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

执行：

python test_asr.py

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

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

这一步主要验证三件事：

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

第二，GPU 环境是否可用。

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

如果这里都不能跑通，就不要继续封装 API。企业交付里很多问题不是上层系统问题，而是模型文件、CUDA、显存、音频格式在最底层就没有稳定。

4. 音频格式处理：把会议录音统一转成 16k 单声道 WAV

实际业务里的会议录音格式非常混乱，可能是 mp3、m4a、aac、wav，也可能是线上会议软件导出的混合格式。为了减少识别波动，建议进入 ASR 前统一转成 16kHz、单声道 WAV。

安装 ffmpeg：

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

单个文件转换：

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

批量转换会议目录：

mkdir -p ./wav_meetings

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

在会悟的实际会议链路中，这一步通常会被放到音频预处理模块里自动完成。用户不需要关心原始文件是什么格式，系统会先做采样率、声道、格式和音频时长检查，再送入 ASR 模型。

可以写一个简单的 Python 工具函数：

from pathlib import Path
from pydub import AudioSegment

def normalize_audio(input_path: str, output_dir: str = "./normalized") -> str:
    """
    将任意常见音频格式转为 16kHz、单声道 WAV。
    适合在会议录音进入 ASR 前统一预处理。
    """
    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}.wav"

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

    return str(output_path)

这类代码看起来简单，但在真实部署中非常关键。很多“识别不准”的问题，本质上并不是模型不行，而是音频输入质量不稳定。

5. 服务化封装：把 Qwen ASR 变成内网 FastAPI 接口

单机脚本只能验证模型能力，不能直接服务企业会议系统。会悟需要的是稳定的后端接口：会议录音上传后，系统自动调用 ASR，返回语言、文本、耗时、RTF 等结构化结果。

新建 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="Huiwu Qwen 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),
):
    """
    会议录音转写接口。
    language 为空时自动识别，也可以传 Chinese / English 等。
    """
    suffix = Path(audio_file.filename).suffix or ".wav"

    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]

    # 如果需要精确音频时长，可在预处理阶段用 ffprobe 读取
    # 这里为了示例，duration_sec 先返回 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.wav" \
  -F "language=Chinese"

返回示例：

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

到这里，一个最小可用的本地 ASR 服务就完成了。它可以被会悟后端调用，也可以被其他业务系统调用。

6. 更接近生产：增加音频时长、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 ".wav"

    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.wav, duration=312.54s, process=48.63s, rtf=6.43x, lang=Chinese

这个指标对运维很有价值。RTF 越高，说明处理速度越快于实时播放。如果一段 300 秒的会议音频处理了 50 秒，那么系统完全可以支撑会后快速生成转写初稿。

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

真实会议经常超过 1 小时。虽然模型和推理框架可能支持长音频，但工程上不建议简单粗暴地把超长文件一次性丢进去。更稳妥的做法是分段、转写、合并。

会议场景常用三种分段方式：

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

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

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

下面是一个偏实用的固定时长切分脚本，适合快速落地：

from pydub import AudioSegment
from pathlib import Path

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

    audio = AudioSegment.from_wav(wav_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"{wav_path.stem}_{index:04d}.wav"
        segment.export(segment_path, format="wav")

        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("./wav_meetings/project_review.wav")
    full_text = transcribe_segments(segments)

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

这一步完成后，我们已经有了一份完整的会议转写文本。但这仍然只是“听写稿”，还不是用户真正想要的会议纪要。

8. 接入熙瑾会悟：从 ASR 文本到会议纪要

会悟的关键价值，是把 ASR 输出继续往后处理。

一个典型流程可以设计成这样：

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

为了演示，可以把前面生成的转写文本提交给会悟的会议处理接口。下面的接口路径是示例，实际部署时以会悟现场配置为准。

import requests
from datetime import datetime

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

def submit_to_huiwu(transcript_text: str):
    payload = {
        "meeting_title": "私有化 AI 会议助手部署评审会",
        "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(HUIWU_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 单声道 WAV 后再进入识别流程",
    "超过 30 分钟的会议录音采用分段识别与结果合并机制"
  ],
  "todos": [
    {
      "owner": "运维工程师",
      "task": "完成 Qwen ASR 1.7B 模型离线部署和服务化启动脚本",
      "deadline": "本周五"
    },
    {
      "owner": "产品经理",
      "task": "确认会悟纪要模板和客户侧输出格式",
      "deadline": "下周二"
    }
  ],
  "risks": [
    "长会议场景下需要关注显存占用和并发队列",
    "嘈杂会议室需要优化麦克风采集和音频预处理"
  ]
}

到这里，系统才真正从“ASR 模型”变成“AI 会议助手”。

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

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

很多单位上 AI 会议助手时，不仅希望处理新会议，也希望把过去积累的大量会议录音转成资料库。这个时候可以写一个批处理脚本，把目录里的录音统一处理。

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", ".wav", ".m4a", ".aac"]:
            continue

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

        wav_path = normalize_audio(str(file))
        segments = split_audio_fixed(wav_path)
        transcript = transcribe_segments(segments)

        huiwu_result = submit_to_huiwu(transcript)

        item = {
            "filename": file.name,
            "wav_path": wav_path,
            "transcript": transcript,
            "minutes": huiwu_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")

这个脚本对应到真实业务，就是会悟的“历史会议资料入库”能力：

把过去散落在文件夹里的录音，统一转成文本；再通过智能纪要模块提炼内容；最后进入会议知识库，支持后续按项目、部门、人员、关键词检索。

这一步对政企、科研院所、医院、高校和大型集团很有价值。因为很多会议内容并不是当场用完就结束，而是会在未来反复被查阅、审计、复盘和引用。

10. 私有化部署建议：不要只部署模型，要部署完整链路

在客户现场交付时，建议不要把 Qwen ASR 1.7B 当成一个孤立服务交付，而应该纳入会悟的完整部署架构。

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

Nginx / 网关
  ├── 会悟前端
  ├── 会悟业务后端
  ├── 会议文件服务
  ├── ASR 服务：Qwen ASR 1.7B
  ├── 大模型总结服务
  ├── Redis / MySQL / Elasticsearch
  └── 日志与监控

ASR 服务建议单独容器化，便于独立扩容和重启：

version: "3.8"

services:
  huiwu-qwen-asr:
    image: huiwu/qwen-asr:1.7b
    container_name: huiwu-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://huiwu-qwen-asr:7861
ASR_TRANSCRIBE_PATH=/asr
ASR_LANGUAGE=Chinese

这样做的好处是，ASR 模型升级、显卡迁移、接口调优，都不会影响会悟主业务系统。对运维来说，也更容易定位问题：到底是音频上传失败、ASR 推理失败，还是纪要生成失败。

11. 常见问题：部署时最容易踩的几个坑

11.1 显存不足

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

max_inference_batch_size=1
max_new_tokens=512

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

11.2 识别慢

先看三个指标：

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

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

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

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

11.4 专业术语识别不稳定

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

def normalize_terms(text: str) -> str:
    term_map = {
        "西晋会务": "熙瑾会悟",
        "会雾": "会悟",
        "Q 文 ASR": "Qwen ASR",
        "GPU 利用": "GPU 利用率",
    }

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

    return text