Merge pull request #2687 from pipecat-ai/memory_leak

Improving memory cleanup

CHANGELOG.md

@@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Added memory cleanup improvements to reduce memory peaks.
+
 - Added `on_before_process_frame`, `on_after_process_frame`,
   `on_before_push_frame` and `on_after_push_frame`. These are synchronous events
   that get called before and after a frame is processed or pushed. Note that

examples/foundational/46-video-processing.py

@@ -0,0 +1,175 @@
+#
+# Copyright (c) 2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+import os
+
+import cv2
+import numpy as np
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import Frame, InputImageRawFrame, LLMRunFrame, OutputImageRawFrame
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.processors.frameworks.rtvi import RTVIObserver, RTVIProcessor
+from pipecat.runner.types import RunnerArguments
+from pipecat.runner.utils import create_transport
+from pipecat.services.gemini_multimodal_live import GeminiMultimodalLiveLLMService
+from pipecat.transports.base_transport import TransportParams
+from pipecat.transports.daily.transport import DailyParams, DailyTransport
+
+load_dotenv(override=True)
+
+transport_params = {
+    "daily": lambda: DailyParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        audio_out_10ms_chunks=2,
+        video_in_enabled=True,
+        video_out_enabled=True,
+        video_out_is_live=True,
+        vad_analyzer=SileroVADAnalyzer(),
+    ),
+    "webrtc": lambda: TransportParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        audio_out_10ms_chunks=2,
+        video_in_enabled=True,
+        video_out_enabled=True,
+        video_out_is_live=True,
+        vad_analyzer=SileroVADAnalyzer(),
+    ),
+}
+
+
+class EdgeDetectionProcessor(FrameProcessor):
+    def __init__(self, video_out_width, video_out_height: int):
+        super().__init__()
+        self._video_out_width = video_out_width
+        self._video_out_height = video_out_height
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        # Send back the user's camera video with edge detection applied
+        if isinstance(frame, InputImageRawFrame) and frame.transport_source == "camera":
+            # Convert bytes to NumPy array
+            img = np.frombuffer(frame.image, dtype=np.uint8).reshape(
+                (frame.size[1], frame.size[0], 3)
+            )
+
+            # perform edge detection only on camera frames
+            img = cv2.cvtColor(cv2.Canny(img, 100, 200), cv2.COLOR_GRAY2BGR)
+
+            # convert the size if needed
+            desired_size = (self._video_out_width, self._video_out_height)
+            if frame.size != desired_size:
+                resized_image = cv2.resize(img, desired_size)
+                out_frame = OutputImageRawFrame(resized_image.tobytes(), desired_size, frame.format)
+                await self.push_frame(out_frame)
+            else:
+                out_frame = OutputImageRawFrame(
+                    image=img.tobytes(), size=frame.size, format=frame.format
+                )
+                await self.push_frame(out_frame)
+        else:
+            await self.push_frame(frame, direction)
+
+
+SYSTEM_INSTRUCTION = f"""
+"You are Gemini Chatbot, a friendly, helpful robot.
+
+Your goal is to demonstrate your capabilities in a succinct way.
+
+Your output will be converted to audio so don't include special characters in your answers.
+
+Respond to what the user said in a creative and helpful way. Keep your responses brief. One or two sentences at most.
+"""
+
+
+async def run_bot(pipecat_transport):
+    llm = GeminiMultimodalLiveLLMService(
+        api_key=os.getenv("GOOGLE_API_KEY"),
+        voice_id="Puck",  # Aoede, Charon, Fenrir, Kore, Puck
+        transcribe_user_audio=True,
+        system_instruction=SYSTEM_INSTRUCTION,
+    )
+
+    messages = [
+        {
+            "role": "user",
+            "content": "Start by greeting the user warmly and introducing yourself.",
+        }
+    ]
+
+    context = OpenAILLMContext(messages)
+    context_aggregator = llm.create_context_aggregator(context)
+
+    # RTVI events for Pipecat client UI
+    rtvi = RTVIProcessor()
+
+    pipeline = Pipeline(
+        [
+            pipecat_transport.input(),
+            context_aggregator.user(),
+            rtvi,
+            llm,  # LLM
+            EdgeDetectionProcessor(
+                pipecat_transport._params.video_out_width,
+                pipecat_transport._params.video_out_height,
+            ),  # Sending the video back to the user
+            pipecat_transport.output(),
+            context_aggregator.assistant(),
+        ]
+    )
+
+    task = PipelineTask(
+        pipeline,
+        params=PipelineParams(
+            enable_metrics=True,
+            enable_usage_metrics=True,
+        ),
+        observers=[RTVIObserver(rtvi)],
+    )
+
+    @rtvi.event_handler("on_client_ready")
+    async def on_client_ready(rtvi):
+        logger.info("Pipecat client ready.")
+        await rtvi.set_bot_ready()
+        # Kick off the conversation.
+        await task.queue_frames([LLMRunFrame()])
+
+    @pipecat_transport.event_handler("on_client_connected")
+    async def on_client_connected(transport, participant):
+        logger.info("Pipecat Client connected")
+        if isinstance(transport, DailyTransport):
+            await pipecat_transport.capture_participant_video(participant["id"], framerate=30)
+        else:
+            await pipecat_transport.capture_participant_video("camera")
+
+    @pipecat_transport.event_handler("on_client_disconnected")
+    async def on_client_disconnected(transport, client):
+        logger.info("Pipecat Client disconnected")
+        await task.cancel()
+
+    runner = PipelineRunner(handle_sigint=False, force_gc=True)
+
+    await runner.run(task)
+
+
+async def bot(runner_args: RunnerArguments):
+    """Main bot entry point compatible with Pipecat Cloud."""
+    transport = await create_transport(runner_args, transport_params)
+    await run_bot(transport)
+
+
+if __name__ == "__main__":
+    from pipecat.runner.run import main
+
+    main()

