create instance per prompt;

app_interface_actor.py

@@ -10,7 +10,7 @@ class AppInterfaceActor:
     def __init__(self):
         self.audio_input_queue = Queue(maxsize=3000)  # Adjust the size as needed
         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.audio_output_queue = Queue(maxsize=50)  # Adjust the size as needed
         self.video_output_queue = Queue(maxsize=10)  # Adjust the size as needed
         self.debug_str = ""
         self.state = "Initializing"

charles_actor.py

@@ -5,6 +5,7 @@ import asyncio
 import os
 from clip_transform import CLIPTransform
 from environment_state_actor import EnvironmentStateActor, EnvironmentState
+from respond_to_prompt_async import RespondToPromptAsync
 import asyncio
 import subprocess
 
@@ -34,7 +35,6 @@ 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)
 
         self.set_state("002 - creating EnvironmentStateActor")
         self._environment_state_actor = EnvironmentStateActor.remote()
@@ -44,35 +44,24 @@ class CharlesActor:
         self._prompt_manager = PromptManager()
 
         self.set_state("004 - creating RespondToPromptAsync")
-        await self._app_interface_actor.set_state.remote(self._state)
-        from respond_to_prompt_async import RespondToPromptAsync
-        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())
+        self._respond_to_prompt = None
+        self._respond_to_prompt_task = None
 
         self.set_state("005 - create 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")
-        
-        self._debug_queue = [
-            # "hello, how are you today?",
-            # "hmm, interesting, tell me more about that.",
-        ]
 
         self.set_state("006 - create Prototypes")
-        await self._app_interface_actor.set_state.remote(self._state)
         from prototypes import Prototypes
         self._prototypes = Prototypes()
 
         self.set_state("007 - create animator")
-        await self._app_interface_actor.set_state.remote(self._state)
         from charles_animator import CharlesAnimator
         self._animator = CharlesAnimator()
 
         self._needs_init = True
         self.set_state("010 - Initialized")
-        await self._app_interface_actor.set_state.remote(self._state)
         
     async def start(self):
         if self._needs_init:
@@ -95,7 +84,6 @@ class CharlesActor:
             await render_debug_output(debug_output_history)
         
         self.set_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
@@ -113,13 +101,7 @@ class CharlesActor:
         is_talking = False
         has_spoken_for_this_prompt = False
 
-
-        while True:
-            if len(self._debug_queue) > 0:
-                prompt = self._debug_queue.pop(0)
-                self._prompt_manager.append_user_message(prompt)
-                await self._respond_to_prompt.enqueue_prompt(prompt, self._prompt_manager.messages)
-            
+        while True:            
             env_state = await self._environment_state_actor.begin_next_step.remote()
             self._environment_state = env_state
             audio_frames = await self._app_interface_actor.dequeue_audio_input_frames_async.remote()    
@@ -171,7 +153,11 @@ class CharlesActor:
                             prompt = additional_prompt + ". " + prompt
                         await add_debug_output(f"👨 {prompt}")
                         self._prompt_manager.append_user_message(prompt)
-                        await self._respond_to_prompt.enqueue_prompt(prompt, self._prompt_manager.messages)
+                        if self._respond_to_prompt_task is not None:
+                            await self._respond_to_prompt.terminate()
+                            self._respond_to_prompt_task.cancel()
+                        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(prompt, self._prompt_manager.messages))
                         additional_prompt = None
                         previous_prompt = prompt
                         is_talking = False
@@ -182,7 +168,11 @@ 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.enqueue_prompt("", self._prompt_manager.messages)
+                            if self._respond_to_prompt_task is not None:
+                                await self._respond_to_prompt.terminate()
+                                self._respond_to_prompt_task.cancel()
+                            self._respond_to_prompt_task = None
+                            self._respond_to_prompt = None
                         if additional_prompt is not None:
                             prompt = additional_prompt + ". " + prompt                        
                         human_preview_text = f"👨❓ {prompt}"
@@ -218,7 +208,7 @@ class CharlesActor:
             await render_debug_output(list_of_strings)
 
 
-            await asyncio.sleep(0.01)
+            await asyncio.sleep(0.001)
 
             # add observations to the environment state
             count = len(self._audio_output_queue)
@@ -236,7 +226,6 @@ class CharlesActor:
                     Is speaking: {is_talking}({count}). \
                     {vector_debug}\
                     ", skip_print=True)
-            await self._app_interface_actor.set_state.remote(self._state)
 
 def init_ray():
     try:

ffmpeg_converter_actor.py

