Speech 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
    language_map:
      en:
        model: flux-general-en
    eot_threshold: 0.7
    eot_timeout_ms: 5000

browser_audio:
  asr:
    name: deepgram
    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:
  server_url: localhost
  asr:
    name: azure
    language: de-DE
    speech_region: germanywestcentral
  tts:
    name: azure
    language: de-DE
    voice: de-DE-KatjaNeural
    speech_region: germanywestcentral

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:
  # ... other configuration
  tts:
    name: azure

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

pipeline:
- name: SingleStepLLMCommandGenerator

policies:
- name: FlowPolicy

assistant_id: 20240418-073244-narrow-archive

browser_audio:
  tts:
    name: azure
    speech_region: germanywestcentral
    language_map:
      en:
        language: en-US
        voice: en-US-JennyMultilingualNeural
      de-DE:
        language: de-DE
        voice: de-DE-KatjaNeural

browser_audio:
  # ... other configuration
  tts:
    name: cartesia

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

pipeline:
- name: SingleStepLLMCommandGenerator

policies:
- name: FlowPolicy

assistant_id: 20240418-073244-narrow-archive

browser_audio:
  tts:
    name: cartesia
    language_map:
      en:
        language: en
      de-DE:
        language: de

browser_audio:
  # ... other configuration
  tts:
    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:
  tts:
    name: deepgram
    language_map:
      en:
        model: aura-2-thalia-en
      de-DE:
        model: aura-2-julius-de

browser_audio:
  # ... other configuration
  tts:
    name: rime

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

pipeline:
- name: SingleStepLLMCommandGenerator

policies:
- name: FlowPolicy

assistant_id: 20240418-073244-narrow-archive

browser_audio:
  tts:
    name: rime
    language_map:
      en:
        language: eng
        voice: abbie
      de-DE:
        language: ger
        voice: amalia

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"}))

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

import aiohttp
from aiohttp import ClientTimeout, WSMsgType

from rasa.core.channels.voice_stream.audio_bytes import RasaAudioBytes
from rasa.core.channels.voice_stream.tts.tts_engine import (
    TTSEngine,
    TTSEngineConfig,
    TTSError,
)

@dataclass
class MyTTSConfig(TTSEngineConfig):
    endpoint: str = "wss://api.example.com/v1/speak"
    model_id: str = "en-US-standard"

class MyTTS(TTSEngine[MyTTSConfig]):
    session: Optional[aiohttp.ClientSession] = None
    required_env_vars = ("MY_TTS_API_KEY",)  # Optional: required environment variables
    required_packages = ("aiohttp",)  # Optional: required Python packages
    streaming_input = True
    ws: Optional[aiohttp.ClientWebSocketResponse] = None

    def __init__(self, config: Optional[MyTTSConfig] = None):
        super().__init__(config)
        timeout = ClientTimeout(total=self.config.timeout)
        # All class instances share the same session
        if self.__class__.session is None or self.__class__.session.closed:
            self.__class__.session = aiohttp.ClientSession(timeout=timeout)

    async def connect(self, config: Optional[MyTTSConfig] = None) -> None:
        """Establish WebSocket connection to the TTS engine."""
        headers = {
            "Authorization": f"Bearer {os.environ['MY_TTS_API_KEY']}",
        }
        
        query_params = {
            "model": merged_config.model_id,
            "language": merged_config.language,
            "voice": merged_config.voice,
            "encoding": "mulaw",
            "sample_rate": "8000",
        }
        
        ws_url = f"{merged_config.endpoint}?{urlencode(query_params)}"
        
        try:
            self.ws = await self.session.ws_connect(
                ws_url,
                headers=headers,
                timeout=float(self.config.timeout) if self.config.timeout else 30,
            )
        except Exception as e:
            raise TTSError(f"Failed to connect to TTS service: {e}")

    async def close_connection(self) -> None:
        """Close WebSocket connection if it exists."""
        if self.ws and not self.ws.closed:
            await self.ws.close()
            self.ws = None

    async def send_text_chunk(self, text: str) -> None:
        """Send text chunk to TTS engine for streaming synthesis."""
        await self.ws.send_json({"text": text})

    async def signal_text_done(self) -> None:
        """Signal TTS engine that all text has been sent."""
        await self.ws.send_json({"operation": "flush"})

    async def stream_audio(self) -> AsyncIterator[RasaAudioBytes]:
        """Stream audio output from the TTS engine."""
        try:
            async for msg in self.ws:
                if msg.type == WSMsgType.TEXT:
                    data = msg.json()
                    if data.get("type") == "audio":
                        # Assume base64 encoded audio data
                        import base64
                        audio_bytes = base64.b64decode(data.get("data", ""))
                        if audio_bytes:
                            yield self.engine_bytes_to_rasa_audio_bytes(audio_bytes)
                    elif data.get("type") == "error":
                        raise TTSError(f"TTS error: {data.get('message')}")
                elif msg.type == WSMsgType.CLOSED:
                    break
                elif msg.type == WSMsgType.ERROR:
                    raise TTSError("WebSocket error during audio streaming")
        except Exception as e:
            raise TTSError(f"Error during audio streaming: {e}")

    async def synthesize(
        self, text: str, config: Optional[MyTTSConfig] = None
    ) -> AsyncIterator[RasaAudioBytes]:
        """Generate speech from text using streaming WebSocket."""
        await self.connect(config)
        
        try:
            await self.send_text_chunk(text)
            await self.signal_text_done()
            
            async for audio_chunk in self.stream_audio():
                yield audio_chunk
        finally:
            await self.close_connection()

    def engine_bytes_to_rasa_audio_bytes(self, chunk: bytes) -> RasaAudioBytes:
        """Convert the generated TTS audio bytes into Rasa audio bytes."""
        # If your service returns audio in mulaw format already, return directly
        return RasaAudioBytes(chunk)
        
        # If your service returns audio in a different format (e.g., PCM linear),
        # you'll need to convert it. For example, to convert from linear PCM:
        # import audioop
        # mulaw_data = audioop.lin2ulaw(chunk, 2)  # 2 = 16-bit samples
        # return RasaAudioBytes(mulaw_data)

    @staticmethod
    def get_default_config() -> MyTTSConfig:
        return MyTTSConfig(
            endpoint="wss://api.example.com/v1/speak",
            model_id="en-US-standard",
            language="en-US",
            voice="female-1",
            timeout=30,
        )

    @classmethod
    def from_config_dict(cls, config: Dict) -> "MyTTS":
        return cls(MyTTSConfig.from_dict(config))

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

browser_audio:
  # ... other configuration
  tts:
    name: addons.custom_tts.MyTTS
    api_key: "your_api_key"
    endpoint: "wss://api.example.com/v1/speak"
    language: "en-US"
    voice: "en-US-JennyNeural"
    timeout: 30
    # any other custom parameters your TTS needs