Spaces:
Runtime error
Runtime error
| """ | |
| Pipecat Flow Handlers for RTVI Integration | |
| This module provides handlers for integrating Pipecat flows with RTVI punt system, | |
| allowing client-side code execution while maintaining server-side flow management. | |
| """ | |
| from datetime import datetime | |
| from typing import Callable, Optional | |
| from pipecat.processors.frameworks.rtvi import RTVIProcessor, RTVIServerMessageFrame | |
| from pipecat_flows import FlowConfig, FlowManager, FlowsFunctionSchema | |
| from loguru import logger | |
| # Define the set of action types that should get no-op handlers | |
| NO_OP_ACTION_TYPES = { | |
| "save_summary_conversation", | |
| "node_summary_conversation", | |
| "save_html_conversation", | |
| } | |
| def create_stub_handler( | |
| node_name: str, | |
| function_name: str, | |
| rtvi_client: RTVIProcessor, | |
| transition_to: Optional[str] = None, | |
| ) -> Callable: | |
| """ | |
| Create a stub handler that forwards function calls to client via RTVI. | |
| This enables "RTVI stubs" - when server-side flow encounters a client function, | |
| it sends the request to the client for execution while maintaining flow state. | |
| Args: | |
| node_name: Name of the node that contains the function | |
| function_name: Name of the function to punt to client | |
| rtvi_client: RTVI client for sending messages | |
| transition_to: Optional node to transition to after function execution | |
| Returns: | |
| Async handler function for use with FlowManager | |
| """ | |
| async def stub_handler(params, flow_manager: FlowManager): | |
| """Fire-and-forget stub handler""" | |
| try: | |
| # Create function request message | |
| message = { | |
| "type": "function_request", | |
| "function_name": function_name, | |
| "params": [params], | |
| "timestamp": datetime.now().isoformat(), | |
| } | |
| # Send to client via RTVI server message frame | |
| frame = RTVIServerMessageFrame(data=message) | |
| await flow_manager.task.queue_frame(frame) | |
| logger.info(f"🚀 Punted function '{function_name}' to client") | |
| if transition_to: | |
| state_message = { | |
| "type": "flow_state_sync", | |
| "current_node": transition_to, | |
| "previous_node": node_name, | |
| "timestamp": datetime.now().isoformat(), | |
| } | |
| frame = RTVIServerMessageFrame(data=state_message) | |
| await flow_manager.task.queue_frame(frame) | |
| logger.info(f"🔄 State sync: {node_name} → {transition_to}") | |
| return {"status": "success"} | |
| except Exception as e: | |
| logger.error(f"❌ Failed to punt function '{function_name}': {e}") | |
| return {"status": "error", "error": str(e)} | |
| return stub_handler | |
| async def send_state_sync(flow_manager: FlowManager, node_name: str): | |
| state_message = { | |
| "type": "flow_state_sync", | |
| "current_node": node_name, | |
| "timestamp": datetime.now().isoformat(), | |
| } | |
| frame = RTVIServerMessageFrame(data=state_message) | |
| await flow_manager.task.queue_frame(frame) | |
| logger.info(f"🔄 State sync: {node_name}") | |
| def create_no_op_action_handler() -> Callable: | |
| """ | |
| Create a no-op handler for actions that should be ignored on server-side. | |
| Returns: | |
| Async handler function that does nothing | |
| """ | |
| async def no_op_handler(action: dict, flow_manager: FlowManager): | |
| """No-op handler for client-side only actions""" | |
| logger.debug( | |
| f"No-op handler called for action type: {action.get('type', 'unknown')}" | |
| ) | |
| pass | |
| return no_op_handler | |
| async def end_conversation_handler(action: dict, flow_manager: FlowManager): | |
| """Handler for end_conversation action - works for both Daily and Twilio""" | |
| try: | |
| # Import here to avoid circular imports | |
| from utils.twilio import terminate_call, get_call_context | |
| # Determine call source | |
| if not action or action == {}: | |
| call_source = "function_call" | |
| else: | |
| call_source = "post_action" | |
| logger.info(f"🏁 End conversation called with source {call_source}") | |
| # Create end_conversation message with source information | |
| message = { | |
| "type": "end_conversation", | |
| "timestamp": datetime.now().isoformat(), | |
| "call_source": call_source, | |
| "immediate_disconnect": call_source == "function_call", | |
| } | |
| # Try to send RTVI message (works for Daily/WebRTC clients) | |
| try: | |
| frame = RTVIServerMessageFrame(data=message) | |
| await flow_manager.task.queue_frame(frame) | |
| logger.info("🏁 Sent end_conversation message to client") | |
| except Exception as rtvi_error: | |
| logger.debug(f"RTVI message not sent (expected for Twilio): {rtvi_error}") | |
| # For Twilio calls, terminate via REST API | |
| call_context = get_call_context() | |
| if call_context and call_context.get("call_sid"): | |
| call_sid = call_context["call_sid"] | |
| logger.info(f"🏁 Terminating Twilio call via REST API: {call_sid}") | |
| await terminate_call(call_sid) | |
| # Cancel the task to end the call (works for both Daily and Twilio) | |
| # This ensures the call is terminated regardless of transport type | |
| if call_source == "function_call": | |
| logger.info("🏁 Cancelling task to end call") | |
| await flow_manager.task.cancel() | |
| except Exception as e: | |
| logger.error(f"❌ Failed to end conversation: {e}") | |
| def process_flow_config_handlers( | |
| llm_flow: FlowConfig, | |
| rtvi_client: RTVIProcessor, | |
| ) -> FlowConfig: | |
| """ | |
| Process flow configuration and replace string handlers with RTVI punt handlers. | |
| This is the core of the RTVI punt system - it finds string handlers like | |
| "__function__:handleTransition" and replaces them with lambda functions | |
| that send function requests to the client. | |
| Args: | |
| llm_flow: LLM flow configuration | |
| rtvi_client: RTVI client for sending messages | |
| Returns: | |
| Processed flow config with punt handlers | |
| """ | |
| processed_config = llm_flow.copy() | |
| # Get the handlers once | |
| no_op_handler = create_no_op_action_handler() | |
| # Process each node | |
| for node_name, node_config in processed_config.get("nodes", {}).items(): | |
| # Process functions | |
| functions = node_config.get("functions", []) | |
| # Inject end_conversation function at every node | |
| end_conversation_function = FlowsFunctionSchema( | |
| name="end_conversation", | |
| description="Call this function when the user explicitly asks to stop, end, or terminate the current interview or conversation. This should be called when user clearly expresses they want to finish the interview session.", | |
| properties={}, | |
| required=[], | |
| handler=end_conversation_handler, | |
| ) | |
| functions.append(end_conversation_function) | |
| logger.info(f"💬 Injected end_conversation function in node: {node_name}") | |
| for i, func_outer in enumerate(functions): | |
| # Handle both dict format and FlowsFunctionSchema | |
| if isinstance(func_outer, FlowsFunctionSchema): | |
| func = func_outer | |
| func_name = func.name | |
| else: | |
| func = func_outer["function"] | |
| func_name = func["name"] | |
| # Skip processing if this is the end_conversation function we just added | |
| if func_name == "end_conversation": | |
| continue | |
| # Convert to FlowsFunctionSchema if it's not already | |
| if isinstance(func_outer, dict): | |
| func_data = func_outer["function"] | |
| name = func_data["name"] | |
| description = func_data.get("description", "") | |
| parameters = func_data.get("parameters", {}) or {} | |
| properties = parameters.get("properties", {}) | |
| required = parameters.get("required", []) | |
| # Extract Flows-specific fields | |
| transition_to = func_data.get("transition_to") | |
| # Create RTVI stub handler | |
| handler = create_stub_handler( | |
| node_name, func_name, rtvi_client, transition_to=transition_to | |
| ) | |
| logger.info(f"📤 Converted {func_name} to RTVI stub") | |
| new_func = FlowsFunctionSchema( | |
| name=name, | |
| description=description, | |
| properties=properties, | |
| required=required, | |
| handler=handler, | |
| transition_to=transition_to, | |
| ) | |
| functions[i] = new_func | |
| # Process pre_actions | |
| pre_actions = node_config.get("pre_actions", []) | |
| for action in pre_actions: | |
| action_type = action.get("type") | |
| if action_type in NO_OP_ACTION_TYPES: | |
| action["handler"] = no_op_handler | |
| logger.info( | |
| f"Added no-op handler to pre_action: {action_type} in node: {node_name}" | |
| ) | |
| # Process post_actions | |
| post_actions = node_config.get("post_actions", []) | |
| for action in post_actions: | |
| action_type = action.get("type") | |
| if action_type in NO_OP_ACTION_TYPES: | |
| action["handler"] = no_op_handler | |
| logger.info( | |
| f"Added no-op handler to post_action: {action_type} in node: {node_name}" | |
| ) | |
| logger.info("✅ Flow config handlers processed for RTVI stubs and actions") | |
| return processed_config | |