Spaces:
Runtime error
Runtime error
File size: 2,892 Bytes
24cdea4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 | """
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
"""
@rtvi.event_handler("on_client_ready")
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")
|