Qwen3-ASR-1.7B实战教程：识别结果API对接→企业微信/钉钉消息自动推送

1. 引言：从语音识别到自动化通知

想象一下这个场景：你正在开会，手机静音。会后，你收到一条企业微信消息，里面是刚才会议录音的完整文字纪要。或者，你的客服团队收到一条钉钉通知，内容是客户电话录音中反馈的关键问题。这听起来像是未来科技，但其实用Qwen3-ASR-1.7B语音识别模型，加上一点简单的代码，今天就能实现。

Qwen3-ASR-1.7B是阿里云通义千问团队推出的开源语音识别模型，它最大的特点就是“准”。支持52种语言和方言，1.7B的参数量让它比小模型识别得更精准，尤其是在有背景噪音或者口音的情况下。但模型本身只是一个工具，它的价值在于如何被使用。

本教程要做的，就是帮你把这个强大的工具“用活”。我们不只讲怎么把语音转成文字，更要讲怎么让这些文字自动“跑”到你需要的地方去——比如企业微信或钉钉的工作群。整个过程，我会手把手带你走一遍，从搭建环境、调用API，到编写自动推送脚本，最后实现一个完整的自动化流程。即使你之前没怎么接触过API对接，跟着步骤走也能搞定。

2. 环境准备与快速部署

在开始写代码之前，我们得先把Qwen3-ASR-1.7B模型服务跑起来。这里假设你已经通过CSDN星图镜像广场部署好了对应的镜像，并获得了访问地址。

2.1 确认服务状态

首先，通过SSH连接到你的服务器，检查语音识别服务是否正常运行。

# 查看服务状态
supervisorctl status qwen3-asr

# 预期看到类似输出：
# qwen3-asr                       RUNNING   pid 12345, uptime 1:23:45

如果服务没有运行，使用以下命令启动它：

# 重启服务
supervisorctl restart qwen3-asr

# 等待几秒后，再次检查状态
supervisorctl status qwen3-asr

2.2 获取Web界面访问地址

服务运行后，你会有一个Web访问地址，格式通常是：

https://gpu-{你的实例ID}-7860.web.gpu.csdn.net/

在浏览器中打开这个地址，你应该能看到一个简洁的上传界面。可以上传一个MP3或WAV文件测试一下，确保“开始识别”按钮能正常工作，并返回转写文本。这一步的成功，证明模型服务的基础API是通畅的，为我们后续的编程调用打下了基础。

2.3 安装必要的Python库

我们需要在本地或另一台服务器上编写调用脚本，因此要安装几个Python库。创建一个新的项目目录，然后安装依赖。

# 创建项目目录并进入
mkdir qwen-asr-notifier && cd qwen-asr-notifier

# 创建虚拟环境（可选但推荐）
python -m venv venv
source venv/bin/activate  # Linux/Mac
# venv\Scripts\activate  # Windows

# 安装核心库
pip install requests python-multipart

• requests：用于发送HTTP请求调用ASR API和消息推送API。
• python-multipart：帮助我们用代码模拟表单上传文件。

3. 核心步骤一：调用ASR API获取识别文本

模型提供了Web界面，但其背后是一个标准的HTTP接口。我们的第一个任务就是用代码去调用它，把语音文件送过去，然后把文字结果拿回来。

3.1 分析API请求格式

通过浏览器的开发者工具（F12），在上传文件并识别时，我们可以观察到网络请求。你会发现，这是一个向 /run/predict 端点发送的 POST 请求，内容类型是 multipart/form-data，里面包含了音频文件和一个语言参数。

我们的代码需要完全模拟这个行为。

3.2 编写Python调用函数

下面是一个完整的函数，用于调用Qwen3-ASR-1.7B的识别接口。

import requests
import json
import time

