First cut at sample 6 rewrite with pipelines

src/dailyai/pipeline/aggregators.py

@@ -170,8 +170,18 @@ class ParallelPipeline(FrameProcessor):
 
         await asyncio.gather(*[pipeline.run_pipeline() for pipeline in self.pipelines])
 
+        seen_ids = set()
         while not self.sink.empty():
             frame = await self.sink.get()
+
+            # de-dup frames. Because the convention is to yield a frame that isn't processed,
+            # each pipeline will likely yield the same frame, so we will end up with _n_ copies
+            # of unprocessed frames where _n_ is the number of parallel pipes that don't
+            # process that frame.
+            if id(frame) in seen_ids:
+                continue
+            seen_ids.add(id(frame))
+
             # Skip passing along EndParallelPipeQueueFrame, because we use them for our own flow control.
             if not isinstance(frame, EndParallelPipeQueueFrame):
                 yield frame

src/dailyai/pipeline/pipeline.py

@@ -39,12 +39,12 @@ class Pipeline:
                 for frame_generator in frame_generators:
                     async for frame in frame_generator:
                         await self.sink.put(frame)
-                    if isinstance(
+                        if isinstance(
-                        frame, EndStreamQueueFrame
+                            frame, EndStreamQueueFrame
-                    ) or isinstance(
+                        ) or isinstance(
-                        frame, EndParallelPipeQueueFrame
+                            frame, EndParallelPipeQueueFrame
-                    ):
+                        ):
-                        return
+                            return
         except asyncio.CancelledError:
             # this means there's been an interruption, do any cleanup necessary here.
             for processor in self.processors:

src/examples/foundational/06-listen-and-respond.py

@@ -1,5 +1,6 @@
 import asyncio
 import os
+from dailyai.pipeline.pipeline import Pipeline
 
 from dailyai.services.daily_transport_service import DailyTransportService
 from dailyai.services.azure_ai_services import AzureLLMService, AzureTTSService
@@ -44,21 +45,19 @@ async def main(room_url: str, token):
 
         tma_in = LLMUserContextAggregator(messages, transport._my_participant_id)
         tma_out = LLMAssistantContextAggregator(messages, transport._my_participant_id)
-        await tts.run_to_queue(
+        pipeline = Pipeline(
-            transport.send_queue,
+            source=transport.receive_queue,
-            tma_out.run(
+            sink=transport.send_queue,
-                fl2.run(
+            processors=[
-                    llm.run(
+                fl,
-                    tma_in.run(
+                tma_in,
-                        fl.run(
+                llm,
-                            transport.get_receive_frames()
+                fl2,
-                            )
+                tma_out,
-                        )
+                tts
-                    )
+            ],
-                )
+        )
-
+        await pipeline.run_pipeline()
-                )
-            )
 
     transport.transcription_settings["extra"]["endpointing"] = True
     transport.transcription_settings["extra"]["punctuate"] = True