File size: 3,008 Bytes
bcea2ea
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
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)