def transcribe_audio(audio_file_path, api_base_url, language="auto"):
    """
    调用Qwen3-ASR-1.7B API进行语音识别
    
    参数:
        audio_file_path: 本地音频文件路径 (e.g., 'meeting.mp3')
        api_base_url: ASR服务的Web地址 (e.g., 'https://gpu-xxx-7860.web.gpu.csdn.net')
        language: 识别语言，默认'auto'自动检测
    
    返回:
        识别出的文本字符串，失败时返回None
    """
    # 构造完整的API端点
    api_url = f"{api_base_url.rstrip('/')}/run/predict"
    
    # 准备表单数据
    files = {
        'files': (open(audio_file_path, 'rb'))
    }
    data = {
        'data': json.dumps([language])  # API期望的格式是列表
    }
    
    try:
        print(f"正在上传并识别文件: {audio_file_path}")
        response = requests.post(api_url, files=files, data=data)
        
        if response.status_code == 200:
            result = response.json()
            # 解析返回的JSON，提取文本
            # 实际结构可能是 result['data'][0] 或类似，需要根据实际返回调整
            if 'data' in result and len(result['data']) > 0:
                transcribed_text = result['data'][0]
                print("识别成功！")
                return transcribed_text
            else:
                print(f"API返回了成功状态，但未找到文本数据。完整响应: {result}")
                return None
        else:
            print(f"识别请求失败，状态码: {response.status_code}, 响应: {response.text}")
            return None
            
    except Exception as e:
        print(f"调用API时发生异常: {e}")
        return None
    finally:
        # 确保文件被关闭
        if 'files' in locals():
            files['files'].close()

# 使用示例
if __name__ == "__main__":
    # 替换成你的实际信息
    MY_API_URL = "https://gpu-你的实例ID-7860.web.gpu.csdn.net"
    AUDIO_FILE = "test_audio.wav"
    
    text = transcribe_audio(AUDIO_FILE, MY_API_URL)
    if text:
        print("识别结果：")
        print(text)

关键点说明：

1. 文件上传：我们用 requests 库的 files 参数来上传音频文件，就像在网页上点击“选择文件”一样。
2. 数据格式：data 参数需要是一个JSON字符串，里面是一个列表，列表的第一个元素是语言代码。auto 表示自动检测。
3. 结果解析：API返回的是JSON格式，我们需要从中提取出 data 字段里的文本。具体路径可能需要你根据第一次运行的结果微调一下。
4. 错误处理：代码包含了基本的网络请求和状态码检查，确保在出错时我们能知道原因。

运行这个脚本，如果一切顺利，你的控制台就会打印出音频文件里的文字内容。恭喜你，最难的部分已经完成了！

4. 核心步骤二：将文本推送到企业微信或钉钉

拿到文字只是第一步，让它自动通知到人才是目的。这里我们分别介绍企业微信和钉钉机器人的配置方法。

4.1 方案A：推送到企业微信群机器人

企业微信的群机器人配置非常简单，不需要复杂的公司认证，适合快速测试和内部使用。

第一步：在企业微信里创建机器人

1. 打开你需要接收消息的企业微信群。
2. 点击右上角... -> 添加群机器人 -> 新建。
3. 给机器人起个名字，比如“会议纪要小助手”，然后点击添加。
4. 系统会生成一个 Webhook地址，格式像 https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxxxxx。务必复制并保存好这个地址，它就是你发送消息的“钥匙”。

第二步：编写推送函数 有了Webhook地址，发送消息就只是一次HTTP POST请求。

def send_to_wechat_work(webhook_url, text_content, title="语音识别结果通知"):
    """
    发送Markdown格式消息到企业微信群机器人
    
    参数:
        webhook_url: 机器人的Webhook地址
        text_content: 要发送的正文内容
        title: 消息标题（可选）
    """
    headers = {
        'Content-Type': 'application/json'
    }
    
    # 构造企业微信机器人要求的消息格式 (Markdown格式)
    data = {
        "msgtype": "markdown",
        "markdown": {
            "content": f"**{title}**\n\n{text_content}"
        }
    }
    
    try:
        response = requests.post(webhook_url, headers=headers, json=data, timeout=5)
        if response.status_code == 200:
            resp_json = response.json()
            if resp_json.get('errcode') == 0:
                print("消息已成功发送到企业微信！")
                return True
            else:
                print(f"企业微信接口返回错误: {resp_json.get('errmsg')}")
                return False
        else:
            print(f"请求企业微信失败，状态码: {response.status_code}")
            return False
    except Exception as e:
        print(f"发送消息到企业微信时发生异常: {e}")
        return False

