Merge pull request #2404 from pipecat-ai/aleix/examples-22-simplify-main-pipeline

examples(foundational): update 22 series with simple main pipelines

CHANGELOG.md

@@ -22,9 +22,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Improving the latency of the `HeyGenVideoService`.
 
-- Updated `15-switch-voices.py` and `15a-switch-languages.py` examples to show
+- Updated foundational examples to show how to enclose complex logic
-  how to enclose complex logic (e.g. `ParallelPipeline`) into a single processor
+  (e.g. `ParallelPipeline`) into a single processor so the main pipeline becomes
-  so the main pipeline becomes simpler.
+  simpler.
 
 ## [0.0.79] - 2025-08-07
 

examples/foundational/22-natural-conversation.py

@@ -25,7 +25,8 @@ from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
 from pipecat.services.cartesia.tts import CartesiaTTSService
 from pipecat.services.deepgram.stt import DeepgramSTTService
-from pipecat.services.openai.llm import OpenAILLMService
+from pipecat.services.llm_service import LLMService
+from pipecat.services.openai.llm import OpenAIContextAggregatorPair, OpenAILLMService
 from pipecat.sync.event_notifier import EventNotifier
 from pipecat.transports.base_transport import BaseTransport, TransportParams
 from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams
@@ -34,6 +35,76 @@ from pipecat.transports.services.daily import DailyParams
 load_dotenv(override=True)
 
 
+class TurnDetectionLLM(Pipeline):
+    def __init__(self, llm: LLMService, context_aggregator: OpenAIContextAggregatorPair):
+        # This is the LLM that will be used to detect if the user has finished a
+        # statement. This doesn't really need to be an LLM, we could use NLP
+        # libraries for that, but it was easier as an example because we
+        # leverage the context aggregators.
+        statement_llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
+
+        statement_messages = [
+            {
+                "role": "system",
+                "content": "Determine if the user's statement is a complete sentence or question, ending in a natural pause or punctuation. Return 'YES' if it is complete and 'NO' if it seems to leave a thought unfinished.",
+            },
+        ]
+
+        statement_context = OpenAILLMContext(statement_messages)
+        statement_context_aggregator = statement_llm.create_context_aggregator(statement_context)
+
+        # We have instructed the LLM to return 'YES' if it thinks the user
+        # completed a sentence. So, if it's 'YES' we will return true in this
+        # predicate which will wake up the notifier.
+        async def wake_check_filter(frame):
+            logger.debug(f"Completeness check frame: {frame}")
+            return frame.text == "YES"
+
+        # This is a notifier that we use to synchronize the two LLMs.
+        notifier = EventNotifier()
+
+        # This a filter that will wake up the notifier if the given predicate
+        # (wake_check_filter) returns true.
+        completness_check = WakeNotifierFilter(
+            notifier, types=(TextFrame,), filter=wake_check_filter
+        )
+
+        # This processor keeps the last context and will let it through once the
+        # notifier is woken up. We start with the gate open because we send an
+        # initial context frame to start the conversation.
+        gated_context_aggregator = GatedOpenAILLMContextAggregator(
+            notifier=notifier, start_open=True
+        )
+
+        # Notify if the user hasn't said anything.
+        async def user_idle_notifier(frame):
+            await notifier.notify()
+
+        # Sometimes the LLM will fail detecting if a user has completed a
+        # sentence, this will wake up the notifier if that happens.
+        user_idle = UserIdleProcessor(callback=user_idle_notifier, timeout=3.0)
+
+        # The ParallePipeline input are the user transcripts. We have two
+        # contexts. The first one will be used to determine if the user finished
+        # a statement and if so the notifier will be woken up. The second
+        # context is simply the regular context but it's gated waiting for the
+        # notifier to be woken up.
+        super().__init__(
+            [
+                ParallelPipeline(
+                    [
+                        statement_context_aggregator.user(),
+                        statement_llm,
+                        completness_check,
+                        NullFilter(),
+                    ],
+                    [context_aggregator.user(), gated_context_aggregator, llm],
+                ),
+                user_idle,
+            ]
+        )
+
+
 # We store functions so objects (e.g. SileroVADAnalyzer) don't get
 # instantiated. The function will be called when the desired transport gets
 # selected.