@@ -3,21 +3,27 @@ import asyncio
 import ray
 from ray.util.queue import Queue
 
-@ray.remote
 class FFMpegConverterActor:
     def __init__(self, output_queue: Queue, buffer_size: int = 1920, output_format: str='s16le'):
         self.output_queue = output_queue
         self.buffer_size = buffer_size
         self.output_format = output_format
-
         self.input_pipe = None
         self.output_pipe = None
-
         self.process = None
+        self.running = True
 
     async def run(self):
-        while True:
-            chunk = await self.output_pipe.readexactly(self.buffer_size)
+        while self.running:
+            try:
+                chunk = await self.output_pipe.readexactly(self.buffer_size)
+            except asyncio.IncompleteReadError:
+                # exit if we have finsihsed the process
+                if self.running == False:
+                    return
+                # If the pipe is broken, restart the process.
+                await self.start_process()
+                continue
             # print(f"FFMpegConverterActor: read {len(chunk)} bytes")
             chunk_ref = ray.put(chunk)
             await self.output_queue.put_async(chunk_ref)       
@@ -55,14 +61,11 @@ class FFMpegConverterActor:
             self.input_pipe.write(chunk)
             await self.input_pipe.drain()
 
-    # def has_processed_all_data(self):
-    #     return self.process.poll() is not None
-
-    async def flush_output_queue(self):
-        while not self.output_queue.empty():
-            await self.output_queue.get_async()   
-
-    def close(self):
-        self.input_pipe.close()
-        self.output_pipe.close()
-        self.process.wait()
+    async def close(self):
+        self.running = False  # Stop the loop inside run()
+        if self.process:
+            self.process.stdin.transport.close()
+            self.process.kill()
+            self.process.terminate()
+        # while not self.output_queue.empty():
+        #     await self.output_queue.get_async()

respond_to_prompt_async.py

@@ -10,7 +10,6 @@ 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__(
@@ -18,47 +17,38 @@ class RespondToPromptAsync:
             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:str, messages:[str]):
-        if len(prompt) > 0:  # handles case where we just want to flush
-            await self.prompt_queue.put((prompt, messages))
-        print("Enqueued prompt")
-
-    async def prompt_to_llm(self):
+    async def prompt_to_llm(self, prompt:str, messages:[str]):
         chat_service = ChatService()
 
         async with TaskGroup() as tg:
-            while True:
-                prompt, messages = await self.prompt_queue.get()
-                agent_response = AgentResponse(prompt)
-                async for text, is_complete_sentance in chat_service.get_responses_as_sentances_async(messages):
-                    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
+            agent_response = AgentResponse(prompt)
+            async for text, is_complete_sentance in chat_service.get_responses_as_sentances_async(messages):
+                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):
@@ -79,9 +69,9 @@ class RespondToPromptAsync:
             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()
+        self.ffmpeg_converter_actor = FFMpegConverterActor(self.audio_output_queue)
+        await self.ffmpeg_converter_actor.start_process()
+        self.ffmpeg_converter_actor_task = asyncio.create_task(self.ffmpeg_converter_actor.run())
         
         while True:
             for i, task in enumerate(self.sentence_tasks):
@@ -93,12 +83,38 @@ class RespondToPromptAsync:
                     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)
+                    await self.ffmpeg_converter_actor.push_chunk(audio_chunk)
                 break
 
             await asyncio.sleep(0.01) 
 
-    async def run(self):
+    async def run(self, prompt:str, messages:[str]):
+        self.task_group_tasks = []
         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())
+            t1 = tg.create_task(self.prompt_to_llm(prompt, messages))
+            t2 = tg.create_task(self.speech_to_converter())
+            self.task_group_tasks.extend([t1, t2])
+
+    async def terminate(self):
+        # Cancel tasks
+        if self.task_group_tasks:
+            for task in self.task_group_tasks:
+                task.cancel()
+        for task in self.sentence_tasks:
+            task.cancel()
+
+        # Close FFmpeg converter actor
+        if self.ffmpeg_converter_actor_task:
+            self.ffmpeg_converter_actor_task.cancel()
+            await self.ffmpeg_converter_actor.close()
+        # ray.kill(self.ffmpeg_converter_actor)
+
+        # Flush all queues
+        while not self.llm_sentence_queue.empty():
+            await self.llm_sentence_queue.get()
+        while not self.speech_chunk_queue.empty():
+            await self.speech_chunk_queue.get()
+        for sentence_queue in self.sentence_queues:
+            while not sentence_queue.empty():
+                await sentence_queue.get()
+