Merge pull request #3692 from pipecat-ai/mb/request-metadata-updates

Rename RequestMetadataFrame to ServiceSwitcherRequestMetadataFrame with service targeting

changelog/3692.changed.md

@@ -0,0 +1 @@
+- Renamed `RequestMetadataFrame` to `ServiceSwitcherRequestMetadataFrame` and added a `service` field to target a specific service. The frame is now pushed downstream by services after handling instead of being silently consumed.

src/pipecat/frames/frames.py

@@ -1721,16 +1721,19 @@ class STTMetadataFrame(ServiceMetadataFrame):
 
 
 @dataclass
-class RequestMetadataFrame(ControlFrame):
-    """Request services to re-emit their metadata frames.
+class ServiceSwitcherRequestMetadataFrame(ControlFrame):
+    """Request a service to re-emit its metadata frames.
 
     Used by ServiceSwitcher when switching active services to ensure
     downstream processors receive updated metadata from the newly active service.
     Services that receive this frame should re-push their metadata frame
     (e.g., STTMetadataFrame for STT services).
+
+    Parameters:
+        service: The target service that should re-emit its metadata.
     """
 
-    pass
+    service: "FrameProcessor"
 
 
 #

src/pipecat/pipeline/service_switcher.py

@@ -12,9 +12,9 @@ from typing import Any, Generic, List, Optional, Type, TypeVar
 from pipecat.frames.frames import (
     Frame,
     ManuallySwitchServiceFrame,
-    RequestMetadataFrame,
     ServiceMetadataFrame,
     ServiceSwitcherFrame,
+    ServiceSwitcherRequestMetadataFrame,
 )
 from pipecat.pipeline.parallel_pipeline import ParallelPipeline
 from pipecat.processors.filters.function_filter import FunctionFilter
@@ -220,16 +220,19 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]):
     async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
         """Push a frame out of the service switcher.
 
-        Suppresses `RequestMetadataFrame` (internal to the switcher) and
-        `ServiceMetadataFrame` from inactive services so only the active
-        service's metadata reaches downstream processors. One case this happens
-        is with `StartFrame` since all the filters let it pass, and `StartFrame`
-        causes the service to generate `ServiceMetadataFrame`.
+        Suppresses `ServiceSwitcherRequestMetadataFrame` targeting the active
+        service (since it has already been handled) and `ServiceMetadataFrame`
+        from inactive services so only the active service's metadata reaches
+        downstream processors. One case this happens is with `StartFrame` since
+        all the filters let it pass, and `StartFrame` causes the service to
+        generate `ServiceMetadataFrame`.
 
         """
-        # Don't let RequestMetadataFrame out.
-        if isinstance(frame, RequestMetadataFrame):
-            return
+        # Consume ServiceSwitcherRequestMetadataFrame once the targeted service
+        # has handled it (i.e. the active service).
+        if isinstance(frame, ServiceSwitcherRequestMetadataFrame):
+            if frame.service == self.strategy.active_service:
+                return
 
         # Only let metadata from the active service escape.
         if isinstance(frame, ServiceMetadataFrame):
@@ -255,6 +258,6 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]):
 
             # If we switched to a new service, request its metadata.
             if service:
-                await service.queue_frame(RequestMetadataFrame())
+                await service.queue_frame(ServiceSwitcherRequestMetadataFrame(service=service))
         else:
             await super().process_frame(frame, direction)

src/pipecat/services/stt_service.py

@@ -21,7 +21,7 @@ from pipecat.frames.frames import (
     Frame,
     InterruptionFrame,
     MetricsFrame,
-    RequestMetadataFrame,
+    ServiceSwitcherRequestMetadataFrame,
     StartFrame,
     STTMetadataFrame,
     STTMuteFrame,
@@ -264,9 +264,9 @@ class STTService(AIService):
             # Push StartFrame first, then metadata so downstream receives them in order
             await self.push_frame(frame, direction)
             await self._push_stt_metadata()
-        elif isinstance(frame, RequestMetadataFrame):
-            # Don't push the RequestMetadataFrame, just push the metadata
+        elif isinstance(frame, ServiceSwitcherRequestMetadataFrame):
             await self._push_stt_metadata()
+            await self.push_frame(frame, direction)
         elif isinstance(frame, AudioRawFrame):
             # In this service we accumulate audio internally and at the end we
             # push a TextFrame. We also push audio downstream in case someone

tests/test_service_switcher.py

@@ -13,8 +13,8 @@ from dataclasses import dataclass
 from pipecat.frames.frames import (
     Frame,
     ManuallySwitchServiceFrame,
-    RequestMetadataFrame,
     ServiceMetadataFrame,
+    ServiceSwitcherRequestMetadataFrame,
     StartFrame,
     SystemFrame,
     TextFrame,
@@ -68,7 +68,7 @@ class MockMetadataFrame(ServiceMetadataFrame):
 class MockMetadataService(FrameProcessor):
     """A mock service that emits ServiceMetadataFrame like STT services.
 
-    Pushes MockMetadataFrame on StartFrame and RequestMetadataFrame.
+    Pushes MockMetadataFrame on StartFrame and ServiceSwitcherRequestMetadataFrame.
     """
 
     def __init__(self, test_name: str, **kwargs):
@@ -84,9 +84,9 @@ class MockMetadataService(FrameProcessor):
         if isinstance(frame, StartFrame):
             await self.push_frame(frame, direction)
             await self._push_metadata()
-        elif isinstance(frame, RequestMetadataFrame):
-            # Don't push RequestMetadataFrame downstream (it's internal)
+        elif isinstance(frame, ServiceSwitcherRequestMetadataFrame):
             await self._push_metadata()
+            await self.push_frame(frame, direction)
         else:
             await self.push_frame(frame, direction)
 
@@ -472,9 +472,11 @@ class TestServiceSwitcherMetadata(unittest.IsolatedAsyncioTestCase):
             expected_up_frames=[],
         )
 
-        # service2 should have received RequestMetadataFrame after becoming active
+        # service2 should have received ServiceSwitcherRequestMetadataFrame after becoming active
         request_frames = [
-            f for f in self.service2.processed_frames if isinstance(f, RequestMetadataFrame)
+            f
+            for f in self.service2.processed_frames
+            if isinstance(f, ServiceSwitcherRequestMetadataFrame)
         ]
         self.assertEqual(len(request_frames), 1)