WechatApi首页价格个人微信API开发文档博客立即注册
首页 / 博客 / 机器人·功能实战

微信语音消息收发与语音转文字实战

分类：机器人·功能实战　·　标签：微信语音消息、语音转文字、微信API

前言

在微信生态的自动化场景里，文字消息的处理相对简单，但语音消息往往是绕不开的障碍：用户习惯发语音，机器人却只能读懂文字。要实现完整的微信客服机器人或消息归档系统，必须打通"接收语音 → 下载音频 → 语音识别"这条链路。

本文围绕三个核心问题展开：

  1. 如何通过 HTTP 接口向指定联系人发送语音消息？
  2. 收到语音消息时，如何从回调中获取音频文件？
  3. 拿到音频文件后，如何调用语音识别服务实现语音转文字？

文章会给出完整可运行的 Python 示例，覆盖接口调用、音频下载、格式转换与 ASR 接入的全流程。

一、微信语音消息的基本格式

1.1 微信语音文件格式

微信在内部使用 SILK 格式（.slk）存储语音，这是腾讯自研的低码率编解码格式，不同于常见的 MP3、WAV、AAC 等。SILK 文件可直接在微信内播放，但通用播放器和大多数语音识别 API 并不支持，因此在进行语音识别之前往往需要做一次格式转换。

几种常见格式的对比：

格式码率识别支持备注
SILK (.slk)极低几乎不支持微信内部格式
PCM (.pcm)高广泛支持原始采样，体积大
WAV (.wav)高广泛支持PCM 加容器头
MP3 (.mp3)中广泛支持有损压缩
AAC (.aac)低广泛支持效率最佳

实际落地时推荐的转换路径是：SILK → PCM → WAV，再送入语音识别服务，这条链路的开源工具最完整、识别效果最稳定。

1.2 语音消息在回调中的结构

当有人向你的微信账号发送语音时，平台会把该事件以 POST 请求的形式推送到你预先设置的回调地址。回调 payload 中与语音相关的字段大致如下（字段名以实际平台文档为准）：

json{
  "appId": "你的appId",
  "fromWxid": "发送方wxid",
  "toWxid": "接收方wxid",
  "type": 34,
  "msgId": "消息唯一ID",
  "content": "语音文件的mediaId或URL",
  "voiceTime": 12,
  "createTime": 1718000000
}

其中 type=34 是微信协议层对语音消息的类型编号，voiceTime 是语音时长（秒），content 字段通常承载文件标识——不同平台的实现方式略有差异，有的直接给下载链接，有的给一个 mediaId 需要再调下载接口换取文件流。

二、发送语音消息

2.1 接口格式说明

发送语音的接口通常是 POST /message/postVoice，请求体为 JSON，需要提供音频文件内容或可访问的 URL，以及语音时长。

示例代码如下（占位符形式，具体字段以官方文档为准）：

pythonimport requests
import base64

BASE    = "https://你的接口域名"   # 注册后在官方文档获取
TOKEN   = "你的Token"
APPID   = "你的appId"
HEADERS = {"token": TOKEN}        # 鉴权字段名以官方文档为准


def send_voice(to_wxid: str, audio_path: str, voice_time: int) -> dict:
    """
    发送语音消息
    :param to_wxid:    接收方 wxid 或群 id
    :param audio_path: 本地 silk 或 wav 文件路径
    :param voice_time: 语音时长（秒）
    """
    with open(audio_path, "rb") as f:
        audio_b64 = base64.b64encode(f.read()).decode()

    payload = {
        "appId":     APPID,
        "toWxid":    to_wxid,
        "voice":     audio_b64,      # base64 编码的音频内容（字段名以文档为准）
        "voiceTime": voice_time,
    }
    resp = requests.post(f"{BASE}/message/postVoice", json=payload, headers=HEADERS)
    return resp.json()


# 示例调用
result = send_voice("wxid_xxxxxx", "./hello.slk", 5)
print(result)  # {"ret": 200, "msg": "操作成功", "data": {...}}
代码为示例，具体接口路径、请求字段及返回结构以官方文档为准。

2.2 发送注意事项

三、接收语音消息并下载音频

3.1 搭建回调服务

使用 Flask 快速搭建一个接收回调的 Web 服务：

pythonfrom flask import Flask, request, jsonify
import json

app = Flask(__name__)

