Add retrieve_conversation_item() method to OpenAIRealtimeBetaLLMService, using the new conversation.item.retrieve introspection message.

src/pipecat/services/openai_realtime_beta/events.py

@@ -179,6 +179,11 @@ class ConversationItemDeleteEvent(ClientEvent):
     item_id: str
 
 
+class ConversationItemRetrieveEvent(ClientEvent):
+    type: Literal["conversation.item.retrieve"] = "conversation.item.retrieve"
+    item_id: str
+
+
 class ResponseCreateEvent(ClientEvent):
     type: Literal["response.create"] = "response.create"
     response: Optional[ResponseProperties] = None
@@ -255,6 +260,11 @@ class ConversationItemDeleted(ServerEvent):
     item_id: str
 
 
+class ConversationItemRetrieved(ServerEvent):
+    type: Literal["conversation.item.retrieved"]
+    item: ConversationItem
+
+
 class ResponseCreated(ServerEvent):
     type: Literal["response.created"]
     response: "Response"
@@ -441,6 +451,7 @@ _server_event_types = {
     "conversation.item.input_audio_transcription.failed": ConversationItemInputAudioTranscriptionFailed,
     "conversation.item.truncated": ConversationItemTruncated,
     "conversation.item.deleted": ConversationItemDeleted,
+    "conversation.item.retrieved": ConversationItemRetrieved,
     "response.created": ResponseCreated,
     "response.done": ResponseDone,
     "response.output_item.added": ResponseOutputItemAdded,

src/pipecat/services/openai_realtime_beta/openai.py

@@ -119,6 +119,7 @@ class OpenAIRealtimeBetaLLMService(LLMService):
 
         self._register_event_handler("on_conversation_item_created")
         self._register_event_handler("on_conversation_item_updated")
+        self._retrieve_conversation_item_futures = {}
 
     def can_generate_metrics(self) -> bool:
         return True
@@ -126,6 +127,12 @@ class OpenAIRealtimeBetaLLMService(LLMService):
     def set_audio_input_paused(self, paused: bool):
         self._audio_input_paused = paused
 
+    async def retrieve_conversation_item(self, item_id: str):
+        future = self.get_event_loop().create_future()
+        self._retrieve_conversation_item_futures[item_id] = future
+        await self.send_client_event(events.ConversationItemRetrieveEvent(item_id=item_id))
+        return await future
+
     #
     # standard AIService frame handling
     #
@@ -363,6 +370,8 @@ class OpenAIRealtimeBetaLLMService(LLMService):
                 await self._handle_evt_input_audio_transcription_delta(evt)
             elif evt.type == "conversation.item.input_audio_transcription.completed":
                 await self.handle_evt_input_audio_transcription_completed(evt)
+            elif evt.type == "conversation.item.retrieved":
+                await self._handle_conversation_item_retrieved(evt)
             elif evt.type == "response.done":
                 await self._handle_evt_response_done(evt)
             elif evt.type == "input_audio_buffer.speech_started":
@@ -461,6 +470,14 @@ class OpenAIRealtimeBetaLLMService(LLMService):
             # User message without preceding conversation.item.created. Bug?
             logger.warning(f"Transcript for unknown user message: {evt}")
 
+    async def _handle_conversation_item_retrieved(self, evt: events.ConversationItemRetrieved):
+        future = self._retrieve_conversation_item_futures.get(evt.item.id)
+        if future:
+            # print(f"[pk] setting result: {evt.item}")
+            future.set_result(evt.item)
+            # TODO: handle error
+            # TODO: what happens if we try to receive bogus item id?
+
     async def _handle_evt_response_done(self, evt):
         # todo: figure out whether there's anything we need to do for "cancelled" events
         # usage metrics