src/pipecat/pipeline/task.py

@@ -777,7 +777,6 @@ class PipelineTask(BasePipelineTask):
         """
         running = True
         last_frame_time = 0
-        frame_buffer = deque(maxlen=10)  # Store last 10 frames
 
         while running:
             try:
@@ -785,9 +784,6 @@ class PipelineTask(BasePipelineTask):
                     self._idle_queue.get(), timeout=self._idle_timeout_secs
                 )
 
-                if not isinstance(frame, InputAudioRawFrame):
-                    frame_buffer.append(frame)
-
                 if isinstance(frame, StartFrame) or isinstance(frame, self._idle_timeout_frames):
                     # If we find a StartFrame or one of the frames that prevents a
                     # time out we update the time.
@@ -798,7 +794,7 @@ class PipelineTask(BasePipelineTask):
                     # valid frames.
                     diff_time = time.time() - last_frame_time
                     if diff_time >= self._idle_timeout_secs:
-                        running = await self._idle_timeout_detected(frame_buffer)
+                        running = await self._idle_timeout_detected()
                         # Reset `last_frame_time` so we don't trigger another
                         # immediate idle timeout if we are not cancelling. For
                         # example, we might want to force the bot to say goodbye
@@ -808,14 +804,11 @@ class PipelineTask(BasePipelineTask):
                 self._idle_queue.task_done()
 
             except asyncio.TimeoutError:
-                running = await self._idle_timeout_detected(frame_buffer)
+                running = await self._idle_timeout_detected()
 
-    async def _idle_timeout_detected(self, last_frames: Deque[Frame]) -> bool:
+    async def _idle_timeout_detected(self) -> bool:
         """Handle idle timeout detection and optional cancellation.
 