VOICE_MSG_TYPE = 34   # 微信语音消息 type，以实际文档为准

@app.route("/callback", methods=["POST"])
def callback():
    data = request.get_json(force=True)
    msg_type = data.get("type")

    if msg_type == VOICE_MSG_TYPE:
        handle_voice(data)

    return jsonify({"code": 200})   # 必须返回 200，否则平台会重试


def handle_voice(data: dict):
    """处理语音消息"""
    msg_id    = data.get("msgId")
    from_wxid = data.get("fromWxid")
    voice_ref = data.get("content")   # mediaId 或直接 URL，视文档而定
    voice_sec = data.get("voiceTime", 0)

    print(f"收到语音消息 from={from_wxid}, time={voice_sec}s, ref={voice_ref}")
    # 异步排队下载，避免阻塞回调响应
    download_voice_async(msg_id, voice_ref)


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080)

3.2 下载语音文件

下载接口通常为 POST /message/downloadVoice（字段名以文档为准）：

pythonimport requests
import os

BASE    = "https://你的接口域名"
TOKEN   = "你的Token"
APPID   = "你的appId"
HEADERS = {"token": TOKEN}

DOWNLOAD_DIR = "/tmp/wechat_voices"
os.makedirs(DOWNLOAD_DIR, exist_ok=True)


def download_voice(msg_id: str, voice_ref: str) -> str:
    """
    下载语音文件，返回本地路径
    :param msg_id:    消息 ID，用于命名文件
    :param voice_ref: 回调中的音频标识（mediaId 或 URL）
    """
    payload = {
        "appId":   APPID,
        "msgId":   msg_id,
        "content": voice_ref,   # 字段名以文档为准
    }
    resp = requests.post(f"{BASE}/message/downloadVoice", json=payload, headers=HEADERS)
    result = resp.json()

    if result.get("ret") != 200:
        raise RuntimeError(f"下载失败: {result}")

    # 假设 data.fileUrl 是可访问的临时文件地址（以文档为准）
    file_url = result["data"]["fileUrl"]
    local_path = os.path.join(DOWNLOAD_DIR, f"{msg_id}.slk")

    audio_resp = requests.get(file_url, timeout=30)
    with open(local_path, "wb") as f:
        f.write(audio_resp.content)

    return local_path
实际返回字段视平台文档而定，部分平台直接在下载接口中返回 base64 编码的音频内容而非 URL。

3.3 异步队列设计

语音下载属于 IO 密集型操作，直接在回调函数中同步执行会超时。建议使用 threading.Thread 或消息队列（Redis + Celery）将下载任务放到后台处理：

pythonimport threading
import time
import random


def download_voice_async(msg_id: str, voice_ref: str):
    """后台线程下载，带随机延迟防封"""
    def _task():
        time.sleep(random.uniform(3, 10))   # 每条间隔 3–10s，防止频率过高
        try:
            local_path = download_voice(msg_id, voice_ref)
            print(f"语音已保存: {local_path}")
            # 下载完成后触发语音识别
            text = voice_to_text(local_path)
            print(f"识别结果: {text}")
        except Exception as e:
            print(f"处理失败: {e}")

    t = threading.Thread(target=_task, daemon=True)
    t.start()

四、SILK 转 WAV 格式转换

4.1 silk-v3-decoder 的使用

开源库 silk-v3-decoder 是目前最成熟的 SILK 解码方案，支持将微信语音转换为 PCM/WAV：

bash# 编译（macOS/Linux）
git clone https://github.com/kn007/silk-v3-decoder
cd silk-v3-decoder
make

# 转换：slk → pcm
./decoder input.slk output.pcm

# pcm → wav（借助 ffmpeg）
ffmpeg -y -f s16le -ar 24000 -ac 1 -i output.pcm output.wav

或者在 Python 中通过 subprocess 调用：

pythonimport subprocess
import os


def silk_to_wav(silk_path: str) -> str:
    """
    将 SILK 文件转为 WAV，返回 WAV 文件路径
    需要提前编译 silk-v3-decoder 并安装 ffmpeg
    """
    base    = silk_path.rsplit(".", 1)[0]
    pcm_path = base + ".pcm"
    wav_path = base + ".wav"

    # Step 1: SILK → PCM
    subprocess.run(
        ["/path/to/silk-v3-decoder/decoder", silk_path, pcm_path],
        check=True, capture_output=True
    )

    # Step 2: PCM → WAV (24000Hz, mono, 16-bit signed little-endian)
    subprocess.run(
        ["ffmpeg", "-y", "-f", "s16le", "-ar", "24000", "-ac", "1",
         "-i", pcm_path, wav_path],
        check=True, capture_output=True
    )

    # 清理中间文件
    os.remove(pcm_path)
    return wav_path
