Merge pull request #2395 from pipecat-ai/aleix/examples-15-inherit-parallel-pipeline

examples(foundational): move 15/15a logic into its own processor

CHANGELOG.md

@@ -5,6 +5,14 @@ All notable changes to **Pipecat** will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## Unreleased
+
+### Other
+
+- Updated `15-switch-voices.py` and `15a-switch-languages.py` examples to show
+  how to enclose complex logic (e.g. `ParallelPipeline`) into a single processor
+  so the main pipeline becomes simpler.
+
 ## [0.0.79] - 2025-08-07
 
 ### Changed

examples/foundational/15-switch-voices.py

@@ -12,6 +12,7 @@ from loguru import logger
 from openai.types.chat import ChatCompletionToolParam
 
 from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import Frame
 from pipecat.pipeline.parallel_pipeline import ParallelPipeline
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
@@ -31,29 +32,54 @@ from pipecat.transports.services.daily import DailyParams
 load_dotenv(override=True)
 
 
-current_voice = "News Lady"
+class SwitchVoices(ParallelPipeline):
+    def __init__(self):
+        self._current_voice = "News Lady"
 
+        news_lady = CartesiaTTSService(
+            api_key=os.getenv("CARTESIA_API_KEY"),
+            voice_id="bf991597-6c13-47e4-8411-91ec2de5c466",  # Newslady
+        )
 
-async def switch_voice(params: FunctionCallParams):
+        british_lady = CartesiaTTSService(
-    global current_voice
+            api_key=os.getenv("CARTESIA_API_KEY"),
-    current_voice = params.arguments["voice"]
+            voice_id="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
-    await params.result_callback(
+        )
-        {
-            "voice": f"You are now using your {current_voice} voice. Your responses should now be as if you were a {current_voice}."
-        }
-    )
 
+        barbershop_man = CartesiaTTSService(
+            api_key=os.getenv("CARTESIA_API_KEY"),
+            voice_id="a0e99841-438c-4a64-b679-ae501e7d6091",  # Barbershop Man
+        )
 
-async def news_lady_filter(frame) -> bool:
+        super().__init__(
-    return current_voice == "News Lady"
+            # News Lady voice
+            [FunctionFilter(self.news_lady_filter), news_lady],
+            # British Reading Lady voice
+            [FunctionFilter(self.british_lady_filter), british_lady],
+            # Barbershop Man voice
+            [FunctionFilter(self.barbershop_man_filter), barbershop_man],
+        )
 
+    @property
+    def current_voice(self):
+        return self._current_voice
 
-async def british_lady_filter(frame) -> bool:
+    async def switch_voice(self, params: FunctionCallParams):
-    return current_voice == "British Lady"
+        self._current_voice = params.arguments["voice"]
+        await params.result_callback(
+            {
+                "voice": f"You are now using your {self.current_voice} voice. Your responses should now be as if you were a {self.current_voice}."
+            }
+        )
 
+    async def news_lady_filter(self, _: Frame) -> bool:
+        return self.current_voice == "News Lady"
 
-async def barbershop_man_filter(frame) -> bool:
+    async def british_lady_filter(self, _: Frame) -> bool:
-    return current_voice == "Barbershop Man"
+        return self.current_voice == "British Lady"
+
+    async def barbershop_man_filter(self, _: Frame) -> bool:
+        return self.current_voice == "Barbershop Man"
 
 
 # We store functions so objects (e.g. SileroVADAnalyzer) don't get
@@ -83,23 +109,10 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
 
     stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
 
-    news_lady = CartesiaTTSService(
+    tts = SwitchVoices()
-        api_key=os.getenv("CARTESIA_API_KEY"),
-        voice_id="bf991597-6c13-47e4-8411-91ec2de5c466",  # Newslady
-    )
-
-    british_lady = CartesiaTTSService(
-        api_key=os.getenv("CARTESIA_API_KEY"),
-        voice_id="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
-    )
-
-    barbershop_man = CartesiaTTSService(
-        api_key=os.getenv("CARTESIA_API_KEY"),
-        voice_id="a0e99841-438c-4a64-b679-ae501e7d6091",  # Barbershop Man
-    )
 
     llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