# 使用示例
if __name__ == "__main__":
    WEBHOOK_URL = "你的企业微信机器人Webhook地址"
    test_text = "这是一条测试消息，来自Qwen3-ASR自动转写。"
    send_to_wechat_work(WEBHOOK_URL, test_text)

4.2 方案B：推送到钉钉群机器人

钉钉机器人的配置逻辑类似，但消息格式略有不同。

第一步：在钉钉里创建机器人

1. 打开钉钉群，点击右上角设置图标 -> 机器人 -> 添加机器人。
2. 选择自定义机器人，点击添加。
3. 设置机器人名字和安全设置（建议先选择“自定义关键词”，比如设定“转写”）。
4. 同意条款后，你会获得一个 Webhook地址，格式像 https://oapi.dingtalk.com/robot/send?access_token=xxxxxx。复制保存。

第二步：编写推送函数 钉钉支持Text和Markdown类型，这里我们用Markdown，格式更丰富。

def send_to_dingtalk(webhook_url, text_content, title="语音识别结果通知"):
    """
    发送Markdown格式消息到钉钉群机器人
    
    参数:
        webhook_url: 钉钉机器人的Webhook地址（需包含access_token）
        text_content: 要发送的正文内容
        title: 消息标题
    """
    headers = {
        'Content-Type': 'application/json'
    }
    
    # 构造钉钉机器人要求的消息格式
    data = {
        "msgtype": "markdown",
        "markdown": {
            "title": title,
            "text": f"### {title}\n\n{text_content}"
        },
        "at": {
            "isAtAll": False  # 是否@所有人，可根据需要修改
        }
    }
    
    try:
        response = requests.post(webhook_url, headers=headers, json=data, timeout=5)
        if response.status_code == 200:
            resp_json = response.json()
            if resp_json.get('errcode') == 0:
                print("消息已成功发送到钉钉！")
                return True
            else:
                print(f"钉钉接口返回错误: {resp_json.get('errmsg')}")
                return False
        else:
            print(f"请求钉钉失败，状态码: {response.status_code}")
            return False
    except Exception as e:
        print(f"发送消息到钉钉时发生异常: {e}")
        return False

# 使用示例
if __name__ == "__main__":
    DING_WEBHOOK = "你的钉钉机器人Webhook地址"
    test_text = "这是一条测试消息，来自Qwen3-ASR自动转写。"
    send_to_dingtalk(DING_WEBHOOK, test_text)

安全设置提醒：钉钉机器人如果设置了“自定义关键词”，那么你发送的消息内容中必须包含这个关键词，否则消息会被拒绝。在上面的例子中，如果你的关键词是“转写”，那么text_content里最好有这个词语。

5. 实战整合：构建自动化流程

现在，我们把“识别”和“推送”两个模块像拼积木一样组合起来，形成一个完整的自动化脚本。这个脚本可以手动运行，也可以被定时任务（如Linux的cron或Windows的任务计划程序）调用。

5.1 创建主程序脚本

创建一个名为 asr_auto_notify.py 的文件，将以下代码复制进去，并填写你自己的配置信息。

#!/usr/bin/env python3
"""
Qwen3-ASR自动识别与消息推送主程序
将指定目录下的新音频文件自动转写并推送到工作群。
"""

import os
import glob
import json
import requests
from datetime import datetime

# ========== 用户配置区域 ==========
# 1. ASR服务配置
ASR_API_BASE_URL = "https://gpu-你的实例ID-7860.web.gpu.csdn.net"  # 替换为你的服务地址
ASR_LANGUAGE = "auto"  # 识别语言，默认自动检测

