Enhance conversation model to include word-level timestamps and improve transcript handling

- Added a new `words` field to the `Conversation` model for storing word-level timestamps.
- Updated methods to handle word data during transcript version creation, ensuring compatibility with speaker recognition.
- Refactored conversation job processing to utilize the new word structure, improving data integrity and access.
- Enhanced speaker recognition job to read words from the new standardized location, ensuring backward compatibility with legacy data.

Author: AnkushMalaker

backends/advanced/src/advanced_omi_backend/models/conversation.py

@@ -63,7 +63,14 @@ class TranscriptVersion(BaseModel):
         """Version of a transcript with processing metadata."""
         version_id: str = Field(description="Unique version identifier")
         transcript: Optional[str] = Field(None, description="Full transcript text")
-        segments: List["Conversation.SpeakerSegment"] = Field(default_factory=list, description="Speaker segments")
+        words: List["Conversation.Word"] = Field(
+            default_factory=list,
+            description="Word-level timestamps for entire transcript"
+        )
+        segments: List["Conversation.SpeakerSegment"] = Field(
+            default_factory=list,
+            description="Speaker segments (filled by speaker recognition)"
+        )
         provider: Optional[str] = Field(None, description="Transcription provider used (deepgram, parakeet, etc.)")
         model: Optional[str] = Field(None, description="Model used (e.g., nova-3, parakeet)")
         created_at: datetime = Field(description="When this version was created")