采样率 24000 Hz 是微信语音的默认值，部分场景可能是 16000 Hz，以实际音频为准。

4.2 pydub 直接处理 PCM

如果环境中已安装 pydub + ffmpeg，也可以用更 Pythonic 的方式处理：

pythonfrom pydub import AudioSegment


def pcm_to_wav_pydub(pcm_path: str, wav_path: str,
                     sample_rate: int = 24000, channels: int = 1):
    """PCM 原始采样转 WAV"""
    audio = AudioSegment.from_raw(
        pcm_path,
        sample_width=2,        # 16-bit = 2 bytes
        frame_rate=sample_rate,
        channels=channels
    )
    audio.export(wav_path, format="wav")

4.3 格式转换实操要点

格式转换是整条链路中最容易出错的环节，上线前务必逐一核实以下细节：

五、语音转文字（ASR 接入）

5.1 方案选择

语音识别（ASR）有离线和在线两条路：

方案优点缺点适合场景
百度 ASR接入简单、中文准确率高有频率限制，付费超量中低频客服机器人
讯飞开放平台方言识别强文档复杂需要方言支持
OpenAI Whisper（本地）完全离线、无费用需要 GPU，速度慢隐私敏感或高频场景
Azure Speech企业级稳定价格较高对准确率要求高

下面以百度语音识别为例给出完整调用示例（接入其他服务只需替换 SDK 部分）。

5.2 百度 ASR 接入

pythonfrom aip import AipSpeech   # pip install baidu-aip

# 百度 AI 平台凭证（需自行注册获取）
BAIDU_APP_ID    = "你的AppId"
BAIDU_API_KEY   = "你的ApiKey"
BAIDU_SECRET    = "你的SecretKey"

asr_client = AipSpeech(BAIDU_APP_ID, BAIDU_API_KEY, BAIDU_SECRET)


def voice_to_text_baidu(wav_path: str) -> str:
    """
    百度 ASR：WAV 文件 → 文字
    :param wav_path: 已转换的 WAV 文件路径（PCM 编码，16000 或 24000 Hz）
    """
    with open(wav_path, "rb") as f:
        audio_data = f.read()

    result = asr_client.asr(
        audio_data,
        "wav",
        16000,              # 采样率，需与文件实际采样率一致
        {"dev_pid": 1537}   # 1537=普通话，其他 pid 见百度文档
    )

    if result.get("err_no") == 0:
        return "".join(result.get("result", []))
    else:
        raise RuntimeError(f"百度 ASR 错误: {result}")

5.3 使用 Whisper 做本地离线识别

若对数据隐私要求较高，或语音消息量大导致云服务费用高，可以使用 OpenAI Whisper 在本地做识别：

pythonimport whisper

# 首次运行会自动下载模型（约 1.5 GB for "medium"）
_whisper_model = whisper.load_model("medium")


def voice_to_text_whisper(audio_path: str, language: str = "zh") -> str:
    """
    Whisper 本地识别，支持 wav/mp3/slk 等多种格式
    :param audio_path: 音频文件路径
    :param language:   语言代码，中文填 "zh"
    """
    result = _whisper_model.transcribe(audio_path, language=language)
    return result["text"].strip()

Whisper 的优势在于直接支持 SILK 文件（内部会调 ffmpeg 解码），省去了格式转换步骤。选用 medium 或 large 模型时普通话识别准确率可达 95% 以上。

5.4 完整流水线整合

将前面各步骤串联，形成从回调到识别结果的完整流水线：

pythondef voice_to_text(silk_path: str) -> str:
    """
    语音消息完整处理流水线
    SILK → WAV → ASR → 文字
    """
    try:
        # Step 1: 格式转换
        wav_path = silk_to_wav(silk_path)

        # Step 2: 语音识别（二选一）
        # text = voice_to_text_baidu(wav_path)  # 云端方案
        text = voice_to_text_whisper(wav_path)  # 本地方案

        return text
    finally:
        # 清理临时文件
        for path in [silk_path, silk_path.replace(".slk", ".wav")]:
            if os.path.exists(path):
                os.remove(path)

