fix: order delayed realtime transcripts by speech start

livekit-agents/livekit/agents/llm/realtime.py

@@ -4,7 +4,7 @@
 import time
 from abc import ABC, abstractmethod
 from collections.abc import AsyncIterable, Awaitable
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from types import TracebackType
 from typing import Generic, Literal, TypeVar
 
@@ -153,6 +153,8 @@ class InputTranscriptionCompleted:
     is_final: bool
     confidence: float | None = None
     """confidence score of the transcript (0.0 to 1.0), derived from model logprobs"""
+    created_at: float = field(default_factory=time.time)
+    """time when the input conversation item was created"""
 
 
 @dataclass

livekit-agents/livekit/agents/voice/agent_activity.py

@@ -1847,9 +1847,12 @@ def _on_input_audio_transcription_completed(self, ev: llm.InputTranscriptionComp
             if self.stt is None and ev.transcript and (amd := self._session._amd) is not None:
                 amd._on_transcript(ev.transcript)
 
-            # TODO: for realtime models, the created_at field is off. it should be set to when the user started speaking.
-            # but we don't have that information here.
-            msg = llm.ChatMessage(role="user", content=[ev.transcript], id=ev.item_id)
+            msg = llm.ChatMessage(
+                role="user",
+                content=[ev.transcript],
+                id=ev.item_id,
+                created_at=ev.created_at,
+            )
             self._agent._chat_ctx._upsert_item(msg)
             self._session._conversation_item_added(msg)
 

livekit-plugins/livekit-plugins-openai/livekit/plugins/openai/realtime/realtime_model.py

@@ -866,6 +866,7 @@ def __init__(self, realtime_model: RealtimeModel) -> None:
 
         # accumulates partial input-audio transcripts per (item_id, content_index)
         self._input_transcript_accumulators: dict[str, dict[int, str]] = {}
+        self._input_speech_started_at = utils.BoundedDict[str, float](maxsize=100)
 
         self._current_generation: _ResponseGeneration | _DiscardedGeneration | None = None
         self._remote_chat_ctx = llm.remote_chat_context.RemoteChatContext()
@@ -915,6 +916,7 @@ async def _reconnect() -> None:
             old_chat_ctx = self._remote_chat_ctx
             self._remote_chat_ctx = llm.remote_chat_context.RemoteChatContext()
             self._input_transcript_accumulators.clear()
+            self._input_speech_started_at.clear()
             events.extend(self._create_update_chat_ctx_events(chat_ctx))
 
             try:
@@ -1760,8 +1762,9 @@ def _resample_audio(self, frame: rtc.AudioFrame) -> Iterator[rtc.AudioFrame]:
             yield frame
 
     def _handle_input_audio_buffer_speech_started(
-        self, _: InputAudioBufferSpeechStartedEvent
+        self, event: InputAudioBufferSpeechStartedEvent
     ) -> None:
+        self._input_speech_started_at[event.item_id] = time.time()
         self.emit("input_speech_started", llm.InputSpeechStartedEvent())
 
     def _handle_input_audio_buffer_speech_stopped(
@@ -1885,6 +1888,7 @@ def _handle_conversion_item_deleted(self, event: ConversationItemDeletedEvent) -
         assert event.item_id is not None, "item_id is None"
 
         self._input_transcript_accumulators.pop(event.item_id, None)
+        self._input_speech_started_at.pop(event.item_id, None)
 
         try:
             self._remote_chat_ctx.delete(event.item_id)
@@ -1913,7 +1917,10 @@ def _handle_conversion_item_input_audio_transcription_delta(
         self.emit(
             "input_audio_transcription_completed",
             llm.InputTranscriptionCompleted(
-                item_id=event.item_id, transcript=accumulated, is_final=False
+                item_id=event.item_id,
+                transcript=accumulated,
+                is_final=False,
+                created_at=self._input_speech_started_at.get(event.item_id, time.time()),
             ),
         )
 
@@ -1930,6 +1937,7 @@ def _handle_conversion_item_input_audio_transcription_completed(
         self, event: ConversationItemInputAudioTranscriptionCompletedEvent
     ) -> None:
         self._clear_transcript_accumulator(event.item_id, event.content_index or 0)
+        created_at = self._input_speech_started_at.pop(event.item_id, time.time())
 
         confidence = calculate_confidence_from_logprobs(event.logprobs)
 
@@ -1945,6 +1953,7 @@ def _handle_conversion_item_input_audio_transcription_completed(
                 transcript=event.transcript,
                 is_final=True,
                 confidence=confidence,
+                created_at=created_at,
             ),
         )
 
@@ -1957,13 +1966,17 @@ def _handle_conversion_item_input_audio_transcription_failed(
         )
 
         # close any open partial stream so consumers waiting for is_final don't hang
+        created_at = self._input_speech_started_at.pop(event.item_id, time.time())
         partial = self._clear_transcript_accumulator(event.item_id, event.content_index or 0)
         if partial is None:
             return
         self.emit(
             "input_audio_transcription_completed",
             llm.InputTranscriptionCompleted(
-                item_id=event.item_id, transcript=partial, is_final=True
+                item_id=event.item_id,
+                transcript=partial,
+                is_final=True,
+                created_at=created_at,
             ),
         )
 

tests/test_agent_session.py

@@ -125,6 +125,55 @@ def _user_input_transcribed(self, ev: UserInputTranscribedEvent) -> None:
     assert captured_events[0].item_id == "item_123"
 
 
+def test_realtime_user_transcript_report_order_preserves_server_context_order() -> None:
+    class DummySession:
+        def __init__(self) -> None:
+            self._amd = None
+            self._chat_ctx = ChatContext.empty()
+            self._text_only = True
+
+        def _user_input_transcribed(self, ev: UserInputTranscribedEvent) -> None:
+            pass
+
+        def _conversation_item_added(self, message: ChatMessage) -> None:
+            self._chat_ctx.insert(message)
+
+    session = DummySession()
+    agent_chat_ctx = ChatContext.empty()
+    assistant = ChatMessage(
+        role="assistant",
+        content=["reply"],
+        id="assistant-1",
+        created_at=200.0,
+    )
+    user_placeholder = ChatMessage(
+        role="user",
+        content=[],
+        id="user-1",
+        created_at=300.0,
+    )
+    session._chat_ctx.insert(assistant)
+    agent_chat_ctx.insert(assistant)
+    agent_chat_ctx.insert(user_placeholder)
+
+    activity = object.__new__(AgentActivity)
+    activity._session = session
+    activity._agent = SimpleNamespace(_chat_ctx=agent_chat_ctx)
+
+    AgentActivity._on_input_audio_transcription_completed(
+        activity,
+        InputTranscriptionCompleted(
+            item_id="user-1",
+            transcript="hello",
+            is_final=True,
+            created_at=100.0,
+        ),
+    )
+
+    assert [item.id for item in agent_chat_ctx.items] == ["assistant-1", "user-1"]
+    assert [item.id for item in session._chat_ctx.items] == ["user-1", "assistant-1"]
+
+
 async def test_events_and_metrics() -> None:
     speed = 1
     actions = FakeActions()

tests/test_realtime/test_openai_realtime_model.py

@@ -2,6 +2,7 @@
 
 from types import SimpleNamespace
 from typing import cast
+from unittest.mock import patch
 
 import pytest
 
@@ -13,6 +14,57 @@
 pytestmark = pytest.mark.unit
 
 
+def test_input_transcripts_keep_their_speech_start_time() -> None:
+    emitted: list[tuple[str, object]] = []
+    session = cast(
+        RealtimeSession,
+        SimpleNamespace(
+            _input_speech_started_at={},
+            _remote_chat_ctx=RemoteChatContext(),
+            _clear_transcript_accumulator=lambda item_id, content_index: None,
+            emit=lambda name, event: emitted.append((name, event)),
+        ),
+    )
+
+    with patch(
+        "livekit.plugins.openai.realtime.realtime_model.time.time",
+        side_effect=[100.0, 250.0, 999.0, 999.0],
+    ):
+        RealtimeSession._handle_input_audio_buffer_speech_started(
+            session, SimpleNamespace(item_id="user-1")
+        )
+        RealtimeSession._handle_input_audio_buffer_speech_started(
+            session, SimpleNamespace(item_id="user-2")
+        )
+        RealtimeSession._handle_conversion_item_input_audio_transcription_completed(
+            session,
+            SimpleNamespace(
+                item_id="user-2",
+                content_index=0,
+                transcript="second",
+                logprobs=None,
+            ),
+        )
+        RealtimeSession._handle_conversion_item_input_audio_transcription_completed(
+            session,
+            SimpleNamespace(
+                item_id="user-1",
+                content_index=0,
+                transcript="first",
+                logprobs=None,
+            ),
+        )
+
+    transcripts = [
+        event for name, event in emitted if name == "input_audio_transcription_completed"
+    ]
+    assert [(event.item_id, event.created_at) for event in transcripts] == [
+        ("user-2", 250.0),
+        ("user-1", 100.0),
+    ]
+    assert session._input_speech_started_at == {}
+
+
 def test_update_chat_ctx_deletes_empty_remote_items() -> None:
     remote_ctx = RemoteChatContext()
     audio_item = llm.ChatMessage(id="audio_item", role="user", content=[])

tests/test_realtime/test_xai_realtime_model.py

@@ -15,6 +15,7 @@ def _make_session() -> tuple[RealtimeSession, list[llm.InputTranscriptionComplet
     session = RealtimeSession.__new__(RealtimeSession)
     session._remote_chat_ctx = RemoteChatContext()  # type: ignore[attr-defined]
     session._input_transcript_accumulators = {}  # type: ignore[attr-defined]
+    session._input_speech_started_at = {}  # type: ignore[attr-defined]
 
     emitted: list[llm.InputTranscriptionCompleted] = []
     session.emit = lambda name, ev: emitted.append(ev)  # type: ignore[method-assign,assignment]