Voice Assistant offline tiếng Việt với Raspberry Pi

Tải Vosk model tiếng Việt

cd ~/voice-assistant
mkdir -p models && cd models
wget https://alphacephei.com/vosk/models/vosk-model-small-vn-0.4.zip
unzip vosk-model-small-vn-0.4.zip
rm vosk-model-small-vn-0.4.zip
cd ..

config.py

VOSK_MODEL_PATH = "models/vosk-model-small-vn-0.4"
SAMPLE_RATE = 16000
BLOCK_SIZE  = 8000

COMMANDS = [
    "bật đèn","tắt đèn","bật quạt","tắt quạt",
    "bật điều hòa","tắt điều hòa","bật ổ cắm","tắt ổ cắm",
]

DEVICE_PINS    = {"light":17,"fan":27,"ac":22,"socket":23}
DEVICE_NAMES_VI= {"light":"đèn","fan":"quạt","ac":"điều hòa","socket":"ổ cắm"}
MQTT_BROKER    = "localhost"
MQTT_PORT      = 1883
MQTT_TOPIC_PREFIX = "home/voice"

command_parser.py

from dataclasses import dataclass
from typing import Optional

@dataclass
class ParsedCommand:
    action: str   # "on" | "off"
    device: str   # "light" | "fan" | "ac" | "socket"

COMMAND_TABLE: dict[str, ParsedCommand] = {
    "bật đèn":      ParsedCommand("on",  "light"),
    "tắt đèn":      ParsedCommand("off", "light"),
    "bật quạt":     ParsedCommand("on",  "fan"),
    "tắt quạt":     ParsedCommand("off", "fan"),
    "bật điều hòa": ParsedCommand("on",  "ac"),
    "tắt điều hòa": ParsedCommand("off", "ac"),
    "bật ổ cắm":    ParsedCommand("on",  "socket"),
    "tắt ổ cắm":    ParsedCommand("off", "socket"),
}

def parse_command(text: str) -> Optional[ParsedCommand]:
    return COMMAND_TABLE.get(text.strip().lower())

listener.py

import json, queue
from typing import Callable
import sounddevice as sd
from vosk import Model, KaldiRecognizer
from config import VOSK_MODEL_PATH, SAMPLE_RATE, BLOCK_SIZE, COMMANDS

def create_recognizer() -> KaldiRecognizer:
    model = Model(VOSK_MODEL_PATH)
    rec   = KaldiRecognizer(model, SAMPLE_RATE)
    rec.SetWords(True)
    rec.SetGrammar(json.dumps(COMMANDS + ["[unk]"]))
    return rec

def listen(on_command: Callable[[str], None]) -> None:
    rec = create_recognizer()
    q: queue.Queue[bytes] = queue.Queue()

    def audio_cb(indata, frames, time_info, status):
        q.put(bytes(indata))

    print("Đang lắng nghe... (Ctrl+C để dừng)")
    with sd.RawInputStream(samplerate=SAMPLE_RATE, blocksize=BLOCK_SIZE,
                            dtype="int16", channels=1, callback=audio_cb):
        while True:
            data = q.get()
            if rec.AcceptWaveform(data):
                text = json.loads(rec.Result()).get("text","").strip()
                if text and text != "[unk]":
                    print(f"[Nhận]: {text}")
                    on_command(text)

Test nhanh

source .venv/bin/activate
python -c "from listener import listen; listen(lambda t: print('===', t, '==='))"