wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Handle the possibility of multiple concurrent calls to retrieve_conversation_item() in the OpenAIRealtimeBetaLLMService

Browse Source
This commit is contained in:
Paul Kompfner
2025-03-18 10:05:41 -04:00
parent 31317ce77d
commit 7b594093dd
1 changed files with 16 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
src/pipecat/services/openai_realtime_beta/openai.py
View File

@@ -131,8 +131,14 @@ class OpenAIRealtimeBetaLLMService(LLMService):
async def retrieve_conversation_item(self, item_id: str): async def retrieve_conversation_item(self, item_id: str):
future = self.get_event_loop().create_future() future = self.get_event_loop().create_future()
self._retrieve_conversation_item_futures[item_id] = future retrieval_in_progress = False
await self.send_client_event(events.ConversationItemRetrieveEvent(item_id=item_id)) if not self._retrieve_conversation_item_futures.get(item_id):
self._retrieve_conversation_item_futures[item_id] = []
else:
retrieval_in_progress = True
self._retrieve_conversation_item_futures[item_id].append(future)
if not retrieval_in_progress:
await self.send_client_event(events.ConversationItemRetrieveEvent(item_id=item_id))
return await future return await future
# #
@@ -474,10 +480,10 @@ class OpenAIRealtimeBetaLLMService(LLMService):
logger.warning(f"Transcript for unknown user message: {evt}") logger.warning(f"Transcript for unknown user message: {evt}")
async def _handle_conversation_item_retrieved(self, evt: events.ConversationItemRetrieved): async def _handle_conversation_item_retrieved(self, evt: events.ConversationItemRetrieved):
future = self._retrieve_conversation_item_futures.pop(evt.item.id, None) futures = self._retrieve_conversation_item_futures.pop(evt.item.id, None)
if future: if futures:
# print(f"[pk] setting result: {evt.item}") for future in futures:
future.set_result(evt.item) future.set_result(evt.item)
async def _handle_evt_response_done(self, evt): async def _handle_evt_response_done(self, evt):
# todo: figure out whether there's anything we need to do for "cancelled" events # todo: figure out whether there's anything we need to do for "cancelled" events
@@ -543,9 +549,10 @@ class OpenAIRealtimeBetaLLMService(LLMService):
) )
if match: if match:
item_id = match.group(1) item_id = match.group(1)
future = self._retrieve_conversation_item_futures.pop(item_id, None) futures = self._retrieve_conversation_item_futures.pop(item_id, None)
if future: if futures:
future.set_exception(Exception(evt.error.message)) for future in futures:
future.set_exception(Exception(evt.error.message))
return True return True
return False return False