create respond to prompt actor

charles_actor.py

@@ -27,10 +27,10 @@ class CharlesActor:
         from speech_to_text_vosk_actor import SpeechToTextVoskActor
         self._speech_to_text_actor = SpeechToTextVoskActor.remote()
 
-        from chat_pipeline import ChatPipeline
-        self._chat_pipeline = ChatPipeline()
-        await self._chat_pipeline.start()
-
+        print("003")
+        from respond_to_prompt_actor import RespondToPromptActor
+        self._respond_to_prompt_actor = RespondToPromptActor.remote()
+        
         self._debug_queue = [
             # "hello, how are you today?",
             # "hmm, interesting, tell me more about that.",
@@ -55,7 +55,7 @@ class CharlesActor:
         while True:
             if len(self._debug_queue) > 0:
                 prompt = self._debug_queue.pop(0)
-                await self._chat_pipeline.enqueue(prompt)
+                await self._respond_to_prompt_actor.enqueue(prompt)
             audio_frames = await self._streamlit_av_queue.get_audio_frames_async()    
             if len(audio_frames) > 0:
                 total_audio_frames += len(audio_frames)
@@ -79,7 +79,7 @@ class CharlesActor:
                         table_content = "| System 1 Audio History |\n| --- |\n"
                         table_content += "\n".join([f"| {item} |" for item in reversed(system_one_audio_history)])
                         self._system_one_audio_history_output = table_content
-                        await self._chat_pipeline.enqueue(prompt)
+                        self._respond_to_prompt_actor.enqueue_prompt.remote(prompt)
                         
             # video_frames = await self._streamlit_av_queue.get_video_frames_async()
             # if len(video_frames) > 0:
@@ -127,5 +127,6 @@ if __name__ == "__main__":
                 time.sleep(1)
                 state = charles_actor.get_state.remote()
                 print(f"Charles is in state: {ray.get(state)}")
-    except KeyboardInterrupt:
+    except KeyboardInterrupt as e:
         print("Script was manually terminated")
+        throw(e)

respond_to_prompt_actor.py

@@ -0,0 +1,77 @@
+import ray
+from ray.util.queue import Queue
+from dotenv import load_dotenv
+from audio_stream_processor import AudioStreamProcessor
+from streaming_chat_service import StreamingChatService
+
+# from ray.actor import ActorHandle
+
+@ray.remote
+class PromptToLLMActor:
+    def __init__(self, input_queue, output_queue, voice_id):
+        load_dotenv()
+        self.input_queue = input_queue
+        self.output_queue = output_queue
+        self.audio_processor = AudioStreamProcessor()
+        self.chat_service = StreamingChatService(self.audio_processor, voice_id=voice_id)
+
+    async def run(self):
+        while True:
+            prompt = self.input_queue.get()
+            async for sentence in self.chat_service.get_responses_as_sentances_async(prompt):
+                if self.chat_service.ignore_sentence(sentence):
+                    continue
+                print(f"{sentence}")
+                self.output_queue.put(sentence)
+
+@ray.remote
+class LLMSentanceToSpeechActor:
+    def __init__(self, input_queue, output_queue, voice_id):
+        load_dotenv()
+        self.input_queue = input_queue
+        self.output_queue = output_queue
+        self.audio_processor = AudioStreamProcessor()
+        self.chat_service = StreamingChatService(self.audio_processor, voice_id=voice_id)
+
+    async def run(self):
+        while True:
+            sentance = self.input_queue.get()
+            async for chunk in self.chat_service.get_speech_chunks_async(sentance):
+                self.output_queue.put(chunk)
+
+@ray.remote
+class SpeechToSpeakerActor:
+    def __init__(self, input_queue, voice_id):
+        load_dotenv()
+        self.input_queue = input_queue
+        self.audio_processor = AudioStreamProcessor()
+        self.chat_service = StreamingChatService(self.audio_processor, voice_id=voice_id)
+
+    async def run(self):
+        while True:
+            audio_chunk = self.input_queue.get()
+            self.chat_service.enqueue_speech_bytes_to_play([audio_chunk])
+
+@ray.remote
+class RespondToPromptActor:
+    def __init__(self):
+        voice_id="2OviOUQc1JsQRQgNkVBj"
+        self.prompt_queue = Queue(maxsize=100)
+        self.llm_sentence_queue = Queue(maxsize=100)
+        self.speech_chunk_queue = Queue(maxsize=100)
+        
+        self.prompt_to_llm = PromptToLLMActor.remote(self.prompt_queue, self.llm_sentence_queue, voice_id)
+        self.llm_sentence_to_speech = LLMSentanceToSpeechActor.remote(self.llm_sentence_queue, self.speech_chunk_queue, voice_id)
+        self.speech_to_speaker = SpeechToSpeakerActor.remote(self.speech_chunk_queue, voice_id)
+
+        # Start the pipeline components.
+        print ("Starting pipeline components")
+        self.prompt_to_llm.run.remote()
+        print ("prompt_to_llm running")
+        self.llm_sentence_to_speech.run.remote()
+        print ("llm_sentence_to_speech running")
+        self.speech_to_speaker.run.remote()
+        print ("speech_to_speaker running")
+            
+    def enqueue_prompt(self, prompt):
+        self.prompt_queue.put(prompt)