六、在消息机器人中落地

6.1 语音消息自动转文字回复

一个常见场景：收到语音消息后，自动将识别结果以文字形式回复给发送者，方便客服归档或 AI 进一步处理：

pythonimport requests

BASE    = "https://你的接口域名"
TOKEN   = "你的Token"
APPID   = "你的appId"
HEADERS = {"token": TOKEN}


def reply_with_text(to_wxid: str, text: str):
    """将识别结果回复给发送方"""
    payload = {
        "appId":   APPID,
        "toWxid":  to_wxid,
        "content": f"[语音转文字] {text}",
    }
    resp = requests.post(f"{BASE}/message/postText", json=payload, headers=HEADERS)
    return resp.json()

如果需要对接企业内部系统，WechatApi 提供扫码登录、消息收发、好友与群管理等 REST 接口，HTTP 调用即可完成上述语音收发的全部操作，具体能力以官方文档为准。

6.2 批量归档历史语音

对于已积压的历史语音消息，可以批量从消息记录中提取 msgId，依次调用下载接口并识别：

pythonimport time
import random


def batch_archive_voices(voice_records: list[dict]) -> list[dict]:
    """
    批量归档语音消息为文字
    :param voice_records: [{"msg_id": ..., "from_wxid": ..., "voice_ref": ...}, ...]
    """
    results = []
    for record in voice_records:
        try:
            silk_path = download_voice(record["msg_id"], record["voice_ref"])
            text = voice_to_text(silk_path)
            results.append({
                "msg_id":   record["msg_id"],
                "from":     record["from_wxid"],
                "text":     text,
                "status":   "ok",
            })
        except Exception as e:
            results.append({
                "msg_id": record["msg_id"],
                "status": "error",
                "error":  str(e),
            })
        # 下载间隔 3–10 秒，避免频率过高
        time.sleep(random.uniform(3, 10))

    return results

6.3 常见问题排查

问题现象可能原因解决方式
下载返回 404 / 失败mediaId 已过期（通常 3 天内有效）实时下载，不要延迟太久
WAV 转换后杂音严重采样率填错（应为 24000，填了 16000）确认 ffmpeg 参数中 -ar 值
ASR 识别率低输入采样率与识别服务不匹配统一转为 16000 Hz 再识别
回调收不到语音消息回调地址不公网可达，或返回非 200检查公网 IP + 端口，确保返回 {"code":200}
silk-v3-decoder 编译失败缺少 gcc / makeapt install build-essential 或 brew install gcc

总结

语音消息转文字在微信自动化场景中是一个高频需求，核心链路不复杂：回调接收语音 → 异步下载 SILK 文件 → 转换为 WAV → ASR 识别。格式转换是最容易踩坑的环节，关键是保证采样率与 ASR 服务的输入要求匹配。离线 Whisper 和云端百度 ASR 各有适用场景，按实际的并发量和隐私要求选择即可。

落地时有几点值得特别关注：一是异步处理必须做，回调响应时间窗口极短，同步阻塞极易导致平台重推甚至断连；二是mediaId 有时效性，务必在消息到达后尽快触发下载，超过有效期将无法获取文件；三是磁盘管理不能忽略，批量处理语音时 PCM 中间文件体积可观，需要在流水线末尾做好清理；四是ASR 采样率对齐，格式转换和识别调用中的采样率参数必须保持一致，否则识别结果会出现乱码或词序混乱。以上几点做扎实，整条语音处理链路就能稳定运行。

想动手试试？

WechatApi 提供扫码登录、消息收发、好友与群管理等 REST 接口，注册后几分钟跑通。

立即免费注册查看开发文档

相关产品页

🔗 微信机器人开发（产品页）🔗 微信客服机器人（产品页）🔗 微信群管理机器人（产品页）

相关文章

30 分钟做一个微信自动回复机器人（完整实战）微信机器人接入 GPT，实现智能自动回复微信群管理机器人开发实战：自动迎新、答疑、踢人微信客服机器人怎么做？7×24自动应答+转人工方案
© 2025 WechatApi · 企业级微信智能机器人接入平台
官网价格帮助文档博客
苏ICP备2024128799号 · 苏ICP备2023038368号