How Conversation Merging and Deduplication Works in Omi: A Technical Deep Dive

Omi combines multiple conversations into a single unified record by validating source data, merging transcripts with adjusted timestamps, deduplicating photos by ID, copying audio chunks to new storage paths, and atomically replacing source documents with the merged result.

The Omi open-source project (available at basedhardware/omi) provides a sophisticated backend system for managing voice conversations. When users need to consolidate fragmented recordings, the conversation merging and deduplication pipeline executes an asynchronous workflow that preserves data integrity while eliminating redundancy. This article examines the complete technical implementation, from the initial API request through final cleanup and notification.

The Merge Workflow Overview

The merging process follows a strict pipeline designed to handle heavy I/O operations without blocking the client. The system validates inputs synchronously, then delegates the intensive work to a background task. The workflow encompasses five major phases: validation, background processing, data merging, persistence, and cleanup. Each phase operates on specific data types—transcript segments, photos, audio chunks, and metadata—ensuring that the resulting conversation represents a seamless chronological continuation of the source material.

API Entry Point and Validation

The merge journey begins in the frontend TypeScript client, which POSTs to the backend endpoint.

// web/app/src/lib/api.ts
export async function mergeConversations(
  conversationIds: string[],
  reprocess: boolean = true
): Promise<MergeConversationsResponse> {
  return fetchWithAuth<MergeConversationsResponse>('/v1/conversations/merge', {
    method: 'POST',
    body: JSON.stringify({
      conversation_ids: conversationIds,
      reprocess,
    }),
  });
}

The backend router at backend/routers/conversations.py handles this request at lines 98-106. Before queuing any work, the system performs rigorous validation through validate_merge_compatibility in backend/utils/conversations/merge_conversations.py.

The validation layer enforces four critical constraints:

• Minimum Count: The request must contain at least two conversation IDs.
• Existence: Each ID must resolve to an existing conversation via conversations_db.get_conversation.
• Status Compatibility: All conversations must be in a completed state and not locked.
• Temporal Warnings: The system generates warnings for time gaps exceeding one hour between conversations.

If any validation fails, the router returns an immediate 400 or 404 error. Only upon passing all checks does the system proceed to background execution.

Background Processing Architecture

Because merging involves copying large audio files from Google Cloud Storage and rebuilding vector embeddings, the operation runs asynchronously to prevent HTTP timeouts. The router adds a background task using FastAPI's BackgroundTasks:

background_tasks.add_task(
    perform_merge_async,
    uid=uid,
    conversation_ids=request.conversation_ids,
    reprocess=request.reprocess,
)

The client receives an immediate response with status "merging" and any temporal warnings. The heavy lifting occurs in perform_merge_async, defined in backend/utils/conversations/merge_conversations.py.

Core Merge Implementation

The perform_merge_async function orchestrates the data consolidation. It begins by reloading and sorting source conversations by started_at to ensure chronological processing.

Merging Transcript Segments

The _merge_transcript_segments function handles the most complex data transformation. It sequentially concatenates transcript segments while adjusting timestamps to create a continuous timeline.

The algorithm maintains a cumulative_offset representing the running end-time of the merged conversation. For each subsequent conversation, segments receive timestamp adjustments:

# backend/utils/conversations/merge_conversations.py

def _merge_transcript_segments(conversations):
    # cumulative_offset tracks the running end-time.

    # Each later segment gets start += offset and end += offset.

Gaps between conversations are handled by applying zero-duration offsets, ensuring that segments from later conversations start precisely where the previous conversation ended.

Photo Deduplication Strategy

Photo merging occurs in _collect_all_photos. The system retrieves all photos from each conversation's sub-collection, then enforces deduplication through a seen_ids set that tracks unique photo identifiers.

def _collect_all_photos(uid, conversations):
    # Deduplicate by photo ID → no duplicate images in the merged conversation.

This ID-based approach ensures that if the same photo appears in multiple source conversations (perhaps through shared media), it appears only once in the final result. The deduplicated list is then sorted by created_at to maintain chronological order.

Audio Chunk Preservation

Audio data requires special handling to prevent data loss while avoiding expensive re-encoding. The _copy_audio_chunks_for_merge function manages this through direct Google Cloud Storage operations.

Each source conversation stores audio chunks at paths like chunks/{uid}/{conv_id}/{timestamp}.bin or .enc. The merge process copies these blobs to the new conversation path while preserving the original timestamps:

def _copy_audio_chunks_for_merge(uid, conversations, new_conversation_id):
    # Copies GCS blobs, keeps original timestamps.