-        Args:
-            last_frames: Recent frames received before timeout for debugging.
-
         Returns:
             Whether the pipeline task should continue running.
         """
@@ -823,10 +816,7 @@ class PipelineTask(BasePipelineTask):
         if self._cancelled:
             return True
 
-        logger.warning("Idle timeout detected. Last 10 frames received:")
+        logger.warning("Idle timeout detected.")
-        for i, frame in enumerate(last_frames, 1):
-            logger.warning(f"Frame {i}: {frame}")
-
         await self._call_event_handler("on_idle_timeout")
         if self._cancel_on_idle_timeout:
             logger.warning(f"Idle pipeline detected, cancelling pipeline task...")

src/pipecat/services/openai/base_llm.py

@@ -281,8 +281,10 @@ class BaseOpenAILLMService(LLMService):
         # base64 encode any images
         for message in messages:
             if message.get("mime_type") == "image/jpeg":
-                encoded_image = base64.b64encode(message["data"].getvalue()).decode("utf-8")
+                # Avoid .getvalue() which makes a full copy of BytesIO
-                text = message["content"]
+                raw_bytes = message["data"].read()
+                encoded_image = base64.b64encode(raw_bytes).decode("utf-8")
+                text = message.get("content", "")
                 message["content"] = [
                     {"type": "text", "text": text},
                     {
@@ -290,6 +292,7 @@ class BaseOpenAILLMService(LLMService):
                         "image_url": {"url": f"data:image/jpeg;base64,{encoded_image}"},
                     },
                 ]
+                # Explicit cleanup
                 del message["data"]
                 del message["mime_type"]
 

src/pipecat/transports/base_output.py

@@ -659,6 +659,7 @@ class BaseOutputTransport(FrameProcessor):
                             self._audio_queue.get(), timeout=vad_stop_secs
                         )
                         yield frame
+                        self._audio_queue.task_done()
                     except asyncio.TimeoutError:
                         # Notify the bot stopped speaking upstream if necessary.
                         await self._bot_stopped_speaking()
@@ -673,6 +674,7 @@ class BaseOutputTransport(FrameProcessor):
                             frame.audio = await self._mixer.mix(frame.audio)
                             last_frame_time = time.time()
                         yield frame
+                        self._audio_queue.task_done()
                     except asyncio.QueueEmpty:
                         # Notify the bot stopped speaking upstream if necessary.
                         diff_time = time.time() - last_frame_time

src/pipecat/transports/smallwebrtc/transport.py

@@ -309,7 +309,7 @@ class SmallWebRTCClient:
                     # self._webrtc_connection.ask_to_renegotiate()
                 frame = None
             except MediaStreamError:
-                logger.warning("Received an unexpected media stream error while reading the audio.")
+                logger.warning("Received an unexpected media stream error while reading the video.")
                 frame = None
 
             if frame is None or not isinstance(frame, VideoFrame):
@@ -321,15 +321,21 @@ class SmallWebRTCClient:
             # Convert frame to NumPy array in its native format
             frame_array = frame.to_ndarray(format=format_name)
             frame_rgb = self._convert_frame(frame_array, format_name)
+            del frame_array  # free intermediate array immediately
+            image_bytes = frame_rgb.tobytes()
+            del frame_rgb  # free RGB array immediately
 
             image_frame = UserImageRawFrame(
                 user_id=self._webrtc_connection.pc_id,
-                image=frame_rgb.tobytes(),
+                image=image_bytes,
                 size=(frame.width, frame.height),
                 format="RGB",
             )
             image_frame.transport_source = video_source
 
+            del frame  # free original VideoFrame
+            del image_bytes  # reference kept in image_frame
+
             yield image_frame
 
     async def read_audio_frame(self):
@@ -364,23 +370,35 @@ class SmallWebRTCClient:
                 resampled_frames = self._pipecat_resampler.resample(frame)
                 for resampled_frame in resampled_frames:
                     # 16-bit PCM bytes
-                    pcm_bytes = resampled_frame.to_ndarray().astype(np.int16).tobytes()
+                    pcm_array = resampled_frame.to_ndarray().astype(np.int16)
+                    pcm_bytes = pcm_array.tobytes()
+                    del pcm_array  # free NumPy array immediately
+
                     audio_frame = InputAudioRawFrame(
                         audio=pcm_bytes,
                         sample_rate=resampled_frame.sample_rate,
                         num_channels=self._audio_in_channels,
                     )
+                    del pcm_bytes  # reference kept in audio_frame
+
                     yield audio_frame
             else:
                 # 16-bit PCM bytes
-                pcm_bytes = frame.to_ndarray().astype(np.int16).tobytes()
+                pcm_array = frame.to_ndarray().astype(np.int16)
+                pcm_bytes = pcm_array.tobytes()
+                del pcm_array  # free NumPy array immediately
+
                 audio_frame = InputAudioRawFrame(
                     audio=pcm_bytes,
                     sample_rate=frame.sample_rate,
                     num_channels=self._audio_in_channels,
                 )
+                del pcm_bytes  # reference kept in audio_frame
+
                 yield audio_frame
 
+            del frame  # free original AudioFrame
+
     async def write_audio_frame(self, frame: OutputAudioRawFrame):
         """Write an audio frame to the WebRTC connection.