wip

src/dailyai/services/base_transport_service.py

@@ -1,5 +1,6 @@
 from abc import abstractmethod
 import asyncio
+import functools
 import itertools
 import logging
 import numpy as np
@@ -11,6 +12,7 @@ import threading
 import time
 from typing import AsyncGenerator
 from enum import Enum
+from typing import AsyncGenerator, AsyncIterable, BinaryIO, Iterable
 
 from dailyai.queue_frame import (
     AudioQueueFrame,
@@ -122,6 +124,50 @@ class BaseTransportService():
         self._is_interrupted = threading.Event()
 
         self._logger: logging.Logger = logging.getLogger()
+    
+    def update_messages(self, new_context: list[dict[str, str]], task: asyncio.Task | None):
+        if task:
+            if not task.cancelled():
+                self._current_phrase = ""
+                self._context = new_context
+    
+    
+    
+    async def run_pipeline(self, frame):
+        # TODO-CB: This exception for missing class gets eaten!
+        await self._runner(frame)
+    
+    async def run_conversation(self, runner: Iterable[QueueFrame]
+                               | AsyncIterable[QueueFrame]
+                               | asyncio.Queue[QueueFrame],
+                               ) -> AsyncGenerator[QueueFrame, None]:
+        current_response_task = None
+        self._runner = runner
+    
+        async for frame in self.get_receive_frames():
+            if isinstance(frame, EndStreamQueueFrame):
+                break
+            # elif not isinstance(frame, TranscriptionQueueFrame):
+                # continue
+            # TODO-CB: Verify this is an accurate replacement
+            # if hasattr(frame, 'participantId') and frame.participantId == self._my_participant_id:
+            # if not isinstance(frame, UserStoppedSpeakingFrame):
+            #     continue
+    
+            if current_response_task and isinstance(frame, UserStartedSpeakingFrame):
+                # TODO-CB: Maybe not always interrupt? Are there frame types we can pass through?
+                current_response_task.cancel()
+                self.interrupt()
+    
+            # self._current_phrase += " " + frame.text
+           # current_llm_context = copy.deepcopy(self._context)
+            current_response_task = asyncio.create_task(
+                self.run_pipeline(
+                    frame)
+            )
+            current_response_task.add_done_callback(
+                functools.partial(self.update_messages, self._context)
+            )
 
     async def run(self):
         self._prerun()

src/examples/foundational/06b-patient-intake.py

@@ -443,7 +443,7 @@ async def main(room_url: str, token):
         fl = FrameLogger("got transcript")
         fl2 = FrameLogger("just above the checklist")
 
-        async def handle_transcriptions(user_speech, tma_in, tma_out):
+        async def run_response(user_speech, tma_in, tma_out):
             tf = TranscriptFilter(transport._my_participant_id)
             await tts.run_to_queue(
                 transport.send_queue,
@@ -458,17 +458,6 @@ async def main(room_url: str, token):
                 )
             )
 
-        async def run_conversation():
-
-            conversation_wrapper = InterruptibleConversationWrapper(
-                frame_generator=transport.get_receive_frames,
-                runner=handle_transcriptions,
-                interrupt=transport.interrupt,
-                my_participant_id=transport._my_participant_id,
-                llm_messages=messages,
-            )
-            await conversation_wrapper.run_conversation()
-
         @transport.event_handler("on_first_other_participant_joined")
         async def on_first_other_participant_joined(transport):
             fl = FrameLogger("first other participant")
@@ -483,8 +472,9 @@ async def main(room_url: str, token):
         transport.transcription_settings["extra"]["endpointing"] = True
         transport.transcription_settings["extra"]["punctuate"] = True
         try:
-            await asyncio.gather(transport.run(), run_conversation())
+            await asyncio.gather(transport.run(), transport.run_conversation(run_response))
         except (asyncio.CancelledError, KeyboardInterrupt):
+            print('whoops')
             transport.stop()