使用 Whisper.cpp 和 aiogram 3.x 的机器人架构

用Whisper.cpp和FSM构建西班牙语语音识别Telegram机器人

一个专为西班牙语学习者设计的Telegram机器人，利用本地whisper.cpp引擎实现语音识别，无需依赖云端API。系统可对比发音与标准语句，支持通过大语言模型（LLM）进行对话式交互，并能在资源有限的VPS环境中高效运行。通过将async aiogram 3.x与whisper.cpp二进制文件集成，机器人能并行处理多个请求。

核心组件包括：音频转为16kHz单声道WAV格式、在独立进程中调用whisper-cli，以及基于逐词比对的评分算法以评估准确率。

非阻塞集成Whisper.cpp

使用标准的subprocess.run()会阻塞aiogram事件循环。解决方案是采用asyncio.create_subprocess_exec实现异步执行，并用asyncio.to_thread将耗时的pydub操作移至独立线程。

import asyncio
from pydub import AudioSegment
import subprocess
import os

async def recognize_voice_async(file_path: str) -> str:
    wav_path = "temp_voice.wav"
    txt_path = wav_path + ".txt"
    
    try:
        # 1. 将音频转换为Whisper兼容格式（16kHz，单声道，WAV）
        # 将pydub处理任务移至线程，避免阻塞事件循环
        audio = AudioSegment.from_file(file_path)
        await asyncio.to_thread(
            lambda: audio.set_frame_rate(16000)
                     .set_channels(1)
                     .set_sample_width(2)
                     .export(wav_path, format="wav")
        )

        if not os.path.exists(wav_path):
            raise FileNotFoundError("WAV文件未生成")

        # 2. 异步运行whisper.cpp
        process = await asyncio.create_subprocess_exec(
            "/root/whisper.cpp/build/bin/whisper-cli", # 二进制路径
            "-m", "/root/whisper.cpp/models/ggml-tiny.bin", # 轻量模型
            "-f", wav_path,
            "--language", "es",
            "--output-txt",
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE
        )
        
        stdout, stderr = await process.communicate()

        if process.returncode != 0:
            print(f"[ERROR] Whisper失败: {stderr.decode()}")
            return ""

        # 3. 读取识别结果
        if os.path.exists(txt_path):
            with open(txt_path, "r", encoding="utf-8") as f:
                text = f.read().strip()
            return text
        return ""
    
    except Exception as e:
        print(f"[CRITICAL ERROR] {e}")
        return ""
    finally:
        # 清理临时文件
        for f in [wav_path, txt_path, file_path]:
            if os.path.exists(f):
                os.remove(f)

该代码确保了实时语音识别且不会导致机器人卡顿。ggml-tiny.bin 模型因其在CPU上的高速表现被选中。

发音评分算法

由于Whisper在冠词和词尾上常出现错误，简单字符串比对不可靠。因此我们采用逐词位置评分算法：

def calculate_accuracy(expected: str, recognized: str) -> int:
    if not recognized:
        return 0
    
    expected_words = expected.lower().split()
    recognized_words = recognized.lower().split()
    
    if not expected_words:
        return 0

    # 按位置比较单词
    min_len = min(len(expected_words), len(recognized_words))
    matches = sum(1 for i in range(min_len) if expected_words[i] == recognized_words[i])
    
    # 公式：(匹配数 / 参考长度) * 100
    accuracy = int((matches / len(expected_words)) * 100)
    
    return max(0, min(100, accuracy))

示例结果：

• 完全匹配：100%
• 缺少最后一个词：80–90%
• 完全不匹配：0–20%

成功阈值设为70%。未来计划引入Levenshtein距离以检测词内拼写错误。

FSM状态管理与路由

aiogram.fsm状态机区分两种模式：自由对话与发音训练。

from aiogram.fsm.state import State, StatesGroup

class UserStates(StatesGroup):
    waiting_for_ai_dialog = State()  # 自由聊天模式

通用语音消息处理器：

@dp.message(F.voice)
async def universal_voice_handler(message: Message, state: FSMContext):
    current_state = await state.get_state()
    
    # 分支1：AI对话模式
    if current_state == "UserStates:waiting_for_ai_dialog":
        from handlers.voice_ai import handle_ai_dialog_voice
        await handle_ai_dialog_voice(message, bot, state)
        return

    # 分支2：发音训练（课程）
    # ... 获取参考语句，识别并评分的逻辑

技术栈与数据存储

• 语音识别（STT）： whisper.cpp（ggml-tiny.bin，西班牙语）
• 语音合成（TTS）： gTTS + pydub
• 大语言模型（LLM）： 外部API
• 存储方式： JSON文件（users.json, lessons.json, phrases.json）

管理员命令（/addphrase, /stats）支持动态内容更新。可通过PostgreSQL和Docker实现可扩展部署。

核心优势：

• 异步集成whisper.cpp，防止机器人冻结。
• 逐词比对提供精准发音反馈，无需依赖云服务。
• FSM机制便于扩展新功能，无需重写处理器。
• 本地化处理保障用户隐私与成本效益。
• tiny.bin模型在CPU上运行迅速，延迟极低。

— Editorial Team