diff --git a/changelog/3692.changed.md b/changelog/3692.changed.md new file mode 100644 index 000000000..adb9e1d62 --- /dev/null +++ b/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. diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index c88180b46..de33c6187 100644 --- a/src/pipecat/frames/frames.py +++ b/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" # diff --git a/src/pipecat/pipeline/service_switcher.py b/src/pipecat/pipeline/service_switcher.py index 2c4d54085..d18f00e7c 100644 --- a/src/pipecat/pipeline/service_switcher.py +++ b/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) diff --git a/src/pipecat/services/stt_service.py b/src/pipecat/services/stt_service.py index 53840221c..7e9ed4c9f 100644 --- a/src/pipecat/services/stt_service.py +++ b/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 diff --git a/tests/test_service_switcher.py b/tests/test_service_switcher.py index 3f1f586fe..4df6696af 100644 --- a/tests/test_service_switcher.py +++ b/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)