@@ -66,24 +137,8 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         voice_id="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
     )
 
-    # This is the LLM that will be used to detect if the user has finished a
-    # statement. This doesn't really need to be an LLM, we could use NLP
-    # libraries for that, but it was easier as an example because we
-    # leverage the context aggregators.
-    statement_llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
-
-    statement_messages = [
-        {
-            "role": "system",
-            "content": "Determine if the user's statement is a complete sentence or question, ending in a natural pause or punctuation. Return 'YES' if it is complete and 'NO' if it seems to leave a thought unfinished.",
-        },
-    ]
-
-    statement_context = OpenAILLMContext(statement_messages)
-    statement_context_aggregator = statement_llm.create_context_aggregator(statement_context)
-
     # This is the regular LLM.
-    llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
+    llm_main = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
 
     messages = [
         {
@@ -93,53 +148,16 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     ]
 
     context = OpenAILLMContext(messages)
-    context_aggregator = llm.create_context_aggregator(context)
+    context_aggregator = llm_main.create_context_aggregator(context)
 
-    # We have instructed the LLM to return 'YES' if it thinks the user
+    # LLM + turn detection (with an extra LLM as a judge)
-    # completed a sentence. So, if it's 'YES' we will return true in this
+    llm = TurnDetectionLLM(llm_main, context_aggregator)
-    # predicate which will wake up the notifier.
-    async def wake_check_filter(frame):
-        return frame.text == "YES"
 
-    # This is a notifier that we use to synchronize the two LLMs.
-    notifier = EventNotifier()
-
-    # This a filter that will wake up the notifier if the given predicate
-    # (wake_check_filter) returns true.
-    completness_check = WakeNotifierFilter(notifier, types=(TextFrame,), filter=wake_check_filter)
-
-    # This processor keeps the last context and will let it through once the
-    # notifier is woken up. We start with the gate open because we send an
-    # initial context frame to start the conversation.
-    gated_context_aggregator = GatedOpenAILLMContextAggregator(notifier=notifier, start_open=True)
-
-    # Notify if the user hasn't said anything.
-    async def user_idle_notifier(frame):
-        await notifier.notify()
-
-    # Sometimes the LLM will fail detecting if a user has completed a
-    # sentence, this will wake up the notifier if that happens.
-    user_idle = UserIdleProcessor(callback=user_idle_notifier, timeout=3.0)
-
-    # The ParallePipeline input are the user transcripts. We have two
-    # contexts. The first one will be used to determine if the user finished
-    # a statement and if so the notifier will be woken up. The second
-    # context is simply the regular context but it's gated waiting for the
-    # notifier to be woken up.
     pipeline = Pipeline(
         [
             transport.input(),  # Transport user input
-            stt,
+            stt,  # STT
-            ParallelPipeline(
+            llm,  # LLM with turn detection
-                [
-                    statement_context_aggregator.user(),
-                    statement_llm,
-                    completness_check,
-                    NullFilter(),
-                ],
-                [context_aggregator.user(), gated_context_aggregator, llm],
-            ),
-            user_idle,
             tts,  # TTS
             transport.output(),  # Transport bot output
             context_aggregator.assistant(),  # Assistant spoken responses

examples/foundational/22b-natural-conversation-proposal.py

@@ -6,7 +6,6 @@
 
 import asyncio
 import os
-import time
 
 from dotenv import load_dotenv
 from loguru import logger
@@ -44,13 +43,14 @@ from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
 from pipecat.services.cartesia.tts import CartesiaTTSService
 from pipecat.services.deepgram.stt import DeepgramSTTService
-from pipecat.services.llm_service import FunctionCallParams
+from pipecat.services.llm_service import FunctionCallParams, LLMService
 from pipecat.services.openai.llm import OpenAILLMService
 from pipecat.sync.base_notifier import BaseNotifier
 from pipecat.sync.event_notifier import EventNotifier
 from pipecat.transports.base_transport import BaseTransport, TransportParams
 from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams
 from pipecat.transports.services.daily import DailyParams
+from pipecat.utils.time import time_now_iso8601
 
 load_dotenv(override=True)
 
@@ -192,6 +192,75 @@ async def fetch_weather_from_api(params: FunctionCallParams):
     await params.result_callback({"conditions": "nice", "temperature": "75"})
 
 
+class TurnDetectionLLM(Pipeline):
+    def __init__(self, llm: LLMService):
+        # This is the LLM that will be used to detect if the user has finished a
+        # statement. This doesn't really need to be an LLM, we could use NLP
+        # libraries for that, but we have the machinery to use an LLM, so we
+        # might as well!
+        statement_llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
+
+        # We have instructed the LLM to return 'YES' if it thinks the user
+        # completed a sentence. So, if it's 'YES' we will return true in this
+        # predicate which will wake up the notifier.
+        async def wake_check_filter(frame):
+            logger.debug(f"Completeness check frame: {frame}")
+            return frame.text == "YES"
+
+        # This is a notifier that we use to synchronize the two LLMs.
+        notifier = EventNotifier()
+
+        # This turns the LLM context into an inference request to classify the user's speech
+        # as complete or incomplete.
+        statement_judge_context_filter = StatementJudgeContextFilter()
+
+        # This sends a UserStoppedSpeakingFrame and triggers the notifier event
+        completeness_check = CompletenessCheck(notifier=notifier)
+
+        # # Notify if the user hasn't said anything.
+        async def user_idle_notifier(frame):
+            await notifier.notify()
+
+        # Sometimes the LLM will fail detecting if a user has completed a
+        # sentence, this will wake up the notifier if that happens.
+        user_idle = UserIdleProcessor(callback=user_idle_notifier, timeout=5.0)
+
+        # We start with the gate open because we send an initial context frame
+        # to start the conversation.
+        bot_output_gate = OutputGate(notifier=notifier, start_open=True)
+
+        async def pass_only_llm_trigger_frames(frame):
+            return (
+                isinstance(frame, OpenAILLMContextFrame)
+                or isinstance(frame, StartInterruptionFrame)
+                or isinstance(frame, StopInterruptionFrame)
+                or isinstance(frame, FunctionCallInProgressFrame)
+                or isinstance(frame, FunctionCallResultFrame)
+            )
+
+        super().__init__(
+            [
+                ParallelPipeline(
+                    [
+                        # Ignore everything except an OpenAILLMContextFrame. Pass a specially constructed
+                        # simplified context frame to the statement classifier LLM. The only frame this
+                        # sub-pipeline will output is a UserStoppedSpeakingFrame.
+                        statement_judge_context_filter,
+                        statement_llm,
+                        completeness_check,
+                    ],
+                    [
+                        # Block everything except frames that trigger LLM inference.
+                        FunctionFilter(filter=pass_only_llm_trigger_frames),
+                        llm,
+                        bot_output_gate,  # Buffer all llm/tts output until notified.
+                    ],
+                ),
+                user_idle,
+            ]
+        )
+
+
 # We store functions so objects (e.g. SileroVADAnalyzer) don't get
 # instantiated. The function will be called when the desired transport gets
 # selected.
@@ -224,18 +293,13 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         voice_id="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
     )
 
-    # This is the LLM that will be used to detect if the user has finished a
-    # statement. This doesn't really need to be an LLM, we could use NLP
-    # libraries for that, but we have the machinery to use an LLM, so we might as well!
-    statement_llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
-
     # This is the regular LLM.
-    llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
+    llm_main = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
     # You can also register a function_name of None to get all functions
     # sent to the same callback with an additional function_name parameter.
-    llm.register_function("get_current_weather", fetch_weather_from_api)
+    llm_main.register_function("get_current_weather", fetch_weather_from_api)
 
-    @llm.event_handler("on_function_calls_started")
+    @llm_main.event_handler("on_function_calls_started")
     async def on_function_calls_started(service, function_calls):
         await tts.queue_frame(TTSSpeakFrame("Let me check on that."))
 
@@ -272,69 +336,18 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     ]
 
     context = OpenAILLMContext(messages, tools)
-    context_aggregator = llm.create_context_aggregator(context)
+    context_aggregator = llm_main.create_context_aggregator(context)
 
-    # We have instructed the LLM to return 'YES' if it thinks the user
+    # LLM + turn detection (with an extra LLM as a judge)
-    # completed a sentence. So, if it's 'YES' we will return true in this
+    llm = TurnDetectionLLM(llm_main)
-    # predicate which will wake up the notifier.
-    async def wake_check_filter(frame):
-        logger.debug(f"Completeness check frame: {frame}")
-        return frame.text == "YES"
-
-    # This is a notifier that we use to synchronize the two LLMs.
-    notifier = EventNotifier()
-
-    # This turns the LLM context into an inference request to classify the user's speech
-    # as complete or incomplete.
-    statement_judge_context_filter = StatementJudgeContextFilter()
-
-    # This sends a UserStoppedSpeakingFrame and triggers the notifier event
-    completeness_check = CompletenessCheck(notifier=notifier)
-
-    # # Notify if the user hasn't said anything.
-    async def user_idle_notifier(frame):
-        await notifier.notify()
-
-    # Sometimes the LLM will fail detecting if a user has completed a
-    # sentence, this will wake up the notifier if that happens.
-    user_idle = UserIdleProcessor(callback=user_idle_notifier, timeout=5.0)
-
-    # We start with the gate open because we send an initial context frame
-    # to start the conversation.
-    bot_output_gate = OutputGate(notifier=notifier, start_open=True)
-
-    async def pass_only_llm_trigger_frames(frame):
-        return (
-            isinstance(frame, OpenAILLMContextFrame)
-            or isinstance(frame, StartInterruptionFrame)
-            or isinstance(frame, StopInterruptionFrame)
-            or isinstance(frame, FunctionCallInProgressFrame)
-            or isinstance(frame, FunctionCallResultFrame)
-        )
 
     pipeline = Pipeline(
         [
             transport.input(),
             stt,
             context_aggregator.user(),
-            ParallelPipeline(
+            llm,
-                [
-                    # Ignore everything except an OpenAILLMContextFrame. Pass a specially constructed
-                    # simplified context frame to the statement classifier LLM. The only frame this
-                    # sub-pipeline will output is a UserStoppedSpeakingFrame.
-                    statement_judge_context_filter,
-                    statement_llm,
-                    completeness_check,
-                ],
-                [
-                    # Block everything except frames that trigger LLM inference.
-                    FunctionFilter(filter=pass_only_llm_trigger_frames),
-                    llm,
-                    bot_output_gate,  # Buffer all llm/tts output until notified.
-                ],
-            ),
             tts,
-            user_idle,
             transport.output(),
             context_aggregator.assistant(),
         ]
@@ -365,7 +378,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         await task.queue_frames(
             [
                 UserStartedSpeakingFrame(),
-                TranscriptionFrame(user_id="", timestamp=time.time(), text=message["message"]),
+                TranscriptionFrame(
+                    user_id="", timestamp=time_now_iso8601(), text=message["message"]
+                ),
                 UserStoppedSpeakingFrame(),
             ]
         )

examples/foundational/22c-natural-conversation-mixed-llms.py

@@ -6,7 +6,6 @@
 
 import asyncio
 import os
-import time
 
 from dotenv import load_dotenv
 from loguru import logger
@@ -45,13 +44,14 @@ from pipecat.runner.utils import create_transport
 from pipecat.services.anthropic.llm import AnthropicLLMService
 from pipecat.services.cartesia.tts import CartesiaTTSService
 from pipecat.services.deepgram.stt import DeepgramSTTService
-from pipecat.services.llm_service import FunctionCallParams
+from pipecat.services.llm_service import FunctionCallParams, LLMService
 from pipecat.services.openai.llm import OpenAILLMService
 from pipecat.sync.base_notifier import BaseNotifier
 from pipecat.sync.event_notifier import EventNotifier
 from pipecat.transports.base_transport import BaseTransport, TransportParams
 from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams
 from pipecat.transports.services.daily import DailyParams
+from pipecat.utils.time import time_now_iso8601
 
 load_dotenv(override=True)
 
@@ -391,6 +391,75 @@ class OutputGate(FrameProcessor):
                 break
 
 
+class TurnDetectionLLM(Pipeline):
+    def __init__(self, llm: LLMService):
+        # This is the LLM that will be used to detect if the user has finished a
+        # statement. This doesn't really need to be an LLM, we could use NLP
+        # libraries for that, but we have the machinery to use an LLM, so we might as well!
+        statement_llm = AnthropicLLMService(api_key=os.getenv("ANTHROPIC_API_KEY"))
+
+        # This is a notifier that we use to synchronize the two LLMs.
+        notifier = EventNotifier()
+
+        # This turns the LLM context into an inference request to classify the user's speech
+        # as complete or incomplete.
+        statement_judge_context_filter = StatementJudgeContextFilter()
+
+        # This sends a UserStoppedSpeakingFrame and triggers the notifier event
+        completeness_check = CompletenessCheck(notifier=notifier)
+
+        # # Notify if the user hasn't said anything.
+        async def user_idle_notifier(frame):
+            await notifier.notify()
+
+        # Sometimes the LLM will fail detecting if a user has completed a
+        # sentence, this will wake up the notifier if that happens.
+        user_idle = UserIdleProcessor(callback=user_idle_notifier, timeout=5.0)
+
+        # We start with the gate open because we send an initial context frame
+        # to start the conversation.
+        bot_output_gate = OutputGate(notifier=notifier, start_open=True)
+
+        async def block_user_stopped_speaking(frame):
+            return not isinstance(frame, UserStoppedSpeakingFrame)
+
+        async def pass_only_llm_trigger_frames(frame):
+            return (
+                isinstance(frame, OpenAILLMContextFrame)
+                or isinstance(frame, StartInterruptionFrame)
+                or isinstance(frame, StopInterruptionFrame)
+                or isinstance(frame, FunctionCallInProgressFrame)
+                or isinstance(frame, FunctionCallResultFrame)
+            )
+
+        super().__init__(
+            [
+                ParallelPipeline(
+                    [
+                        # Pass everything except UserStoppedSpeaking to the elements after
+                        # this ParallelPipeline
+                        FunctionFilter(filter=block_user_stopped_speaking),
+                    ],
+                    [
+                        # Ignore everything except an OpenAILLMContextFrame. Pass a specially constructed
+                        # simplified context frame to the statement classifier LLM. The only frame this
+                        # sub-pipeline will output is a UserStoppedSpeakingFrame.
+                        statement_judge_context_filter,
+                        statement_llm,
+                        completeness_check,
+                    ],
+                    [
+                        # Block everything except frames that trigger LLM inference.
+                        FunctionFilter(filter=pass_only_llm_trigger_frames),
+                        llm,
+                        bot_output_gate,  # Buffer all llm/tts output until notified.
+                    ],
+                ),
+                user_idle,
+            ]
+        )
+
+
 async def fetch_weather_from_api(params: FunctionCallParams):
     await params.result_callback({"conditions": "nice", "temperature": "75"})
 
@@ -427,18 +496,13 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         voice_id="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
     )
 
-    # This is the LLM that will be used to detect if the user has finished a
-    # statement. This doesn't really need to be an LLM, we could use NLP
-    # libraries for that, but we have the machinery to use an LLM, so we might as well!
-    statement_llm = AnthropicLLMService(api_key=os.getenv("ANTHROPIC_API_KEY"))
-
     # This is the regular LLM.
-    llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
+    llm_main = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
     # Register a function_name of None to get all functions
     # sent to the same callback with an additional function_name parameter.
-    llm.register_function("get_current_weather", fetch_weather_from_api)
+    llm_main.register_function("get_current_weather", fetch_weather_from_api)
 
-    @llm.event_handler("on_function_calls_started")
+    @llm_main.event_handler("on_function_calls_started")
     async def on_function_calls_started(service, function_calls):
         await tts.queue_frame(TTSSpeakFrame("Let me check on that."))
 
@@ -475,76 +539,18 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     ]
 
     context = OpenAILLMContext(messages, tools)
