ACS Flows

ACS Call Automation & Media Flows¶

Azure Communication Services Integration¶

Speech Recognition Capabilities¶

Three-Thread Processing Architecture¶

Thread Responsibilities & Communication¶

Core Design Principles¶

The three-thread architecture follows these key principles:

🎤 Speech SDK Thread - Never Blocks¶

• Continuous audio recognition using Azure Speech SDK
• Immediate barge-in detection via on_partial callbacks
• Cross-thread communication via run_coroutine_threadsafe
• Performance: < 10ms response time for barge-in detection

🔄 Route Turn Thread - Blocks Only on Queue¶

• AI processing and response generation through orchestrator
• Queue-based serialization of conversation turns
• Safe cancellation without affecting speech recognition
• Performance: Processes one turn at a time, can be cancelled

🌐 Main Event Loop - Never Blocks¶

• WebSocket handling for real-time media streaming
• Task cancellation for barge-in scenarios
• Non-blocking coordination between threads
• Performance: < 50ms for task cancellation and stop commands

Thread Performance Matrix¶

Implementation Flow¶

Barge-in Detection and Handling¶

1. User speaks during AI response:
2. on_partial() callback fires immediately (< 10ms)
3. ThreadBridge.schedule_barge_in() schedules handler on main event loop

4. MainEventLoop.handle_barge_in() cancels current processing

5. Task cancellation chain:

   on_partial() → schedule_barge_in() → cancel_current_processing() → send_stop_audio()
   

6. Speech finalization:

7. on_final() callback queues completed speech via ThreadBridge.queue_speech_result()
8. RouteTurnThread picks up speech from queue
9. New AI processing task created for response generation

Key Components¶

ThreadBridge¶

Provides thread-safe communication between Speech SDK Thread and Main Event Loop: - schedule_barge_in() - Schedules barge-in handler execution - queue_speech_result() - Queues final speech for processing - Uses run_coroutine_threadsafe and asyncio.Queue for safe cross-thread communication

SpeechSDKThread¶

Manages Speech SDK in dedicated background thread: - Pre-initializes push_stream to prevent audio data loss - Never blocks on AI processing or network operations - Provides immediate callback execution for barge-in detection

RouteTurnThread¶

Handles AI processing in isolated thread: - Blocks only on speech_queue.get() operations - Processes speech through orchestrator - Creates and manages TTS playback tasks

MainEventLoop¶

Coordinates WebSocket operations and task management: - Handles incoming media messages and audio data - Manages barge-in interruption and task cancellation - Never blocks to ensure real-time responsiveness

🔄 Non-Blocking Thread Communication Sequence¶

🚀 Critical Non-Blocking Characteristics¶

🎯 Key Insight: Only the Route Turn Thread blocks (on queue.get()), ensuring Speech SDK and Main Event Loop remain responsive for real-time barge-in detection.

Key Implementation Details¶

This section provides concrete implementation specifics for developers working with the ACS Media Handler threading architecture.

🚨 Barge-In Detection¶

• Trigger: on_partial callback from Speech Recognizer detects user speech
• Immediate Action: Synchronous cancellation of playback_task using asyncio.Task.cancel()
• Stop Signal: Send {"Kind": "StopAudio", "StopAudio": {}} JSON command to ACS via WebSocket
• Logging: Comprehensive logging with emojis for real-time debugging

🔄 Async Background Task Management¶

• Route Turn Queue: Serializes final speech processing using asyncio.Queue()
• Playback Task: Tracks current AI response generation/playback with self.playback_task
• Task Lifecycle: Clean creation, cancellation, and cleanup of background tasks
• Cancellation Safety: Proper try/except asyncio.CancelledError handling

🛑 Stop Audio Signal Protocol¶

{
  "Kind": "StopAudio",
  "AudioData": null,
  "StopAudio": {}
}

⚡ Error Handling & Resilience¶

• Event Loop Detection: Graceful handling when no event loop is available
• WebSocket Validation: Connection state checks before sending messages
• Task Cancellation: Proper cleanup with await task after cancellation
• Queue Management: Full queue detection and message dropping strategies

📊 Performance Optimizations¶

• Immediate Cancellation: Barge-in triggers instant playback stop (< 50ms)
• Background Processing: Non-blocking AI response generation
• Memory Management: Proper task cleanup prevents memory leaks
• Concurrent Safety: Thread-safe queue operations for speech processing