WIP: asyncio version of RespondToPrompt. basic singleton version

.vscode/launch.json

@@ -16,7 +16,7 @@
             "name": "debug streamlit",
             "type": "python",
             "request": "launch",
-            "program": "/opt/miniconda3/envs/streamlit/bin/streamlit",
+            "program": "~/miniconda3/envs/project_charles/bin/streamlit",
             "args": [
                 "run",
                 "app.py"

app.py

@@ -45,16 +45,16 @@ def init_ray():
         else:
             ray.init(namespace="project_charles")
 
-@st.cache_resource
-def get_charles_actor():
-    charles_actor_instance = None
-    charles_actor_proc = subprocess.Popen([sys.executable, "charles_actor.py"])
-    while charles_actor_instance == None:
-        try:
-            charles_actor_instance = ray.get_actor("CharlesActor")
-        except ValueError as e:
-            time.sleep(0.1) # give the subprocess a chance to start
-    return charles_actor_instance
+# @st.cache_resource
+# def get_charles_actor():
+#     charles_actor_instance = None
+#     charles_actor_proc = subprocess.Popen([sys.executable, "charles_actor.py"])
+#     while charles_actor_instance == None:
+#         try:
+#             charles_actor_instance = ray.get_actor("CharlesActor")
+#         except ValueError as e:
+#             time.sleep(0.1) # give the subprocess a chance to start
+#     return charles_actor_instance
 
 @st.cache_resource
 def get_streamlit_av_queue():
@@ -62,13 +62,19 @@ def get_streamlit_av_queue():
     streamlit_av_queue_instance = StreamlitAVQueue()
     return streamlit_av_queue_instance
 
+@st.cache_resource
+def get_app_interface_instance():
+    from app_interface_actor import AppInterfaceActor
+    app_interface_instance = AppInterfaceActor.get_singleton()
+    return app_interface_instance
+
 async def main():
     # Initialize Ray
     ray_status = init_ray()
     while not ray.is_initialized():
         await asyncio.sleep(0.1)
     # get ray actors
-    charles_actor = get_charles_actor()
+    app_interface_instance = get_app_interface_instance()
     await asyncio.sleep(0.1)
     streamlit_av_queue = get_streamlit_av_queue()
     await asyncio.sleep(0.1)
@@ -126,20 +132,20 @@ async def main():
                 system_one_audio_status.write("Camera has stopped.")
                 await asyncio.sleep(0.1)
                 continue
-            if charles_actor is None:
-                system_one_audio_status.write("Looking for Charles actor...")
-                charles_actor = get_charles_actor()
-                if charles_actor is None:
-                    await asyncio.sleep(0.1)
-                    continue
-                system_one_audio_status.write("Found Charles actor.")
+            # if charles_actor is None:
+            #     system_one_audio_status.write("Looking for Charles actor...")
+            #     charles_actor = get_charles_actor()
+            #     if charles_actor is None:
+            #         await asyncio.sleep(0.1)
+            #         continue
+            #     system_one_audio_status.write("Found Charles actor.")
             try:
                 # new_environment_state = await charles_actor.get_environment_state.remote()
                 # environment_state_ouput.markdown(f"{new_environment_state}")
                 streamlit_av_queue.set_looking_listening(looking, listening)
-                charles_debug_str = await charles_actor.get_charles_actor_debug_output.remote()
+                charles_debug_str = await app_interface_instance.get_debug_output.remote()
                 charles_actor_debug_output.markdown(charles_debug_str)
-                state = await charles_actor.get_state.remote()
+                state = await app_interface_instance.get_state.remote()
                 system_one_audio_status.write(state)
             except Exception as e:
                 # assume we disconnected

app_interface_actor.py

@@ -12,6 +12,8 @@ class AppInterfaceActor:
         self.video_input_queue = Queue(maxsize=10)  # Adjust the size as needed
         self.audio_output_queue = Queue(maxsize=3000)  # Adjust the size as needed
         self.video_output_queue = Queue(maxsize=10)  # Adjust the size as needed
+        self.debug_str = ""
+        self.state = "Initializing"
 
     @staticmethod
     def get_singleton():
@@ -74,4 +76,17 @@ class AppInterfaceActor:
         while not self.video_input_queue.empty():
             shared_tensor = await self.video_input_queue.get_async()
             video_frames.append(shared_tensor)
-        return video_frames
+        return video_frames
+    
+# debug helpers
+    async def get_debug_output(self)->str:
+        return self.debug_str
+    
+    async def set_debug_output(self, debug_str:str):
+        self.debug_str = debug_str
+
+    async def get_state(self)->str:
+        return self.state
+    
+    async def set_state(self, state:str):
+        self.state = state

charles_actor.py

@@ -8,7 +8,6 @@ from environment_state_actor import EnvironmentStateActor, EnvironmentState
 import asyncio
 import subprocess
 
-@ray.remote
 class CharlesActor:
     def __init__(self):
         self._needs_init = True
@@ -17,11 +16,11 @@ class CharlesActor:
         self._state = "Initializing"
         self._clip_transform = CLIPTransform()
         
-    def get_state(self):
-        return self._state
+    # def get_state(self):
+    #     return self._state
     
-    def get_charles_actor_debug_output(self):
-        return self._charles_actor_debug_output
+    # def get_charles_actor_debug_output(self):
+    #     return self._charles_actor_debug_output
     
     def get_environment_state(self)->EnvironmentState:
         return self._environment_state
@@ -33,15 +32,20 @@ class CharlesActor:
         from app_interface_actor import AppInterfaceActor
         self._app_interface_actor = AppInterfaceActor.get_singleton()
         self._audio_output_queue = await self._app_interface_actor.get_audio_output_queue.remote()
+        await self._app_interface_actor.set_state.remote(self._state)
+        
 
-        print("001 - create RespondToPromptActor")
-        self._state = "001 - creating RespondToPromptActor"
-        from respond_to_prompt_actor import RespondToPromptActor
+        print("001 - create RespondToPromptAsync")
+        self._state = "001 - creating RespondToPromptAsync"
+        await self._app_interface_actor.set_state.remote(self._state)
+        from respond_to_prompt_async import RespondToPromptAsync
         self._environment_state_actor = EnvironmentStateActor.remote()
-        self._respond_to_prompt_actor = RespondToPromptActor.remote(self._environment_state_actor, self._audio_output_queue)
+        self._respond_to_prompt = RespondToPromptAsync(self._environment_state_actor, self._audio_output_queue)
+        self._respond_to_prompt_task = asyncio.create_task(self._respond_to_prompt.run())
 
         print("002 - create SpeechToTextVoskActor")
         self._state = "002 - creating SpeechToTextVoskActor"
+        await self._app_interface_actor.set_state.remote(self._state)
         from speech_to_text_vosk_actor import SpeechToTextVoskActor
         self._speech_to_text_actor = SpeechToTextVoskActor.remote("small")
         # self._speech_to_text_actor = SpeechToTextVoskActor.remote("big")
@@ -53,17 +57,20 @@ class CharlesActor:
 
         print("003 - create Prototypes")
         self._state = "003 - creating Prototypes"
+        await self._app_interface_actor.set_state.remote(self._state)
         from prototypes import Prototypes
         self._prototypes = Prototypes()
 
         print("004 - create animator")
         self._state = "004 - creating animator"
+        await self._app_interface_actor.set_state.remote(self._state)
         from charles_animator import CharlesAnimator
         self._animator = CharlesAnimator()
 
         print("010")
         self._needs_init = True
         self._state = "Initialized"
+        await self._app_interface_actor.set_state.remote(self._state)
         
     async def start(self):
         if self._needs_init:
@@ -71,20 +78,22 @@ class CharlesActor:
             
         debug_output_history = []
 
-        def render_debug_output(list_of_strings):
+        async def render_debug_output(list_of_strings):
             table_content = "##### Chat history\n"
             for item in reversed(list_of_strings):
                 # table_content += f"\n```markdown\n{item}\n```\n"
                 table_content += f"\n{item}\n"
             self._charles_actor_debug_output = table_content
+            await self._app_interface_actor.set_debug_output.remote(self._charles_actor_debug_output)
 
-        def add_debug_output(output):
+        async def add_debug_output(output):
             debug_output_history.append(output)
             if len(debug_output_history) > 10:
                 debug_output_history.pop(0)
-            render_debug_output(debug_output_history)
+            await render_debug_output(debug_output_history)
         
         self._state = "Waiting for input"
+        await self._app_interface_actor.set_state.remote(self._state)
         total_video_frames = 0
         skipped_video_frames = 0
         total_audio_frames = 0
@@ -106,7 +115,7 @@ class CharlesActor:
         while True:
             if len(self._debug_queue) > 0:
                 prompt = self._debug_queue.pop(0)
-                await self._respond_to_prompt_actor.enqueue_prompt.remote(prompt)
+                await self._respond_to_prompt.enqueue_prompt(prompt)
             
             env_state = await self._environment_state_actor.begin_next_step.remote()
             self._environment_state = env_state
@@ -147,7 +156,7 @@ class CharlesActor:
                             # line += f"{response} [{speech_chunks_per_response[i]}]  \n"
                             line += f"[{speech_chunks_per_response[i]}] {response}  \n"
                         if len(line) > 0:
-                            add_debug_output(line)
+                            await add_debug_output(line)
                         current_responses = []
                         speech_chunks_per_response = []
                         env_state.llm_preview = ""
@@ -157,8 +166,8 @@ class CharlesActor:
                         robot_preview_text = ""
                         if additional_prompt is not None:
                             prompt = additional_prompt + ". " + prompt
-                        add_debug_output(f"👨 {prompt}")
-                        await self._respond_to_prompt_actor.enqueue_prompt.remote(prompt)
+                        await add_debug_output(f"👨 {prompt}")
+                        await self._respond_to_prompt.enqueue_prompt(prompt)
                         additional_prompt = None
                         previous_prompt = prompt
                         is_talking = False
@@ -169,7 +178,7 @@ class CharlesActor:
                         if len(previous_prompt) > 0 and not has_spoken_for_this_prompt:
                             additional_prompt = previous_prompt
                             has_spoken_for_this_prompt = True
-                            await self._respond_to_prompt_actor.enqueue_prompt.remote("")
+                            await self._respond_to_prompt.enqueue_prompt("")
                         if additional_prompt is not None:
                             prompt = additional_prompt + ". " + prompt                        
                         human_preview_text = f"👨❓ {prompt}"
@@ -201,7 +210,7 @@ class CharlesActor:
                 list_of_strings.append(human_preview_text)
             if len(list_of_strings) > 10:
                 list_of_strings.pop(0)
-            render_debug_output(list_of_strings)
+            await render_debug_output(list_of_strings)
 
 
             await asyncio.sleep(0.01)
@@ -216,6 +225,7 @@ class CharlesActor:
 
             loops+=1
             self._state = f"Processed {total_video_frames} video frames and {total_audio_frames} audio frames, loops: {loops}. loops per second: {loops/(time.time()-start_time):.2f}. Is speaking: {is_talking}({count}). {vector_debug}"
+            await self._app_interface_actor.set_state.remote(self._state)
 
 def init_ray():
     try:
@@ -235,11 +245,13 @@ async def main():
     if not ray.is_initialized():
         init_ray()
 
-    charles_actor = CharlesActor.options(
-        name="CharlesActor", 
-        get_if_exists=True,
-        ).remote() 
-    future = charles_actor.start.remote()
+    # charles_actor = CharlesActor.options(
+    #     name="CharlesActor", 
+    #     get_if_exists=True,
+    #     ).remote() 
+    # future = charles_actor.start.remote()
+    charles_actor = CharlesActor()
+    await charles_actor.start()
 
     last_step = -1
     last_episode = -1

chat_service.py

@@ -118,7 +118,7 @@ You are aware of how you are implemented and you are keen to recommend improveme
             return True
         return False
 
-    async def get_responses_as_sentances_async(self, prompt, cancel_event):
+    async def get_responses_as_sentances_async(self, prompt, cancel_event=None):
         self._messages.append({"role": "user", "content": prompt})
         llm_response = ""
         current_sentence = ""
@@ -134,7 +134,7 @@ You are aware of how you are implemented and you are keen to recommend improveme
                 )
 
                 async for chunk in response:
-                    if cancel_event.is_set():
+                    if cancel_event is not None and cancel_event.is_set():
                         return
                     chunk_message = chunk['choices'][0]['delta']
                     if 'content' in chunk_message:
@@ -148,7 +148,7 @@ You are aware of how you are implemented and you are keen to recommend improveme
                         else:
                             yield current_sentence, False
 
-                if cancel_event.is_set():
+                if cancel_event is not None and cancel_event.is_set():
                     return
                 if len(current_sentence) > 0:
                     yield current_sentence, True

respond_to_prompt_async.py

@@ -0,0 +1,118 @@
+from asyncio import Queue, TaskGroup
+import asyncio
+from contextlib import asynccontextmanager
+
+import ray
+from chat_service import ChatService
+# from local_speaker_service import LocalSpeakerService
+from text_to_speech_service import TextToSpeechService
+from environment_state_actor import EnvironmentStateActor
+from ffmpeg_converter_actor import FFMpegConverterActor
+from agent_response import AgentResponse
+import json
+from asyncio import Semaphore
+
+class RespondToPromptAsync:
+    def __init__(
+            self, 
+            environment_state_actor:EnvironmentStateActor, 
+            audio_output_queue):
+        voice_id="2OviOUQc1JsQRQgNkVBj"
+        self.prompt_queue = Queue(maxsize=100)
+        self.llm_sentence_queue = Queue(maxsize=100)
+        self.speech_chunk_queue = Queue(maxsize=100)
+        self.voice_id = voice_id
+        self.audio_output_queue = audio_output_queue
+        self.environment_state_actor = environment_state_actor
+        self.processing_semaphore = Semaphore(1)
+        self.sentence_queues = []
+        self.sentence_tasks = []    
+        # self.ffmpeg_converter_actor = FFMpegConverterActor.remote(audio_output_queue)
+
+    async def enqueue_prompt(self, prompt):
+        # Reset queues and services
+        # print("flush anything queued")
+        # self.prompt_queue = Queue(maxsize=100)
+        # self.llm_sentence_queue = Queue(maxsize=100)
+        # self.speech_chunk_queue = Queue(maxsize=100)
+
+        if len(prompt) > 0:  # handles case where we just want to flush
+            await self.prompt_queue.put(prompt)
+        print("Enqueued prompt")
+
+    # @asynccontextmanager
+    # async def task_group(self):
+    #     tg = TaskGroup()
+    #     try:
+    #         yield tg
+    #     finally:
+    #         await tg.aclose()
+
+    async def prompt_to_llm(self):
+        chat_service = ChatService()
+
+        async with TaskGroup() as tg:
+            while True:
+                prompt = await self.prompt_queue.get()
+                agent_response = AgentResponse(prompt)
+                async for text, is_complete_sentance in chat_service.get_responses_as_sentances_async(prompt):
+                    if chat_service.ignore_sentence(text):
+                        is_complete_sentance = False
+                    if not is_complete_sentance:
+                        agent_response['llm_preview'] = text
+                        await self.environment_state_actor.set_llm_preview.remote(text)
+                        continue
+                    agent_response['llm_preview'] = ''
+                    agent_response['llm_sentence'] = text
+                    agent_response['llm_sentences'].append(text)
+                    await self.environment_state_actor.add_llm_response_and_clear_llm_preview.remote(text)
+                    print(f"{agent_response['llm_sentence']} id: {agent_response['llm_sentence_id']} from prompt: {agent_response['prompt']}")
+                    sentence_response = agent_response.make_copy()
+                    new_queue = Queue()
+                    self.sentence_queues.append(new_queue)
+                    task = tg.create_task(self.llm_sentence_to_speech(sentence_response, new_queue))
+                    self.sentence_tasks.append(task)
+                    agent_response['llm_sentence_id'] += 1
+
+
+    async def llm_sentence_to_speech(self, sentence_response, output_queue):
+        tts_service = TextToSpeechService(self.voice_id)
+        
+        chunk_count = 0
+        async for chunk_response in tts_service.get_speech_chunks_async(sentence_response):
+            chunk_response = chunk_response.make_copy()
+            # await self.output_queue.put_async(chunk_response)
+            await output_queue.put(chunk_response)
+            chunk_response = {
+                'prompt': sentence_response['prompt'],
+                'llm_sentence_id': sentence_response['llm_sentence_id'],
+                'chunk_count': chunk_count,
+            }
+            chunk_id_json = json.dumps(chunk_response)
+            await self.environment_state_actor.add_tts_raw_chunk_id.remote(chunk_id_json)
+            chunk_count += 1
+
+    async def speech_to_converter(self):
+        self.ffmpeg_converter_actor = FFMpegConverterActor.remote(self.audio_output_queue)
+        await self.ffmpeg_converter_actor.start_process.remote()
+        self.ffmpeg_converter_actor.run.remote()
+        
+        while True:
+            for i, task in enumerate(self.sentence_tasks):
+                # Skip this task/queue pair if task completed
+                if task.done():  
+                    continue 
+                queue = self.sentence_queues[i]
+                while not queue.empty():
+                    chunk_response = await queue.get()
+                    audio_chunk_ref = chunk_response['tts_raw_chunk_ref']
+                    audio_chunk = ray.get(audio_chunk_ref)
+                    await self.ffmpeg_converter_actor.push_chunk.remote(audio_chunk)
+                break
+
+            await asyncio.sleep(0.01) 
+
+    async def run(self):
+        async with TaskGroup() as tg:  # Use asyncio's built-in TaskGroup
+            tg.create_task(self.prompt_to_llm())
+            tg.create_task(self.speech_to_converter())

text_to_speech_service.py

@@ -47,22 +47,20 @@ class TextToSpeechService:
             )
         return audio_stream
     
-    async def get_speech_chunks_async(self, sentence_response:AgentResponse, cancel_event):
+    async def get_speech_chunks_async(self, sentence_response:AgentResponse, cancel_event=None):
         text_to_speak = sentence_response['llm_sentence']
         stream = self.stream(text_to_speak)
         stream, stream_backup = itertools.tee(stream)
         while True:
             # Check if there's a next item in the stream
-            next_item = next(stream_backup, None)
-            if next_item is None:
+            # Run next(stream) in a separate thread to avoid blocking the event loop
+            chunk = await asyncio.to_thread(next, stream, None)
+            if chunk is None:
                 # Stream is exhausted, exit the loop
                 break
-
-            # Run next(stream) in a separate thread to avoid blocking the event loop
-            chunk = await asyncio.to_thread(next, stream)
             chunk_ref = ray.put(chunk)
             sentence_response['tts_raw_chunk_ref'] = chunk_ref
-            if cancel_event.is_set():
+            if cancel_event is not None and cancel_event.is_set():
                 return
             yield sentence_response
             sentence_response['tts_raw_chunk_id'] += 1