From e726f15c4ef7aa63db3e3a307e281c73003b0632 Mon Sep 17 00:00:00 2001 From: Chad Bailey Date: Tue, 19 Mar 2024 15:31:19 +0000 Subject: [PATCH] wip: telestrator --- src/dailyai/pipeline/frames.py | 4 + .../services/base_transport_service.py | 9 +- .../services/daily_transport_service.py | 3 +- src/dailyai/services/fal_ai_services.py | 5 +- src/dailyai/services/open_ai_services.py | 29 +++-- .../starter-apps/telestrator/describer.py | 100 +++++++++++++++ .../starter-apps/telestrator/illustrator.py | 112 ++++++++++++++++ .../starter-apps/telestrator/telestrator.py | 120 ++++++++++++++++++ 8 files changed, 368 insertions(+), 14 deletions(-) create mode 100644 src/examples/starter-apps/telestrator/describer.py create mode 100644 src/examples/starter-apps/telestrator/illustrator.py create mode 100644 src/examples/starter-apps/telestrator/telestrator.py diff --git a/src/dailyai/pipeline/frames.py b/src/dailyai/pipeline/frames.py index b93d5bf5f..b9ca8ae5c 100644 --- a/src/dailyai/pipeline/frames.py +++ b/src/dailyai/pipeline/frames.py @@ -191,6 +191,10 @@ class VideoImageFrame(Frame): # return f"{self.__class__.__name__}, participantId: {self.participantId}, image size: {len(self.image)} B" +class TelestratorImageFrame(ImageFrame): + pass + + @dataclass() class VisionFrame(Frame): prompt: str diff --git a/src/dailyai/services/base_transport_service.py b/src/dailyai/services/base_transport_service.py index cfdea5b94..741709489 100644 --- a/src/dailyai/services/base_transport_service.py +++ b/src/dailyai/services/base_transport_service.py @@ -24,7 +24,8 @@ from dailyai.pipeline.frames import ( TextFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame, - RequestVideoImageFrame + RequestVideoImageFrame, + TelestratorImageFrame ) from dailyai.pipeline.pipeline import Pipeline from dailyai.services.ai_services import TTSService @@ -456,6 +457,12 @@ class BaseTransportService: self.write_frame_to_mic( bytes(b[:truncated_length])) b = b[truncated_length:] + elif isinstance(frame, TelestratorImageFrame): + self._set_image(frame.image) + asyncio.run_coroutine_threadsafe( + self.receive_queue.put(frame), + self._loop, + ) elif isinstance(frame, ImageFrame): self._set_image(frame.image) elif isinstance(frame, SpriteFrame): diff --git a/src/dailyai/services/daily_transport_service.py b/src/dailyai/services/daily_transport_service.py index 660b8de76..25d45afca 100644 --- a/src/dailyai/services/daily_transport_service.py +++ b/src/dailyai/services/daily_transport_service.py @@ -12,7 +12,8 @@ from typing import Any from dailyai.pipeline.frames import ( ReceivedAppMessageFrame, TranscriptionQueueFrame, - VideoImageFrame + VideoImageFrame, + TelestratorImageFrame ) from threading import Event diff --git a/src/dailyai/services/fal_ai_services.py b/src/dailyai/services/fal_ai_services.py index 10343f97c..bc881b357 100644 --- a/src/dailyai/services/fal_ai_services.py +++ b/src/dailyai/services/fal_ai_services.py @@ -53,4 +53,7 @@ class FalImageGenService(ImageGenService): async with self._aiohttp_session.get(image_url) as response: image_stream = io.BytesIO(await response.content.read()) image = Image.open(image_stream) - return (image_url, image.tobytes()) + image_bytes = image.tobytes() + print(f"!!! fal image tobytes is:") + print(image) + return (image_url, image_bytes) diff --git a/src/dailyai/services/open_ai_services.py b/src/dailyai/services/open_ai_services.py index 817ec21cf..afcb62979 100644 --- a/src/dailyai/services/open_ai_services.py +++ b/src/dailyai/services/open_ai_services.py @@ -14,6 +14,7 @@ from openai.types.chat import ( ChatCompletionMessageParam, ) +from daily import VideoFrame from dailyai.services.ai_services import LLMService, ImageGenService, VisionService from dailyai.services.openai_api_llm_service import BaseOpenAILLMService from dailyai.pipeline.frames import TextFrame @@ -71,17 +72,22 @@ class OpenAIVisionService(VisionService): self._client = AsyncOpenAI(api_key=api_key) async def run_vision(self, prompt: str, image: bytes): - IMAGE_WIDTH = image.width - IMAGE_HEIGHT = image.height - COLOR_FORMAT = image.color_format - a_image = Image.frombytes( - 'RGBA', (IMAGE_WIDTH, IMAGE_HEIGHT), image.buffer) - new_image = a_image.convert('RGB') - - # Uncomment these lines to write the frame to a jpg in the same directory. - # current_path = os.getcwd() - # image_path = os.path.join(current_path, "image.jpg") - # image.save(image_path, format="JPEG") + if isinstance(image, VideoFrame): + # Then it's from a daily video frame + print("### processing daily video frame for recognition") + IMAGE_WIDTH = image.width + IMAGE_HEIGHT = image.height + COLOR_FORMAT = image.color_format + a_image = Image.frombytes( + 'RGBA', (IMAGE_WIDTH, IMAGE_HEIGHT), image.buffer) + new_image = a_image.convert('RGB') + else: + # handle it as a byte stream from image gen + new_image = Image.frombytes('RGB', (1024, 1024), image) + # Uncomment these lines to write the frame to a jpg in the same directory. + # current_path = os.getcwd() + # image_path = os.path.join(current_path, "image.jpg") + # image.save(image_path, format="JPEG") jpeg_buffer = io.BytesIO() @@ -89,6 +95,7 @@ class OpenAIVisionService(VisionService): jpeg_bytes = jpeg_buffer.getvalue() base64_image = base64.b64encode(jpeg_bytes).decode('utf-8') + messages = [ { "role": "user", diff --git a/src/examples/starter-apps/telestrator/describer.py b/src/examples/starter-apps/telestrator/describer.py new file mode 100644 index 000000000..39f5ca70b --- /dev/null +++ b/src/examples/starter-apps/telestrator/describer.py @@ -0,0 +1,100 @@ +import asyncio +import aiohttp +import logging +import os +from typing import AsyncGenerator + +from dailyai.pipeline.frames import Frame, LLMMessagesQueueFrame, RequestVideoImageFrame, LLMResponseEndFrame +from dailyai.pipeline.pipeline import Pipeline +from dailyai.pipeline.frame_processor import FrameProcessor +from dailyai.services.daily_transport_service import DailyTransportService +from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService +from dailyai.services.open_ai_services import OpenAILLMService, OpenAIVisionService +from dailyai.services.fal_ai_services import FalImageGenService +from dailyai.services.deepgram_ai_services import DeepgramTTSService +from dailyai.services.ai_services import FrameLogger +from dailyai.pipeline.aggregators import ( + LLMAssistantContextAggregator, + LLMUserContextAggregator, +) +from dailyai.pipeline.frames import VideoImageFrame, VisionFrame +from examples.support.runner import configure + +logging.basicConfig(format=f"%(levelno)s %(asctime)s %(message)s") +logger = logging.getLogger("dailyai") +logger.setLevel(logging.DEBUG) + + +class VideoImageFrameProcessor(FrameProcessor): + def __init__(self): + pass + + async def process_frame(self, frame: Frame) -> AsyncGenerator[Frame, None]: + if isinstance(frame, VideoImageFrame): + yield VisionFrame("Describe the image in one sentence.", frame.image) + else: + yield frame + + +class ImageRefresher(FrameProcessor): + async def process_frame(self, frame: Frame) -> AsyncGenerator[Frame, None]: + if isinstance(frame, LLMResponseEndFrame): + yield RequestVideoImageFrame(participantId=None) + yield frame + else: + yield frame + + +async def main(room_url: str, token): + async with aiohttp.ClientSession() as session: + transport = DailyTransportService( + room_url, + token, + "Respond bot", + duration_minutes=5, + start_transcription=True, + mic_enabled=True, + mic_sample_rate=16000, + camera_enabled=True, + camera_width=1024, + camera_height=1024, + vad_enabled=False, + receive_video=True, + receive_video_fps=0 + ) + + tts = ElevenLabsTTSService( + aiohttp_session=session, + api_key=os.getenv("ELEVENLABS_API_KEY"), + voice_id=os.getenv("ELEVENLABS_VOICE_ID"), + ) + + llm = OpenAILLMService( + api_key=os.getenv("OPENAI_CHATGPT_API_KEY"), + model="gpt-4-turbo-preview") + + vs = OpenAIVisionService(api_key=os.getenv("OPENAI_CHATGPT_API_KEY")) + vifp = VideoImageFrameProcessor() + ir = ImageRefresher() + + pipeline = Pipeline( + processors=[ + vifp, + vs, + tts, + ir, + ], + ) + + @transport.event_handler("on_first_other_participant_joined") + async def on_first_other_participant_joined(transport): + await pipeline.queue_frames([RequestVideoImageFrame(participantId=None)]) + + transport.transcription_settings["extra"]["endpointing"] = True + transport.transcription_settings["extra"]["punctuate"] = True + await transport.run(pipeline) + + +if __name__ == "__main__": + (url, token) = configure() + asyncio.run(main(url, token)) diff --git a/src/examples/starter-apps/telestrator/illustrator.py b/src/examples/starter-apps/telestrator/illustrator.py new file mode 100644 index 000000000..4e787c7b5 --- /dev/null +++ b/src/examples/starter-apps/telestrator/illustrator.py @@ -0,0 +1,112 @@ +import asyncio +import aiohttp +import logging +import os +from typing import AsyncGenerator + +from dailyai.pipeline.frames import Frame, LLMMessagesQueueFrame, RequestVideoImageFrame, LLMResponseEndFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame, TranscriptionQueueFrame, TextFrame +from dailyai.pipeline.pipeline import Pipeline +from dailyai.pipeline.frame_processor import FrameProcessor +from dailyai.services.daily_transport_service import DailyTransportService +from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService +from dailyai.services.open_ai_services import OpenAILLMService, OpenAIVisionService +from dailyai.services.fal_ai_services import FalImageGenService +from dailyai.services.deepgram_ai_services import DeepgramTTSService +from dailyai.services.ai_services import FrameLogger +from dailyai.pipeline.aggregators import ( + LLMAssistantContextAggregator, + LLMUserContextAggregator, +) +from dailyai.pipeline.frames import VideoImageFrame, VisionFrame +from examples.support.runner import configure + +logging.basicConfig(format=f"%(levelno)s %(asctime)s %(message)s") +logger = logging.getLogger("dailyai") +logger.setLevel(logging.DEBUG) + + +class VADAggregator(FrameProcessor): + def __init__(self): + self.aggregating = False + self.aggregation = "" + + async def process_frame(self, frame: Frame) -> AsyncGenerator[Frame, None]: + if isinstance(frame, UserStartedSpeakingFrame): + self.aggregating = True + elif isinstance(frame, UserStoppedSpeakingFrame): + self.aggregating = False + # Sometimes VAD triggers quickly on and off. If we don't get any transcription, + # it creates empty LLM message queue frames + if len(self.aggregation) > 0: + yield TextFrame(self.aggregation) + + self.aggregation = "" + yield frame + elif isinstance(frame, TranscriptionQueueFrame) and self.aggregating: + self.aggregation += f" {frame.text}" + else: + yield frame + + +async def main(room_url: str, token): + async with aiohttp.ClientSession() as session: + transport = DailyTransportService( + room_url, + token, + "Respond bot", + duration_minutes=5, + start_transcription=True, + mic_enabled=True, + mic_sample_rate=16000, + camera_enabled=True, + camera_width=1024, + camera_height=1024, + vad_enabled=True, + receive_video=True, + receive_video_fps=0, + vad_timeout_s=1.0 + ) + + tts = ElevenLabsTTSService( + aiohttp_session=session, + api_key=os.getenv("ELEVENLABS_API_KEY"), + voice_id=os.getenv("ELEVENLABS_VOICE_ID"), + ) + + llm = OpenAILLMService( + api_key=os.getenv("OPENAI_CHATGPT_API_KEY"), + model="gpt-4-turbo-preview") + + vs = OpenAIVisionService(api_key=os.getenv("OPENAI_CHATGPT_API_KEY")) + vad = VADAggregator() + img = FalImageGenService( + image_size="1024x1024", + aiohttp_session=session, + key_id=os.getenv("FAL_KEY_ID"), + key_secret=os.getenv("FAL_KEY_SECRET"), + ) + fl = FrameLogger("!!! Start") + fl2 = FrameLogger("!!! AFTER VAD") + fl3 = FrameLogger("!!! After img") + pipeline = Pipeline( + processors=[ + fl, + vad, + fl2, + img, + fl3 + ], + ) + + @transport.event_handler("on_first_other_participant_joined") + async def on_first_other_participant_joined(transport): + await pipeline.queue_frames([RequestVideoImageFrame(participantId=None)]) + + transport.transcription_settings["extra"]["endpointing"] = True + transport.transcription_settings["extra"]["punctuate"] = True + await transport.run(pipeline) + + +if __name__ == "__main__": + (url, token) = configure() + asyncio.run(main(url, token)) diff --git a/src/examples/starter-apps/telestrator/telestrator.py b/src/examples/starter-apps/telestrator/telestrator.py new file mode 100644 index 000000000..8c405b015 --- /dev/null +++ b/src/examples/starter-apps/telestrator/telestrator.py @@ -0,0 +1,120 @@ +import asyncio +import aiohttp +import logging +import os +from typing import AsyncGenerator + +from dailyai.pipeline.frames import Frame, LLMMessagesQueueFrame, RequestVideoImageFrame, LLMResponseEndFrame, TelestratorImageFrame, ImageFrame +from dailyai.pipeline.pipeline import Pipeline +from dailyai.pipeline.frame_processor import FrameProcessor +from dailyai.services.daily_transport_service import DailyTransportService +from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService +from dailyai.services.open_ai_services import OpenAILLMService, OpenAIVisionService +from dailyai.services.fal_ai_services import FalImageGenService +from dailyai.services.deepgram_ai_services import DeepgramTTSService +from dailyai.services.ai_services import FrameLogger +from dailyai.pipeline.aggregators import ( + LLMAssistantContextAggregator, + LLMUserContextAggregator, + LLMFullResponseAggregator +) +from dailyai.pipeline.frames import VideoImageFrame, VisionFrame +from examples.support.runner import configure + +logging.basicConfig(format=f"%(levelno)s %(asctime)s %(message)s") +logger = logging.getLogger("dailyai") +logger.setLevel(logging.DEBUG) + + +class VideoImageFrameProcessor(FrameProcessor): + def __init__(self): + pass + + async def process_frame(self, frame: Frame) -> AsyncGenerator[Frame, None]: + if isinstance(frame, VideoImageFrame) or isinstance(frame, TelestratorImageFrame): + yield VisionFrame("Describe the image in one sentence, in the style of David Attenborough.", frame.image) + else: + yield frame + + +class ImageRefresher(FrameProcessor): + async def process_frame(self, frame: Frame) -> AsyncGenerator[Frame, None]: + if isinstance(frame, LLMResponseEndFrame): + yield RequestVideoImageFrame(participantId=None) + yield frame + else: + yield frame + + +class TelestratorImageWrapper(FrameProcessor): + async def process_frame(self, frame: Frame) -> AsyncGenerator[Frame, None]: + if isinstance(frame, ImageFrame): + yield TelestratorImageFrame(None, frame.image) + else: + yield frame + + +async def main(room_url: str, token): + async with aiohttp.ClientSession() as session: + transport = DailyTransportService( + room_url, + token, + "Respond bot", + duration_minutes=5, + start_transcription=True, + mic_enabled=True, + mic_sample_rate=16000, + camera_enabled=True, + camera_width=1024, + camera_height=1024, + vad_enabled=False, + receive_video=True, + receive_video_fps=0 + ) + + tts = ElevenLabsTTSService( + aiohttp_session=session, + api_key=os.getenv("ELEVENLABS_API_KEY"), + voice_id=os.getenv("ELEVENLABS_VOICE_ID"), + ) + + llm = OpenAILLMService( + api_key=os.getenv("OPENAI_CHATGPT_API_KEY"), + model="gpt-4-turbo-preview") + + vs = OpenAIVisionService(api_key=os.getenv("OPENAI_CHATGPT_API_KEY")) + vifp = VideoImageFrameProcessor() + ir = ImageRefresher() + img = FalImageGenService( + image_size="1024x1024", + aiohttp_session=session, + key_id=os.getenv("FAL_KEY_ID"), + key_secret=os.getenv("FAL_KEY_SECRET"), + ) + tiw = TelestratorImageWrapper() + lfra = LLMFullResponseAggregator() + fl1 = FrameLogger("!!! About to image gen") + pipeline = Pipeline( + processors=[ + vifp, + vs, + tts, + lfra, + fl1, + img, + tiw, + ], + ) + + @transport.event_handler("on_first_other_participant_joined") + async def on_first_other_participant_joined(transport): + await pipeline.queue_frames([RequestVideoImageFrame(participantId=None)]) + + transport.transcription_settings["extra"]["endpointing"] = True + transport.transcription_settings["extra"]["punctuate"] = True + await transport.run(pipeline) + + +if __name__ == "__main__": + (url, token) = configure() + asyncio.run(main(url, token))