ASR

robot.asr is a plugin namespace for continuous speech recognition, backed by your choice of engine — Azure, Groq Whisper, or a local Parakeet or Riva server. Enable the matching plugin, configure_* it once, then subscribe to its streams. These examples assume you already have a connected robot — see Connection if you haven't set one up yet.

Setup

mkdir ~/example
cd ~/example
python -m venv .venv
# or: uv venv .venv

source .venv/bin/activate

pip install luxai-robot
# or: uv pip install luxai-robot

Connect

from luxai.robot.core import Robot

robot = Robot.connect_zmq(robot_id="QTRD000123")
print(f"connected to {robot.robot_id} ({robot.robot_type})")

See Connection for MQTT, WebRTC, and other connection options.

Azure

from luxai.magpie.frames import StringFrame, DictFrame

def asr_event_callback(event: StringFrame):
    Logger.info(f"event: {event.value}")

def asr_speech_callback(speech: DictFrame):
    Logger.info(f"speech: {speech.value}")

robot.enable_plugin_local("asr-azure")
ret = robot.asr.configure_azure(
    region="",         # your Azure speech region, e.g. westeurope
    subscription="",   # your Azure speech subscription key
    continuous_mode=True,
    use_vad=True,
)
Logger.info(f"configure_azure returned {ret}")

robot.asr.stream.on_azure_event(asr_event_callback)
robot.asr.stream.on_azure_speech(asr_speech_callback)

# Or perform a single recognition (non-blocking):
# h = robot.asr.recognize_azure_async()
# Logger.info(h.result())

Groq Whisper

from luxai.magpie.frames import StringFrame, DictFrame

def asr_event_callback(event: StringFrame):
    Logger.info(f"event: {event.value}")

def asr_speech_callback(speech: DictFrame):
    Logger.info(f"speech: {speech.value}")

robot.enable_plugin_local("asr-groq")
ret = robot.asr.configure_groq(
    api_key="",            # your Groq API key (https://console.groq.com)
    language="en",         # ISO-639-1 language code, e.g. 'en', 'fr', 'de'
    silence_timeout=0.5,   # seconds of silence that end an utterance
    continuous_mode=True,
)
Logger.info(f"configure_groq returned {ret}")

robot.asr.stream.on_groq_event(asr_event_callback)
robot.asr.stream.on_groq_speech(asr_speech_callback)

# Or perform a single recognition (non-blocking):
# h = robot.asr.recognize_groq_async()
# Logger.info(h.result())

Parakeet (local)

Runs against qtrobot-parakeet-asr-driver (e.g. on a Jetson Orin) rather than a cloud API — no API key needed.

from luxai.magpie.frames import StringFrame, DictFrame

def asr_event_callback(event: StringFrame):
    Logger.info(f"event: {event.value}")

def asr_speech_callback(speech: DictFrame):
    Logger.info(f"speech: {speech.value}")

robot.enable_plugin_local("asr-parakeet")
ret = robot.asr.configure_parakeet(
    endpoint="tcp://10.231.0.1:50860",  # qtrobot-parakeet-asr-server
    language="en",                       # auto-detected, used only for reporting
    use_vad=True,
    silence_timeout=0.3,
    max_buffer_seconds=20.0,
    continuous_mode=True,
)
Logger.info(f"configure_parakeet returned {ret}")

# speech stream carries both interim (is_final=False) and final (is_final=True) results
robot.asr.stream.on_parakeet_event(asr_event_callback)
robot.asr.stream.on_parakeet_speech(asr_speech_callback)

# Or perform a single recognition (non-blocking):
# h = robot.asr.recognize_parakeet_async()
# Logger.info(h.result())

Riva (local)

Runs against a self-hosted NVIDIA Riva ASR server (typically a Docker container on the same network) rather than a cloud API. Requires the asr-riva extra: pip install "luxai-robot[asr-riva]".

from luxai.magpie.frames import StringFrame, DictFrame

def asr_event_callback(event: StringFrame):
    Logger.info(f"event: {event.value}")

def asr_speech_callback(speech: DictFrame):
    Logger.info(f"speech: {speech.value}")

robot.enable_plugin_local("asr-riva")
ret = robot.asr.configure_riva(
    server="localhost:50051",  # address of the NVIDIA Riva ASR docker server
    language="en-US",
    continuous_mode=True,
    use_vad=True,
)
Logger.info(f"configure_riva returned {ret}")

robot.asr.stream.on_riva_event(asr_event_callback)
robot.asr.stream.on_riva_speech(asr_speech_callback)

# Or perform a single recognition (non-blocking):
# h = robot.asr.recognize_riva_async()
# Logger.info(h.result())

Next steps

Continue with the Motor tutorial, or see the full robot.asr namespace in the Python API Reference.