# 2. 消息推送配置 (二选一即可，注释掉不用的那个)
# 企业微信机器人配置
WECHAT_WORK_WEBHOOK = "你的企业微信机器人Webhook地址"
USE_WECHAT_WORK = True  # 设置为True则启用企业微信推送

# 钉钉机器人配置
DINGTALK_WEBHOOK = "你的钉钉机器人Webhook地址"
USE_DINGTALK = False   # 设置为True则启用钉钉推送

# 3. 监控目录配置
AUDIO_MONITOR_DIR = "./audio_to_process"  # 存放待处理音频文件的目录
PROCESSED_DIR = "./audio_processed"      # 处理完成后移动到的目录
# ========== 配置结束 ==========

# 确保目录存在
os.makedirs(AUDIO_MONITOR_DIR, exist_ok=True)
os.makedirs(PROCESSED_DIR, exist_ok=True)

def transcribe_audio(audio_file_path):
    """调用ASR API（复用之前的函数，此处略作简化）"""
    api_url = f"{ASR_API_BASE_URL.rstrip('/')}/run/predict"
    try:
        with open(audio_file_path, 'rb') as f:
            files = {'files': f}
            data = {'data': json.dumps([ASR_LANGUAGE])}
            response = requests.post(api_url, files=files, data=data, timeout=30)
        
        if response.status_code == 200:
            result = response.json()
            # 注意：根据你的API实际返回结构调整下面的提取逻辑
            if 'data' in result and len(result['data']) > 0:
                return result['data'][0]
    except Exception as e:
        print(f"[错误] 处理文件 {audio_file_path} 时调用ASR API失败: {e}")
    return None

def send_notification(text_content, audio_filename):
    """根据配置发送通知"""
    title = f"语音转写完成: {audio_filename}"
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    full_message = f"**文件**: {audio_filename}\n**时间**: {timestamp}\n\n**转写内容**:\n{text_content[:500]}..." if len(text_content) > 500 else text_content
    
    success = False
    if USE_WECHAT_WORK and WECHAT_WORK_WEBHOOK:
        success = send_to_wechat_work(WECHAT_WORK_WEBHOOK, full_message, title)
    elif USE_DINGTALK and DINGTALK_WEBHOOK:
        success = send_to_dingtalk(DINGTALK_WEBHOOK, full_message, title)
    else:
        print("[警告] 未启用任何消息推送方式，仅本地记录。")
    return success

def send_to_wechat_work(webhook_url, content, title):
    """发送到企业微信（复用之前的函数）"""
    # ... 函数体同上，此处省略以节省篇幅 ...
    pass

def send_to_dingtalk(webhook_url, content, title):
    """发送到钉钉（复用之前的函数）"""
    # ... 函数体同上，此处省略以节省篇幅 ...
    pass

def main():
    print(f"[{datetime.now()}] 开始扫描目录: {AUDIO_MONITOR_DIR}")
    
    # 查找所有支持的音频文件
    audio_extensions = ('*.wav', '*.mp3', '*.flac', '*.ogg', '*.m4a')
    audio_files = []
    for ext in audio_extensions:
        audio_files.extend(glob.glob(os.path.join(AUDIO_MONITOR_DIR, ext)))
    
    if not audio_files:
        print("未发现新的音频文件。")
        return
    
    print(f"发现 {len(audio_files)} 个待处理文件。")
    
    for audio_file in audio_files:
        filename = os.path.basename(audio_file)
        print(f"正在处理: {filename}")
        
        # 步骤1: 语音识别
        transcribed_text = transcribe_audio(audio_file)
        
        if transcribed_text:
            print(f"  识别成功，字符数: {len(transcribed_text)}")
            # 步骤2: 发送通知
            if send_notification(transcribed_text, filename):
                print(f"  通知发送成功。")
            else:
                print(f"  通知发送失败。")
            
            # 步骤3: 保存结果到本地文件（可选）
            result_file = os.path.join(PROCESSED_DIR, f"{filename}.txt")
            with open(result_file, 'w', encoding='utf-8') as f:
                f.write(transcribed_text)
            print(f"  文本已保存至: {result_file}")
        else:
            print(f"  识别失败。")
        
        # 步骤4: 移动已处理的音频文件
        processed_audio_path = os.path.join(PROCESSED_DIR, filename)
        os.rename(audio_file, processed_audio_path)
        print(f"  文件已移动至: {processed_audio_path}")
        print("-" * 50)

