ASR Integrations

browser_audio:
  # ... other configuration
  asr:
    name: deepgram

recipe: default.v1
language: en
additional_languages:
  - de-DE

pipeline:
- name: SingleStepLLMCommandGenerator

policies:
- name: FlowPolicy

assistant_id: 20240418-073244-narrow-archive

browser_audio:
  asr:
    name: deepgram
    language_map:
      en:
        language: eng
        model: mistv2
      de-DE:
        language: de
        model: mistv2

browser_audio:
  asr:
    name: deepgram_flux
    language_map:
      en:
        model: flux-general-en
    eot_threshold: 0.7
    eot_timeout_ms: 5000

browser_audio:
  asr:
    name: deepgram_flux
    language_map:
      en:
        language: en
        model: flux-general-multi
      de-DE:
        language: de
        model: flux-general-multi

browser_audio:
  # ... other configuration
  asr:
    name: azure

browser_audio:
  asr:
    name: azure
    speech_region: germanywestcentral
    language_map:
      de-DE:
        language: de-DE

recipe: default.v1
language: en
additional_languages:
  - de-DE

pipeline:
- name: SingleStepLLMCommandGenerator

policies:
- name: FlowPolicy

assistant_id: 20240418-073244-narrow-archive

browser_audio:
  asr:
    name: azure
    speech_region: germanywestcentral
    language_map:
      en:
        language: en-US
      de-DE:
        language: de-DE

browser_audio:
  asr:
    name: elevenlabs

import json
import os
from dataclasses import dataclass
from typing import Any, Dict, Optional
from urllib.parse import urlencode

import websockets
from websockets.legacy.client import WebSocketClientProtocol

from rasa.core.channels.voice_stream.asr.asr_engine import ASREngine, ASREngineConfig
from rasa.core.channels.voice_stream.asr.asr_event import (
    ASREvent,
    NewTranscript,
    UserIsSpeaking,
)
from rasa.core.channels.voice_stream.audio_bytes import RasaAudioBytes

@dataclass
class MyASRConfig(ASREngineConfig):
    api_key: str = ""
    endpoint: str = "wss://api.example.com/v1/speech"
    language: str = "en-US"

class MyASR(ASREngine[MyASRConfig]):
    required_env_vars = ("MY_ASR_API_KEY",)  # Optional: required environment variables
    required_packages = ("my_asr_package",)  # Optional: required Python packages

    def __init__(self, config: Optional[MyASRConfig] = None):
        super().__init__(config)
        self.accumulated_transcript = ""

    async def open_websocket_connection(self) -> WebSocketClientProtocol:
        """Connect to the ASR system."""
        api_key = os.environ["MY_ASR_API_KEY"]
        headers = {"Authorization": f"Bearer {api_key}"}
        return await websockets.connect(
            self._get_api_url_with_params(),
            extra_headers=headers
        )

    def _get_api_url_with_params(self) -> str:
        """Build API URL with query parameters."""
        query_params = {
            "language": self.config.language,
            "encoding": "mulaw",
            "sample_rate": "8000",
            "interim_results": "true"
        }
        return f"{self.config.endpoint}?{urlencode(query_params)}"

    @classmethod
    def from_config_dict(cls, config: Dict) -> "MyASR":
        """Create instance from configuration dictionary."""
        asr_config = MyASRConfig.from_dict(config)
        return cls(asr_config)

    async def signal_audio_done(self) -> None:
        """Signal to the ASR service that audio input has ended."""
        await self.asr_socket.send(json.dumps({"type": "stop_audio"}))

    def rasa_audio_bytes_to_engine_bytes(self, chunk: RasaAudioBytes) -> bytes:
        """Convert Rasa audio format to engine format."""
        # For most services, you can return the chunk directly
        # since it's already in mulaw format
        return chunk

    def engine_event_to_asr_event(self, event: Any) -> Optional[ASREvent]:
        """Convert engine response to ASREvent."""
        data = json.loads(event)
        
        if data.get("type") == "transcript":
            transcript = data.get("text", "")
            
            if data.get("is_final"):
                # Final transcript - user finished speaking
                full_transcript = self.accumulated_transcript + " " + transcript
                self.accumulated_transcript = ""
                return NewTranscript(full_transcript.strip())
            elif transcript:
                # Interim transcript - user is still speaking
                return UserIsSpeaking(transcript)
                
        return None

    @staticmethod
    def get_default_config() -> MyASRConfig:
        """Get default configuration."""
        return MyASRConfig(
            endpoint="wss://api.example.com/v1/speech",
            language="en-US"
        )

    async def send_keep_alive(self) -> None:
        """Send keep-alive message if supported by your service."""
        if self.asr_socket is not None:
            await self.asr_socket.send(json.dumps({"type": "keep_alive"}))

browser_audio:
  # ... other configuration
  asr:
    name: addons.custom_asr.MyASR
    api_key: "your_api_key"
    endpoint: "wss://api.example.com/v1/speech"
    language: "en-US"
    # any other custom parameters your ASR needs