import json import ray import time import asyncio import os from clip_transform import CLIPTransform from response_state_manager import ResponseStateManager from respond_to_prompt_async import RespondToPromptAsync import asyncio import subprocess class CharlesApp: def __init__(self): self._needs_init = True self._charles_actor_debug_output = "" self._state = "Initializing" self._clip_transform = CLIPTransform() def set_state(self, state, skip_print=False): self._state = state if not skip_print: print(state) # check if self._app_interface_actor exists if hasattr(self, '_app_interface_actor'): self._app_interface_actor.set_state.remote(self._state) async def _initalize_resources(self): # Initialize resources self.set_state("001 - creating AppInterfaceActor") 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() self.set_state("002 - creating ResponseStateManager") self._response_state_manager = ResponseStateManager() self.set_state("003 - creating PromptManager") from prompt_manager import PromptManager self._prompt_manager = PromptManager() self.set_state("004 - creating RespondToPromptAsync") self._respond_to_prompt = None self._respond_to_prompt_task = None self.set_state("005 - create SpeechToTextVoskActor") 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.set_state("006 - create Prototypes") from prototypes import Prototypes self._prototypes = Prototypes() self.set_state("007 - create animator") from charles_animator import CharlesAnimator self._animator = CharlesAnimator() self._needs_init = True self.set_state("010 - Initialized") async def start(self): if self._needs_init: await self._initalize_resources() debug_output_history = [] 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) async def add_debug_output(output): debug_output_history.append(output) if len(debug_output_history) > 10: debug_output_history.pop(0) await render_debug_output(debug_output_history) self.set_state("Waiting for input") total_video_frames = 0 skipped_video_frames = 0 total_audio_frames = 0 loops = 0 start_time = time.time() vector_debug = "--n/a--" process_speech_to_text_future = [] human_preview_text = "" additional_prompt = None previous_prompt = "" is_talking = False has_spoken_for_this_prompt = False while True: response_step_obs, response_state = self._response_state_manager.begin_next_step() audio_frames = await self._app_interface_actor.dequeue_audio_input_frames_async.remote() video_frames = await self._app_interface_actor.dequeue_video_input_frames_async.remote() if len(audio_frames) > 0: total_audio_frames += len(audio_frames) # Concatenate all audio frames into a single buffer audio_buffer = b"".join([buffer.tobytes() for buffer in audio_frames]) future = self._speech_to_text_actor.process_speech.remote(audio_buffer) process_speech_to_text_future.append(future) # audio_frames_task = None if len(video_frames) > 0: vector_debug = f"found {len(video_frames)} video frames" total_video_frames += 1 skipped_video_frames += (len(video_frames) -1) image_as_array = video_frames[-1] image_vector = self._clip_transform.image_to_embeddings(image_as_array) image_vector = image_vector[0] distances, closest_item_key, distance_debug_str = self._prototypes.get_distances(image_vector) vector_debug = f"{closest_item_key} {distance_debug_str}" if len(process_speech_to_text_future) > 0: ready, _ = ray.wait([process_speech_to_text_future[0]], timeout=0) if ready: prompt, speaker_finished, raw_json = await process_speech_to_text_future[0] del process_speech_to_text_future[0] prompts_to_ignore = ["um", "uh", "ah", "huh", "hmm", "the", "but", "by", "just", "i'm"] if speaker_finished and len(prompt) > 0 and prompt not in prompts_to_ignore: print(f"Prompt: {prompt}") response_preview_text = self._response_state_manager.pretty_print_current_responses() if len(response_preview_text) > 0: await add_debug_output(response_preview_text) human_preview_text = "" if additional_prompt is not None: prompt = additional_prompt + ". " + prompt await add_debug_output(f"👨 {prompt}") self._prompt_manager.replace_or_append_user_message(prompt) 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._response_state_manager, 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 has_spoken_for_this_prompt = False response_step_obs, response_state = self._response_state_manager.reset_episode() elif len(prompt) > 0 and prompt not in prompts_to_ignore: # sometimes we get a false signal of speaker_finsihed # in which case we get new prompts before we have spoken if len(previous_prompt) > 0 and not has_spoken_for_this_prompt: additional_prompt = previous_prompt has_spoken_for_this_prompt = True 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 response_step_obs, response_state = self._response_state_manager.reset_episode() if additional_prompt is not None: prompt = additional_prompt + ". " + prompt human_preview_text = f"👨❓ {prompt}" # i choose to add each line of responce one at a time as them come in for new_response in response_step_obs.llm_responses: self._prompt_manager.append_assistant_message(new_response) list_of_strings = debug_output_history.copy() robot_preview_text = self._response_state_manager.pretty_print_preview_text() response_preview_text = self._response_state_manager.pretty_print_current_responses() if len(robot_preview_text) > 0: response_preview_text += robot_preview_text+" \n" list_of_strings.append(response_preview_text) if len(human_preview_text) > 0: list_of_strings.append(human_preview_text) if len(list_of_strings) > 10: list_of_strings.pop(0) await render_debug_output(list_of_strings) await asyncio.sleep(0.001) # add observations to the environment state count = len(self._audio_output_queue) is_talking = bool(count > 0) has_spoken_for_this_prompt = has_spoken_for_this_prompt or is_talking frame = self._animator.update(is_talking) frame_ref = ray.put(frame) await self._app_interface_actor.enqueue_video_output_frame.remote(frame_ref) loops+=1 self.set_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}\ ", skip_print=True) def init_ray(): try: subprocess.check_output(["ray", "start", "--include-dashboard=True", "--head"]) except Exception as e: print (f"charles_actor.py init_ray: {e}") # Connect to a running Ray cluster while not ray.is_initialized(): time.sleep(0.1) ray_address = os.getenv('RAY_ADDRESS') if ray_address: ray.init(ray_address, namespace="project_charles") else: ray.init(namespace="project_charles") async def main(): if not ray.is_initialized(): init_ray() charles_actor = CharlesApp() await charles_actor.start() if __name__ == "__main__": loop = asyncio.get_event_loop() loop.run_until_complete(main())