if __name__ == "__main__":
    main()

5.2 如何使用这个自动化脚本

1. 配置：打开脚本，在“用户配置区域”填写你的ASR服务地址、机器人Webhook地址，并设置好监控目录。
2. 运行：将需要识别的音频文件（如 meeting_recording.mp3）放入 ./audio_to_process 目录。
3. 执行：在命令行运行 python asr_auto_notify.py。
4. 查看结果：脚本会自动识别音频，将文本推送到你设定的工作群，并把处理完的音频文件移动到 ./audio_processed 目录，同时在同目录下生成一个同名的 .txt 文件保存完整文本。

5.3 进阶：设置定时任务（Linux为例）

如果你想让它定时运行，比如每5分钟检查一次新录音，可以使用cron。

# 编辑当前用户的cron任务
crontab -e

# 在文件末尾添加一行，表示每5分钟运行一次脚本
*/5 * * * * cd /你的/项目/路径 && /usr/bin/python3 /你的/项目/路径/asr_auto_notify.py >> /你的/项目/路径/run.log 2>&1

这样，一个完整的、从语音识别到消息推送的自动化流程就搭建完成了。你的手机或电脑上会定期收到来自“机器人同事”的语音转写通知。

6. 总结与扩展思路

通过这个教程，我们完成了一个从技术工具到业务场景的闭环。Qwen3-ASR-1.7B提供了精准的识别能力，而简单的API调用和消息推送则让这个能力变得触手可及，直接融入日常工作流。

6.1 核心流程回顾

让我们再快速梳理一下整个链路，确保你掌握了关键点：

1. 服务就绪：确保Qwen3-ASR-1.7B镜像服务正常运行，并记住它的Web访问地址。
2. 调用识别：使用Python的requests库，模拟表单上传，向服务的/run/predict接口发送音频文件，并接收返回的文本。
3. 配置机器人：在企业微信或钉钉的群聊中创建一个“自定义机器人”，获取独一无二的Webhook地址。
4. 推送消息：将识别得到的文本，按照企业微信或钉钉要求的JSON格式包装好，通过POST请求发送到Webhook地址。
5. 自动化整合：将前两步写成函数，并创建一个主脚本，实现“监控目录->识别->推送->归档”的全自动流程。

6.2 还能怎么玩？扩展应用场景

这个基础框架就像乐高底座，你可以在此基础上搭建更复杂的应用：

• 会议纪要自动化：搭配录音笔或会议软件，自动录制会议音频，会后立即将文字纪要发到项目群。
• 客服质检助手：将客服通话录音自动转写，并提取关键词（如“投诉”、“不满意”），触发提醒给主管。
• 媒体内容速记：自动处理采访录音、播客音频，快速生成文字初稿，极大提升内容创作效率。
• 教育培训辅助：将课程录音自动转成文字笔记，方便学生复习和检索。
• 结合其他AI模型：将识别出的文本，再接入大语言模型（如Qwen、ChatGLM）进行摘要、提取待办事项、翻译或情感分析，让推送的消息更有价值。

6.3 遇到问题怎么办？

• 识别不准：检查音频质量，尝试在调用API时手动指定语言（如zh中文）而非auto。
• 推送失败：检查Webhook地址是否正确；对于钉钉，检查消息内容是否包含设定的安全“关键词”；查看脚本打印的错误信息。
• 服务无响应：登录服务器，使用 supervisorctl status qwen3-asr 和 tail -f /root/workspace/qwen3-asr.log 查看服务状态和日志。

技术的价值在于解决实际问题。希望这个教程不仅让你学会了如何对接API，更启发了你利用现有工具去优化工作流程的思路。动手试一试，从自动化处理下一段会议录音开始吧。

---

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。