From 25ca8b751e1d00f51f302f4122a57e8f7f4f6850 Mon Sep 17 00:00:00 2001 From: Chad Bailey Date: Tue, 19 Mar 2024 03:08:04 +0000 Subject: [PATCH] cleanup --- src/dailyai/pipeline/frames.py | 7 ++++ src/dailyai/services/ai_services.py | 1 + .../services/base_transport_service.py | 14 +++++++- .../services/daily_transport_service.py | 22 ++++++++++-- .../foundational/12-describe-video.py | 35 ++++++++++--------- 5 files changed, 58 insertions(+), 21 deletions(-) diff --git a/src/dailyai/pipeline/frames.py b/src/dailyai/pipeline/frames.py index 0b40ba6cc..b93d5bf5f 100644 --- a/src/dailyai/pipeline/frames.py +++ b/src/dailyai/pipeline/frames.py @@ -198,3 +198,10 @@ class VisionFrame(Frame): # def __str__(self): # return f"{self.__class__.__name__}, prompt: {self.prompt}, image size: {len(self.image)} B" + + +@dataclass() +class RequestVideoImageFrame(Frame): + """Send to the transport to request a new video image from a specific participant. Leave participantId + empty to request a frame from all participants.""" + participantId: str | None diff --git a/src/dailyai/services/ai_services.py b/src/dailyai/services/ai_services.py index c8cb2e314..0edbb53ae 100644 --- a/src/dailyai/services/ai_services.py +++ b/src/dailyai/services/ai_services.py @@ -148,6 +148,7 @@ class VisionService(AIService): if isinstance(frame, VisionFrame): async for frame in self.run_vision(frame.prompt, frame.image): yield frame + yield LLMResponseEndFrame() else: yield frame diff --git a/src/dailyai/services/base_transport_service.py b/src/dailyai/services/base_transport_service.py index fcba24ae7..cfdea5b94 100644 --- a/src/dailyai/services/base_transport_service.py +++ b/src/dailyai/services/base_transport_service.py @@ -24,6 +24,7 @@ from dailyai.pipeline.frames import ( TextFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame, + RequestVideoImageFrame ) from dailyai.pipeline.pipeline import Pipeline from dailyai.services.ai_services import TTSService @@ -91,7 +92,8 @@ class BaseTransportService: self._context = kwargs.get("context") or [] self._vad_enabled = kwargs.get("vad_enabled") or False self._receive_video = kwargs.get("receive_video") or False - self._receive_video_fps = kwargs.get("receive_video_fps") or 1.0 + self._receive_video_fps = kwargs.get("receive_video_fps") or 0.0 + self._participant_frame_times = {} if self._vad_enabled and self._speaker_enabled: raise Exception( "Sorry, you can't use speaker_enabled and vad_enabled at the same time. Please set one to False." @@ -442,6 +444,7 @@ class BaseTransportService: # discard them if not self._is_interrupted.is_set(): if frame: + if isinstance(frame, AudioFrame): chunk = frame.data @@ -460,6 +463,15 @@ class BaseTransportService: elif isinstance(frame, SendAppMessageFrame): self.send_app_message( frame.message, frame.participantId) + elif isinstance(frame, RequestVideoImageFrame): + # removing one or all participant IDs from _participant_frame_times + # will cause the transport to send the next available frame from + # that participant + if frame.participantId: + self._participant_frame_times.pop( + frame.participantId, None) + else: + self._participant_frame_times.clear() elif len(b): self.write_frame_to_mic(bytes(b)) b = bytearray() diff --git a/src/dailyai/services/daily_transport_service.py b/src/dailyai/services/daily_transport_service.py index b6c66fd7e..660b8de76 100644 --- a/src/dailyai/services/daily_transport_service.py +++ b/src/dailyai/services/daily_transport_service.py @@ -64,7 +64,6 @@ class DailyTransportService(BaseTransportService, EventHandler): self._other_participant_has_joined = False self._my_participant_id = None - self._participant_frame_times = {} self.transcription_settings = { "language": "en", @@ -230,9 +229,26 @@ class DailyTransportService(BaseTransportService, EventHandler): self.client.release() def _handle_video_frame(self, participant_id, video_frame): - if (not participant_id in self._participant_frame_times) or (time.time() > self._participant_frame_times[participant_id] + 1.0/self._receive_video_fps): - self._participant_frame_times[participant_id] = time.time() + """If receive_video is true, this function is called once for each frame from each participant. We + don't need to send every frame to the pipeline, so there are two ways to decide how to send frames: + 1. Set a greater-than-zero value for receive_video_fps. The transport will track the last send time + for each participant and send a new frame when the requested frame rate has elapsed. This + guarantees an image every second, for example. + 2. Set receive_video_fps less than or equal to zero to disable timed frame sending. Then, put a + RequestVideoImageFrame in the pipeline to get a new frame for one or all participants. By + sending a RequestVideoImageFrame immediately after successfully processing an image, you can + ensure you don't end up queueing up frames faster than you can process them. + """ + send_frame = False + if not participant_id in self._participant_frame_times: + # then it's a new participant; send the first frame + send_frame = True + elif self._receive_video_fps > 0 and time.time() > self._participant_frame_times[participant_id] + 1.0/self._receive_video_fps: + # Then it's an existing participant who is due to send a new frame + send_frame = True + if send_frame: + self._participant_frame_times[participant_id] = time.time() future = asyncio.run_coroutine_threadsafe( self.receive_queue.put( VideoImageFrame(participant_id, video_frame)), self._loop) diff --git a/src/examples/foundational/12-describe-video.py b/src/examples/foundational/12-describe-video.py index da0cc7f95..898d35a1d 100644 --- a/src/examples/foundational/12-describe-video.py +++ b/src/examples/foundational/12-describe-video.py @@ -4,7 +4,7 @@ import logging import os from typing import AsyncGenerator -from dailyai.pipeline.frames import Frame, LLMMessagesQueueFrame +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 @@ -30,7 +30,16 @@ class VideoImageFrameProcessor(FrameProcessor): async def process_frame(self, frame: Frame) -> AsyncGenerator[Frame, None]: if isinstance(frame, VideoImageFrame): - yield VisionFrame("What is in this image?", frame.image) + 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 @@ -48,7 +57,7 @@ async def main(room_url: str, token): camera_enabled=False, vad_enabled=True, receive_video=True, - receive_video_fps=1/10.0 + receive_video_fps=0 ) tts = ElevenLabsTTSService( @@ -61,31 +70,23 @@ async def main(room_url: str, token): api_key=os.getenv("OPENAI_CHATGPT_API_KEY"), model="gpt-4-turbo-preview") - messages = [ - { - "role": "system", - "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio. Respond to what the user said in a creative and helpful way.", - }, - ] - - tma_in = LLMUserContextAggregator( - messages, transport._my_participant_id) - tma_out = LLMAssistantContextAggregator( - messages, transport._my_participant_id - ) vs = OpenAIVisionService(api_key=os.getenv("OPENAI_CHATGPT_API_KEY")) - vifp = VideoImageFrameProcessor() + ir = ImageRefresher() pipeline = Pipeline( processors=[ vifp, vs, llm, tts, - tma_out, + 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)