examples: use Cartesia TTS in most examples

src/pipecat/pipeline/parallel_task.py

@@ -57,6 +57,7 @@ class ParallelTask(BasePipeline):
             raise Exception(f"ParallelTask needs at least one argument")
 
         self._sinks = []
+        self._sources = []
         self._pipelines = []
 
         self._up_queue = asyncio.Queue()
@@ -70,10 +71,10 @@ class ParallelTask(BasePipeline):
             # We add a source at the beginning of the pipeline and a sink at the end.
             source = Source(self._up_queue)
             sink = Sink(self._down_queue)
-            processors: List[FrameProcessor] = [source] + processors
-            processors.append(sink)
+            processors: List[FrameProcessor] = [source] + processors + [sink]
 
-            # Keep track of sinks. We access the source through the pipeline.
+            # Keep track of sources and sinks.
+            self._sources.append(source)
             self._sinks.append(sink)
 
             # Create pipeline
@@ -99,8 +100,8 @@ class ParallelTask(BasePipeline):
             # If we get an upstream frame we process it in each sink.
             await asyncio.gather(*[s.process_frame(frame, direction) for s in self._sinks])
         elif direction == FrameDirection.DOWNSTREAM:
-            # If we get a downstream frame we process it in each source (using the pipeline).
-            await asyncio.gather(*[p.process_frame(frame, direction) for p in self._pipelines])
+            # If we get a downstream frame we process it in each source.
+            await asyncio.gather(*[s.process_frame(frame, direction) for s in self._sources])
 
         seen_ids = set()
         while not self._up_queue.empty():

src/pipecat/processors/aggregators/gated.py

@@ -4,7 +4,7 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-from typing import List
+from typing import List, Tuple
 
 from pipecat.frames.frames import Frame, SystemFrame
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
@@ -40,12 +40,14 @@ class GatedAggregator(FrameProcessor):
     Goodbye.
     """
 
-    def __init__(self, gate_open_fn, gate_close_fn, start_open):
+    def __init__(self, gate_open_fn, gate_close_fn, start_open,
+                 direction: FrameDirection = FrameDirection.DOWNSTREAM):
         super().__init__()
         self._gate_open_fn = gate_open_fn
         self._gate_close_fn = gate_close_fn
         self._gate_open = start_open
-        self._accumulator: List[Frame] = []
+        self._direction = direction
+        self._accumulator: List[Tuple[Frame, FrameDirection]] = []
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
@@ -55,6 +57,11 @@ class GatedAggregator(FrameProcessor):
             await self.push_frame(frame, direction)
             return
 
+        # Ignore frames that are not following the direction of this gate.
+        if direction != self._direction:
+            await self.push_frame(frame, direction)
+            return
+
         old_state = self._gate_open
         if self._gate_open:
             self._gate_open = not self._gate_close_fn(frame)
@@ -67,8 +74,8 @@ class GatedAggregator(FrameProcessor):
 
         if self._gate_open:
             await self.push_frame(frame, direction)
-            for frame in self._accumulator:
-                await self.push_frame(frame, direction)
+            for (f, d) in self._accumulator:
+                await self.push_frame(f, d)
             self._accumulator = []
         else:
-            self._accumulator.append(frame)
+            self._accumulator.append((frame, direction))