-    llm.register_function("switch_voice", switch_voice)
+    llm.register_function("switch_voice", tts.switch_voice)
 
     tools = [
         ChatCompletionToolParam(
@@ -136,14 +149,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
             stt,
             context_aggregator.user(),  # User responses
             llm,  # LLM
-            ParallelPipeline(  # TTS (one of the following vocies)
+            tts,  # TTS with switch voice functionality
-                [FunctionFilter(news_lady_filter), news_lady],  # News Lady voice
-                [
-                    FunctionFilter(british_lady_filter),
-                    british_lady,
-                ],  # British Reading Lady voice
-                [FunctionFilter(barbershop_man_filter), barbershop_man],  # Barbershop Man voice
-            ),
             transport.output(),  # Transport bot output
             context_aggregator.assistant(),  # Assistant spoken responses
         ]
@@ -165,7 +171,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         messages.append(
             {
                 "role": "system",
-                "content": f"Please introduce yourself to the user and let them know the voices you can do. Your initial responses should be as if you were a {current_voice}.",
+                "content": f"Please introduce yourself to the user and let them know the voices you can do. Your initial responses should be as if you were a {tts.current_voice}.",
             }
         )
         await task.queue_frames([context_aggregator.user().get_context_frame()])

examples/foundational/15a-switch-languages.py

@@ -13,6 +13,7 @@ from loguru import logger
 from openai.types.chat import ChatCompletionToolParam
 
 from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import Frame
 from pipecat.pipeline.parallel_pipeline import ParallelPipeline
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
@@ -32,23 +33,42 @@ from pipecat.transports.services.daily import DailyParams
 load_dotenv(override=True)
 
 
-current_language = "English"
+class SwitchLanguage(ParallelPipeline):
+    def __init__(self):
+        self._current_language = "English"
 
+        english_tts = CartesiaTTSService(
+            api_key=os.getenv("CARTESIA_API_KEY"),
+            voice_id="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
+        )
 
-async def switch_language(params: FunctionCallParams):
+        spanish_tts = CartesiaTTSService(
-    global current_language
+            api_key=os.getenv("CARTESIA_API_KEY"),
-    current_language = params.arguments["language"]
+            voice_id="d4db5fb9-f44b-4bd1-85fa-192e0f0d75f9",  # Spanish-speaking Lady
-    await params.result_callback(
+        )
-        {"voice": f"Your answers from now on should be in {current_language}."}
-    )
 
+        super().__init__(
+            # English
+            [FunctionFilter(self.english_filter), english_tts],
+            # Spanish
+            [FunctionFilter(self.spanish_filter), spanish_tts],
+        )
 
-async def english_filter(frame) -> bool:
+    @property
-    return current_language == "English"
+    def current_language(self):
+        return self._current_language
 
+    async def switch_language(self, params: FunctionCallParams):
+        self._current_language = params.arguments["language"]
+        await params.result_callback(
+            {"voice": f"Your answers from now on should be in {self.current_language}."}
+        )
 
-async def spanish_filter(frame) -> bool:
+    async def english_filter(self, _: Frame) -> bool:
-    return current_language == "Spanish"
+        return self.current_language == "English"
+
+    async def spanish_filter(self, _: Frame) -> bool:
+        return self.current_language == "Spanish"
 
 
 # We store functions so objects (e.g. SileroVADAnalyzer) don't get
@@ -80,18 +100,10 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         api_key=os.getenv("DEEPGRAM_API_KEY"), live_options=LiveOptions(language="multi")
     )
 
-    english_tts = CartesiaTTSService(
+    tts = SwitchLanguage()
-        api_key=os.getenv("CARTESIA_API_KEY"),
-        voice_id="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
-    )
-
-    spanish_tts = CartesiaTTSService(
-        api_key=os.getenv("CARTESIA_API_KEY"),
-        voice_id="d4db5fb9-f44b-4bd1-85fa-192e0f0d75f9",  # Spanish-speaking Lady
-    )
 
     llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
-    llm.register_function("switch_language", switch_language)
+    llm.register_function("switch_language", tts.switch_language)
 
     tools = [
         ChatCompletionToolParam(
@@ -128,10 +140,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
             stt,  # STT
             context_aggregator.user(),  # User responses
             llm,  # LLM
-            ParallelPipeline(  # TTS (bot will speak the chosen language)
+            tts,  # TTS (bot will speak the chosen language)
-                [FunctionFilter(english_filter), english_tts],  # English
-                [FunctionFilter(spanish_filter), spanish_tts],  # Spanish
-            ),
             transport.output(),  # Transport bot output
             context_aggregator.assistant(),  # Assistant spoken responses
         ]
@@ -153,7 +162,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         messages.append(
             {
                 "role": "system",
-                "content": f"Please introduce yourself to the user and let them know the languages you speak. Your initial responses should be in {current_language}.",
+                "content": f"Please introduce yourself to the user and let them know the languages you speak. Your initial responses should be in {tts.current_language}.",
             }
         )
         await task.queue_frames([context_aggregator.user().get_context_frame()])

scripts/evals/run-release-evals.py

@@ -93,7 +93,7 @@ TESTS_14 = [
     ("14p-function-calling-gemini-vertex-ai.py", PROMPT_WEATHER, EVAL_WEATHER),
     ("14q-function-calling-qwen.py", PROMPT_WEATHER, EVAL_WEATHER),
     ("14r-function-calling-aws.py", PROMPT_WEATHER, EVAL_WEATHER),
-    ("14v-function-calling-openai.py.py", PROMPT_WEATHER, EVAL_WEATHER),
+    ("14v-function-calling-openai.py", PROMPT_WEATHER, EVAL_WEATHER),
     # Currently not working.
     # ("14c-function-calling-together.py", PROMPT_WEATHER, EVAL_WEATHER),
     # ("14k-function-calling-cerebras.py", PROMPT_WEATHER, EVAL_WEATHER),