created chat_pipeline

audio_stream_processor.py

@@ -10,6 +10,7 @@ class AudioStreamProcessor:
         self.queue = Queue()
         self._is_running = threading.Event()
         self._is_running.set()
+        self.mpv_process = None
         self.thread = Thread(target=self._process_audio_streams)
         self.thread.start()
 
@@ -18,30 +19,34 @@ class AudioStreamProcessor:
             self.queue.put(audio_stream)
 
     def _process_audio_streams(self):
+        self._start_mpv()
         while self._is_running.is_set() or not self.queue.empty():
             try:
                 audio_stream = self.queue.get(timeout=1)
             except Empty:
                 continue
             self._stream(audio_stream)
+        self._close_mpv()
 
     def _stream(self, audio_stream: Iterator[bytes]):
+        for chunk in audio_stream:
+            if chunk is not None:
+                self.mpv_process.stdin.write(chunk)
+                self.mpv_process.stdin.flush()
+
+    def _start_mpv(self):
         mpv_command = ["mpv", "--no-cache", "--no-terminal", "--", "fd://0"]
-        mpv_process = subprocess.Popen(
+        self.mpv_process = subprocess.Popen(
             mpv_command,
             stdin=subprocess.PIPE,
             stdout=subprocess.DEVNULL,
             stderr=subprocess.DEVNULL,
         )
 
-        for chunk in audio_stream:
-            if chunk is not None:
-                mpv_process.stdin.write(chunk)
-                mpv_process.stdin.flush()
-
-        if mpv_process.stdin:
-            mpv_process.stdin.close()
-        mpv_process.wait()
+    def _close_mpv(self):
+        if self.mpv_process.stdin:
+            self.mpv_process.stdin.close()
+        self.mpv_process.wait()
 
     def close(self):
         self._is_running.clear()

chat_pipeline.py

@@ -0,0 +1,96 @@
+import asyncio
+import time
+from clip_transform import CLIPTransform
+from chat_service import ChatService
+from dotenv import load_dotenv
+from speech_service import SpeechService
+from concurrent.futures import ThreadPoolExecutor
+from audio_stream_processor import AudioStreamProcessor
+from streaming_chat_service import StreamingChatService
+from pipeline import Pipeline, Node, Job
+from typing import List
+
+class ChatJob(Job):
+    def __init__(self, data, chat_service: ChatService):
+        super().__init__(data)
+        self.chat_service = chat_service
+
+class Node1(Node):
+    next_id = 0
+
+    async def process_job(self, job: ChatJob):
+        # input job.data is the input string
+        # output job.data is the next sentance
+        async for sentence in job.chat_service.get_responses_as_sentances_async(job.data):
+            if job.chat_service.ignore_sentence(sentence):
+                continue
+            print(f"{sentence}")
+            new_job = ChatJob(sentence, job.chat_service)
+            new_job.id = self.next_id
+            self.next_id += 1
+            yield new_job
+    
+class Node2(Node):
+    next_id = 0
+
+    async def process_job(self, job: ChatJob):
+        # input job.data is the sentance
+        # output job.data is the streamed speech bytes
+        async for chunk in job.chat_service.get_speech_chunks_async(job.data):
+            new_job = ChatJob(chunk, job.chat_service)
+            new_job.id = self.next_id
+            self.next_id += 1
+            yield new_job
+
+
+class Node3(Node):
+    # sync_size = 64
+    # sync = []
+
+    async def process_job(self, job: ChatJob):
+        # input job.data is the streamed speech bytes
+        # Node3.sync.append(job.data)
+        job.chat_service.enqueue_speech_bytes_to_play([job.data])
+        yield job
+        # if len(Node3.sync) >= Node3.sync_size:
+        #     audio_chunks = Node3.sync[:Node3.sync_size]
+        #     Node3.sync = Node3.sync[Node3.sync_size:]
+        #     job.chat_service.enqueue_speech_bytes_to_play(audio_chunks)
+        # yield job
+
+class ChatPipeline():
+    def __init__(self):
+        load_dotenv()
+        self.pipeline = Pipeline()
+        self.audio_processor = AudioStreamProcessor()
+        self.chat_service = StreamingChatService(self.audio_processor, voice_id="2OviOUQc1JsQRQgNkVBj") # Chales003
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        self.audio_processor.close()
+        self.audio_processor = None
+    
+    def __del__(self):
+        if self.audio_processor:
+            self.audio_processor.close()
+            self.audio_processor = None
+
+    async def start(self):
+        self.node1_queue = asyncio.Queue()
+        self.node2_queue = asyncio.Queue()
+        self.node3_queue = asyncio.Queue()
+        self.sync = []
+        await self.pipeline.add_node(Node1, 1, self.node1_queue, self.node2_queue, sequential_node=True)
+        await self.pipeline.add_node(Node2, 1, self.node2_queue, self.node3_queue, sequential_node=True)
+        await self.pipeline.add_node(Node3, 1, self.node3_queue, None, sequential_node=True)
+
+    async def enqueue(self, prompt):
+        job = ChatJob(prompt, self.chat_service)
+        await self.pipeline.enqueue_job(job)
+
+    async def wait_until_all_jobs_idle(self):
+        # TODO - implement this
+        while True:
+            await asyncio.sleep(0.1)

debug.py

@@ -1,4 +1,6 @@
+import asyncio
 import time
+from chat_pipeline import ChatPipeline
 from clip_transform import CLIPTransform
 from chat_service import ChatService
 from dotenv import load_dotenv
@@ -125,8 +127,29 @@ def run_debug_code():
     audio_processor.close()
     print ("Chat success")
 
+async def run_pipeline():
+    load_dotenv()
+
+    try:
+        chat_pipeline = ChatPipeline()
+        await chat_pipeline.start()
+        prompts = [
+            "hello, how are you today?",
+            "tell me about your shadow self?",
+            "hmm, interesting, tell me more about that.",
+            "wait, that is so interesting, what else?",
+        ]
+        for prompt in prompts:
+            await chat_pipeline.enqueue(prompt)
+        await chat_pipeline.wait_until_all_jobs_idle()
+    except KeyboardInterrupt:
+        print("Pipeline interrupted by user")
+    except Exception as e:
+        print(f"An error occurred: {e}")
 
 if __name__ == '__main__':
     # time_sentance_lenghts()
     # test_sentance_lenghts()
-    run_debug_code()
+    # run_debug_code()
+    asyncio.run(run_pipeline())
+    

pipeline.py

@@ -1,4 +1,5 @@
 import asyncio
+import traceback
 
 class Job:
     def __init__(self, data):
@@ -23,28 +24,33 @@ class Node:
             raise ValueError('job_sync is not None and sequential_node is False')
 
     async def run(self):
-        while True:
-            job: Job = await self.input_queue.get()
-            self._jobs_dequeued += 1
-            if self.sequential_node == False:
-                async for job in self.process_job(job):
-                    if self.output_queue is not None:
-                        await self.output_queue.put(job)
-                    if self.job_sync is not None:
-                        self.job_sync.append(job)
-                    self._jobs_processed += 1
-            else:
-                # ensure that jobs are processed in order
-                self.buffer[job.id] = job
-                while self.next_i in self.buffer:
-                    job = self.buffer.pop(self.next_i)
+        try:
+            while True:
+                job: Job = await self.input_queue.get()
+                self._jobs_dequeued += 1
+                if self.sequential_node == False:
                     async for job in self.process_job(job):
                         if self.output_queue is not None:
                             await self.output_queue.put(job)
                         if self.job_sync is not None:
                             self.job_sync.append(job)
+                    self._jobs_processed += 1
+                else:
+                    # ensure that jobs are processed in order
+                    self.buffer[job.id] = job
+                    while self.next_i in self.buffer:
+                        job = self.buffer.pop(self.next_i)
+                        async for job in self.process_job(job):
+                            if self.output_queue is not None:
+                                await self.output_queue.put(job)
+                            if self.job_sync is not None:
+                                self.job_sync.append(job)
                         self._jobs_processed += 1
-                    self.next_i += 1
+                        self.next_i += 1
+        except Exception as e:
+            print(f"An error occurred in node: {self.__class__.__name__} worker: {self.worker_id}: {e}")
+            traceback.print_exc()
+            raise  # Re-raises the last exception.
 
     async def process_job(self, job: Job):
         raise NotImplementedError()
@@ -79,7 +85,7 @@ class Pipeline:
             self.nodes.append(node_name)
 
         # if input_queue is None then this is the root node
-        if len(self.input_queues) is 0:
+        if len(self.input_queues) == 0:
             self.root_queue = input_queue
 
         self.input_queues.append(input_queue)

speech_service.py

@@ -38,7 +38,8 @@ class SpeechService:
             text=prompt,
             voice=self._voice_id,
             model=self._model_id,
-            stream=True
+            stream_chunk_size=2048,
+            stream=True,
             )
         return audio_stream
     

