Spaces:
Runtime error
Runtime error
| """ | |
| RTVI processor setup and configuration. | |
| Provides RTVI (Real-Time Voice Interface) processor initialization | |
| with standard actions like TTS say and interrupt. | |
| """ | |
| from typing import Dict, Any, Optional, Callable | |
| from loguru import logger | |
| from pipecat.frames.frames import BotInterruptionFrame, TTSSpeakFrame | |
| from pipecat.pipeline.pipeline import Pipeline | |
| from pipecat.processors.frameworks.rtvi import ( | |
| RTVIConfig, | |
| RTVIAction, | |
| RTVIActionArgument, | |
| RTVIProcessor, | |
| ActionResult, | |
| ) | |
| def setup_rtvi_processor( | |
| pipeline_getter: Callable[[], Pipeline], | |
| log_func: Optional[Callable[[str], None]] = None, | |
| ) -> RTVIProcessor: | |
| """ | |
| Create and configure an RTVI processor with standard actions. | |
| Args: | |
| pipeline_getter: Callable that returns the pipeline for frame queueing | |
| log_func: Optional logging function for action events | |
| Returns: | |
| Configured RTVIProcessor with tts:say and tts:interrupt actions | |
| """ | |
| rtvi = RTVIProcessor(config=RTVIConfig(config=[])) | |
| async def handle_tts_say( | |
| processor: RTVIProcessor, service: str, arguments: Dict[str, Any] | |
| ) -> ActionResult: | |
| """Handle tts:say action - queue text for speech.""" | |
| text = arguments.get("text", "") | |
| if text: | |
| pipeline = pipeline_getter() | |
| await pipeline.queue_frame(TTSSpeakFrame(text=text)) | |
| if log_func: | |
| log_func(f"TTS Say triggered [text={text}]") | |
| return True | |
| rtvi.register_action(RTVIAction( | |
| service="tts", | |
| action="say", | |
| arguments=[RTVIActionArgument(name="text", type="string")], | |
| result="bool", | |
| handler=handle_tts_say, | |
| )) | |
| async def handle_tts_interrupt( | |
| processor: RTVIProcessor, service: str, arguments: Dict[str, Any] | |
| ) -> ActionResult: | |
| """Handle tts:interrupt action - interrupt current speech.""" | |
| pipeline = pipeline_getter() | |
| await pipeline.queue_frame(BotInterruptionFrame()) | |
| if log_func: | |
| log_func("TTS Interrupt triggered") | |
| return True | |
| rtvi.register_action(RTVIAction( | |
| service="tts", | |
| action="interrupt", | |
| arguments=[], | |
| result="bool", | |
| handler=handle_tts_interrupt, | |
| )) | |
| logger.info("RTVI processor configured with tts:say and tts:interrupt actions") | |
| return rtvi | |
| def setup_client_ready_handler(rtvi: RTVIProcessor, log_func: Optional[Callable[[str], None]] = None): | |
| """ | |
| Setup on_client_ready event handler for RTVI. | |
| Args: | |
| rtvi: RTVI processor to attach handler to | |
| log_func: Optional logging function | |
| """ | |
| async def on_client_ready(rtvi_instance): | |
| if log_func: | |
| log_func("Client ready") | |
| await rtvi_instance.set_bot_ready() | |
| logger.debug("RTVI client ready handler configured") | |