fix(bedrock): share one item_id across a user utterance's realtime events

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/llms/bedrock/realtime/handler.py

@@ -156,7 +156,9 @@ class BedrockRealtime(BaseAWSLLM):
                 )
             )
 
-            logged_events: Final[list[OpenAIRealtimeEvents]] = []  # mutable-ok: events accumulate across stream loop iterations for spend logging
+            logged_events: Final[
+                list[OpenAIRealtimeEvents]
+            ] = []  # mutable-ok: events accumulate across stream loop iterations for spend logging
             bedrock_to_client_task: Final = asyncio.create_task(
                 self._forward_bedrock_to_client(
                     bedrock_stream,

litellm/llms/bedrock/realtime/transformation.py

@@ -117,6 +117,7 @@ class BedrockRealtimeConfig(BaseRealtimeConfig):
         # so the USER/ASSISTANT split from contentStart is tracked here)
         self._user_transcript_active = False
         self._user_transcript_generation_stage: str | None = None
+        self._user_item_id: str | None = None
         self._latest_usage: OpenAIRealtimeResponseUsage | None = None
 
     def validate_environment(self, headers: dict, model: str, api_key: str | None = None) -> dict:
@@ -818,13 +819,19 @@ class BedrockRealtimeConfig(BaseRealtimeConfig):
         stage: Final = parsed.get("generationStage") if isinstance(parsed, dict) else None
         return stage if isinstance(stage, str) else None
 
+    def _current_user_item_id(self, new_utterance: bool = False) -> str:
+        """Item id shared by all events of one user utterance (speech boundaries and transcript)."""
+        if new_utterance or self._user_item_id is None:
+            self._user_item_id = f"item_{uuid.uuid4()}"
+        return self._user_item_id
+
     def transform_user_speech_event(self, is_speech_start: bool) -> tuple[OpenAIRealtimeEvents, ...]:
         """Transform Bedrock userSpeechStart/userSpeechEnd to OpenAI speech boundary events."""
         verbose_logger.debug("Handling userSpeech%s", "Start" if is_speech_start else "End")
         speech_event: Final[OpenAIRealtimeInputAudioBufferSpeechEvent] = {
             "type": "input_audio_buffer.speech_started" if is_speech_start else "input_audio_buffer.speech_stopped",
             "event_id": f"event_{uuid.uuid4()}",
-            "item_id": f"item_{uuid.uuid4()}",
+            "item_id": self._current_user_item_id(new_utterance=is_speech_start),
         }
         return (speech_event,)
 
@@ -852,7 +859,7 @@ class BedrockRealtimeConfig(BaseRealtimeConfig):
     def transform_user_transcript_event(self, transcript: str) -> tuple[OpenAIRealtimeEvents, ...]:
         """Transform a USER-role Bedrock textOutput (ASR transcript) to OpenAI transcription events."""
         verbose_logger.debug("Handling USER textOutput (ASR transcript)")
-        item_id: Final = f"item_{uuid.uuid4()}"
+        item_id: Final = self._current_user_item_id()
         delta_event: Final[OpenAIRealtimeInputAudioTranscriptionDelta] = {
             "type": "conversation.item.input_audio_transcription.delta",
             "event_id": f"event_{uuid.uuid4()}",

tests/test_litellm/llms/bedrock/realtime/test_bedrock_realtime_handler.py

@@ -368,9 +368,7 @@ class TestBedrockRealtimeSessionLifecycle:
             [json.dumps({"type": "session.update", "session": {"instructions": "hi", "modalities": ["text"]}})]
         )
 
-        await handler._forward_client_to_bedrock(
-            client_ws, stream, config, "amazon.nova-sonic-v1:0", {}, FakeLogging()
-        )
+        await handler._forward_client_to_bedrock(client_ws, stream, config, "amazon.nova-sonic-v1:0", {}, FakeLogging())
 
         acked = [json.loads(message) for message in client_ws.sent_to_client]
         updated = [event for event in acked if event["type"] == "session.updated"]
@@ -382,9 +380,7 @@ class TestBedrockRealtimeSessionLifecycle:
         handler = BedrockRealtime()
         config = BedrockRealtimeConfig()
         stream = FakeBedrockStream()
-        client_ws = DisconnectingClientWS(
-            [json.dumps({"type": "session.update", "session": {"instructions": "hi"}})]
-        )
+        client_ws = DisconnectingClientWS([json.dumps({"type": "session.update", "session": {"instructions": "hi"}})])
 
         await handler._forward_client_to_bedrock(client_ws, stream, config, "amazon.nova-sonic-v1:0", {})
 

tests/test_litellm/llms/bedrock/realtime/test_bedrock_realtime_transformation.py

@@ -874,6 +874,37 @@ class TestBedrockRealtimeUserEventsAndUsage:
             "input_audio_buffer.speech_stopped",
         ]
         assert all(e["event_id"] and e["item_id"] for e in events)
+        assert events[0]["item_id"] == events[1]["item_id"]
+
+    def test_utterance_lifecycle_shares_one_item_id(self):
+        events = self._run(
+            BedrockRealtimeConfig(),
+            [
+                {"event": {"userSpeechStart": {}}},
+                {"event": {"userSpeechEnd": {}}},
+                {
+                    "event": {
+                        "contentStart": {
+                            "role": "USER",
+                            "type": "TEXT",
+                            "additionalModelFields": json.dumps({"generationStage": "FINAL"}),
+                        }
+                    }
+                },
+                {"event": {"textOutput": {"content": "ready"}}},
+                {"event": {"contentEnd": {"stopReason": "PARTIAL_TURN"}}},
+            ],
+        )
+        item_ids = {e["item_id"] for e in events if "item_id" in e}
+        assert len(item_ids) == 1
+
+    def test_new_utterance_gets_new_item_id(self):
+        config = BedrockRealtimeConfig()
+        first = self._run(config, [{"event": {"userSpeechStart": {}}}, {"event": {"userSpeechEnd": {}}}])
+        second = self._run(config, [{"event": {"userSpeechStart": {}}}, {"event": {"userSpeechEnd": {}}}])
+        assert first[0]["item_id"] == first[1]["item_id"]
+        assert second[0]["item_id"] == second[1]["item_id"]
+        assert first[0]["item_id"] != second[0]["item_id"]
 
     def test_user_transcript_emits_input_audio_transcription_events(self):
         events = self._run(