-    context_aggregator = llm.create_context_aggregator(context)
+    context_aggregator = llm_main.create_context_aggregator(context)
 
-    # We have instructed the LLM to return 'YES' if it thinks the user
+    # LLM + turn detection (with an extra LLM as a judge)
-    # completed a sentence. So, if it's 'YES' we will return true in this
+    llm = TurnDetectionLLM(llm_main)
-    # predicate which will wake up the notifier.
-    async def wake_check_filter(frame):
-        return frame.text == "YES"
-
-    # This is a notifier that we use to synchronize the two LLMs.
-    notifier = EventNotifier()
-
-    # This turns the LLM context into an inference request to classify the user's speech
-    # as complete or incomplete.
-    statement_judge_context_filter = StatementJudgeContextFilter()
-
-    # This sends a UserStoppedSpeakingFrame and triggers the notifier event
-    completeness_check = CompletenessCheck(notifier=notifier)
-
-    # # Notify if the user hasn't said anything.
-    async def user_idle_notifier(frame):
-        await notifier.notify()
-
-    # Sometimes the LLM will fail detecting if a user has completed a
-    # sentence, this will wake up the notifier if that happens.
-    user_idle = UserIdleProcessor(callback=user_idle_notifier, timeout=5.0)
-
-    # We start with the gate open because we send an initial context frame
-    # to start the conversation.
-    bot_output_gate = OutputGate(notifier=notifier, start_open=True)
-
-    async def block_user_stopped_speaking(frame):
-        return not isinstance(frame, UserStoppedSpeakingFrame)
-
-    async def pass_only_llm_trigger_frames(frame):
-        return (
-            isinstance(frame, OpenAILLMContextFrame)
-            or isinstance(frame, StartInterruptionFrame)
-            or isinstance(frame, StopInterruptionFrame)
-            or isinstance(frame, FunctionCallInProgressFrame)
-            or isinstance(frame, FunctionCallResultFrame)
-        )
 
     pipeline = Pipeline(
         [
             transport.input(),
             stt,
             context_aggregator.user(),
-            ParallelPipeline(
+            llm,
-                [
-                    # Pass everything except UserStoppedSpeaking to the elements after
-                    # this ParallelPipeline
-                    FunctionFilter(filter=block_user_stopped_speaking),
-                ],
-                [
-                    # Ignore everything except an OpenAILLMContextFrame. Pass a specially constructed
-                    # simplified context frame to the statement classifier LLM. The only frame this
-                    # sub-pipeline will output is a UserStoppedSpeakingFrame.
-                    statement_judge_context_filter,
-                    statement_llm,
-                    completeness_check,
-                ],
-                [
-                    # Block everything except frames that trigger LLM inference.
-                    FunctionFilter(filter=pass_only_llm_trigger_frames),
-                    llm,
-                    bot_output_gate,  # Buffer all llm/tts output until notified.
-                ],
-            ),
             tts,
-            user_idle,
             transport.output(),
             context_aggregator.assistant(),
         ]
@@ -580,7 +586,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         await task.queue_frames(
             [
                 UserStartedSpeakingFrame(),
-                TranscriptionFrame(user_id="", timestamp=time.time(), text=message["message"]),
+                TranscriptionFrame(
+                    user_id="", timestamp=time_now_iso8601(), text=message["message"]
+                ),
                 UserStoppedSpeakingFrame(),
             ]
         )

examples/foundational/22d-natural-conversation-gemini-audio.py

@@ -47,11 +47,13 @@ from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
 from pipecat.services.cartesia.tts import CartesiaTTSService
 from pipecat.services.google.llm import GoogleLLMContext, GoogleLLMService
+from pipecat.services.llm_service import LLMService
 from pipecat.sync.base_notifier import BaseNotifier
 from pipecat.sync.event_notifier import EventNotifier
 from pipecat.transports.base_transport import BaseTransport, TransportParams
 from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams
 from pipecat.transports.services.daily import DailyParams
+from pipecat.utils.time import time_now_iso8601
 
 load_dotenv(override=True)
 
@@ -607,23 +609,90 @@ class OutputGate(FrameProcessor):
             self._gate_task = None
 
     async def _gate_task_handler(self):
-        while True:
+        await self._notifier.wait()
-            try:
-                await self._notifier.wait()
 
-                transcription = await self._transcription_buffer.wait_for_transcription() or "-"
+        transcription = await self._transcription_buffer.wait_for_transcription() or "-"
-                self._context.add_message(Content(role="user", parts=[Part(text=transcription)]))
+        self._context.add_message(Content(role="user", parts=[Part(text=transcription)]))
 
-                self.open_gate()
+        self.open_gate()
-                for frame, direction in self._frames_buffer:
+        for frame, direction in self._frames_buffer:
-                    await self.push_frame(frame, direction)
+            await self.push_frame(frame, direction)
-                self._frames_buffer = []
+        self._frames_buffer = []
-            except asyncio.CancelledError:
+
-                break
+
-            except Exception as e:
+class TurnDetectionLLM(Pipeline):
-                logger.error(f"OutputGate error: {e}")
+    def __init__(self, llm: LLMService, context: OpenAILLMContext):
-                raise e
+        # This is the LLM that will transcribe user speech.
-                break
+        tx_llm = GoogleLLMService(
+            name="Transcriber",
+            model=TRANSCRIBER_MODEL,
+            api_key=os.getenv("GOOGLE_API_KEY"),
+            temperature=0.0,
+            system_instruction=transcriber_system_instruction,
+        )
+
+        # This is the LLM that will classify user speech as complete or incomplete.
+        classifier_llm = GoogleLLMService(
+            name="Classifier",
+            model=CLASSIFIER_MODEL,
+            api_key=os.getenv("GOOGLE_API_KEY"),
+            temperature=0.0,
+            system_instruction=classifier_system_instruction,
+        )
+
+        # This is a notifier that we use to synchronize the two LLMs.
+        notifier = EventNotifier()
+
+        # This turns the LLM context into an inference request to classify the user's speech
+        # as complete or incomplete.
+        # statement_judge_context_filter = StatementJudgeAudioContextAccumulator(notifier=notifier)
+
+        audio_accumulater = AudioAccumulator()
+        # This sends a UserStoppedSpeakingFrame and triggers the notifier event
+        completeness_check = CompletenessCheck(
+            notifier=notifier, audio_accumulator=audio_accumulater
+        )
+
+        async def block_user_stopped_speaking(frame):
+            return not isinstance(frame, UserStoppedSpeakingFrame)
+
+        conversation_audio_context_assembler = ConversationAudioContextAssembler(context=context)
+
+        llm_aggregator_buffer = LLMAggregatorBuffer()
+
+        bot_output_gate = OutputGate(
+            notifier=notifier, context=context, llm_transcription_buffer=llm_aggregator_buffer
+        )
+
+        super().__init__(
+            [
+                audio_accumulater,
+                ParallelPipeline(
+                    [
+                        # Pass everything except UserStoppedSpeaking to the elements after
+                        # this ParallelPipeline
+                        FunctionFilter(filter=block_user_stopped_speaking),
+                    ],
+                    [
+                        ParallelPipeline(
+                            [
+                                classifier_llm,
+                                completeness_check,
+                            ],
+                            [
+                                tx_llm,
+                                llm_aggregator_buffer,
+                            ],
+                        )
+                    ],
+                    [
+                        conversation_audio_context_assembler,
+                        llm,
+                        bot_output_gate,  # buffer output until notified, then flush frames and update context
+                    ],
+                ),
+            ]
+        )
 
 
 # We store functions so objects (e.g. SileroVADAnalyzer) don't get
@@ -656,24 +725,6 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         voice_id="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
     )
 
