Audio Processing Pipeline
Relevant source files
The following files were used as context for generating this wiki page:
The Audio Processing Pipeline is a sophisticated multi-stage system designed to handle real-time audio ingestion, transcription, and intelligent analysis for Omi devices and applications. It transforms raw audio captured by hardware wearables or mobile apps into structured data, such as summaries, action items, and plugin-specific triggers.
The pipeline operates across several layers: hardware (Omi Wearable), mobile (Flutter), and a distributed backend (Python/FastAPI) utilizing specialized GPU services for Voice Activity Detection (VAD) and Diarization.
System Architecture
The pipeline follows a tiered architecture where audio moves from the edge device to specialized backend services.
The architecture leverages a "Pusher" service for binary WebSocket handling and specialized GPU-accelerated nodes for VAD and diarization to ensure low-latency processing.
Sources: README.md:68-88, AGENTS.md:73-108
Component Overview
| Component | Responsibility | Technology |
|---|---|---|
| Pusher | Receives binary audio via WebSocket; coordinates with STT. | Python / WebSocket |
| VAD Service | Performs Voice Activity Detection and speaker identification. | Python / GPU (Modal) |
| Diarizer | Generates speaker embeddings for identifying distinct voices. | Python / GPU |
| Deepgram / Parakeet | Speech-to-Text (STT) services for real-time transcription. | External API / Self-hosted |
| LLM Gateway | Orchestrates structured extraction (Action Items, Structured summaries). | Python / GPT-4o |
Sources: AGENTS.md:73-108, backend/utils/llm/conversation_processing.py:535-645
Audio Ingestion & Buffering
Audio is typically ingested in real-time as raw PCM16 bytes. Some integrations, such as the Hume AI plugin, handle audio in 5-second intervals, while the core pipeline supports continuous streaming.
Real-time Audio Bytes
Plugins can receive audio via specialized endpoints (e.g., /audio). The Omi device sends transcripts in segments or raw bytes. For storage-heavy operations, these bytes are often buffered and converted into standard formats like WAV.
def create_wav_file(audio_bytes: bytes, sample_rate: int = 16000) -> bytes:
"""Convert raw PCM16 audio to WAV format."""
buffer = io.BytesIO()
with wave.open(buffer, 'wb') as wav_file:
wav_file.setnchannels(1) # Mono
wav_file.setsampwidth(2) # 16-bit = 2 bytes
wav_file.setframerate(sample_rate)
wav_file.writeframes(audio_bytes)
return buffer.getvalue()
Sources: plugins/omi-dropbox-app/main.py:46-54, plugins/hume-ai/README.md:46-60
Intelligent Segment Collection
The pipeline employs "Smart Segment Collection" to determine when a user has finished speaking and when the system has enough information to process a command.
- Trigger Detection: The pipeline listens for specific phrases (e.g., "Feedback Post", "Send Slack message").
- Dynamic Collection:
- Minimum: 2 segments are required for quick issues.
- Maximum: Up to 10 segments for detailed requests.
- AI Validation: After each segment, the AI checks if the information is "complete".
- Timeout: Automatic processing occurs after 30 seconds of silence if at least 2 segments were collected.
The logic ensures that trivial conversation snippets are discarded while complex instructions are fully captured.
Sources: plugins/omi-github-app/README.md:143-157, plugins/omi-github-app/SETUP_COMPLETE.md:27-58
Transcription & Semantic Analysis
Once segments are collected, they are passed through the LLM processing stage to extract structure and action items.
Conversation Structuring
The system generates a title, overview, emoji, and category for every non-discarded conversation. It also identifies calendar events and filters out hypothetical mentions.
Sources: backend/utils/llm/conversation_processing.py:648-732
Action Item Extraction
The pipeline identifies concrete tasks using ownership and timing signals. It follows strict deduplication rules by comparing candidate items against "Potentially Related Open Tasks" from the user's current task list.
Filtering Criteria for Action Items:
- Concrete Action: Must describe a specific next step.
- Timing Signal: Must include explicit or relative dates (e.g., "tomorrow").
- Real Importance: Must have consequences (financial, health, deadlines).
- Not Immediate: Items the user is currently doing are skipped.
Sources: backend/utils/llm/conversation_processing.py:410-480
Discard Logic
To prevent clutter, the pipeline evaluates whether a conversation should be saved or discarded based on word count, duration, and content significance.
- Keep: Tasks, decisions, plans, personal facts, or visually significant scenes.
- Discard: Trivial snippets ("hello", "hmm"), filler words, or fragmented speech under 2 minutes.
Sources: backend/utils/llm/conversation_processing.py:330-365
Service Protocol & Concurrency
The backend utilizes asyncio.wait(FIRST_COMPLETED) supervisor loops for WebSocket handlers (transcribe.py, pusher.py) to manage concurrent audio streams.
- Task Naming: WebSocket tasks are named using the pattern
ws:{uid}:{task_name}for easier debugging. - Executors: Blocking calls (Firestore, GCS, LLM) are offloaded to specialized executors:
db_executor: Firestore/Redis operations.llm_executor: AI processing.storage_executor: File I/O and GCS.sync_executor: STT and VAD operations.
Sources: AGENTS.md:54-69
Pipeline Summary
The Audio Processing Pipeline is a robust framework that transforms raw audio into actionable intelligence. By combining low-level audio engineering (PCM to WAV, buffering) with high-level AI orchestration (dynamic segment collection, structured extraction), Omi provides a "second brain" that accurately captures and organizes the user's digital and verbal life. The system's ability to filter noise, deduplicate tasks, and diarize speakers makes it an essential component for wearable AI productivity.
Sources: README.md:1-15, backend/utils/llm/conversation_processing.py:12-25