@@ -253,8 +260,9 @@ def add_transcript_version(
         self,
         version_id: str,
         transcript: str,
-        segments: List["Conversation.SpeakerSegment"],
-        provider: str,  # Provider name from config.yml (deepgram, parakeet, etc.)
+        words: Optional[List["Conversation.Word"]] = None,
+        segments: Optional[List["Conversation.SpeakerSegment"]] = None,
+        provider: str = None,  # Provider name from config.yml (deepgram, parakeet, etc.)
         model: Optional[str] = None,
         processing_time_seconds: Optional[float] = None,
         metadata: Optional[Dict[str, Any]] = None,
@@ -264,7 +272,8 @@ def add_transcript_version(
         new_version = Conversation.TranscriptVersion(
             version_id=version_id,
             transcript=transcript,
-            segments=segments,
+            words=words or [],
+            segments=segments or [],
             provider=provider,
             model=model,
             created_at=datetime.now(),

backends/advanced/src/advanced_omi_backend/workers/conversation_jobs.py

@@ -588,74 +588,39 @@ async def open_conversation_job(
     # Create transcript version from streaming results
     version_id = f"streaming_{session_id[:12]}"
     transcript_text = final_transcript.get("text", "")
-    segments_data = final_transcript.get("segments", [])
-
-    # If streaming provider didn't provide segments (e.g., Deepgram streaming),
-    # create segments from individual final results with word-level data
-    if not segments_data:
-        logger.info(f"📝 No segments in streaming results, creating from word-level data")
-        results = await aggregator.get_session_results(session_id)
-
-        for result in results:
-            words = result.get("words", [])
-            text = result.get("text", "").strip()
-
-            # Skip empty results or results without timing data
-            # WARNING: We don't support results without word-level timing data.
-            # Ideally should error, but skipping for now to handle edge cases gracefully.
-            if not words or not text:
-                continue
-
-            # Create segment dict from this result chunk
-            # Each "final" result becomes one segment with generic speaker label
-            segment_dict = {
-                "start": words[0]["start"],
-                "end": words[-1]["end"],
-                "text": text,
-                "speaker": "SPEAKER_00",  # Generic label, updated by speaker recognition
-                "confidence": result.get("confidence"),
-                "words": words  # Already in correct format from aggregator
-            }
-            segments_data.append(segment_dict)
-
-        logger.info(f"✅ Created {len(segments_data)} segments from streaming results")
-
-    # Convert segments to SpeakerSegment format with word-level timestamps
-    segments = [
-        Conversation.SpeakerSegment(
-            start=seg.get("start", 0.0),
-            end=seg.get("end", 0.0),
-            text=seg.get("text", ""),
-            speaker=seg.get("speaker", "SPEAKER_00"),
-            confidence=seg.get("confidence"),
-            words=[
-                Conversation.Word(
-                    word=w.get("word", ""),
-                    start=w.get("start", 0.0),
-                    end=w.get("end", 0.0),
-                    confidence=w.get("confidence")
-                )
-                for w in seg.get("words", [])
-            ]
+    words_data = final_transcript.get("words", [])  # All words from aggregator
+
+    # Convert words to Word objects
+    words = [
+        Conversation.Word(
+            word=w.get("word", ""),
+            start=w.get("start", 0.0),
+            end=w.get("end", 0.0),
+            confidence=w.get("confidence")
         )
-        for seg in segments_data
+        for w in words_data
     ]
 
+    # Segments remain EMPTY until speaker recognition service processes them
+    # Per Chronicle architecture: segments ONLY come from speaker service
+    segments = []
+
     # Determine provider from streaming results
     provider = final_transcript.get("provider", "deepgram")
 
-    # Add streaming transcript as the initial version
+    # Add streaming transcript with words at version level
     conversation.add_transcript_version(
         version_id=version_id,
         transcript=transcript_text,
-        segments=segments,
+        words=words,  # Store at version level
+        segments=segments,  # Empty - only speaker service creates segments
         provider=provider,
         model=provider,  # Provider name as model
         processing_time_seconds=None,  # Not applicable for streaming
         metadata={
             "source": "streaming",
             "chunk_count": final_transcript.get("chunk_count", 0),
-            "word_count": len(final_transcript.get("words", []))
+            "word_count": len(words),
         },
         set_as_active=True
     )
@@ -664,7 +629,7 @@ async def open_conversation_job(
     await conversation.save()
     logger.info(
         f"✅ Saved streaming transcript: {len(transcript_text)} chars, "
-        f"{len(segments)} segments, {len(final_transcript.get('words', []))} words "
+        f"{len(segments)} segments (empty until speaker recognition), {len(words)} words "
         f"for conversation {conversation_id[:12]}"
     )
 

backends/advanced/src/advanced_omi_backend/workers/speaker_jobs.py

@@ -207,11 +207,36 @@ async def recognise_speakers_job(
     actual_transcript_text = transcript_text or transcript_version.transcript or ""
     actual_words = words if words else []
 
-    # If words not provided, read from transcript version metadata
-    # (Transcription job stores words in metadata since segments are created by speaker service)
-    if not actual_words and transcript_version.metadata:
+    # If words not provided as parameter, read from version.words field (standardized location)
+    if not actual_words and transcript_version.words:
+        # Convert Word objects to dicts for speaker service API
+        actual_words = [
+            {
+                "word": w.word,
+                "start": w.start,
+                "end": w.end,
+                "confidence": w.confidence
+            }
+            for w in transcript_version.words
+        ]
+        logger.info(f"🔤 Loaded {len(actual_words)} words from transcript version.words field")
+    # Backward compatibility: Fall back to metadata if words field is empty (old data)
+    elif not actual_words and transcript_version.metadata.get("words"):
         actual_words = transcript_version.metadata.get("words", [])
-        logger.info(f"🔤 Loaded {len(actual_words)} words from transcript version metadata")
+        logger.info(f"🔤 Loaded {len(actual_words)} words from transcript version metadata (legacy)")
+    # Backward compatibility: Extract from segments if that's all we have (old streaming data)
+    elif not actual_words and transcript_version.segments:
+        for segment in transcript_version.segments:
+            if segment.words:
+                for w in segment.words:
+                    actual_words.append({
+                        "word": w.word,
+                        "start": w.start,
+                        "end": w.end,
+                        "confidence": w.confidence
+                    })
+        if actual_words:
+            logger.info(f"🔤 Extracted {len(actual_words)} words from segments (legacy)")
 
     if not actual_transcript_text:
         logger.warning(f"🎤 No transcript text found in version {version_id}")
@@ -223,6 +248,16 @@ async def recognise_speakers_job(
             "processing_time_seconds": 0
         }
 
+    if not actual_words:
+        logger.warning(f"🎤 No words found in version {version_id}")
+        return {
+            "success": False,
+            "conversation_id": conversation_id,
+            "version_id": version_id,
+            "error": "No word-level timing data available",
+            "processing_time_seconds": 0
+        }
+
     transcript_data = {
         "text": actual_transcript_text,
         "words": actual_words

backends/advanced/src/advanced_omi_backend/workers/transcription_jobs.py

@@ -365,20 +365,30 @@ async def transcribe_full_audio_job(
     # Add new transcript version
     provider_normalized = provider_name.lower() if provider_name else "unknown"
 
+    # Convert words to Word objects
+    word_objects = [
+        Conversation.Word(
+            word=w.get("word", ""),
+            start=w.get("start", 0.0),
+            end=w.get("end", 0.0),
+            confidence=w.get("confidence")
+        )
+        for w in words
+    ]
+
     # Prepare metadata (transcription only - speaker service will add segments and metadata)
-    # Store words in metadata so speaker job can access them
     metadata = {
         "trigger": trigger,
         "audio_file_size": len(wav_data),
         "word_count": len(words),
         "segments_created_by": "speaker_service",  # Speaker service creates segments via diarization
-        "words": words,  # Store word-level timing data for speaker job
     }
 
     conversation.add_transcript_version(
         version_id=version_id,
         transcript=transcript_text,
-        segments=speaker_segments,
+        words=word_objects,  # Store at version level (not in metadata!)
+        segments=speaker_segments,  # Empty - will be filled by speaker recognition
         provider=provider_normalized,  # Now just a string, no enum constructor needed
         model=provider.name,
         processing_time_seconds=processing_time,