-    # This is the LLM that will transcribe user speech.
-    tx_llm = GoogleLLMService(
-        name="Transcriber",
-        model=TRANSCRIBER_MODEL,
-        api_key=os.getenv("GOOGLE_API_KEY"),
-        temperature=0.0,
-        system_instruction=transcriber_system_instruction,
-    )
-
-    # This is the LLM that will classify user speech as complete or incomplete.
-    classifier_llm = GoogleLLMService(
-        name="Classifier",
-        model=CLASSIFIER_MODEL,
-        api_key=os.getenv("GOOGLE_API_KEY"),
-        temperature=0.0,
-        system_instruction=classifier_system_instruction,
-    )
-
     # This is the regular LLM that responds conversationally.
     conversation_llm = GoogleLLMService(
         name="Conversation",
@@ -685,57 +736,12 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     context = OpenAILLMContext()
     context_aggregator = conversation_llm.create_context_aggregator(context)
 
-    # This is a notifier that we use to synchronize the two LLMs.
+    llm = TurnDetectionLLM(conversation_llm, context)
-    notifier = EventNotifier()
-
-    # This turns the LLM context into an inference request to classify the user's speech
-    # as complete or incomplete.
-    # statement_judge_context_filter = StatementJudgeAudioContextAccumulator(notifier=notifier)
-
-    audio_accumulater = AudioAccumulator()
-    # This sends a UserStoppedSpeakingFrame and triggers the notifier event
-    completeness_check = CompletenessCheck(notifier=notifier, audio_accumulator=audio_accumulater)
-
-    async def block_user_stopped_speaking(frame):
-        return not isinstance(frame, UserStoppedSpeakingFrame)
-
-    conversation_audio_context_assembler = ConversationAudioContextAssembler(context=context)
-
-    llm_aggregator_buffer = LLMAggregatorBuffer()
-
-    bot_output_gate = OutputGate(
-        notifier=notifier, context=context, llm_transcription_buffer=llm_aggregator_buffer
-    )
 
     pipeline = Pipeline(
         [
             transport.input(),
-            audio_accumulater,
+            llm,
-            ParallelPipeline(
-                [
-                    # Pass everything except UserStoppedSpeaking to the elements after
-                    # this ParallelPipeline
-                    FunctionFilter(filter=block_user_stopped_speaking),
-                ],
-                [
-                    ParallelPipeline(
-                        [
-                            classifier_llm,
-                            completeness_check,
-                        ],
-                        [
-                            tx_llm,
-                            llm_aggregator_buffer,
-                        ],
-                    )
-                ],
-                [
-                    conversation_audio_context_assembler,
-                    conversation_llm,
-                    bot_output_gate,  # buffer output until notified, then flush frames and update context
-                    # TempPrinter(),
-                ],
-            ),
             tts,
             transport.output(),
             context_aggregator.assistant(),
@@ -766,7 +772,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         await task.queue_frames(
             [
                 UserStartedSpeakingFrame(),
-                TranscriptionFrame(user_id="", timestamp=time.time(), text=message["message"]),
+                TranscriptionFrame(
+                    user_id="", timestamp=time_now_iso8601(), text=message["message"]
+                ),
                 UserStoppedSpeakingFrame(),
             ]
         )