After copying, conversations_db.create_audio_files_from_chunks generates new AudioFile records pointing to the copied chunks. This approach maintains audio integrity while updating database references.

Metadata Consolidation

The merge logic applies specific business rules to combine conversation metadata:

• Visibility: The most restrictive level wins (private > shared > public), determined by _determine_visibility.
• Private Cloud Sync: Enabled if any source conversation has it enabled.
• Discarded Status: Marked as discarded only if all source conversations are discarded.
• Language, Source, Geolocation: Inherited from the earliest conversation (first by started_at).

These rules ensure that security settings (like private visibility) are never downgraded during a merge, while preserving useful metadata from the primary conversation.

Finalization and Cleanup

After data consolidation, the system executes the final persistence and cleanup sequence.

Creating the New Conversation

The merged data instantiates a new Conversation model (defined in models/conversation.py):

new_conversation = Conversation(
    id=new_conversation_id,
    created_at=created_at,
    started_at=started_at,
    finished_at=finished_at,
    transcript_segments=merged_segments,
    photos=merged_photos,
    audio_files=merged_audio_files,
    ...
)
conversations_db.upsert_conversation(uid, new_conversation.dict())

Optional Reprocessing

If the request specified reprocess=True, the system invokes process_conversation to regenerate the title, summary, action items, and memories. Should this processing fail, the conversation remains in completed status, ensuring the user retains access to the merged transcript.

Source Cleanup

The _delete_conversation_and_related_data function performs atomic cleanup of all source conversations:

• Deletes memories and action items
• Removes photos and audio chunks from storage
• Deletes vector embeddings
• Removes the Firestore document

This executes in a loop over all sorted source conversations, ensuring no orphaned data remains.

Client Notification

Finally, send_merge_completed_message (from utils/notifications.py) dispatches a Firebase Cloud Messaging (FCM) notification to the client, allowing the UI to refresh and display the newly merged conversation.

Summary

• Validation Layer: The system enforces minimum conversation counts, existence checks, and compatibility rules (completed status, no locks) before accepting a merge request.
• Asynchronous Processing: Heavy I/O operations (audio copying, vector updates) execute in background tasks to prevent HTTP timeouts.
• Transcript Continuity: The _merge_transcript_segments function adjusts timestamps using cumulative offsets to create seamless chronological flow.
• Deduplication Strategy: Photos deduplicate by ID in _collect_all_photos, while audio chunks copy with preserved timestamps in _copy_audio_chunks_for_merge.
• Security Preservation: Metadata consolidation applies the most restrictive visibility settings and preserves privacy flags.
• Atomic Cleanup: Source conversations and all related data (memories, photos, audio, vectors) are fully deleted after successful merge.

Frequently Asked Questions

How does Omi prevent duplicate photos when merging conversations?

Omi deduplicates photos by maintaining a seen_ids set during the _collect_all_photos phase in backend/utils/conversations/merge_conversations.py. As the system iterates through each source conversation's photo collection, it checks each photo ID against this set. Only photos with unique IDs are added to the final merged list, which is then sorted by created_at to maintain chronological order.

What happens to audio recordings during a conversation merge?

Audio chunks are preserved through direct Google Cloud Storage operations in _copy_audio_chunks_for_merge. The system copies raw audio blobs from chunks/{uid}/{source_conv_id}/{timestamp}.bin (or .enc) to the new conversation's storage path while maintaining the original timestamps. After copying, conversations_db.create_audio_files_from_chunks generates new database records referencing these copied chunks, ensuring no audio data is lost or re-encoded.

Why does Omi use a background task for conversation merging?

The merge operation executes asynchronously via FastAPI's BackgroundTasks because it involves I/O-heavy operations that could exceed HTTP timeout limits. These include copying large audio files between Google Cloud Storage buckets, updating vector embeddings, and deleting multiple Firestore documents. By queuing perform_merge_async as a background task, the API returns an immediate "merging" status to the client while the heavy lifting continues server-side, with completion signaled via Firebase Cloud Messaging.

How does Omi handle timestamps when combining transcript segments?

The system maintains chronological continuity through cumulative timestamp offsets in _merge_transcript_segments. As each source conversation is processed sequentially, the algorithm tracks a cumulative_offset representing the running end-time of the merged conversation. Each segment from subsequent conversations receives this offset added to both its start and end timestamps, creating a seamless timeline where the first segment of conversation B starts exactly when the last segment of conversation A ended.

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