streaming_chat_service.py

@@ -1,3 +1,5 @@
+import asyncio
+import itertools
 import json
 import os
 import torch
@@ -51,14 +53,19 @@ class StreamingChatService:
         text_to_speak = sentence[:last_termination_index+1]
         return text_to_speak
     
-    def _safe_enqueue_text_to_speak(self, text_to_speak):
+    def ignore_sentence(self, text_to_speak):
         # exit if empty, white space or an single breaket
         if text_to_speak.isspace():
-            return
+            return True
         # exit if not letters or numbers
         has_letters = any(char.isalpha() for char in text_to_speak)
         has_numbers = any(char.isdigit() for char in text_to_speak)
         if not has_letters and not has_numbers:
+            return True
+        return False
+    
+    def _safe_enqueue_text_to_speak(self, text_to_speak):
+        if self.ignore_sentence(text_to_speak):
             return
         stream = self._speech_service.stream(text_to_speak)
         self._audio_processor.add_audio_stream(stream)        
@@ -93,3 +100,47 @@ class StreamingChatService:
             print(current_sentence)
         self._messages.append({"role": "assistant", "content": agent_response})
         return agent_response
+
+    async def get_responses_as_sentances_async(self, prompt):
+        self._messages.append({"role": "user", "content": prompt})
+        agent_response = ""
+        current_sentence = ""
+
+        response = await openai.ChatCompletion.acreate(
+            model=self._model_id,
+            messages=self._messages,
+            temperature=1.0,  # use 1.0 for debugging/deterministic results
+            stream=True
+        )
+
+        async for chunk in response:
+            chunk_message = chunk['choices'][0]['delta']
+            if 'content' in chunk_message:
+                chunk_text = chunk_message['content']
+                current_sentence += chunk_text
+                agent_response += chunk_text
+                text_to_speak = self._should_we_send_to_voice(current_sentence)
+                if text_to_speak:
+                    yield text_to_speak
+                    current_sentence = current_sentence[len(text_to_speak):]
+
+        if len(current_sentence) > 0:
+            yield current_sentence
+        self._messages.append({"role": "assistant", "content": agent_response})
+     
+    async def get_speech_chunks_async(self, text_to_speak):
+        stream = self._speech_service.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:
+                # 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)
+            yield chunk
+
+    def enqueue_speech_bytes_to_play(self, speech_bytes):
+        self._audio_processor.add_audio_stream(speech_bytes)