""" Complete Wakeword to Voice Recognition Workflow for Trixy ML System. This module implements the complete conversation workflow from wakeword detection through voice recognition and server communication. It orchestrates the entire process according to the specifications in project_description.md and CLAUDE.md. Key Features: - Complete wakeword detection -> voice recognition workflow - Server communication and arbitration integration - Audio buffer management during detection phases - Conversation session management - Multi-satellite arbitration support - Timeout handling and error recovery - Real-time audio streaming to server - Integration with network protocol Workflow Steps: 1. Continuous wakeword detection 2. Wakeword detected -> pause detection, start buffering 3. Enhance wakeword with speaker identification 4. Send wakeword command to server with speaker info 5. Wait for server selection (1 second arbitration window) 6. If selected: start recording and streaming audio 7. Handle silence detection and recording timeout 8. Send recording completion command 9. Return to wakeword detection mode Usage: from trixy_core.ml.workflow import MLWorkflowManager, WorkflowConfig # Create workflow manager workflow = MLWorkflowManager( ml_manager=ml_manager, network_manager=network_manager, config=workflow_config ) # Start workflow workflow.start() """ import threading import time import queue import logging from typing import Dict, List, Optional, Any, Callable, Union, Tuple from dataclasses import dataclass, field from datetime import datetime from enum import Enum import uuid import numpy as np # Import ML components from .ml_manager import MLManager, MLProcessingState as MLState from .wakeword.audio_buffer import AudioBuffer from .audio_pipeline import AudioPipeline, AudioChunk, AudioPipelineState as SilenceState # Import network components try: from ..network import NetworkManager, ClientSocket NETWORK_AVAILABLE = True except ImportError: NETWORK_AVAILABLE = False NetworkManager = object ClientSocket = object # Import event system try: from ..events import EventHandler, TrixyEvent, EventDataFactory EVENTS_AVAILABLE = True except ImportError: EVENTS_AVAILABLE = False EventHandler = object TrixyEvent = lambda events: lambda func: func EventDataFactory = object # Configure logging logger = logging.getLogger(__name__) class WorkflowState(Enum): """Workflow execution states.""" IDLE = "idle" DETECTING = "detecting" WAKEWORD_PROCESSING = "wakeword_processing" WAITING_FOR_SERVER = "waiting_for_server" RECORDING = "recording" STREAMING = "streaming" COMPLETING = "completing" ERROR = "error" class WorkflowStep(Enum): """Individual workflow steps.""" WAKEWORD_DETECTION = "wakeword_detection" SPEAKER_IDENTIFICATION = "speaker_identification" SERVER_COMMUNICATION = "server_communication" ARBITRATION_WAIT = "arbitration_wait" RECORDING_START = "recording_start" AUDIO_STREAMING = "audio_streaming" SILENCE_DETECTION = "silence_detection" RECORDING_COMPLETION = "recording_completion" CLEANUP = "cleanup" class TimeoutType(Enum): """Types of timeouts in the workflow.""" SERVER_SELECTION = "server_selection" RECORDING_MAX_DURATION = "recording_max_duration" SILENCE_TIMEOUT = "silence_timeout" NETWORK_RESPONSE = "network_response" @dataclass class WorkflowConfig: """Configuration for the workflow manager.""" # Timing settings server_selection_timeout: float = 10.0 arbitration_window_seconds: float = 1.0 max_recording_duration: float = 60.0 silence_timeout_seconds: float = 3.0 network_timeout_seconds: float = 5.0 # Buffer settings pre_trigger_buffer_seconds: float = 1.0 post_trigger_buffer_seconds: float = 1.0 streaming_chunk_size: int = 4096 # Quality settings min_wakeword_confidence: float = 0.7 min_speaker_confidence: float = 0.5 enable_quality_gates: bool = True # Recovery settings max_retry_attempts: int = 3 retry_delay_seconds: float = 1.0 enable_automatic_recovery: bool = True # Debug settings enable_workflow_logging: bool = True log_audio_metrics: bool = False save_debug_audio: bool = False @dataclass class ConversationSession: """Represents an active conversation session.""" session_id: str start_time: float wakeword_type: str wakeword_confidence: float speaker_id: Optional[str] speaker_name: Optional[str] speaker_confidence: float audio_buffer: Optional[AudioBuffer] is_selected: bool = False is_recording: bool = False is_streaming: bool = False bytes_streamed: int = 0 last_activity: float = field(default_factory=time.time) metadata: Dict[str, Any] = field(default_factory=dict) @dataclass class WorkflowMetrics: """Metrics for workflow performance monitoring.""" # Workflow execution metrics total_wakewords: int = 0 successful_workflows: int = 0 failed_workflows: int = 0 timeout_workflows: int = 0 # Timing metrics avg_wakeword_to_recording_ms: float = 0.0 avg_recording_duration_seconds: float = 0.0 avg_silence_detection_ms: float = 0.0 avg_server_response_ms: float = 0.0 # Quality metrics avg_wakeword_confidence: float = 0.0 avg_speaker_confidence: float = 0.0 quality_gate_failures: int = 0 # Network metrics server_selections: int = 0 server_rejections: int = 0 network_errors: int = 0 bytes_transmitted: int = 0 # Error tracking error_counts: Dict[str, int] = field(default_factory=dict) last_error: Optional[str] = None # Timestamp last_update: str = field(default_factory=lambda: datetime.now().isoformat()) class WorkflowError(Exception): """Workflow execution errors.""" pass class MLWorkflowManager: """ Complete ML workflow manager implementing the full conversation flow. This class orchestrates the entire process from wakeword detection through voice recognition and server communication, handling all the steps outlined in the project specification. """ def __init__( self, ml_manager: MLManager, network_manager: Optional[Any] = None, event_handler: Optional[EventHandler] = None, config: Optional[WorkflowConfig] = None ): """ Initialize workflow manager. Args: ml_manager: ML manager instance network_manager: Network manager for server communication event_handler: Event handler for system integration config: Workflow configuration """ self.ml_manager = ml_manager self.network_manager = network_manager self.event_handler = event_handler self.config = config or WorkflowConfig() # Workflow state self._state = WorkflowState.IDLE self._current_step = WorkflowStep.WAKEWORD_DETECTION self._state_lock = threading.RLock() # Active session self._current_session: Optional[ConversationSession] = None self._session_lock = threading.RLock() # Workflow control self._is_running = False self._workflow_thread: Optional[threading.Thread] = None self._stop_event = threading.Event() # Timeout management self._active_timeouts: Dict[str, threading.Timer] = {} self._timeout_lock = threading.RLock() # Metrics self._metrics = WorkflowMetrics() self._start_time = time.time() # Audio streaming self._streaming_queue = queue.Queue() self._streaming_thread: Optional[threading.Thread] = None # Callbacks self.on_workflow_started: Optional[Callable] = None self.on_workflow_completed: Optional[Callable] = None self.on_workflow_failed: Optional[Callable] = None self.on_session_created: Optional[Callable] = None self.on_recording_started: Optional[Callable] = None self.on_recording_completed: Optional[Callable] = None logger.info("ML Workflow Manager initialized") # Set up ML manager callbacks self._setup_ml_callbacks() # Register event handlers if EVENTS_AVAILABLE and self.event_handler: self._register_event_handlers() def start(self) -> None: """Start the workflow manager.""" if self._is_running: logger.warning("Workflow manager already running") return logger.info("Starting ML workflow manager...") self._is_running = True self._stop_event.clear() try: # Start workflow monitoring thread self._workflow_thread = threading.Thread( target=self._workflow_monitoring_loop, daemon=True, name="MLWorkflow-Monitor" ) self._workflow_thread.start() # Start streaming thread self._streaming_thread = threading.Thread( target=self._audio_streaming_loop, daemon=True, name="MLWorkflow-Streaming" ) self._streaming_thread.start() # Set initial state with self._state_lock: self._state = WorkflowState.DETECTING self._current_step = WorkflowStep.WAKEWORD_DETECTION # Trigger started callback if self.on_workflow_started: self.on_workflow_started() logger.info("ML workflow manager started successfully") except Exception as e: self._is_running = False logger.error(f"Failed to start workflow manager: {e}") raise WorkflowError(f"Failed to start workflow manager: {e}") from e def stop(self) -> None: """Stop the workflow manager.""" if not self._is_running: return logger.info("Stopping ML workflow manager...") self._is_running = False self._stop_event.set() try: # Cancel any active timeouts self._cancel_all_timeouts() # Complete current session if active with self._session_lock: if self._current_session: self._complete_session("workflow_stopped") # Stop monitoring thread if self._workflow_thread and self._workflow_thread.is_alive(): self._workflow_thread.join(timeout=5.0) # Stop streaming thread if self._streaming_thread and self._streaming_thread.is_alive(): self._streaming_thread.join(timeout=2.0) with self._state_lock: self._state = WorkflowState.IDLE logger.info("ML workflow manager stopped") except Exception as e: logger.error(f"Error stopping workflow manager: {e}") def get_current_state(self) -> WorkflowState: """Get current workflow state.""" with self._state_lock: return self._state def get_current_session(self) -> Optional[ConversationSession]: """Get current conversation session.""" with self._session_lock: return self._current_session def get_metrics(self) -> WorkflowMetrics: """Get workflow metrics.""" self._metrics.last_update = datetime.now().isoformat() return self._metrics def handle_server_selection_response(self, session_id: str, selected: bool) -> None: """ Handle server selection response. Args: session_id: Session ID from server selected: Whether this client was selected """ logger.info(f"Received server selection: session_id={session_id}, selected={selected}") with self._session_lock: if not self._current_session or self._current_session.session_id != session_id: logger.warning(f"Received selection for unknown session: {session_id}") return # Cancel selection timeout self._cancel_timeout("server_selection") if selected: # We were selected - start recording self._current_session.is_selected = True self._metrics.server_selections += 1 self._start_recording_phase() else: # We were not selected - clean up and return to detection self._metrics.server_rejections += 1 self._complete_session("not_selected") def _setup_ml_callbacks(self) -> None: """Set up callbacks for ML manager events.""" # Set wakeword detection callback self.ml_manager.on_wakeword_detected = self._handle_wakeword_detected # Set speaker identification callback self.ml_manager.on_speaker_identified = self._handle_speaker_identified # Set audio ready callback self.ml_manager.on_audio_ready = self._handle_audio_ready # Set error callback self.ml_manager.on_error = self._handle_ml_error def _handle_wakeword_detected(self, detection_result, speaker_id: Optional[str], speaker_name: Optional[str]) -> None: """Handle wakeword detection from ML manager.""" logger.info(f"Workflow handling wakeword detection: {detection_result.wakeword_type.value}") with self._state_lock: if self._state != WorkflowState.DETECTING: logger.warning("Ignoring wakeword - not in detecting state") return self._state = WorkflowState.WAKEWORD_PROCESSING self._current_step = WorkflowStep.SPEAKER_IDENTIFICATION # Update metrics self._metrics.total_wakewords += 1 self._metrics.avg_wakeword_confidence = ( (self._metrics.avg_wakeword_confidence * (self._metrics.total_wakewords - 1) + detection_result.confidence) / self._metrics.total_wakewords ) # Check quality gates if self.config.enable_quality_gates: if detection_result.confidence < self.config.min_wakeword_confidence: logger.warning(f"Wakeword confidence too low: {detection_result.confidence}") self._metrics.quality_gate_failures += 1 self._return_to_detection("quality_gate_failure") return # Create conversation session session = ConversationSession( session_id=str(uuid.uuid4()), start_time=time.time(), wakeword_type=detection_result.wakeword_type.value, wakeword_confidence=detection_result.confidence, speaker_id=speaker_id, speaker_name=speaker_name, speaker_confidence=0.0, # Will be updated by speaker identification audio_buffer=None ) with self._session_lock: self._current_session = session # Trigger session created callback if self.on_session_created: try: self.on_session_created(session) except Exception as e: logger.error(f"Error in session created callback: {e}") # Continue with workflow self._continue_wakeword_workflow(session, detection_result) def _continue_wakeword_workflow(self, session: ConversationSession, detection_result) -> None: """Continue workflow after wakeword detection.""" start_time = time.time() try: # Step 1: Speaker identification (already done in ML manager) if session.speaker_id: # Update metrics if hasattr(detection_result, 'speaker_confidence'): session.speaker_confidence = detection_result.speaker_confidence self._metrics.avg_speaker_confidence = ( (self._metrics.avg_speaker_confidence * self._metrics.total_wakewords + session.speaker_confidence) / (self._metrics.total_wakewords + 1) ) # Check speaker confidence quality gate if self.config.enable_quality_gates: if session.speaker_confidence < self.config.min_speaker_confidence: logger.warning(f"Speaker confidence too low: {session.speaker_confidence}") self._metrics.quality_gate_failures += 1 self._complete_session("speaker_quality_gate_failure") return # Step 2: Send wakeword command to server with self._state_lock: self._current_step = WorkflowStep.SERVER_COMMUNICATION self._send_wakeword_command(session) # Step 3: Wait for server selection with self._state_lock: self._state = WorkflowState.WAITING_FOR_SERVER self._current_step = WorkflowStep.ARBITRATION_WAIT # Set timeout for server selection self._set_timeout( "server_selection", self.config.server_selection_timeout, lambda: self._handle_timeout(TimeoutType.SERVER_SELECTION) ) # Update timing metrics processing_time = (time.time() - start_time) * 1000 self._metrics.avg_wakeword_to_recording_ms = ( (self._metrics.avg_wakeword_to_recording_ms * (self._metrics.total_wakewords - 1) + processing_time) / self._metrics.total_wakewords ) except Exception as e: logger.error(f"Error in wakeword workflow: {e}") self._handle_workflow_error(f"Wakeword workflow error: {e}") def _send_wakeword_command(self, session: ConversationSession) -> None: """Send wakeword detected command to server.""" if not self.network_manager: logger.warning("No network manager - cannot send wakeword command") return try: # Prepare command data command_data = { 'command': 'wakeword_detected', 'session_id': session.session_id, 'wakeword_type': session.wakeword_type, 'wakeword_confidence': session.wakeword_confidence, 'speaker_id': session.speaker_id, 'speaker_name': session.speaker_name, 'speaker_confidence': session.speaker_confidence, 'timestamp': datetime.now().isoformat(), 'client_id': getattr(self.network_manager, 'client_id', 'unknown') } # Send command via network manager # In a real implementation, this would use the actual Trixy protocol logger.info(f"Sending wakeword command: {command_data}") # Simulate network send for now # self.network_manager.send_command(command_data) except Exception as e: logger.error(f"Error sending wakeword command: {e}") self._metrics.network_errors += 1 raise def _start_recording_phase(self) -> None: """Start the recording phase after server selection.""" logger.info("Starting recording phase") with self._state_lock: self._state = WorkflowState.RECORDING self._current_step = WorkflowStep.RECORDING_START with self._session_lock: if not self._current_session: logger.error("No active session for recording") return self._current_session.is_recording = True self._current_session.last_activity = time.time() # Set recording timeout self._set_timeout( "recording_duration", self.config.max_recording_duration, lambda: self._handle_timeout(TimeoutType.RECORDING_MAX_DURATION) ) # Start recording in ML manager (this would start the audio buffer) # In practice, the ML manager would handle the actual recording logger.info("Recording started - audio buffer active") # Trigger recording started callback if self.on_recording_started: try: self.on_recording_started(self._current_session) except Exception as e: logger.error(f"Error in recording started callback: {e}") # Transition to streaming self._start_streaming_phase() def _start_streaming_phase(self) -> None: """Start streaming audio to server.""" logger.info("Starting audio streaming phase") with self._state_lock: self._state = WorkflowState.STREAMING self._current_step = WorkflowStep.AUDIO_STREAMING with self._session_lock: if self._current_session: self._current_session.is_streaming = True # In a real implementation, this would start streaming audio chunks # to the server via the raw audio input socket logger.info("Audio streaming to server started") def _handle_audio_ready(self, audio_data: np.ndarray, conversation_id: str) -> None: """Handle audio ready for streaming.""" if not self._current_session or self._current_session.session_id != conversation_id: return # Add audio to streaming queue try: audio_chunk = { 'data': audio_data, 'session_id': conversation_id, 'timestamp': time.time(), 'sequence': getattr(self, '_stream_sequence', 0) } if not self._streaming_queue.full(): self._streaming_queue.put(audio_chunk) self._stream_sequence = getattr(self, '_stream_sequence', 0) + 1 else: logger.warning("Streaming queue full - dropping audio chunk") except Exception as e: logger.error(f"Error queuing audio for streaming: {e}") def _audio_streaming_loop(self) -> None: """Audio streaming loop for sending data to server.""" logger.debug("Audio streaming loop started") while not self._stop_event.is_set(): try: # Get audio chunk from queue try: chunk = self._streaming_queue.get(timeout=0.1) except queue.Empty: continue # Stream to server self._stream_audio_chunk(chunk) # Update metrics with self._session_lock: if self._current_session: self._current_session.bytes_streamed += len(chunk['data']) * 4 # float32 self._metrics.bytes_transmitted += len(chunk['data']) * 4 # Mark task done self._streaming_queue.task_done() except Exception as e: logger.error(f"Error in audio streaming loop: {e}") time.sleep(0.1) logger.debug("Audio streaming loop stopped") def _stream_audio_chunk(self, chunk: Dict[str, Any]) -> None: """Stream individual audio chunk to server.""" try: # In a real implementation, this would send the audio data # through the raw audio input socket to the server logger.debug(f"Streaming audio chunk: {len(chunk['data'])} samples") # Simulate network transmission # self.network_manager.stream_audio(chunk['data']) except Exception as e: logger.error(f"Error streaming audio chunk: {e}") self._metrics.network_errors += 1 def _handle_silence_timeout(self) -> None: """Handle silence timeout during recording.""" logger.info("Silence timeout detected - completing recording") with self._session_lock: if self._current_session: self._complete_recording("silence_timeout") def _complete_recording(self, reason: str) -> None: """Complete the recording phase.""" logger.info(f"Completing recording: {reason}") with self._state_lock: self._state = WorkflowState.COMPLETING self._current_step = WorkflowStep.RECORDING_COMPLETION # Cancel recording timeout self._cancel_timeout("recording_duration") with self._session_lock: if not self._current_session: return recording_duration = time.time() - self._current_session.start_time self._current_session.is_recording = False self._current_session.is_streaming = False # Send recording completion command to server self._send_recording_completion_command(reason) # Update metrics self._metrics.avg_recording_duration_seconds = ( (self._metrics.avg_recording_duration_seconds * self._metrics.successful_workflows + recording_duration) / (self._metrics.successful_workflows + 1) ) # Trigger recording completed callback if self.on_recording_completed: try: self.on_recording_completed(self._current_session, reason) except Exception as e: logger.error(f"Error in recording completed callback: {e}") # Complete the session self._complete_session(reason) def _send_recording_completion_command(self, reason: str) -> None: """Send recording completion command to server.""" if not self.network_manager or not self._current_session: return try: command_data = { 'command': 'recording_completed', 'session_id': self._current_session.session_id, 'reason': reason, 'duration_seconds': time.time() - self._current_session.start_time, 'bytes_streamed': self._current_session.bytes_streamed, 'timestamp': datetime.now().isoformat() } logger.info(f"Sending recording completion command: {command_data}") # In real implementation: self.network_manager.send_command(command_data) except Exception as e: logger.error(f"Error sending recording completion command: {e}") self._metrics.network_errors += 1 def _complete_session(self, reason: str) -> None: """Complete the current conversation session.""" logger.info(f"Completing session: {reason}") with self._session_lock: if not self._current_session: return session = self._current_session self._current_session = None # Update metrics based on completion reason if reason in ["silence_timeout", "recording_completed"]: self._metrics.successful_workflows += 1 elif "timeout" in reason: self._metrics.timeout_workflows += 1 else: self._metrics.failed_workflows += 1 # Trigger workflow completed callback if self.on_workflow_completed: try: self.on_workflow_completed(session, reason) except Exception as e: logger.error(f"Error in workflow completed callback: {e}") # Return to detection mode self._return_to_detection(reason) def _return_to_detection(self, reason: str) -> None: """Return to wakeword detection mode.""" logger.info(f"Returning to detection mode: {reason}") # Cancel all timeouts self._cancel_all_timeouts() # Clear streaming queue while not self._streaming_queue.empty(): try: self._streaming_queue.get_nowait() except queue.Empty: break with self._state_lock: self._state = WorkflowState.DETECTING self._current_step = WorkflowStep.WAKEWORD_DETECTION logger.info("Returned to wakeword detection mode") def _handle_timeout(self, timeout_type: TimeoutType) -> None: """Handle workflow timeouts.""" logger.warning(f"Workflow timeout: {timeout_type.value}") if timeout_type == TimeoutType.SERVER_SELECTION: self._complete_session("server_selection_timeout") elif timeout_type == TimeoutType.RECORDING_MAX_DURATION: self._complete_recording("max_duration_timeout") elif timeout_type == TimeoutType.SILENCE_TIMEOUT: self._handle_silence_timeout() # Update timeout metrics self._metrics.timeout_workflows += 1 def _handle_workflow_error(self, error_message: str) -> None: """Handle workflow errors.""" logger.error(f"Workflow error: {error_message}") # Update error metrics error_type = error_message.split(':')[0] if ':' in error_message else "unknown" self._metrics.error_counts[error_type] = self._metrics.error_counts.get(error_type, 0) + 1 self._metrics.last_error = error_message # Trigger workflow failed callback if self.on_workflow_failed: try: self.on_workflow_failed(error_message) except Exception as e: logger.error(f"Error in workflow failed callback: {e}") # Complete current session if active with self._session_lock: if self._current_session: self._complete_session(f"error: {error_message}") else: self._return_to_detection(f"error: {error_message}") def _handle_speaker_identified(self, result) -> None: """Handle continuous speaker identification.""" # This is for continuous recognition mode if self._state == WorkflowState.DETECTING and self.config.enable_workflow_logging: logger.debug(f"Continuous speaker identification: {result.speaker_name}") def _handle_ml_error(self, error_message: str) -> None: """Handle ML manager errors.""" self._handle_workflow_error(f"ML error: {error_message}") def _set_timeout(self, name: str, delay: float, callback: Callable) -> None: """Set a named timeout.""" with self._timeout_lock: # Cancel existing timeout with same name if name in self._active_timeouts: self._active_timeouts[name].cancel() # Create new timeout timer = threading.Timer(delay, callback) timer.start() self._active_timeouts[name] = timer logger.debug(f"Set timeout '{name}' for {delay} seconds") def _cancel_timeout(self, name: str) -> None: """Cancel a named timeout.""" with self._timeout_lock: if name in self._active_timeouts: self._active_timeouts[name].cancel() del self._active_timeouts[name] logger.debug(f"Cancelled timeout '{name}'") def _cancel_all_timeouts(self) -> None: """Cancel all active timeouts.""" with self._timeout_lock: for name, timer in self._active_timeouts.items(): timer.cancel() logger.debug(f"Cancelled timeout '{name}'") self._active_timeouts.clear() def _workflow_monitoring_loop(self) -> None: """Background monitoring loop for workflow health.""" logger.debug("Workflow monitoring loop started") while not self._stop_event.is_set(): try: # Check for stuck states current_time = time.time() with self._session_lock: if self._current_session: session_age = current_time - self._current_session.start_time activity_age = current_time - self._current_session.last_activity # Check for session timeout if session_age > 300: # 5 minutes logger.warning("Session timeout - completing session") self._complete_session("session_timeout") # Check for activity timeout elif activity_age > 60: # 1 minute logger.warning("Activity timeout - completing session") self._complete_session("activity_timeout") # Sleep for monitoring interval self._stop_event.wait(5.0) except Exception as e: logger.error(f"Error in workflow monitoring: {e}") time.sleep(5.0) logger.debug("Workflow monitoring loop stopped") def _register_event_handlers(self) -> None: """Register event handlers with the event system.""" if not EVENTS_AVAILABLE or not self.event_handler: return try: # In a real implementation, these would be proper event handler methods # with @TrixyEvent decorators for handling server responses logger.info("Event handlers registered for workflow manager") except Exception as e: logger.error(f"Error registering event handlers: {e}") def create_workflow_manager( ml_manager: MLManager, network_manager: Optional[Any] = None, event_handler: Optional[EventHandler] = None, config: Optional[WorkflowConfig] = None ) -> MLWorkflowManager: """ Factory function to create a workflow manager. Args: ml_manager: ML manager instance network_manager: Network manager for server communication event_handler: Event handler for system integration config: Workflow configuration Returns: Configured workflow manager """ return MLWorkflowManager( ml_manager=ml_manager, network_manager=network_manager, event_handler=event_handler, config=config ) def create_default_workflow_config(**kwargs) -> WorkflowConfig: """ Create default workflow configuration. Args: **kwargs: Configuration overrides Returns: Workflow configuration """ return WorkflowConfig(**kwargs) # Export main classes and functions __all__ = [ 'MLWorkflowManager', 'WorkflowConfig', 'ConversationSession', 'WorkflowMetrics', 'WorkflowState', 'WorkflowStep', 'TimeoutType', 'WorkflowError', 'create_workflow_manager', 'create_default_workflow_config' ]