cleanup

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

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
 

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()

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):
+        """If receive_video is true, this function is called once for each frame from each participant. We
-            self._participant_frame_times[participant_id] = time.time()
+         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)

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)