How the Omi Conversation Processing Pipeline Works: From Transcription to Storage

The Omi conversation processing pipeline transforms live audio streams into structured, searchable conversations through four distinct stages: real-time speech-to-text capture, voice activity detection and speaker diarization, LLM-based enrichment, and persistent storage abstraction.

The basedhardware/omi open-source platform implements this pipeline using dedicated Python modules that handle everything from WebSocket audio ingestion to Firestore persistence. Each stage operates asynchronously to ensure low-latency transcription while simultaneously identifying speakers and enriching content. Developers can trace the entire flow from raw PCM chunks to final conversation records stored under user-specific namespaces.

Stage 1: Audio Capture and Streaming Speech-to-Text

Audio ingestion begins when the frontend opens a binary WebSocket connection handled by backend/utils/pusher.py. This bridge forwards raw PCM audio chunks to the StreamingSTTClient class defined in backend/utils/stt/streaming.py.

The streaming client authenticates with the Deepgram service—either cloud-hosted or self-hosted—and maintains a persistent connection that returns partial transcripts in real time. As audio frames arrive, the client buffers them and emits transcript segments downstream without waiting for the stream to close. This design ensures that transcription latency remains minimal even during extended recording sessions.

Stage 2: Voice Activity Detection and Speaker Identification

While transcripts flow from Deepgram, a parallel processing thread analyzes the same audio frames for speaker characteristics. The pipeline uses three specialized modules to achieve accurate diarization:

• backend/utils/stt/vad.py: Runs a lightweight Voice Activity Detection (VAD) model to isolate speech segments from silence or background noise, emitting precise timestamps for each utterance.

• backend/utils/stt/speaker_embedding.py: For each detected speech segment, this module calls the hosted diarizer service at HOSTED_SPEAKER_EMBEDDING_API_URL to generate a 256-dimensional speaker embedding vector.

• backend/utils/speaker_identification.py: Compares incoming embeddings against stored speaker profiles maintained in Redis. When a match exceeds the similarity threshold, the module assigns a consistent speaker_id; otherwise, it registers a new profile.

This parallel architecture ensures that speaker identification happens concurrently with transcription, avoiding bottlenecks in the main audio processing thread.

Stage 3: Conversation Assembly and LLM Enrichment

Once transcripts and speaker labels are available, backend/utils/llm/conversation_processing.py orchestrates the assembly phase. This module aligns raw transcript chunks with VAD timestamps and injects the corresponding speaker_id into each utterance, creating a structured JSON payload that adheres to the schema defined in backend/utils/conversations/process_conversation.py.

The pipeline supports optional enrichment through the LLM utility layer:

• backend/utils/llm/chat.py: Generates conversation summaries and extracts key topics from the assembled transcript.
• backend/utils/llm/app_generator.py: Creates structured action items or application-specific metadata based on conversation content.

These enrichment steps occur after initial assembly but before persistence, allowing the system to store both raw transcripts and AI-generated insights within the same conversation object.

Stage 4: Persistence and Storage

The final stage delegates persistence to backend/utils/other/storage.py, which provides a backend-agnostic interface to Firestore, Redis, or other configured datastores. By default, the system serializes the enriched conversation JSON and writes it to Firestore under the user's namespace.

The storage layer returns a unique conversation ID upon successful write, enabling immediate retrieval through the APIs defined in backend/utils/retrieval/. This abstraction ensures that upstream processing logic remains decoupled from specific database implementations, facilitating testing and infrastructure migrations.

End-to-End Implementation Example

The following examples demonstrate how to ingest a live stream and apply post-processing enrichment using the pipeline components:

from backend.utils.pusher import AudioPusher
from backend.utils.other.storage import ConversationStore

pusher = AudioPusher(websocket_url="wss://api.omi.io/audio")
store = ConversationStore()

# Start streaming – internally uses streaming.py, vad.py, and speaker identification

conversation = pusher.start_stream(user_id="user_123")

# When the stream ends, persist to Firestore

conversation_id = store.save(conversation, user_id="user_123")
print(f"Conversation saved with ID: {conversation_id}")

from backend.utils.llm.chat import LLMChat
from backend.utils.conversations.process_conversation import enrich_conversation

# Load and enrich a stored conversation

raw = store.load(conversation_id, user_id="user_123")
enriched = enrich_conversation(raw)  # Adds speaker IDs and timestamps

# Generate insights

summary = LLMChat().summarise(enriched["transcript"])
actions = LLMChat().extract_action_items(enriched["transcript"])

print("Summary:", summary)
print("Action items:", actions)

Summary

• The pipeline processes audio through four stages: streaming STT, VAD/speaker identification, conversation assembly, and storage abstraction.
• backend/utils/stt/streaming.py handles Deepgram integration and real-time transcript generation.
• Speaker diarization relies on backend/utils/stt/vad.py for segmentation, backend/utils/stt/speaker_embedding.py for 256-dimensional vectors, and backend/utils/speaker_identification.py for Redis-based profile matching.
• backend/utils/llm/conversation_processing.py merges transcripts with metadata and coordinates optional LLM enrichment via backend/utils/llm/chat.py.
• backend/utils/other/storage.py abstracts persistence to Firestore or Redis, returning unique conversation IDs for retrieval.

Frequently Asked Questions

How does Omi handle real-time audio streaming without latency bottlenecks?

The platform uses asynchronous WebSocket connections managed by backend/utils/pusher.py to stream PCM chunks directly to backend/utils/stt/streaming.py. By processing transcripts incrementally and running speaker identification in parallel threads via backend/utils/stt/vad.py, the system avoids blocking operations, ensuring that transcription and diarization keep pace with live audio input.

What technology identifies speakers in the conversation processing pipeline?

Speaker identification uses a three-step process: backend/utils/stt/vad.py detects speech segments, backend/utils/stt/speaker_embedding.py generates 256-dimensional embeddings via the hosted diarizer API, and backend/utils/speaker_identification.py matches these vectors against existing profiles stored in Redis. This embedding-based approach allows the system to recognize returning speakers across multiple conversation sessions.

Can the storage backend be customized or run entirely self-hosted?

Yes. While backend/utils/other/storage.py defaults to Firestore, it implements an abstraction layer that supports Redis and other backends through configuration changes. The storage interface accepts any implementation following the same contract, enabling self-hosted deployments or hybrid cloud setups without modifying the conversation processing logic upstream.

Where does LLM enrichment occur in the pipeline, and what functions does it perform?

LLM enrichment happens during Stage 3 in backend/utils/llm/conversation_processing.py, which optionally invokes backend/utils/llm/chat.py to generate summaries, extract topics, or identify action items. This processing occurs after speaker identification but before persistence, ensuring that AI-generated metadata is stored atomically with the raw transcript in the final conversation object.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}