fix: add missing asyncio import to SentryMetrics

* Added Sarvam TTS Websocket Implementation

* Addressed some of the comments on PR

* added change voice logic

* added changes from main

* pushing text frames and added flush audio

* updated docs string for better docs

* Addressed comments and added some improvements

* pushed optional args down

* removed new line

* made aiohttp session mandatory in http service

* added push frame and removed unused function

* removed pong message

* added disconnecting logic

---------

Co-authored-by: vinayak-sarvam <vinayak@sarvam.ai>

CHANGELOG.md

@@ -7,8 +7,94 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- Added a `cancel_timeout_secs` argument to `PipelineTask` which defines how
+  long the pipeline has to complete cancellation. When `PipelineTask.cancel()`
+  is called, a `CancelFrame` is pushed through the pipeline and must reach the
+  end. If it does not reach the end within the specified time, a warning is
+  shown and the wait is aborted.
+
+- Added a new "universal" (LLM-agnostic) `LLMContext` and accompanying
+  `LLMContextAggregatorPair`, which will eventually replace `OpenAILLMContext`
+  (and the other under-the-hood contexts) and the other context aggregators.
+  The new universal `LLMContext` machinery allows a single context to be shared
+  between different LLMs, enabling runtime LLM switching and scenarios like
+  failover.
+
+  From the developer's point of view, switching to using the new universal
+  context machinery will usually be a matter of going from this:
+
+  ```python
+  context = OpenAILLMContext(messages, tools)
+  context_aggregator = llm.create_context_aggregator(context)
+  ```
+
+  To this:
+
+  ```python
+  context = LLMContext(messages, tools)
+  context_aggregator = LLMContextAggregatorPair(context)
+  ```
+
+  To start, the universal `LLMContext` is supported with the following LLM
+  services:
+
+  - `OpenAILLMService`
+  - `GoogleLLMService`
+
+- Added a new `LLMSwitcher` class to enable runtime LLM switching, built atop a
+  new generic `ServiceSwitcher`.
+
+  Switchers take a switching strategy. The first available strategy is
+  `ServiceSwitcherStrategyManual`.
+
+  To switch LLMs at runtime, the LLMs must be sharing one instance of the new
+  universal `LLMContext` (see above bullet).
+
+  ```python
+  # Instantiate your LLM services
+  llm_openai = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
+  llm_google = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY"))
+
+  # Instantiate a switcher
+  # (ServiceSwitcherStrategyManual defaults to OpenAI, as it's first in the list)
+  llm_switcher = LLMSwitcher(
+      llms=[llm_openai, llm_google], strategy_type=ServiceSwitcherStrategyManual
+  )
+
+  # Create your pipeline
+  pipeline = Pipeline(
+    [
+        transport.input(),
+        stt,
+        context_aggregator.user(),
+        llm_switcher,
+        tts,
+        transport.output(),
+        context_aggregator.assistant(),
+    ]
+  )
+  task = PipelineTask(pipeline, params=PipelineParams(allow_interruptions=True))
+
+  # ...
+  # Whenever is appropriate, switch LLMs!
+  await task.queue_frames([ManuallySwitchServiceFrame(service=llm_google)])
+  ```
+
+- Added an `LLMService.run_inference()` method to LLM services to enable
+  direct, out-of-band (i.e. out-of-pipeline) inference.
+
+### Changed
+
+- Updated `SarvamTTSService` to use WebSocket streaming for real-time audio
+  generation with multiple Indian languages, with HTTP support still available
+  via `SarvamHttpTTSService`.
+
 ### Fixed
 
+- Fixed a missing import in `SentryMetrics`.
+
 - Fixed a `CartesiaTTSService` issue that was causing the application to hang
   after Cartesia's 5 minutes timed out.
 
@@ -62,7 +148,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Deprecated
 
 - `FrameProcessor.wait_for_task()` is deprecated. Use `await task` or `await
-  asyncio.wait_for(task, timeout)` instead.
+asyncio.wait_for(task, timeout)` instead.
 
 ### Removed
 

env.example

@@ -59,6 +59,9 @@ GOOGLE_VERTEX_TEST_CREDENTIALS=...
 LMNT_API_KEY=...
 LMNT_VOICE_ID=...
 
+# Perplexity
+PERPLEXITY_API_KEY=...
+
 # PlayHT
 PLAY_HT_USER_ID=...
 PLAY_HT_API_KEY=...

examples/foundational/07z-interruptible-sarvam-http.py

@@ -0,0 +1,126 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+
+import os
+
+import aiohttp
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.runner.types import RunnerArguments
+from pipecat.runner.utils import create_transport
+from pipecat.services.deepgram.stt import DeepgramSTTService
+from pipecat.services.openai.llm import OpenAILLMService
+from pipecat.services.sarvam.tts import SarvamHttpTTSService
+from pipecat.transcriptions.language import Language
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams
+from pipecat.transports.services.daily import DailyParams
+
+load_dotenv(override=True)
+
+
+# We store functions so objects (e.g. SileroVADAnalyzer) don't get
+# instantiated. The function will be called when the desired transport gets
+# selected.
+transport_params = {
+    "daily": lambda: DailyParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(),
+    ),
+    "twilio": lambda: FastAPIWebsocketParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(),
+    ),
+    "webrtc": lambda: TransportParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(),
+    ),
+}
+
+
+async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
+    logger.info(f"Starting bot")
+
+    # Create an HTTP session
+    async with aiohttp.ClientSession() as session:
+        stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+        tts = SarvamHttpTTSService(
+            api_key=os.getenv("SARVAM_API_KEY"),
+            aiohttp_session=session,
+            params=SarvamHttpTTSService.InputParams(language=Language.EN),
+        )
+
+        llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
+
+        messages = [
+            {
+                "role": "system",
+                "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
+            },
+        ]
+
+        context = OpenAILLMContext(messages)
+        context_aggregator = llm.create_context_aggregator(context)
+
+        pipeline = Pipeline(
+            [
+                transport.input(),  # Transport user input
+                stt,
+                context_aggregator.user(),  # User responses
+                llm,  # LLM
+                tts,  # TTS
+                transport.output(),  # Transport bot output
+                context_aggregator.assistant(),  # Assistant spoken responses
+            ]
+        )
+
+        task = PipelineTask(
+            pipeline,
+            params=PipelineParams(
+                enable_metrics=True,
+                enable_usage_metrics=True,
+            ),
+            idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
+        )
+
+        @transport.event_handler("on_client_connected")
+        async def on_client_connected(transport, client):
+            logger.info(f"Client connected")
+            # Kick off the conversation.
+            messages.append({"role": "system", "content": "Please introduce yourself to the user."})
+            await task.queue_frames([context_aggregator.user().get_context_frame()])
+
+        @transport.event_handler("on_client_disconnected")
+        async def on_client_disconnected(transport, client):
+            logger.info(f"Client disconnected")
+            await task.cancel()
+
+        runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
+
+        await runner.run(task)
+
+
+async def bot(runner_args: RunnerArguments):
+    """Main bot entry point compatible with Pipecat Cloud."""
+    transport = await create_transport(runner_args, transport_params)
+    await run_bot(transport, runner_args)
+
+
+if __name__ == "__main__":
+    from pipecat.runner.run import main
+
+    main()

examples/foundational/07z-interruptible-sarvam.py

@@ -5,6 +5,7 @@
 #
 
 
+import asyncio
 import os
 
 import aiohttp
@@ -12,6 +13,7 @@ from dotenv import load_dotenv
 from loguru import logger
 
 from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import TTSUpdateSettingsFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
@@ -21,7 +23,6 @@ from pipecat.runner.utils import create_transport
 from pipecat.services.deepgram.stt import DeepgramSTTService
 from pipecat.services.openai.llm import OpenAILLMService
 from pipecat.services.sarvam.tts import SarvamTTSService
-from pipecat.transcriptions.language import Language
 from pipecat.transports.base_transport import BaseTransport, TransportParams
 from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams
 from pipecat.transports.services.daily import DailyParams
@@ -54,64 +55,64 @@ transport_params = {
 async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     logger.info(f"Starting bot")
 
-    # Create an HTTP session
-    async with aiohttp.ClientSession() as session:
-        stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
 
-        tts = SarvamTTSService(
-            api_key=os.getenv("SARVAM_API_KEY"),
-            aiohttp_session=session,
-            params=SarvamTTSService.InputParams(language=Language.EN),
-        )
+    tts = SarvamTTSService(
+        api_key=os.getenv("SARVAM_API_KEY"),
+        model="bulbul:v2",
+        voice_id="manisha",
+    )
+    llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
 
-        llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
+    messages = [
+        {
+            "role": "system",
+            "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
+        },
+    ]
 
-        messages = [
-            {
-                "role": "system",
-                "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
-            },
+    context = OpenAILLMContext(messages)
+    context_aggregator = llm.create_context_aggregator(context)
+
+    pipeline = Pipeline(
+        [
+            transport.input(),  # Transport user input
+            stt,
+            context_aggregator.user(),  # User responses
+            llm,  # LLM
+            tts,  # TTS
+            transport.output(),  # Transport bot output
+            context_aggregator.assistant(),  # Assistant spoken responses
         ]
+    )
 
-        context = OpenAILLMContext(messages)
-        context_aggregator = llm.create_context_aggregator(context)
+    task = PipelineTask(
+        pipeline,
+        params=PipelineParams(
+            enable_metrics=True,
+            enable_usage_metrics=True,
+        ),
+    )
 
-        pipeline = Pipeline(
-            [
-                transport.input(),  # Transport user input
-                stt,
-                context_aggregator.user(),  # User responses
-                llm,  # LLM
-                tts,  # TTS
-                transport.output(),  # Transport bot output
-                context_aggregator.assistant(),  # Assistant spoken responses
-            ]
-        )
+    @transport.event_handler("on_client_connected")
+    async def on_client_connected(transport, client):
+        logger.info(f"Client connected")
+        # Kick off the conversation.
+        messages.append({"role": "system", "content": "Please introduce yourself to the user."})
+        await task.queue_frames([context_aggregator.user().get_context_frame()])
 
-        task = PipelineTask(
-            pipeline,
-            params=PipelineParams(
-                enable_metrics=True,
-                enable_usage_metrics=True,
-            ),
-            idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
-        )
+        # Optionally, you can wait for 30 seconds and then change the voice.
+        # await asyncio.sleep(30)
+        # await task.queue_frame(TTSUpdateSettingsFrame(settings={"voice": "anushka"}))
 
-        @transport.event_handler("on_client_connected")
-        async def on_client_connected(transport, client):
-            logger.info(f"Client connected")
-            # Kick off the conversation.
-            messages.append({"role": "system", "content": "Please introduce yourself to the user."})
-            await task.queue_frames([context_aggregator.user().get_context_frame()])
+    @transport.event_handler("on_client_disconnected")
+    async def on_client_disconnected(transport, client):
+        logger.info(f"Client disconnected")
+        await task.cancel()
 
-        @transport.event_handler("on_client_disconnected")
-        async def on_client_disconnected(transport, client):
-            logger.info(f"Client disconnected")
-            await task.cancel()
+    runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
 
-        runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
-
-        await runner.run(task)
+    await runner.run(task)
 
 
 async def bot(runner_args: RunnerArguments):

examples/foundational/14x-function-calling-universal-context.py

@@ -0,0 +1,170 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import os
+
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.adapters.schemas.function_schema import FunctionSchema
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import TTSSpeakFrame
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
+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.openai.llm import OpenAILLMService
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams
+from pipecat.transports.services.daily import DailyParams
+
+load_dotenv(override=True)
+
+
+async def fetch_weather_from_api(params: FunctionCallParams):
+    await params.result_callback({"conditions": "nice", "temperature": "75"})
+
+
+async def fetch_restaurant_recommendation(params: FunctionCallParams):
+    await params.result_callback({"name": "The Golden Dragon"})
+
+
+# We store functions so objects (e.g. SileroVADAnalyzer) don't get
+# instantiated. The function will be called when the desired transport gets
+# selected.
+transport_params = {
+    "daily": lambda: DailyParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(),
+    ),
+    "twilio": lambda: FastAPIWebsocketParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(),
+    ),
+    "webrtc": lambda: TransportParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(),
+    ),
+}
+
+
+async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
+    logger.info(f"Starting bot")
+
+    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+    tts = CartesiaTTSService(
+        api_key=os.getenv("CARTESIA_API_KEY"),
+        voice_id="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
+    )
+
+    llm = 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.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation)
+
+    @llm.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."))
+
+    weather_function = FunctionSchema(
+        name="get_current_weather",
+        description="Get the current weather",
+        properties={
+            "location": {
+                "type": "string",
+                "description": "The city and state, e.g. San Francisco, CA",
+            },
+            "format": {
+                "type": "string",
+                "enum": ["celsius", "fahrenheit"],
+                "description": "The temperature unit to use. Infer this from the user's location.",
+            },
+        },
+        required=["location", "format"],
+    )
+    restaurant_function = FunctionSchema(
+        name="get_restaurant_recommendation",
+        description="Get a restaurant recommendation",
+        properties={
+            "location": {
+                "type": "string",
+                "description": "The city and state, e.g. San Francisco, CA",
+            },
+        },
+        required=["location"],
+    )
+    tools = ToolsSchema(standard_tools=[weather_function, restaurant_function])
+
+    messages = [
+        {
+            "role": "system",
+            "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
+        },
+    ]
+
+    context = LLMContext(messages, tools)
+    context_aggregator = LLMContextAggregatorPair(context)
+
+    pipeline = Pipeline(
+        [
+            transport.input(),
+            stt,
+            context_aggregator.user(),
+            llm,
+            tts,
+            transport.output(),
+            context_aggregator.assistant(),
+        ]
+    )
+
+    task = PipelineTask(
+        pipeline,
+        params=PipelineParams(
+            enable_metrics=True,
+            enable_usage_metrics=True,
+        ),
+        idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
+    )
+
+    @transport.event_handler("on_client_connected")
+    async def on_client_connected(transport, client):
+        logger.info(f"Client connected")
+        # Kick off the conversation.
+        await task.queue_frames([context_aggregator.user().get_context_frame()])
+
+    @transport.event_handler("on_client_disconnected")
+    async def on_client_disconnected(transport, client):
+        logger.info(f"Client disconnected")
+        await task.cancel()
+
+    runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
+
+    await runner.run(task)
+
+
+async def bot(runner_args: RunnerArguments):
+    """Main bot entry point compatible with Pipecat Cloud."""
+    transport = await create_transport(runner_args, transport_params)
+    await run_bot(transport, runner_args)
+
+
+if __name__ == "__main__":
+    from pipecat.runner.run import main
+
+    main()

examples/foundational/14y-function-calling-google-universal-context.py

@@ -0,0 +1,229 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+
+import asyncio
+import os
+
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.adapters.schemas.function_schema import FunctionSchema
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import TTSSpeakFrame
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
+from pipecat.runner.types import RunnerArguments
+from pipecat.runner.utils import (
+    create_transport,
+    get_transport_client_id,
+    maybe_capture_participant_camera,
+)
+from pipecat.services.cartesia.tts import CartesiaTTSService
+from pipecat.services.deepgram.stt import DeepgramSTTService
+from pipecat.services.google.llm import GoogleLLMService
+from pipecat.services.llm_service import FunctionCallParams
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.services.daily import DailyParams
+
+load_dotenv(override=True)
+
+
+# Global variable to store the client ID
+client_id = ""
+
+
+async def get_weather(params: FunctionCallParams):
+    location = params.arguments["location"]
+    await params.result_callback(f"The weather in {location} is currently 72 degrees and sunny.")
+
+
+async def fetch_restaurant_recommendation(params: FunctionCallParams):
+    await params.result_callback({"name": "The Golden Dragon"})
+
+
+async def get_image(params: FunctionCallParams):
+    question = params.arguments["question"]
+    logger.debug(f"Requesting image with user_id={client_id}, question={question}")
+
+    # Request the image frame
+    await params.llm.request_image_frame(
+        user_id=client_id,
+        function_name=params.function_name,
+        tool_call_id=params.tool_call_id,
+        text_content=question,
+    )
+
+    # Wait a short time for the frame to be processed
+    await asyncio.sleep(0.5)
+
+    # Return a result to complete the function call
+    await params.result_callback(
+        f"I've captured an image from your camera and I'm analyzing what you asked about: {question}"
+    )
+
+
+# We store functions so objects (e.g. SileroVADAnalyzer) don't get
+# instantiated. The function will be called when the desired transport gets
+# selected.
+transport_params = {
+    "daily": lambda: DailyParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        video_in_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(),
+    ),
+    "webrtc": lambda: TransportParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        video_in_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(),
+    ),
+}
+
+
+async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
+    logger.info(f"Starting bot")
+
+    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+    tts = CartesiaTTSService(
+        api_key=os.getenv("CARTESIA_API_KEY"),
+        voice_id="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
+    )
+
+    llm = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY"), model="gemini-2.0-flash-001")
+    llm.register_function("get_weather", get_weather)
+    llm.register_function("get_image", get_image)
+    llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation)
+
+    @llm.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."))
+
+    weather_function = FunctionSchema(
+        name="get_weather",
+        description="Get the current weather",
+        properties={
+            "location": {
+                "type": "string",
+                "description": "The city and state, e.g. San Francisco, CA",
+            },
+            "format": {
+                "type": "string",
+                "enum": ["celsius", "fahrenheit"],
+                "description": "The temperature unit to use. Infer this from the user's location.",
+            },
+        },
+        required=["location", "format"],
+    )
+    restaurant_function = FunctionSchema(
+        name="get_restaurant_recommendation",
+        description="Get a restaurant recommendation",
+        properties={
+            "location": {
+                "type": "string",
+                "description": "The city and state, e.g. San Francisco, CA",
+            },
+        },
+        required=["location"],
+    )
+    get_image_function = FunctionSchema(
+        name="get_image",
+        description="Get an image from the video stream.",
+        properties={
+            "question": {
+                "type": "string",
+                "description": "The question that the user is asking about the image.",
+            }
+        },
+        required=["question"],
+    )
+    tools = ToolsSchema(standard_tools=[weather_function, get_image_function, restaurant_function])
+
+    system_prompt = """\
+You are a helpful assistant who converses with a user and answers questions. Respond concisely to general questions.
+
+Your response will be turned into speech so use only simple words and punctuation.
+
+You have access to three tools: get_weather, get_restaurant_recommendation, and get_image.
+
+You can respond to questions about the weather using the get_weather tool.
+
+You can answer questions about the user's video stream using the get_image tool. Some examples of phrases that \
+indicate you should use the get_image tool are:
+- What do you see?
+- What's in the video?
+- Can you describe the video?
+- Tell me about what you see.
+- Tell me something interesting about what you see.
+- What's happening in the video?
+"""
+    messages = [
+        {"role": "system", "content": system_prompt},
+        {"role": "user", "content": "Say hello."},
+    ]
+
+    context = LLMContext(messages, tools)
+    context_aggregator = LLMContextAggregatorPair(context)
+
+    pipeline = Pipeline(
+        [
+            transport.input(),
+            stt,
+            context_aggregator.user(),
+            llm,
+            tts,
+            transport.output(),
+            context_aggregator.assistant(),
+        ]
+    )
+
+    task = PipelineTask(
+        pipeline,
+        params=PipelineParams(
+            enable_metrics=True,
+            enable_usage_metrics=True,
+        ),
+        idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
+    )
+
+    @transport.event_handler("on_client_connected")
+    async def on_client_connected(transport, client):
+        logger.info(f"Client connected: {client}")
+
+        await maybe_capture_participant_camera(transport, client)
+
+        global client_id
+        client_id = get_transport_client_id(transport, client)
+
+        # Kick off the conversation.
+        await task.queue_frames([context_aggregator.user().get_context_frame()])
+
+    @transport.event_handler("on_client_disconnected")
+    async def on_client_disconnected(transport, client):
+        logger.info(f"Client disconnected")
+        await task.cancel()
+
+    runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
+
+    await runner.run(task)
+
+
+async def bot(runner_args: RunnerArguments):
+    """Main bot entry point compatible with Pipecat Cloud."""
+    transport = await create_transport(runner_args, transport_params)
+    await run_bot(transport, runner_args)
+
+
+if __name__ == "__main__":
+    from pipecat.runner.run import main
+
+    main()

pyproject.toml

@@ -90,6 +90,7 @@ rime = [ "websockets>=13.1,<15.0" ]
 riva = [ "nvidia-riva-client~=2.21.1" ]
 runner = [ "python-dotenv>=1.0.0,<2.0.0", "uvicorn>=0.32.0,<1.0.0", "fastapi>=0.115.6,<0.117.0", "pipecat-ai-small-webrtc-prebuilt>=1.0.0"]
 sambanova = []
+sarvam = [ "websockets>=13.1,<15.0" ]
 sentry = [ "sentry-sdk~=2.23.1" ]
 local-smart-turn = [ "coremltools>=8.0", "transformers", "torch>=2.5.0,<3", "torchaudio>=2.5.0,<3" ]
 remote-smart-turn = []

src/pipecat/adapters/base_llm_adapter.py

@@ -11,21 +11,45 @@ adapters that handle tool format conversion and standardization.
 """
 
 from abc import ABC, abstractmethod
-from typing import Any, List, Union, cast
+from typing import Any, Generic, List, TypeVar, Union, cast
 
 from loguru import logger
 
 from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.processors.aggregators.llm_context import LLMContext, NotGiven
+
+# Should be a TypedDict
+TLLMInvocationParams = TypeVar("TLLMInvocationParams", bound=dict[str, Any])
 
 
-class BaseLLMAdapter(ABC):
+class BaseLLMAdapter(ABC, Generic[TLLMInvocationParams]):
     """Abstract base class for LLM provider adapters.
 
-    Provides a standard interface for converting between Pipecat's standardized
-    tool schemas and provider-specific tool formats. Subclasses must implement
-    provider-specific conversion logic.
+    Provides a standard interface for converting to provider-specific formats.
+
+    Handles:
+
+    - Extracting provider-specific parameters for LLM invocation from a
+      universal LLM context
+    - Converting standardized tools schema to provider-specific tool formats.
+    - Extracting messages from the LLM context for the purposes of logging
+      about the specific provider.
+
+    Subclasses must implement provider-specific conversion logic.
     """
 
+    @abstractmethod
+    def get_llm_invocation_params(self, context: LLMContext) -> TLLMInvocationParams:
+        """Get provider-specific LLM invocation parameters from a universal LLM context.
+
+        Args:
+            context: The LLM context containing messages, tools, etc.
+
+        Returns:
+            Provider-specific parameters for invoking the LLM.
+        """
+        pass
+
     @abstractmethod
     def to_provider_tools_format(self, tools_schema: ToolsSchema) -> List[Any]:
         """Convert tools schema to the provider's specific format.
@@ -38,7 +62,20 @@ class BaseLLMAdapter(ABC):
         """
         pass
 
-    def from_standard_tools(self, tools: Any) -> List[Any]:
+    @abstractmethod
+    def get_messages_for_logging(self, context: LLMContext) -> List[dict[str, Any]]:
+        """Get messages from a universal LLM context in a format ready for logging about this provider.
+
+        Args:
+            context: The LLM context containing messages.
+
+        Returns:
+            List of messages in a format ready for logging about this
+            provider.
+        """
+        pass
+
+    def from_standard_tools(self, tools: Any) -> List[Any] | NotGiven:
         """Convert tools from standard format to provider format.
 
         Args:

src/pipecat/adapters/services/anthropic_adapter.py

@@ -6,20 +6,58 @@
 
 """Anthropic LLM adapter for Pipecat."""
 
-from typing import Any, Dict, List
+from typing import Any, Dict, List, TypedDict
 
 from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
 from pipecat.adapters.schemas.function_schema import FunctionSchema
 from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.processors.aggregators.llm_context import LLMContext
 
 
-class AnthropicLLMAdapter(BaseLLMAdapter):
+class AnthropicLLMInvocationParams(TypedDict):
+    """Context-based parameters for invoking Anthropic's LLM API.
+
+    This is a placeholder until support for universal LLMContext machinery is added for Anthropic.
+    """
+
+    pass
+
+
+class AnthropicLLMAdapter(BaseLLMAdapter[AnthropicLLMInvocationParams]):
     """Adapter for converting tool schemas to Anthropic's function-calling format.
 
     This adapter handles the conversion of Pipecat's standard function schemas
     to the specific format required by Anthropic's Claude models for function calling.
     """
 
+    def get_llm_invocation_params(self, context: LLMContext) -> AnthropicLLMInvocationParams:
+        """Get Anthropic-specific LLM invocation parameters from a universal LLM context.
+
+        This is a placeholder until support for universal LLMContext machinery is added for Anthropic.
+
+        Args:
+            context: The LLM context containing messages, tools, etc.
+
+        Returns:
+            Dictionary of parameters for invoking Anthropic's LLM API.
+        """
+        raise NotImplementedError("Universal LLMContext is not yet supported for Anthropic.")
+
+    def get_messages_for_logging(self, context) -> List[dict[str, Any]]:
+        """Get messages from a universal LLM context in a format ready for logging about Anthropic.
+
+        Removes or truncates sensitive data like image content for safe logging.
+
+        This is a placeholder until support for universal LLMContext machinery is added for Anthropic.
+
+        Args:
+            context: The LLM context containing messages.
+
+        Returns:
+            List of messages in a format ready for logging about Anthropic.
+        """
+        raise NotImplementedError("Universal LLMContext is not yet supported for Anthropic.")
+
     @staticmethod
     def _to_anthropic_function_format(function: FunctionSchema) -> Dict[str, Any]:
         """Convert a single function schema to Anthropic's format.

src/pipecat/adapters/services/aws_nova_sonic_adapter.py

@@ -7,20 +7,58 @@
 """AWS Nova Sonic LLM adapter for Pipecat."""
 
 import json
-from typing import Any, Dict, List
+from typing import Any, Dict, List, TypedDict
 
 from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
 from pipecat.adapters.schemas.function_schema import FunctionSchema
 from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.processors.aggregators.llm_context import LLMContext
 
 
-class AWSNovaSonicLLMAdapter(BaseLLMAdapter):
+class AWSNovaSonicLLMInvocationParams(TypedDict):
+    """Context-based parameters for invoking AWS Nova Sonic LLM API.
+
+    This is a placeholder until support for universal LLMContext machinery is added for AWS Nova Sonic.
+    """
+
+    pass
+
+
+class AWSNovaSonicLLMAdapter(BaseLLMAdapter[AWSNovaSonicLLMInvocationParams]):
     """Adapter for AWS Nova Sonic language models.
 
     Converts Pipecat's standard function schemas into AWS Nova Sonic's
     specific function-calling format, enabling tool use with Nova Sonic models.
     """
 
+    def get_llm_invocation_params(self, context: LLMContext) -> AWSNovaSonicLLMInvocationParams:
+        """Get AWS Nova Sonic-specific LLM invocation parameters from a universal LLM context.
+
+        This is a placeholder until support for universal LLMContext machinery is added for AWS Nova Sonic.
+
+        Args:
+            context: The LLM context containing messages, tools, etc.
+
+        Returns:
+            Dictionary of parameters for invoking AWS Nova Sonic's LLM API.
+        """
+        raise NotImplementedError("Universal LLMContext is not yet supported for AWS Nova Sonic.")
+
+    def get_messages_for_logging(self, context) -> List[dict[str, Any]]:
+        """Get messages from a universal LLM context in a format ready for logging about AWS Nova Sonic.
+
+        Removes or truncates sensitive data like image content for safe logging.
+
+        This is a placeholder until support for universal LLMContext machinery is added for AWS Nova Sonic.
+
+        Args:
+            context: The LLM context containing messages.
+
+        Returns:
+            List of messages in a format ready for logging about AWS Nova Sonic.
+        """
+        raise NotImplementedError("Universal LLMContext is not yet supported for AWS Nova Sonic.")
+
     @staticmethod
     def _to_aws_nova_sonic_function_format(function: FunctionSchema) -> Dict[str, Any]:
         """Convert a function schema to AWS Nova Sonic format.

src/pipecat/adapters/services/bedrock_adapter.py

@@ -6,20 +6,58 @@
 
 """AWS Bedrock LLM adapter for Pipecat."""
 
-from typing import Any, Dict, List
+from typing import Any, Dict, List, TypedDict
 
 from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
 from pipecat.adapters.schemas.function_schema import FunctionSchema
 from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.processors.aggregators.llm_context import LLMContext
 
 
-class AWSBedrockLLMAdapter(BaseLLMAdapter):
+class AWSBedrockLLMInvocationParams(TypedDict):
+    """Context-based parameters for invoking AWS Bedrock's LLM API.
+
+    This is a placeholder until support for universal LLMContext machinery is added for Bedrock.
+    """
+
+    pass
+
+
+class AWSBedrockLLMAdapter(BaseLLMAdapter[AWSBedrockLLMInvocationParams]):
     """Adapter for AWS Bedrock LLM integration with Pipecat.
 
     Provides conversion utilities for transforming Pipecat function schemas
     into AWS Bedrock's expected tool format for function calling capabilities.
     """
 
+    def get_llm_invocation_params(self, context: LLMContext) -> AWSBedrockLLMInvocationParams:
+        """Get AWS Bedrock-specific LLM invocation parameters from a universal LLM context.
+
+        This is a placeholder until support for universal LLMContext machinery is added for Bedrock.
+
+        Args:
+            context: The LLM context containing messages, tools, etc.
+
+        Returns:
+            Dictionary of parameters for invoking AWS Bedrock's LLM API.
+        """
+        raise NotImplementedError("Universal LLMContext is not yet supported for AWS Bedrock.")
+
+    def get_messages_for_logging(self, context) -> List[dict[str, Any]]:
+        """Get messages from a universal LLM context in a format ready for logging about AWS Bedrock.
+
+        Removes or truncates sensitive data like image content for safe logging.
+
+        This is a placeholder until support for universal LLMContext machinery is added for Bedrock.
+
+        Args:
+            context: The LLM context containing messages.
+
+        Returns:
+            List of messages in a format ready for logging about AWS Bedrock.
+        """
+        raise NotImplementedError("Universal LLMContext is not yet supported for AWS Bedrock.")
+
     @staticmethod
     def _to_bedrock_function_format(function: FunctionSchema) -> Dict[str, Any]:
         """Convert a function schema to Bedrock's tool format.

src/pipecat/adapters/services/gemini_adapter.py

@@ -6,20 +6,71 @@
 
 """Gemini LLM adapter for Pipecat."""
 
-from typing import Any, Dict, List, Union
+import base64
+import json
+from dataclasses import dataclass
+from typing import Any, Dict, List, Optional, TypedDict
+
+from loguru import logger
+from openai import NotGiven
 
 from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
 from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema
+from pipecat.processors.aggregators.llm_context import (
+    LLMContext,
+    LLMContextMessage,
+    LLMSpecificMessage,
+    LLMStandardMessage,
+)
+
+try:
+    from google.genai.types import (
+        Blob,
+        Content,
+        FunctionCall,
+        FunctionResponse,
+        Part,
+    )
+except ModuleNotFoundError as e:
+    logger.error(f"Exception: {e}")
+    logger.error("In order to use Google AI, you need to `pip install pipecat-ai[google]`.")
+    raise Exception(f"Missing module: {e}")
 
 
-class GeminiLLMAdapter(BaseLLMAdapter):
-    """LLM adapter for Google's Gemini service.
+class GeminiLLMInvocationParams(TypedDict):
+    """Context-based parameters for invoking Gemini LLM."""
 
-    Provides tool schema conversion functionality to transform standard tool
-    definitions into Gemini's specific function-calling format for use with
-    Gemini LLM models.
+    system_instruction: Optional[str]
+    messages: List[Content]
+    tools: List[Any] | NotGiven
+
+
+class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
+    """Gemini-specific adapter for Pipecat.
+
+    Handles:
+    - Extracting parameters for Gemini's API from a universal LLM context
+    - Converting Pipecat's standardized tools schema to Gemini's function-calling format.
+    - Extracting and sanitizing messages from the LLM context for logging with Gemini.
     """
 
+    def get_llm_invocation_params(self, context: LLMContext) -> GeminiLLMInvocationParams:
+        """Get Gemini-specific LLM invocation parameters from a universal LLM context.
+
+        Args:
+            context: The LLM context containing messages, tools, etc.
+
+        Returns:
+            Dictionary of parameters for Gemini's API.
+        """
+        messages = self._from_universal_context_messages(self._get_messages(context))
+        return {
+            "system_instruction": messages.system_instruction,
+            "messages": messages.messages,
+            # NOTE; LLMContext's tools are guaranteed to be a ToolsSchema (or NOT_GIVEN)
+            "tools": self.from_standard_tools(context.tools),
+        }
+
     def to_provider_tools_format(self, tools_schema: ToolsSchema) -> List[Dict[str, Any]]:
         """Convert tool schemas to Gemini's function-calling format.
 
@@ -39,3 +90,223 @@ class GeminiLLMAdapter(BaseLLMAdapter):
             custom_gemini_tools = tools_schema.custom_tools.get(AdapterType.GEMINI, [])
 
         return formatted_standard_tools + custom_gemini_tools
+
+    def get_messages_for_logging(self, context: LLMContext) -> List[dict[str, Any]]:
+        """Get messages from a universal LLM context in a format ready for logging about Gemini.
+
+        Removes or truncates sensitive data like image content for safe logging.
+
+        Args:
+            context: The LLM context containing messages.
+
+        Returns:
+            List of messages in a format ready for logging about Gemini.
+        """
+        # Get messages in Gemini's format
+        messages = self._from_universal_context_messages(self._get_messages(context)).messages
+
+        # Sanitize messages for logging
+        messages_for_logging = []
+        for message in messages:
+            obj = message.to_json_dict()
+            try:
+                if "parts" in obj:
+                    for part in obj["parts"]:
+                        if "inline_data" in part:
+                            part["inline_data"]["data"] = "..."
+            except Exception as e:
+                logger.debug(f"Error: {e}")
+            messages_for_logging.append(obj)
+        return messages_for_logging
+
+    def _get_messages(self, context: LLMContext) -> List[LLMContextMessage]:
+        return context.get_messages("google")
+
+    @dataclass
+    class ConvertedMessages:
+        """Container for Google-formatted messages converted from universal context."""
+
+        messages: List[Content]
+        system_instruction: Optional[str] = None
+
+    def _from_universal_context_messages(
+        self, universal_context_messages: List[LLMContextMessage]
+    ) -> ConvertedMessages:
+        """Restructures messages to ensure proper Google format and message ordering.
+
+        This method handles conversion of OpenAI-formatted messages to Google format,
+        with special handling for function calls, function responses, and system messages.
+        System messages are added back to the context as user messages when needed.
+
+        The final message order is preserved as:
+
+        1. Function calls (from model)
+        2. Function responses (from user)
+        3. Text messages (converted from system messages)
+
+        Note::
+
+            System messages are only added back when there are no regular text
+            messages in the context, ensuring proper conversation continuity
+            after function calls.
+        """
+        system_instruction = None
+        messages = []
+
+        # Process each message, preserving Google-formatted messages and converting others
+        for message in universal_context_messages:
+            if isinstance(message, LLMSpecificMessage):
+                # Assume that LLMSpecificMessage wraps a message in Google format
+                messages.append(message.message)
+                continue
+
+            # Convert standard format to Google format
+            converted = self._from_standard_message(
+                message, already_have_system_instruction=bool(system_instruction)
+            )
+            if isinstance(converted, Content):
+                # Regular (non-system) message
+                messages.append(converted)
+            else:
+                # System instruction
+                system_instruction = converted
+
+        # Check if we only have function-related messages (no regular text)
+        has_regular_messages = any(
+            len(msg.parts) == 1
+            and getattr(msg.parts[0], "text", None)
+            and not getattr(msg.parts[0], "function_call", None)
+            and not getattr(msg.parts[0], "function_response", None)
+            for msg in messages
+        )
+
+        # Add system instruction back as a user message if we only have function messages
+        if system_instruction and not has_regular_messages:
+            messages.append(Content(role="user", parts=[Part(text=system_instruction)]))
+
+        # Remove any empty messages
+        messages = [m for m in messages if m.parts]
+
+        return self.ConvertedMessages(messages=messages, system_instruction=system_instruction)
+
+    def _from_standard_message(
+        self, message: LLMStandardMessage, already_have_system_instruction: bool
+    ) -> Content | str:
+        """Convert universal context message to Google Content object.
+
+        Handles conversion of text, images, and function calls to Google's
+        format.
+        System instructions are returned as a plain string.
+
+        Args:
+            message: Message in universal context format.
+            already_have_system_instruction: Whether we already have a system instruction
+
+        Returns:
+            Content object with role and parts, or a plain string for system
+            messages.
+
+        Examples:
+            Standard text message::
+
+                {
+                    "role": "user",
+                    "content": "Hello there"
+                }
+
+            Converts to Google Content with::
+
+                Content(
+                    role="user",
+                    parts=[Part(text="Hello there")]
+                )
+
+            Standard function call message::
+
+                {
+                    "role": "assistant",
+                    "tool_calls": [
+                        {
+                            "function": {
+                                "name": "search",
+                                "arguments": '{"query": "test"}'
+                            }
+                        }
+                    ]
+                }
+
+            Converts to Google Content with::
+
+                Content(
+                    role="model",
+                    parts=[Part(function_call=FunctionCall(name="search", args={"query": "test"}))]
+                )
+        """
+        role = message["role"]
+        content = message.get("content", [])
+        if role == "system":
+            if already_have_system_instruction:
+                role = "user"  # Convert system message to user role if we already have a system instruction
+            else:
+                # System instructions are returned as plain text
+                if isinstance(content, str):
+                    return content
+                elif isinstance(content, list):
+                    # If content is a list, we assume it's a list of text parts, per the standard
+                    return " ".join(part["text"] for part in content if part.get("type") == "text")
+        elif role == "assistant":
+            role = "model"
+
+        parts = []
+        if message.get("tool_calls"):
+            for tc in message["tool_calls"]:
+                parts.append(
+                    Part(
+                        function_call=FunctionCall(
+                            name=tc["function"]["name"],
+                            args=json.loads(tc["function"]["arguments"]),
+                        )
+                    )
+                )
+        elif role == "tool":
+            role = "model"
+            try:
+                response = json.loads(message["content"])
+                if isinstance(response, dict):
+                    response_dict = response
+                else:
+                    response_dict = {"value": response}
+            except Exception as e:
+                # Response might not be JSON-deserializable.
+                # This occurs with a UserImageFrame, for example, where we get a plain "COMPLETED" string.
+                response_dict = {"value": message["content"]}
+            parts.append(
+                Part(
+                    function_response=FunctionResponse(
+                        name="tool_call_result",  # seems to work to hard-code the same name every time
+                        response=response_dict,
+                    )
+                )
+            )
+        elif isinstance(content, str):
+            parts.append(Part(text=content))
+        elif isinstance(content, list):
+            for c in content:
+                if c["type"] == "text":
+                    parts.append(Part(text=c["text"]))
+                elif c["type"] == "image_url":
+                    parts.append(
+                        Part(
+                            inline_data=Blob(
+                                mime_type="image/jpeg",
+                                data=base64.b64decode(c["image_url"]["url"].split(",")[1]),
+                            )
+                        )
+                    )
+                elif c["type"] == "input_audio":
+                    input_audio = c["input_audio"]
+                    audio_bytes = base64.b64decode(input_audio["data"])
+                    parts.append(Part(inline_data=Blob(mime_type="audio/wav", data=audio_bytes)))
+
+        message = Content(role=role, parts=parts)
+        return message

src/pipecat/adapters/services/open_ai_adapter.py

@@ -6,22 +6,63 @@
 
 """OpenAI LLM adapter for Pipecat."""
 
-from typing import List
+import copy
+import json
+from typing import Any, List, TypedDict
 
-from openai.types.chat import ChatCompletionToolParam
+from openai._types import NOT_GIVEN as OPEN_AI_NOT_GIVEN
+from openai._types import NotGiven as OpenAINotGiven
+from openai.types.chat import (
+    ChatCompletionMessageParam,
+    ChatCompletionToolChoiceOptionParam,
+    ChatCompletionToolParam,
+)
 
 from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
 from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.processors.aggregators.llm_context import (
+    LLMContext,
+    LLMContextMessage,
+    LLMContextToolChoice,
+    NotGiven,
+)
 
 
-class OpenAILLMAdapter(BaseLLMAdapter):
-    """Adapter for converting tool schemas to OpenAI's format.
+class OpenAILLMInvocationParams(TypedDict):
+    """Context-based parameters for invoking OpenAI ChatCompletion API."""
 
-    Provides conversion utilities for transforming Pipecat's standard tool
-    schemas into the format expected by OpenAI's ChatCompletion API for
-    function calling capabilities.
+    messages: List[ChatCompletionMessageParam]
+    tools: List[ChatCompletionToolParam] | OpenAINotGiven
+    tool_choice: ChatCompletionToolChoiceOptionParam | OpenAINotGiven
+
+
+class OpenAILLMAdapter(BaseLLMAdapter[OpenAILLMInvocationParams]):
+    """OpenAI-specific adapter for Pipecat.
+
+    Handles:
+
+    - Extracting parameters for OpenAI's ChatCompletion API from a universal
+      LLM context
+    - Converting Pipecat's standardized tools schema to OpenAI's function-calling format.
+    - Extracting and sanitizing messages from the LLM context for logging about OpenAI.
     """
 
+    def get_llm_invocation_params(self, context: LLMContext) -> OpenAILLMInvocationParams:
+        """Get OpenAI-specific LLM invocation parameters from a universal LLM context.
+
+        Args:
+            context: The LLM context containing messages, tools, etc.
+
+        Returns:
+            Dictionary of parameters for OpenAI's ChatCompletion API.
+        """
+        return {
+            "messages": self._from_universal_context_messages(self._get_messages(context)),
+            # NOTE; LLMContext's tools are guaranteed to be a ToolsSchema (or NOT_GIVEN)
+            "tools": self.from_standard_tools(context.tools),
+            "tool_choice": context.tool_choice,
+        }
+
     def to_provider_tools_format(self, tools_schema: ToolsSchema) -> List[ChatCompletionToolParam]:
         """Convert function schemas to OpenAI's function-calling format.
 
@@ -37,3 +78,43 @@ class OpenAILLMAdapter(BaseLLMAdapter):
             ChatCompletionToolParam(type="function", function=func.to_default_dict())
             for func in functions_schema
         ]
+
+    def get_messages_for_logging(self, context: LLMContext) -> List[dict[str, Any]]:
+        """Get messages from a universal LLM context in a format ready for logging about OpenAI.
+
+        Removes or truncates sensitive data like image content for safe logging.
+
+        Args:
+            context: The LLM context containing messages.
+
+        Returns:
+            List of messages in a format ready for logging about OpenAI.
+        """
+        msgs = []
+        for message in self._get_messages(context):
+            msg = copy.deepcopy(message)
+            if "content" in msg:
+                if isinstance(msg["content"], list):
+                    for item in msg["content"]:
+                        if item["type"] == "image_url":
+                            if item["image_url"]["url"].startswith("data:image/"):
+                                item["image_url"]["url"] = "data:image/..."
+            if "mime_type" in msg and msg["mime_type"].startswith("image/"):
+                msg["data"] = "..."
+            msgs.append(msg)
+        return json.dumps(msgs, ensure_ascii=False)
+
+    def _get_messages(self, context: LLMContext) -> List[LLMContextMessage]:
+        return context.get_messages("openai")
+
+    def _from_universal_context_messages(
+        self, messages: List[LLMContextMessage]
+    ) -> List[ChatCompletionMessageParam]:
+        # Just a pass-through: messages are already the right type
+        return messages
+
+    def _from_standard_tool_choice(
+        self, tool_choice: LLMContextToolChoice | NotGiven
+    ) -> ChatCompletionToolChoiceOptionParam | OpenAINotGiven:
+        # Just a pass-through: tool_choice is already the right type
+        return tool_choice

src/pipecat/adapters/services/open_ai_realtime_adapter.py

@@ -6,11 +6,21 @@
 
 """OpenAI Realtime LLM adapter for Pipecat."""
 
-from typing import Any, Dict, List, Union
+from typing import Any, Dict, List, TypedDict, Union
 
 from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
 from pipecat.adapters.schemas.function_schema import FunctionSchema
 from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.processors.aggregators.llm_context import LLMContext
+
+
+class OpenAIRealtimeLLMInvocationParams(TypedDict):
+    """Context-based parameters for invoking OpenAI Realtime API.
+
+    This is a placeholder until support for universal LLMContext machinery is added for OpenAI Realtime.
+    """
+
+    pass
 
 
 class OpenAIRealtimeLLMAdapter(BaseLLMAdapter):
@@ -20,6 +30,34 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter):
     OpenAI's Realtime API for function calling capabilities.
     """
 
+    def get_llm_invocation_params(self, context: LLMContext) -> OpenAIRealtimeLLMInvocationParams:
+        """Get OpenAI Realtime-specific LLM invocation parameters from a universal LLM context.
+
+        This is a placeholder until support for universal LLMContext machinery is added for OpenAI Realtime.
+
+        Args:
+            context: The LLM context containing messages, tools, etc.
+
+        Returns:
+            Dictionary of parameters for invoking OpenAI Realtime's API.
+        """
+        raise NotImplementedError("Universal LLMContext is not yet supported for OpenAI Realtime.")
+
+    def get_messages_for_logging(self, context) -> List[dict[str, Any]]:
+        """Get messages from a universal LLM context in a format ready for logging about OpenAI Realtime.
+
+        Removes or truncates sensitive data like image content for safe logging.
+
+        This is a placeholder until support for universal LLMContext machinery is added for OpenAI Realtime.
+
+        Args:
+            context: The LLM context containing messages.
+
+        Returns:
+            List of messages in a format ready for logging about OpenAI Realtime.
+        """
+        raise NotImplementedError("Universal LLMContext is not yet supported for OpenAI Realtime.")
+
     @staticmethod
     def _to_openai_realtime_function_format(function: FunctionSchema) -> Dict[str, Any]:
         """Convert a function schema to OpenAI Realtime format.

src/pipecat/frames/frames.py

@@ -27,6 +27,7 @@ from typing import (
     Tuple,
 )
 
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
 from pipecat.audio.interruptions.base_interruption_strategy import BaseInterruptionStrategy
 from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams
 from pipecat.audio.vad.vad_analyzer import VADParams
@@ -36,6 +37,7 @@ from pipecat.utils.time import nanoseconds_to_str
 from pipecat.utils.utils import obj_count, obj_id
 
 if TYPE_CHECKING:
+    from pipecat.processors.aggregators.llm_context import LLMContext, NotGiven
     from pipecat.processors.frame_processor import FrameProcessor
 
 
@@ -403,6 +405,11 @@ class OpenAILLMContextAssistantTimestampFrame(DataFrame):
     timestamp: str
 
 
+# A more universal (LLM-agnostic) name for
+# OpenAILLMContextAssistantTimestampFrame, matching LLMContext
+LLMContextAssistantTimestampFrame = OpenAILLMContextAssistantTimestampFrame
+
+
 @dataclass
 class TranscriptionMessage:
     """A message in a conversation transcript.
@@ -474,6 +481,20 @@ class TranscriptionUpdateFrame(DataFrame):
         return f"{self.name}(pts: {pts}, messages: {len(self.messages)})"
 
 
+@dataclass
+class LLMContextFrame(Frame):
+    """Frame containing a universal LLM context.
+
+    Used as a signal to LLM services to ingest the provided context and
+    generate a response based on it.
+
+    Parameters:
+        context: The LLM context containing messages, tools, and configuration.
+    """
+
+    context: "LLMContext"
+
+
 @dataclass
 class LLMMessagesFrame(DataFrame):
     """Frame containing LLM messages for chat completion.
@@ -556,7 +577,7 @@ class LLMSetToolsFrame(DataFrame):
         tools: List of tool/function definitions for the LLM.
     """
 
-    tools: List[dict]
+    tools: List[dict] | ToolsSchema | "NotGiven"
 
 
 @dataclass
@@ -1445,3 +1466,20 @@ class MixerEnableFrame(MixerControlFrame):
     """
 
     enable: bool
+
+
+@dataclass
+class ServiceSwitcherFrame(ControlFrame):
+    """A base class for frames that control ServiceSwitcher behavior."""
+
+    pass
+
+
+@dataclass
+class ManuallySwitchServiceFrame(ServiceSwitcherFrame):
+    """A frame to request a manual switch in the active service in a ServiceSwitcher.
+
+    Handled by ServiceSwitcherStrategyManual to switch the active service.
+    """
+
+    service: "FrameProcessor"

src/pipecat/pipeline/llm_switcher.py

@@ -0,0 +1,84 @@
+#
+# Copyright (c) 2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""LLM switcher for switching between different LLMs at runtime, with different switching strategies."""
+
+from typing import Any, List, Optional, Type
+
+from pipecat.pipeline.service_switcher import ServiceSwitcher, StrategyType
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.services.llm_service import LLMService
+
+
+class LLMSwitcher(ServiceSwitcher[StrategyType]):
+    """A pipeline that switches between different LLMs at runtime."""
+
+    def __init__(self, llms: List[LLMService], strategy_type: Type[StrategyType]):
+        """Initialize the service switcher with a list of LLMs and a switching strategy."""
+        super().__init__(llms, strategy_type)
+
+    @property
+    def llms(self) -> List[LLMService]:
+        """Get the list of LLMs managed by this switcher."""
+        return self.services
+
+    @property
+    def active_llm(self) -> Optional[LLMService]:
+        """Get the currently active LLM, if any."""
+        return self.strategy.active_service
+
+    async def run_inference(
+        self, context: LLMContext, system_instruction: Optional[str] = None
+    ) -> Optional[str]:
+        """Run a one-shot, out-of-band (i.e. out-of-pipeline) inference with the given LLM context, using the currently active LLM.
+
+        Args:
+            context: The LLM context containing conversation history.
+            system_instruction: Optional system instruction to guide the LLM's
+              behavior. You could also (again, optionally) provide a system
+              instruction directly in the context. If both are provided, the
+              one in the context takes precedence.
+
+        Returns:
+            The LLM's response as a string, or None if no response is generated.
+        """
+        if self.active_llm:
+            return await self.active_llm.run_inference(
+                context=context, system_instruction=system_instruction
+            )
+        return None
+
+    def register_function(
+        self,
+        function_name: Optional[str],
+        handler: Any,
+        start_callback=None,
+        *,
+        cancel_on_interruption: bool = True,
+    ):
+        """Register a function handler for LLM function calls, on all LLMs, active or not.
+
+        Args:
+            function_name: The name of the function to handle. Use None to handle
+                all function calls with a catch-all handler.
+            handler: The function handler. Should accept a single FunctionCallParams
+                parameter.
+            start_callback: Legacy callback function (deprecated). Put initialization
+                code at the top of your handler instead.
+
+                .. deprecated:: 0.0.59
+                    The `start_callback` parameter is deprecated and will be removed in a future version.
+
+            cancel_on_interruption: Whether to cancel this function call when an
+                interruption occurs. Defaults to True.
+        """
+        for llm in self.llms:
+            llm.register_function(
+                function_name=function_name,
+                handler=handler,
+                start_callback=start_callback,
+                cancel_on_interruption=cancel_on_interruption,
+            )

src/pipecat/pipeline/service_switcher.py

@@ -0,0 +1,144 @@
+#
+# Copyright (c) 2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Service switcher for switching between different services at runtime, with different switching strategies."""
+
+from typing import Any, Generic, List, Optional, Type, TypeVar
+
+from pipecat.frames.frames import Frame, ManuallySwitchServiceFrame, ServiceSwitcherFrame
+from pipecat.pipeline.parallel_pipeline import ParallelPipeline
+from pipecat.processors.filters.function_filter import FunctionFilter
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+
+
+class ServiceSwitcherStrategy:
+    """Base class for service switching strategies."""
+
+    def __init__(self, services: List[FrameProcessor]):
+        """Initialize the service switcher strategy with a list of services."""
+        self.services = services
+        self.active_service: Optional[FrameProcessor] = None
+
+    def is_active(self, service: FrameProcessor) -> bool:
+        """Determine if the given service is the currently active one.
+
+        This method should be overridden by subclasses to implement specific logic.
+
+        Args:
+            service: The service to check.
+
+        Returns:
+            True if the given service is the active one, False otherwise.
+        """
+        raise NotImplementedError("Subclasses must implement this method.")
+
+    def handle_frame(self, frame: ServiceSwitcherFrame, direction: FrameDirection):
+        """Handle a frame that controls service switching.
+
+        This method can be overridden by subclasses to implement specific logic
+        for handling frames that control service switching.
+
+        Args:
+            frame: The frame to handle.
+            direction: The direction of the frame (upstream or downstream).
+        """
+        raise NotImplementedError("Subclasses must implement this method.")
+
+
+class ServiceSwitcherStrategyManual(ServiceSwitcherStrategy):
+    """A strategy for switching between services manually.
+
+    This strategy allows the user to manually select which service is active.
+    The initial active service is the first one in the list.
+    """
+
+    def __init__(self, services: List[FrameProcessor]):
+        """Initialize the manual service switcher strategy with a list of services."""
+        super().__init__(services)
+        self.active_service = services[0] if services else None
+
+    def is_active(self, service: FrameProcessor) -> bool:
+        """Check if the given service is the currently active one.
+
+        Args:
+            service: The service to check.
+
+        Returns:
+            True if the given service is the active one, False otherwise.
+        """
+        return service == self.active_service
+
+    def handle_frame(self, frame: ServiceSwitcherFrame, direction: FrameDirection):
+        """Handle a frame that controls service switching.
+
+        Args:
+            frame: The frame to handle.
+            direction: The direction of the frame (upstream or downstream).
+        """
+        if isinstance(frame, ManuallySwitchServiceFrame):
+            self._set_active(frame.service)
+        else:
+            raise ValueError(f"Unsupported frame type: {type(frame)}")
+
+    def _set_active(self, service: FrameProcessor):
+        """Set the active service to the given one.
+
+        Args:
+            service: The service to set as active.
+        """
+        if service in self.services:
+            self.active_service = service
+        else:
+            raise ValueError(f"Service {service} is not in the list of available services.")
+
+
+StrategyType = TypeVar("StrategyType", bound=ServiceSwitcherStrategy)
+
+
+class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]):
+    """A pipeline that switches between different services at runtime."""
+
+    def __init__(self, services: List[FrameProcessor], strategy_type: Type[StrategyType]):
+        """Initialize the service switcher with a list of services and a switching strategy."""
+        strategy = strategy_type(services)
+        super().__init__(*self._make_pipeline_definitions(services, strategy))
+        self.services = services
+        self.strategy = strategy
+
+    @staticmethod
+    def _make_pipeline_definitions(
+        services: List[FrameProcessor], strategy: ServiceSwitcherStrategy
+    ) -> List[Any]:
+        pipelines = []
+        for service in services:
+            pipelines.append(ServiceSwitcher._make_pipeline_definition(service, strategy))
+        return pipelines
+
+    @staticmethod
+    def _make_pipeline_definition(
+        service: FrameProcessor, strategy: ServiceSwitcherStrategy
+    ) -> Any:
+        async def filter(frame) -> bool:
+            _ = frame
+            return strategy.is_active(service)
+
+        return [
+            FunctionFilter(filter, direction=FrameDirection.DOWNSTREAM),
+            service,
+            FunctionFilter(filter, direction=FrameDirection.UPSTREAM),
+        ]
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        """Process a frame, handling frames which affect service switching.
+
+        Args:
+            frame: The frame to process.
+            direction: The direction of the frame (upstream or downstream).
+        """
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, ServiceSwitcherFrame):
+            self.strategy.handle_frame(frame, direction)

src/pipecat/pipeline/task.py

@@ -53,8 +53,12 @@ from pipecat.utils.asyncio.task_manager import BaseTaskManager, TaskManager, Tas
 from pipecat.utils.tracing.setup import is_tracing_available
 from pipecat.utils.tracing.turn_trace_observer import TurnTraceObserver
 
-HEARTBEAT_SECONDS = 1.0
-HEARTBEAT_MONITOR_SECONDS = HEARTBEAT_SECONDS * 10
+HEARTBEAT_SECS = 1.0
+HEARTBEAT_MONITOR_SECS = HEARTBEAT_SECS * 10
+
+IDLE_TIMEOUT_SECS = 300
+
+CANCEL_TIMEOUT_SECS = 10.0
 
 
 class PipelineParams(BaseModel):
@@ -91,7 +95,7 @@ class PipelineParams(BaseModel):
     enable_heartbeats: bool = False
     enable_metrics: bool = False
     enable_usage_metrics: bool = False
-    heartbeats_period_secs: float = HEARTBEAT_SECONDS
+    heartbeats_period_secs: float = HEARTBEAT_SECS
     interruption_strategies: List[BaseInterruptionStrategy] = Field(default_factory=list)
     observers: List[BaseObserver] = Field(default_factory=list)
     report_only_initial_ttfb: bool = False
@@ -135,6 +139,7 @@ class PipelineTask(BasePipelineTask):
         params: Optional[PipelineParams] = None,
         additional_span_attributes: Optional[dict] = None,
         cancel_on_idle_timeout: bool = True,
+        cancel_timeout_secs: float = CANCEL_TIMEOUT_SECS,
         check_dangling_tasks: bool = True,
         clock: Optional[BaseClock] = None,
         conversation_id: Optional[str] = None,
@@ -148,7 +153,7 @@ class PipelineTask(BasePipelineTask):
             UserStartedSpeakingFrame,
             UserStoppedSpeakingFrame,
         ),
-        idle_timeout_secs: Optional[float] = 300,
+        idle_timeout_secs: Optional[float] = IDLE_TIMEOUT_SECS,
         observers: Optional[List[BaseObserver]] = None,
         task_manager: Optional[BaseTaskManager] = None,
     ):
@@ -161,6 +166,8 @@ class PipelineTask(BasePipelineTask):
                 OpenTelemetry conversation span attributes.
             cancel_on_idle_timeout: Whether the pipeline task should be cancelled if
                 the idle timeout is reached.
+            cancel_timeout_secs: Timeout (in seconds) to wait for cancellation to happen
+                cleanly.
             check_dangling_tasks: Whether to check for processors' tasks finishing properly.
             clock: Clock implementation for timing operations.
             conversation_id: Optional custom ID for the conversation.
@@ -178,6 +185,7 @@ class PipelineTask(BasePipelineTask):
         self._params = params or PipelineParams()
         self._additional_span_attributes = additional_span_attributes or {}
         self._cancel_on_idle_timeout = cancel_on_idle_timeout
+        self._cancel_timeout_secs = cancel_timeout_secs
         self._check_dangling_tasks = check_dangling_tasks
         self._clock = clock or SystemClock()
         self._conversation_id = conversation_id
@@ -394,12 +402,13 @@ class PipelineTask(BasePipelineTask):
             # `PipelineRunner` which will call `PipelineTask.cancel()` and
             # therefore becoming case (2).
             if self._finished or self._cancelled:
-                logger.debug(f"Pipeline task {self} has finished, cleaning up resources")
+                logger.debug(f"Pipeline task {self} is finishing cleanup...")
                 await self._cancel_tasks()
                 await self._cleanup(cleanup_pipeline)
                 if self._check_dangling_tasks:
                     self._print_dangling_tasks()
                 self._finished = True
+                logger.debug(f"Pipeline task {self} has finished")
 
     async def queue_frame(self, frame: Frame):
         """Queue a single frame to be pushed down the pipeline.
@@ -427,12 +436,13 @@ class PipelineTask(BasePipelineTask):
         if not self._cancelled:
             logger.debug(f"Cancelling pipeline task {self}")
             self._cancelled = True
+            cancel_frame = CancelFrame()
             # Make sure everything is cleaned up downstream. This is sent
             # out-of-band from the main streaming task which is what we want since
             # we want to cancel right away.
-            await self._pipeline.queue_frame(CancelFrame())
-            # Wait for CancelFrame to make it throught the pipeline.
-            await self._wait_for_pipeline_end()
+            await self._pipeline.queue_frame(cancel_frame)
+            # Wait for CancelFrame to make it through the pipeline.
+            await self._wait_for_pipeline_end(cancel_frame)
             # Only cancel the push task, we don't want to be able to process any
             # other frame after cancel. Everything else will be cancelled in
             # run().
@@ -506,9 +516,30 @@ class PipelineTask(BasePipelineTask):
             data.append(ProcessingMetricsData(processor=p.name, value=0.0))
         return MetricsFrame(data=data)
 
-    async def _wait_for_pipeline_end(self):
+    async def _wait_for_pipeline_end(self, frame: Frame):
         """Wait for the pipeline to signal completion."""
-        await self._pipeline_end_event.wait()
+
+        async def wait_for_cancel():
+            try:
+                await asyncio.wait_for(
+                    self._pipeline_end_event.wait(), timeout=self._cancel_timeout_secs
+                )
+                logger.debug(f"{self}: {frame} reached the end of the pipeline.")
+            except asyncio.TimeoutError:
+                logger.warning(
+                    f"{self}: timeout waiting for {frame} to reach the end of the pipeline (being blocked somewhere?)."
+                )
+            finally:
+                await self._call_event_handler("on_pipeline_cancelled", frame)
+
+        logger.debug(f"{self}: waiting for {frame} to reach the end of the pipeline...")
+
+        if isinstance(frame, CancelFrame):
+            await wait_for_cancel()
+        else:
+            await self._pipeline_end_event.wait()
+            logger.debug(f"{self}: {frame} reached the end of the pipeline.")
+
         self._pipeline_end_event.clear()
 
     async def _setup(self, params: PipelineTaskParams):
@@ -569,7 +600,7 @@ class PipelineTask(BasePipelineTask):
             frame = await self._push_queue.get()
             await self._pipeline.queue_frame(frame)
             if isinstance(frame, (CancelFrame, EndFrame, StopFrame)):
-                await self._wait_for_pipeline_end()
+                await self._wait_for_pipeline_end(frame)
             running = not isinstance(frame, (CancelFrame, EndFrame, StopFrame))
             cleanup_pipeline = not isinstance(frame, StopFrame)
             self._push_queue.task_done()
@@ -633,7 +664,6 @@ class PipelineTask(BasePipelineTask):
             await self._call_event_handler("on_pipeline_stopped", frame)
             self._pipeline_end_event.set()
         elif isinstance(frame, CancelFrame):
-            await self._call_event_handler("on_pipeline_cancelled", frame)
             self._pipeline_end_event.set()
         elif isinstance(frame, HeartbeatFrame):
             await self._heartbeat_queue.put(frame)
@@ -655,7 +685,7 @@ class PipelineTask(BasePipelineTask):
         the time that a heartbeat frame takes to processes, that is how long it
         takes for the heartbeat frame to traverse all the pipeline.
         """
-        wait_time = HEARTBEAT_MONITOR_SECONDS
+        wait_time = HEARTBEAT_MONITOR_SECS
         while True:
             try:
                 frame = await asyncio.wait_for(self._heartbeat_queue.get(), timeout=wait_time)

src/pipecat/processors/aggregators/llm_context.py

@@ -0,0 +1,277 @@
+#
+# Copyright (c) 2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Universal LLM context management for LLM services in Pipecat.
+
+Context contents are represented in a universal format (based on OpenAI)
+that supports a union of known Pipecat LLM service functionality.
+
+Whenever an LLM service needs to access context, it does a just-in-time
+translation from this universal context into whatever format it needs, using a
+service-specific adapter.
+"""
+
+import base64
+import io
+from dataclasses import dataclass
+from typing import Any, List, Optional, TypeAlias, Union
+
+from loguru import logger
+from openai._types import NOT_GIVEN as OPEN_AI_NOT_GIVEN
+from openai._types import NotGiven as OpenAINotGiven
+from openai.types.chat import (
+    ChatCompletionMessageParam,
+    ChatCompletionToolChoiceOptionParam,
+)
+from PIL import Image
+
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.frames.frames import AudioRawFrame
+
+# "Re-export" types from OpenAI that we're using as universal context types.
+# NOTE: if universal message types need to someday diverge from OpenAI's, we
+# should consider managing our own definitions. But we should do so carefully,
+# as the OpenAI messages are somewhat of a standard and we want to continue
+# supporting them.
+LLMStandardMessage = ChatCompletionMessageParam
+LLMContextToolChoice = ChatCompletionToolChoiceOptionParam
+NOT_GIVEN = OPEN_AI_NOT_GIVEN
+NotGiven = OpenAINotGiven
+
+
+@dataclass
+class LLMSpecificMessage:
+    """A container for a context message that is specific to a particular LLM service.
+
+    Enables the use of service-specific message types while maintaining
+    compatibility with the universal LLM context format.
+    """
+
+    llm: str
+    message: Any
+
+
+LLMContextMessage: TypeAlias = Union[LLMStandardMessage, LLMSpecificMessage]
+
+
+class LLMContext:
+    """Manages conversation context for LLM interactions.
+
+    Handles message history, tool definitions, tool choices, and multimedia
+    content for LLM conversations. Provides methods for message manipulation,
+    and content formatting.
+    """
+
+    def __init__(
+        self,
+        messages: Optional[List[LLMContextMessage]] = None,
+        tools: ToolsSchema | NotGiven = NOT_GIVEN,
+        tool_choice: LLMContextToolChoice | NotGiven = NOT_GIVEN,
+    ):
+        """Initialize the LLM context.
+
+        Args:
+            messages: Initial list of conversation messages.
+            tools: Available tools for the LLM to use.
+            tool_choice: Tool selection strategy for the LLM.
+        """
+        self._messages: List[LLMContextMessage] = messages if messages else []
+        self._tools: ToolsSchema | NotGiven = LLMContext._normalize_and_validate_tools(tools)
+        self._tool_choice: LLMContextToolChoice | NotGiven = tool_choice
+
+    def get_messages(self, llm_specific_filter: Optional[str] = None) -> List[LLMContextMessage]:
+        """Get the current messages list.
+
+        Args:
+            llm_specific_filter: Optional filter to return LLM-specific
+                messages for the given LLM, in addition to the standard
+                messages. If messages end up being filtered, an error will be
+                logged.
+
+        Returns:
+            List of conversation messages.
+        """
+        if llm_specific_filter is None:
+            return self._messages
+        filtered_messages = [
+            msg
+            for msg in self._messages
+            if not isinstance(msg, LLMSpecificMessage) or msg.llm == llm_specific_filter
+        ]
+        if len(filtered_messages) < len(self._messages):
+            logger.error(
+                f"Attempted to use incompatible LLMSpecificMessages with LLM '{llm_specific_filter}'."
+            )
+        return filtered_messages
+
+    @property
+    def tools(self) -> ToolsSchema | NotGiven:
+        """Get the tools list.
+
+        Returns:
+            Tools list.
+        """
+        return self._tools
+
+    @property
+    def tool_choice(self) -> LLMContextToolChoice | NotGiven:
+        """Get the current tool choice setting.
+
+        Returns:
+            The tool choice configuration.
+        """
+        return self._tool_choice
+
+    def add_message(self, message: LLMContextMessage):
+        """Add a single message to the context.
+
+        Args:
+            message: The message to add to the conversation history.
+        """
+        self._messages.append(message)
+
+    def add_messages(self, messages: List[LLMContextMessage]):
+        """Add multiple messages to the context.
+
+        Args:
+            messages: List of messages to add to the conversation history.
+        """
+        self._messages.extend(messages)
+
+    def set_messages(self, messages: List[LLMContextMessage]):
+        """Replace all messages in the context.
+
+        Args:
+            messages: New list of messages to replace the current history.
+        """
+        self._messages[:] = messages
+
+    def set_tools(self, tools: ToolsSchema | NotGiven = NOT_GIVEN):
+        """Set the available tools for the LLM.
+
+        Args:
+            tools: A ToolsSchema or NOT_GIVEN to disable tools.
+        """
+        self._tools = LLMContext._normalize_and_validate_tools(tools)
+
+    def set_tool_choice(self, tool_choice: LLMContextToolChoice | NotGiven):
+        """Set the tool choice configuration.
+
+        Args:
+            tool_choice: Tool selection strategy for the LLM.
+        """
+        self._tool_choice = tool_choice
+
+    def add_image_frame_message(
+        self, *, format: str, size: tuple[int, int], image: bytes, text: str = None
+    ):
+        """Add a message containing an image frame.
+
+        Args:
+            format: Image format (e.g., 'RGB', 'RGBA').
+            size: Image dimensions as (width, height) tuple.
+            image: Raw image bytes.
+            text: Optional text to include with the image.
+        """
+        buffer = io.BytesIO()
+        Image.frombytes(format, size, image).save(buffer, format="JPEG")
+        encoded_image = base64.b64encode(buffer.getvalue()).decode("utf-8")
+
+        content = []
+        if text:
+            content.append({"type": "text", "text": text})
+        content.append(
+            {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{encoded_image}"}},
+        )
+        self.add_message({"role": "user", "content": content})
+
+    def add_audio_frames_message(
+        self, *, audio_frames: list[AudioRawFrame], text: str = "Audio follows"
+    ):
+        """Add a message containing audio frames.
+
+        Args:
+            audio_frames: List of audio frame objects to include.
+            text: Optional text to include with the audio.
+        """
+        if not audio_frames:
+            return
+
+        sample_rate = audio_frames[0].sample_rate
+        num_channels = audio_frames[0].num_channels
+
+        content = []
+        content.append({"type": "text", "text": text})
+        data = b"".join(frame.audio for frame in audio_frames)
+        data = bytes(
+            self._create_wav_header(
+                sample_rate,
+                num_channels,
+                16,
+                len(data),
+            )
+            + data
+        )
+        encoded_audio = base64.b64encode(data).decode("utf-8")
+        content.append(
+            {
+                "type": "input_audio",
+                "input_audio": {"data": encoded_audio, "format": "wav"},
+            }
+        )
+        self.add_message({"role": "user", "content": content})
+
+    def _create_wav_header(self, sample_rate, num_channels, bits_per_sample, data_size):
+        """Create a WAV file header for audio data.
+
+        Args:
+            sample_rate: Audio sample rate in Hz.
+            num_channels: Number of audio channels.
+            bits_per_sample: Bits per audio sample.
+            data_size: Size of audio data in bytes.
+
+        Returns:
+            WAV header as a bytearray.
+        """
+        # RIFF chunk descriptor
+        header = bytearray()
+        header.extend(b"RIFF")  # ChunkID
+        header.extend((data_size + 36).to_bytes(4, "little"))  # ChunkSize: total size - 8
+        header.extend(b"WAVE")  # Format
+        # "fmt " sub-chunk
+        header.extend(b"fmt ")  # Subchunk1ID
+        header.extend((16).to_bytes(4, "little"))  # Subchunk1Size (16 for PCM)
+        header.extend((1).to_bytes(2, "little"))  # AudioFormat (1 for PCM)
+        header.extend(num_channels.to_bytes(2, "little"))  # NumChannels
+        header.extend(sample_rate.to_bytes(4, "little"))  # SampleRate
+        # Calculate byte rate and block align
+        byte_rate = sample_rate * num_channels * (bits_per_sample // 8)
+        block_align = num_channels * (bits_per_sample // 8)
+        header.extend(byte_rate.to_bytes(4, "little"))  # ByteRate
+        header.extend(block_align.to_bytes(2, "little"))  # BlockAlign
+        header.extend(bits_per_sample.to_bytes(2, "little"))  # BitsPerSample
+        # "data" sub-chunk
+        header.extend(b"data")  # Subchunk2ID
+        header.extend(data_size.to_bytes(4, "little"))  # Subchunk2Size
+        return header
+
+    @staticmethod
+    def _normalize_and_validate_tools(tools: ToolsSchema | NotGiven) -> ToolsSchema | NotGiven:
+        """Normalize and validate the given tools.
+
+        Raises:
+            TypeError: If tools are not a ToolsSchema or NotGiven.
+        """
+        if isinstance(tools, ToolsSchema):
+            if not tools.standard_tools and not tools.custom_tools:
+                return NOT_GIVEN
+            return tools
+        elif tools is NOT_GIVEN:
+            return NOT_GIVEN
+        else:
+            raise TypeError(
+                f"In LLMContext, tools must be a ToolsSchema object or NOT_GIVEN. Got type: {type(tools)}",
+            )

src/pipecat/processors/aggregators/llm_response_universal.py

@@ -0,0 +1,829 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""LLM response aggregators for handling conversation context and message aggregation.
+
+This module provides aggregators that process and accumulate LLM responses, user inputs,
+and conversation context. These aggregators handle the flow between speech-to-text,
+LLM processing, and text-to-speech components in conversational AI pipelines.
+"""
+
+import asyncio
+import json
+from dataclasses import dataclass
+from typing import Any, Dict, List, Literal, Optional, Set
+
+from loguru import logger
+
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.audio.interruptions.base_interruption_strategy import BaseInterruptionStrategy
+from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams
+from pipecat.audio.vad.vad_analyzer import VADParams
+from pipecat.frames.frames import (
+    BotInterruptionFrame,
+    BotStartedSpeakingFrame,
+    BotStoppedSpeakingFrame,
+    CancelFrame,
+    EmulateUserStartedSpeakingFrame,
+    EmulateUserStoppedSpeakingFrame,
+    EndFrame,
+    Frame,
+    FunctionCallCancelFrame,
+    FunctionCallInProgressFrame,
+    FunctionCallResultFrame,
+    FunctionCallsStartedFrame,
+    InputAudioRawFrame,
+    InterimTranscriptionFrame,
+    LLMContextAssistantTimestampFrame,
+    LLMContextFrame,
+    LLMFullResponseEndFrame,
+    LLMFullResponseStartFrame,
+    LLMMessagesAppendFrame,
+    LLMMessagesUpdateFrame,
+    LLMSetToolChoiceFrame,
+    LLMSetToolsFrame,
+    SpeechControlParamsFrame,
+    StartFrame,
+    StartInterruptionFrame,
+    TextFrame,
+    TranscriptionFrame,
+    UserImageRawFrame,
+    UserStartedSpeakingFrame,
+    UserStoppedSpeakingFrame,
+)
+from pipecat.processors.aggregators.llm_context import (
+    LLMContext,
+    LLMContextMessage,
+    LLMSpecificMessage,
+    NotGiven,
+)
+from pipecat.processors.aggregators.llm_response import (
+    LLMAssistantAggregatorParams,
+    LLMUserAggregatorParams,
+)
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.utils.time import time_now_iso8601
+
+
+class LLMContextAggregator(FrameProcessor):
+    """Base LLM aggregator that uses an LLMContext for conversation storage.
+
+    This aggregator maintains conversation state using an LLMContext and
+    pushes LLMContextFrame objects as aggregation frames. It provides
+    common functionality for context-based conversation management.
+    """
+
+    def __init__(self, *, context: LLMContext, role: str, **kwargs):
+        """Initialize the context response aggregator.
+
+        Args:
+            context: The LLM context to use for conversation storage.
+            role: The role this aggregator represents (e.g. "user", "assistant").
+            **kwargs: Additional arguments passed to parent class.
+        """
+        super().__init__(**kwargs)
+        self._context = context
+        self._role = role
+
+        self._aggregation: str = ""
+
+    @property
+    def messages(self) -> List[LLMContextMessage]:
+        """Get messages from the LLM context.
+
+        Returns:
+            List of message dictionaries from the context.
+        """
+        return self._context.get_messages()
+
+    @property
+    def role(self) -> str:
+        """Get the role for this aggregator.
+
+        Returns:
+            The role string for this aggregator.
+        """
+        return self._role
+
+    @property
+    def context(self):
+        """Get the LLM context.
+
+        Returns:
+            The LLMContext instance used by this aggregator.
+        """
+        return self._context
+
+    def get_context_frame(self) -> LLMContextFrame:
+        """Create a context frame with the current context.
+
+        Returns:
+            LLMContextFrame containing the current context.
+        """
+        return LLMContextFrame(context=self._context)
+
+    async def push_context_frame(self, direction: FrameDirection = FrameDirection.DOWNSTREAM):
+        """Push a context frame in the specified direction.
+
+        Args:
+            direction: The direction to push the frame (upstream or downstream).
+        """
+        frame = self.get_context_frame()
+        await self.push_frame(frame, direction)
+
+    def add_messages(self, messages):
+        """Add messages to the context.
+
+        Args:
+            messages: Messages to add to the conversation context.
+        """
+        self._context.add_messages(messages)
+
+    def set_messages(self, messages):
+        """Set the context messages.
+
+        Args:
+            messages: Messages to replace the current context messages.
+        """
+        self._context.set_messages(messages)
+
+    def set_tools(self, tools: ToolsSchema | NotGiven):
+        """Set tools in the context.
+
+        Args:
+            tools: List of tool definitions to set in the context.
+        """
+        self._context.set_tools(tools)
+
+    def set_tool_choice(self, tool_choice: Literal["none", "auto", "required"] | dict):
+        """Set tool choice in the context.
+
+        Args:
+            tool_choice: Tool choice configuration for the context.
+        """
+        self._context.set_tool_choice(tool_choice)
+
+    async def reset(self):
+        """Reset the aggregation state."""
+        self._aggregation = ""
+
+
+class LLMUserAggregator(LLMContextAggregator):
+    """User LLM aggregator that processes speech-to-text transcriptions.
+
+    This aggregator handles the complex logic of aggregating user speech transcriptions
+    from STT services. It manages multiple scenarios including:
+
+    - Transcriptions received between VAD events
+    - Transcriptions received outside VAD events
+    - Interim vs final transcriptions
+    - User interruptions during bot speech
+    - Emulated VAD for whispered or short utterances
+
+    The aggregator uses timeouts to handle cases where transcriptions arrive
+    after VAD events or when no VAD is available.
+    """
+
+    def __init__(
+        self,
+        context: LLMContext,
+        *,
+        params: Optional[LLMUserAggregatorParams] = None,
+        **kwargs,
+    ):
+        """Initialize the user context aggregator.
+
+        Args:
+            context: The LLM context for conversation storage.
+            params: Configuration parameters for aggregation behavior.
+            **kwargs: Additional arguments. Supports deprecated 'aggregation_timeout'.
+        """
+        super().__init__(context=context, role="user", **kwargs)
+        self._params = params or LLMUserAggregatorParams()
+        self._vad_params: Optional[VADParams] = None
+        self._turn_params: Optional[SmartTurnParams] = None
+
+        if "aggregation_timeout" in kwargs:
+            import warnings
+
+            with warnings.catch_warnings():
+                warnings.simplefilter("always")
+                warnings.warn(
+                    "Parameter 'aggregation_timeout' is deprecated, use 'params' instead.",
+                    DeprecationWarning,
+                )
+
+            self._params.aggregation_timeout = kwargs["aggregation_timeout"]
+
+        self._user_speaking = False
+        self._bot_speaking = False
+        self._was_bot_speaking = False
+        self._emulating_vad = False
+        self._seen_interim_results = False
+        self._waiting_for_aggregation = False
+
+        self._aggregation_event = asyncio.Event()
+        self._aggregation_task = None
+
+    async def reset(self):
+        """Reset the aggregation state and interruption strategies."""
+        await super().reset()
+        self._was_bot_speaking = False
+        self._seen_interim_results = False
+        self._waiting_for_aggregation = False
+        [await s.reset() for s in self._interruption_strategies]
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        """Process frames for user speech aggregation and context management.
+
+        Args:
+            frame: The frame to process.
+            direction: The direction of frame flow in the pipeline.
+        """
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, StartFrame):
+            # Push StartFrame before start(), because we want StartFrame to be
+            # processed by every processor before any other frame is processed.
+            await self.push_frame(frame, direction)
+            await self._start(frame)
+        elif isinstance(frame, EndFrame):
+            # Push EndFrame before stop(), because stop() waits on the task to
+            # finish and the task finishes when EndFrame is processed.
+            await self.push_frame(frame, direction)
+            await self._stop(frame)
+        elif isinstance(frame, CancelFrame):
+            await self._cancel(frame)
+            await self.push_frame(frame, direction)
+        elif isinstance(frame, InputAudioRawFrame):
+            await self._handle_input_audio(frame)
+            await self.push_frame(frame, direction)
+        elif isinstance(frame, UserStartedSpeakingFrame):
+            await self._handle_user_started_speaking(frame)
+            await self.push_frame(frame, direction)
+        elif isinstance(frame, UserStoppedSpeakingFrame):
+            await self._handle_user_stopped_speaking(frame)
+            await self.push_frame(frame, direction)
+        elif isinstance(frame, BotStartedSpeakingFrame):
+            await self._handle_bot_started_speaking(frame)
+            await self.push_frame(frame, direction)
+        elif isinstance(frame, BotStoppedSpeakingFrame):
+            await self._handle_bot_stopped_speaking(frame)
+            await self.push_frame(frame, direction)
+        elif isinstance(frame, TranscriptionFrame):
+            await self._handle_transcription(frame)
+        elif isinstance(frame, InterimTranscriptionFrame):
+            await self._handle_interim_transcription(frame)
+        elif isinstance(frame, LLMMessagesAppendFrame):
+            await self._handle_llm_messages_append(frame)
+        elif isinstance(frame, LLMMessagesUpdateFrame):
+            await self._handle_llm_messages_update(frame)
+        elif isinstance(frame, LLMSetToolsFrame):
+            self.set_tools(frame.tools)
+        elif isinstance(frame, LLMSetToolChoiceFrame):
+            self.set_tool_choice(frame.tool_choice)
+        elif isinstance(frame, SpeechControlParamsFrame):
+            self._vad_params = frame.vad_params
+            self._turn_params = frame.turn_params
+            await self.push_frame(frame, direction)
+        else:
+            await self.push_frame(frame, direction)
+
+    async def _process_aggregation(self):
+        """Process the current aggregation and push it downstream."""
+        aggregation = self._aggregation
+        await self.reset()
+        self._context.add_message({"role": self.role, "content": aggregation})
+        frame = LLMContextFrame(self._context)
+        await self.push_frame(frame)
+
+    async def _push_aggregation(self):
+        """Push the current aggregation based on interruption strategies and conditions."""
+        if len(self._aggregation) > 0:
+            if self.interruption_strategies and self._bot_speaking:
+                should_interrupt = await self._should_interrupt_based_on_strategies()
+
+                if should_interrupt:
+                    logger.debug(
+                        "Interruption conditions met - pushing BotInterruptionFrame and aggregation"
+                    )
+                    await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM)
+                    await self._process_aggregation()
+                else:
+                    logger.debug("Interruption conditions not met - not pushing aggregation")
+                    # Don't process aggregation, just reset it
+                    await self.reset()
+            else:
+                # No interruption config - normal behavior (always push aggregation)
+                await self._process_aggregation()
+        # Handles the case where both the user and the bot are not speaking,
+        # and the bot was previously speaking before the user interruption.
+        # Normally, when the user stops speaking, new text is expected,
+        # which triggers the bot to respond. However, if no new text
+        # is received, this safeguard ensures
+        # the bot doesn't hang indefinitely while waiting to speak again.
+        elif not self._seen_interim_results and self._was_bot_speaking and not self._bot_speaking:
+            logger.warning("User stopped speaking but no new aggregation received.")
+            # Resetting it so we don't trigger this twice
+            self._was_bot_speaking = False
+            # TODO: we are not enabling this for now, due to some STT services which can take as long as 2 seconds two return a transcription
+            # So we need more tests and probably make this feature configurable, disabled it by default.
+            # We are just pushing the same previous context to be processed again in this case
+            # await self.push_frame(LLMContextFrame(self._context))
+
+    async def _should_interrupt_based_on_strategies(self) -> bool:
+        """Check if interruption should occur based on configured strategies.
+
+        Returns:
+            True if any interruption strategy indicates interruption should occur.
+        """
+
+        async def should_interrupt(strategy: BaseInterruptionStrategy):
+            await strategy.append_text(self._aggregation)
+            return await strategy.should_interrupt()
+
+        return any([await should_interrupt(s) for s in self._interruption_strategies])
+
+    async def _start(self, frame: StartFrame):
+        self._create_aggregation_task()
+
+    async def _stop(self, frame: EndFrame):
+        await self._cancel_aggregation_task()
+
+    async def _cancel(self, frame: CancelFrame):
+        await self._cancel_aggregation_task()
+
+    async def _handle_llm_messages_append(self, frame: LLMMessagesAppendFrame):
+        self.add_messages(frame.messages)
+        if frame.run_llm:
+            await self.push_context_frame()
+
+    async def _handle_llm_messages_update(self, frame: LLMMessagesUpdateFrame):
+        self.set_messages(frame.messages)
+        if frame.run_llm:
+            await self.push_context_frame()
+
+    async def _handle_input_audio(self, frame: InputAudioRawFrame):
+        for s in self.interruption_strategies:
+            await s.append_audio(frame.audio, frame.sample_rate)
+
+    async def _handle_user_started_speaking(self, frame: UserStartedSpeakingFrame):
+        self._user_speaking = True
+        self._waiting_for_aggregation = True
+        self._was_bot_speaking = self._bot_speaking
+
+        # If we get a non-emulated UserStartedSpeakingFrame but we are in the
+        # middle of emulating VAD, let's stop emulating VAD (i.e. don't send the
+        # EmulateUserStoppedSpeakingFrame).
+        if not frame.emulated and self._emulating_vad:
+            self._emulating_vad = False
+
+    async def _handle_user_stopped_speaking(self, _: UserStoppedSpeakingFrame):
+        self._user_speaking = False
+        # We just stopped speaking. Let's see if there's some aggregation to
+        # push. If the last thing we saw is an interim transcription, let's wait
+        # pushing the aggregation as we will probably get a final transcription.
+        if len(self._aggregation) > 0:
+            if not self._seen_interim_results:
+                await self._push_aggregation()
+        # Handles the case where both the user and the bot are not speaking,
+        # and the bot was previously speaking before the user interruption.
+        # So in this case we are resetting the aggregation timer
+        elif not self._seen_interim_results and self._was_bot_speaking and not self._bot_speaking:
+            # Reset aggregation timer.
+            self._aggregation_event.set()
+
+    async def _handle_bot_started_speaking(self, _: BotStartedSpeakingFrame):
+        self._bot_speaking = True
+
+    async def _handle_bot_stopped_speaking(self, _: BotStoppedSpeakingFrame):
+        self._bot_speaking = False
+
+    async def _handle_transcription(self, frame: TranscriptionFrame):
+        text = frame.text
+
+        # Make sure we really have some text.
+        if not text.strip():
+            return
+
+        self._aggregation += f" {text}" if self._aggregation else text
+        # We just got a final result, so let's reset interim results.
+        self._seen_interim_results = False
+        # Reset aggregation timer.
+        self._aggregation_event.set()
+
+    async def _handle_interim_transcription(self, _: InterimTranscriptionFrame):
+        self._seen_interim_results = True
+
+    def _create_aggregation_task(self):
+        if not self._aggregation_task:
+            self._aggregation_task = self.create_task(self._aggregation_task_handler())
+
+    async def _cancel_aggregation_task(self):
+        if self._aggregation_task:
+            await self.cancel_task(self._aggregation_task)
+            self._aggregation_task = None
+
+    async def _aggregation_task_handler(self):
+        while True:
+            try:
+                # The _aggregation_task_handler handles two distinct timeout scenarios:
+                #
+                # 1. When emulating_vad=True: Wait for emulated VAD timeout before
+                #    pushing aggregation (simulating VAD behavior when no actual VAD
+                #    detection occurred).
+                #
+                # 2. When emulating_vad=False: Use aggregation_timeout as a buffer
+                #    to wait for potential late-arriving transcription frames after
+                #    a real VAD event.
+                #
+                # For emulated VAD scenarios, the timeout strategy depends on whether
+                # a turn analyzer is configured:
+                #
+                # - WITH turn analyzer: Use turn_emulated_vad_timeout parameter because
+                #   the VAD's stop_secs is set very low (e.g. 0.2s) for rapid speech
+                #   chunking to feed the turn analyzer. This low value is too fast
+                #   for emulated VAD scenarios where we need to allow users time to
+                #   finish speaking (e.g. 0.8s).
+                #
+                # - WITHOUT turn analyzer: Use VAD's stop_secs directly to maintain
+                #   consistent user experience between real VAD detection and
+                #   emulated VAD scenarios.
+                if not self._emulating_vad:
+                    timeout = self._params.aggregation_timeout
+                elif self._turn_params:
+                    timeout = self._params.turn_emulated_vad_timeout
+                else:
+                    # Use VAD stop_secs when no turn analyzer is present, fallback if no VAD params
+                    timeout = (
+                        self._vad_params.stop_secs
+                        if self._vad_params
+                        else self._params.turn_emulated_vad_timeout
+                    )
+                await asyncio.wait_for(self._aggregation_event.wait(), timeout=timeout)
+                await self._maybe_emulate_user_speaking()
+            except asyncio.TimeoutError:
+                if not self._user_speaking:
+                    await self._push_aggregation()
+
+                # If we are emulating VAD we still need to send the user stopped
+                # speaking frame.
+                if self._emulating_vad:
+                    await self.push_frame(
+                        EmulateUserStoppedSpeakingFrame(), FrameDirection.UPSTREAM
+                    )
+                    self._emulating_vad = False
+            finally:
+                self._aggregation_event.clear()
+
+    async def _maybe_emulate_user_speaking(self):
+        """Maybe emulate user speaking based on transcription.
+
+        Emulate user speaking if we got a transcription but it was not
+        detected by VAD. Behavior when bot is speaking depends on the
+        enable_emulated_vad_interruptions parameter.
+        """
+        # Check if we received a transcription but VAD was not able to detect
+        # voice (e.g. when you whisper a short utterance). In that case, we need
+        # to emulate VAD (i.e. user start/stopped speaking), but we do it only
+        # if the bot is not speaking. If the bot is speaking and we really have
+        # a short utterance we don't really want to interrupt the bot.
+        if (
+            not self._user_speaking
+            and not self._waiting_for_aggregation
+            and len(self._aggregation) > 0
+        ):
+            if self._bot_speaking and not self._params.enable_emulated_vad_interruptions:
+                # If emulated VAD interruptions are disabled and bot is speaking, ignore
+                logger.debug("Ignoring user speaking emulation, bot is speaking.")
+                await self.reset()
+            else:
+                # Either bot is not speaking, or emulated VAD interruptions are enabled
+                # - trigger user speaking emulation.
+                await self.push_frame(EmulateUserStartedSpeakingFrame(), FrameDirection.UPSTREAM)
+                self._emulating_vad = True
+
+
+class LLMAssistantAggregator(LLMContextAggregator):
+    """Assistant LLM aggregator that processes bot responses and function calls.
+
+    This aggregator handles the complex logic of processing assistant responses including:
+
+    - Text frame aggregation between response start/end markers
+    - Function call lifecycle management
+    - Context updates with timestamps
+    - Tool execution and result handling
+    - Interruption handling during responses
+
+    The aggregator manages function calls in progress and coordinates between
+    text generation and tool execution phases of LLM responses.
+    """
+
+    def __init__(
+        self,
+        context: LLMContext,
+        *,
+        params: Optional[LLMAssistantAggregatorParams] = None,
+        **kwargs,
+    ):
+        """Initialize the assistant context aggregator.
+
+        Args:
+            context: The OpenAI LLM context for conversation storage.
+            params: Configuration parameters for aggregation behavior.
+            **kwargs: Additional arguments. Supports deprecated 'expect_stripped_words'.
+        """
+        super().__init__(context=context, role="assistant", **kwargs)
+        self._params = params or LLMAssistantAggregatorParams()
+
+        if "expect_stripped_words" in kwargs:
+            import warnings
+
+            with warnings.catch_warnings():
+                warnings.simplefilter("always")
+                warnings.warn(
+                    "Parameter 'expect_stripped_words' is deprecated, use 'params' instead.",
+                    DeprecationWarning,
+                )
+
+            self._params.expect_stripped_words = kwargs["expect_stripped_words"]
+
+        self._started = 0
+        self._function_calls_in_progress: Dict[str, Optional[FunctionCallInProgressFrame]] = {}
+        self._context_updated_tasks: Set[asyncio.Task] = set()
+
+    @property
+    def has_function_calls_in_progress(self) -> bool:
+        """Check if there are any function calls currently in progress.
+
+        Returns:
+            True if function calls are in progress, False otherwise.
+        """
+        return bool(self._function_calls_in_progress)
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        """Process frames for assistant response aggregation and function call management.
+
+        Args:
+            frame: The frame to process.
+            direction: The direction of frame flow in the pipeline.
+        """
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, StartInterruptionFrame):
+            await self._handle_interruptions(frame)
+            await self.push_frame(frame, direction)
+        elif isinstance(frame, LLMFullResponseStartFrame):
+            await self._handle_llm_start(frame)
+        elif isinstance(frame, LLMFullResponseEndFrame):
+            await self._handle_llm_end(frame)
+        elif isinstance(frame, TextFrame):
+            await self._handle_text(frame)
+        elif isinstance(frame, LLMMessagesAppendFrame):
+            await self._handle_llm_messages_append(frame)
+        elif isinstance(frame, LLMMessagesUpdateFrame):
+            await self._handle_llm_messages_update(frame)
+        elif isinstance(frame, LLMSetToolsFrame):
+            self.set_tools(frame.tools)
+        elif isinstance(frame, LLMSetToolChoiceFrame):
+            self.set_tool_choice(frame.tool_choice)
+        elif isinstance(frame, FunctionCallsStartedFrame):
+            await self._handle_function_calls_started(frame)
+        elif isinstance(frame, FunctionCallInProgressFrame):
+            await self._handle_function_call_in_progress(frame)
+        elif isinstance(frame, FunctionCallResultFrame):
+            await self._handle_function_call_result(frame)
+        elif isinstance(frame, FunctionCallCancelFrame):
+            await self._handle_function_call_cancel(frame)
+        elif isinstance(frame, UserImageRawFrame) and frame.request and frame.request.tool_call_id:
+            await self._handle_user_image_frame(frame)
+        elif isinstance(frame, BotStoppedSpeakingFrame):
+            await self._push_aggregation()
+            await self.push_frame(frame, direction)
+        else:
+            await self.push_frame(frame, direction)
+
+    async def _push_aggregation(self):
+        """Push the current assistant aggregation with timestamp."""
+        if not self._aggregation:
+            return
+
+        aggregation = self._aggregation.strip()
+        await self.reset()
+
+        if aggregation:
+            self._context.add_message({"role": "assistant", "content": aggregation})
+
+        # Push context frame
+        await self.push_context_frame()
+
+        # Push timestamp frame with current time
+        timestamp_frame = LLMContextAssistantTimestampFrame(timestamp=time_now_iso8601())
+        await self.push_frame(timestamp_frame)
+
+    async def _handle_llm_messages_append(self, frame: LLMMessagesAppendFrame):
+        self.add_messages(frame.messages)
+        if frame.run_llm:
+            await self.push_context_frame(FrameDirection.UPSTREAM)
+
+    async def _handle_llm_messages_update(self, frame: LLMMessagesUpdateFrame):
+        self.set_messages(frame.messages)
+        if frame.run_llm:
+            await self.push_context_frame(FrameDirection.UPSTREAM)
+
+    async def _handle_interruptions(self, frame: StartInterruptionFrame):
+        await self._push_aggregation()
+        self._started = 0
+        await self.reset()
+
+    async def _handle_function_calls_started(self, frame: FunctionCallsStartedFrame):
+        function_names = [f"{f.function_name}:{f.tool_call_id}" for f in frame.function_calls]
+        logger.debug(f"{self} FunctionCallsStartedFrame: {function_names}")
+        for function_call in frame.function_calls:
+            self._function_calls_in_progress[function_call.tool_call_id] = None
+
+    async def _handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame):
+        logger.debug(
+            f"{self} FunctionCallInProgressFrame: [{frame.function_name}:{frame.tool_call_id}]"
+        )
+
+        # Update context with the in-progress function call
+        self._context.add_message(
+            {
+                "role": "assistant",
+                "tool_calls": [
+                    {
+                        "id": frame.tool_call_id,
+                        "function": {
+                            "name": frame.function_name,
+                            "arguments": json.dumps(frame.arguments),
+                        },
+                        "type": "function",
+                    }
+                ],
+            }
+        )
+        self._context.add_message(
+            {
+                "role": "tool",
+                "content": "IN_PROGRESS",
+                "tool_call_id": frame.tool_call_id,
+            }
+        )
+
+        self._function_calls_in_progress[frame.tool_call_id] = frame
+
+    async def _handle_function_call_result(self, frame: FunctionCallResultFrame):
+        logger.debug(
+            f"{self} FunctionCallResultFrame: [{frame.function_name}:{frame.tool_call_id}]"
+        )
+        if frame.tool_call_id not in self._function_calls_in_progress:
+            logger.warning(
+                f"FunctionCallResultFrame tool_call_id [{frame.tool_call_id}] is not running"
+            )
+            return
+
+        del self._function_calls_in_progress[frame.tool_call_id]
+
+        properties = frame.properties
+
+        # Update context with the function call result
+        if frame.result:
+            result = json.dumps(frame.result)
+            self._update_function_call_result(frame.function_name, frame.tool_call_id, result)
+        else:
+            self._update_function_call_result(frame.function_name, frame.tool_call_id, "COMPLETED")
+
+        run_llm = False
+
+        # Run inference if the function call result requires it.
+        if frame.result:
+            if properties and properties.run_llm is not None:
+                # If the tool call result has a run_llm property, use it.
+                run_llm = properties.run_llm
+            elif frame.run_llm is not None:
+                # If the frame is indicating we should run the LLM, do it.
+                run_llm = frame.run_llm
+            else:
+                # If this is the last function call in progress, run the LLM.
+                run_llm = not bool(self._function_calls_in_progress)
+
+        if run_llm:
+            await self.push_context_frame(FrameDirection.UPSTREAM)
+
+        # Call the `on_context_updated` callback once the function call result
+        # is added to the context. Also, run this in a separate task to make
+        # sure we don't block the pipeline.
+        if properties and properties.on_context_updated:
+            task_name = f"{frame.function_name}:{frame.tool_call_id}:on_context_updated"
+            task = self.create_task(properties.on_context_updated(), task_name)
+            self._context_updated_tasks.add(task)
+            task.add_done_callback(self._context_updated_task_finished)
+
+    async def _handle_function_call_cancel(self, frame: FunctionCallCancelFrame):
+        logger.debug(
+            f"{self} FunctionCallCancelFrame: [{frame.function_name}:{frame.tool_call_id}]"
+        )
+        if frame.tool_call_id not in self._function_calls_in_progress:
+            return
+
+        if self._function_calls_in_progress[frame.tool_call_id].cancel_on_interruption:
+            # Update context with the function call cancellation
+            self._update_function_call_result(frame.function_name, frame.tool_call_id, "CANCELLED")
+            del self._function_calls_in_progress[frame.tool_call_id]
+
+    def _update_function_call_result(self, function_name: str, tool_call_id: str, result: Any):
+        for message in self._context.get_messages():
+            if (
+                not isinstance(message, LLMSpecificMessage)
+                and message["role"] == "tool"
+                and message["tool_call_id"]
+                and message["tool_call_id"] == tool_call_id
+            ):
+                message["content"] = result
+
+    async def _handle_user_image_frame(self, frame: UserImageRawFrame):
+        logger.debug(
+            f"{self} UserImageRawFrame: [{frame.request.function_name}:{frame.request.tool_call_id}]"
+        )
+
+        if frame.request.tool_call_id not in self._function_calls_in_progress:
+            logger.warning(
+                f"UserImageRawFrame tool_call_id [{frame.request.tool_call_id}] is not running"
+            )
+            return
+
+        del self._function_calls_in_progress[frame.request.tool_call_id]
+
+        # Update context with the image frame
+        self._update_function_call_result(
+            frame.request.function_name, frame.request.tool_call_id, "COMPLETED"
+        )
+        self._context.add_image_frame_message(
+            format=frame.format,
+            size=frame.size,
+            image=frame.image,
+            text=frame.request.context,
+        )
+
+        await self._push_aggregation()
+        await self.push_context_frame(FrameDirection.UPSTREAM)
+
+    async def _handle_llm_start(self, _: LLMFullResponseStartFrame):
+        self._started += 1
+
+    async def _handle_llm_end(self, _: LLMFullResponseEndFrame):
+        self._started -= 1
+        await self._push_aggregation()
+
+    async def _handle_text(self, frame: TextFrame):
+        if not self._started:
+            return
+
+        if self._params.expect_stripped_words:
+            self._aggregation += f" {frame.text}" if self._aggregation else frame.text
+        else:
+            self._aggregation += frame.text
+
+    def _context_updated_task_finished(self, task: asyncio.Task):
+        self._context_updated_tasks.discard(task)
+
+
+class LLMContextAggregatorPair:
+    """Pair of LLM context aggregators for updating context with user and assistant messages."""
+
+    def __init__(
+        self,
+        context: LLMContext,
+        *,
+        user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
+        assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
+    ):
+        """Initialize the LLM context aggregator pair.
+
+        Args:
+            context: The context to be managed by the aggregators.
+            user_params: Parameters for the user context aggregator.
+            assistant_params: Parameters for the assistant context aggregator.
+        """
+        self._user = LLMUserAggregator(context, params=user_params)
+        self._assistant = LLMAssistantAggregator(context, params=assistant_params)
+
+    def user(self) -> LLMUserAggregator:
+        """Get the user context aggregator.
+
+        Returns:
+            The user context aggregator instance.
+        """
+        return self._user
+
+    def assistant(self) -> LLMAssistantAggregator:
+        """Get the assistant context aggregator.
+
+        Returns:
+            The assistant context aggregator instance.
+        """
+        return self._assistant

src/pipecat/processors/frameworks/rtvi.py

@@ -42,6 +42,7 @@ from pipecat.frames.frames import (
     FunctionCallResultFrame,
     InputAudioRawFrame,
     InterimTranscriptionFrame,
+    LLMContextFrame,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
     LLMMessagesAppendFrame,
@@ -916,7 +917,10 @@ class RTVIObserver(BaseObserver):
             and self._params.user_transcription_enabled
         ):
             await self._handle_user_transcriptions(frame)
-        elif isinstance(frame, OpenAILLMContextFrame) and self._params.user_llm_enabled:
+        elif (
+            isinstance(frame, (OpenAILLMContextFrame, LLMContextFrame))
+            and self._params.user_llm_enabled
+        ):
             await self._handle_context(frame)
         elif isinstance(frame, LLMFullResponseStartFrame) and self._params.bot_llm_enabled:
             await self.push_transport_message_urgent(RTVIBotLLMStartedMessage())
@@ -1017,16 +1021,20 @@ class RTVIObserver(BaseObserver):
         if message:
             await self.push_transport_message_urgent(message)
 
-    async def _handle_context(self, frame: OpenAILLMContextFrame):
+    async def _handle_context(self, frame: OpenAILLMContextFrame | LLMContextFrame):
         """Process LLM context frames to extract user messages for the RTVI client."""
         try:
-            messages = frame.context.messages
+            if isinstance(frame, OpenAILLMContextFrame):
+                messages = frame.context.messages
+            else:
+                messages = frame.context.get_messages()
             if not messages:
                 return
 
             message = messages[-1]
 
             # Handle Google LLM format (protobuf objects with attributes)
+            # Note: not possible if frame is a universal LLMContextFrame
             if hasattr(message, "role") and message.role == "user" and hasattr(message, "parts"):
                 text = "".join(part.text for part in message.parts if hasattr(part, "text"))
                 if text:

src/pipecat/processors/metrics/sentry.py

@@ -6,6 +6,8 @@
 
 """Sentry integration for frame processor metrics."""
 
+import asyncio
+
 from loguru import logger
 
 from pipecat.utils.asyncio.task_manager import BaseTaskManager

src/pipecat/services/anthropic/llm.py

@@ -31,6 +31,7 @@ from pipecat.frames.frames import (
     FunctionCallCancelFrame,
     FunctionCallInProgressFrame,
     FunctionCallResultFrame,
+    LLMContextFrame,
     LLMEnablePromptCachingFrame,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
@@ -41,6 +42,7 @@ from pipecat.frames.frames import (
     VisionImageRawFrame,
 )
 from pipecat.metrics.metrics import LLMTokenUsage
+from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.processors.aggregators.llm_response import (
     LLMAssistantAggregatorParams,
     LLMAssistantContextAggregator,
@@ -197,6 +199,46 @@ class AnthropicLLMService(LLMService):
             response = await api_call(**params)
             return response
 
+    async def run_inference(
+        self, context: LLMContext | OpenAILLMContext, system_instruction: Optional[str] = None
+    ) -> Optional[str]:
+        """Run a one-shot, out-of-band (i.e. out-of-pipeline) inference with the given LLM context.
+
+        Args:
+            context: The LLM context containing conversation history.
+            system_instruction: Optional system instruction to guide the LLM's
+              behavior. You could also (again, optionally) provide a system
+              instruction directly in the context. If both are provided, the
+              one in the context takes precedence.
+
+        Returns:
+            The LLM's response as a string, or None if no response is generated.
+        """
+        messages = []
+        system = []
+        if isinstance(context, LLMContext):
+            # Future code will be something like this:
+            # adapter = self.get_llm_adapter()
+            # params: AnthropicLLMInvocationParams = adapter.get_llm_invocation_params(context)
+            # messages = params["messages"]
+            # system = params["system_instruction"]
+            raise NotImplementedError("Universal LLMContext is not yet supported for Anthropic.")
+        else:
+            context = AnthropicLLMContext.upgrade_to_anthropic(context)
+            messages = context.messages
+            system = getattr(context, "system", None) or system_instruction
+
+        # LLM completion
+        response = await self._client.messages.create(
+            model=self.model_name,
+            messages=messages,
+            system=system,
+            max_tokens=8192,
+            stream=False,
+        )
+
+        return response.content[0].text
+
     @property
     def enable_prompt_caching_beta(self) -> bool:
         """Check if prompt caching beta feature is enabled.
@@ -408,6 +450,8 @@ class AnthropicLLMService(LLMService):
         context = None
         if isinstance(frame, OpenAILLMContextFrame):
             context: "AnthropicLLMContext" = AnthropicLLMContext.upgrade_to_anthropic(frame.context)
+        elif isinstance(frame, LLMContextFrame):
+            raise NotImplementedError("Universal LLMContext is not yet supported for Anthropic.")
         elif isinstance(frame, LLMMessagesFrame):
             context = AnthropicLLMContext.from_messages(frame.messages)
         elif isinstance(frame, VisionImageRawFrame):

src/pipecat/services/aws/llm.py

@@ -31,6 +31,7 @@ from pipecat.frames.frames import (
     FunctionCallFromLLM,
     FunctionCallInProgressFrame,
     FunctionCallResultFrame,
+    LLMContextFrame,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
     LLMMessagesFrame,
@@ -40,6 +41,7 @@ from pipecat.frames.frames import (
     VisionImageRawFrame,
 )
 from pipecat.metrics.metrics import LLMTokenUsage
+from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.processors.aggregators.llm_response import (
     LLMAssistantAggregatorParams,
     LLMAssistantContextAggregator,
@@ -789,6 +791,81 @@ class AWSBedrockLLMService(LLMService):
         """
         return True
 
+    async def run_inference(
+        self, context: LLMContext | OpenAILLMContext, system_instruction: Optional[str] = None
+    ) -> Optional[str]:
+        """Run a one-shot, out-of-band (i.e. out-of-pipeline) inference with the given LLM context.
+
+        Args:
+            context: The LLM context containing conversation history.
+            system_instruction: Optional system instruction to guide the LLM's
+              behavior. You could also (again, optionally) provide a system
+              instruction directly in the context. If both are provided, the
+              one in the context takes precedence.
+
+        Returns:
+            The LLM's response as a string, or None if no response is generated.
+        """
+        try:
+            messages = []
+            system = []
+            if isinstance(context, LLMContext):
+                # Future code will be something like this:
+                # adapter = self.get_llm_adapter()
+                # params: AWSBedrockLLMInvocationParams = adapter.get_llm_invocation_params(context)
+                # messages = params["messages"]
+                # system = params["system_instruction"]
+                raise NotImplementedError(
+                    "Universal LLMContext is not yet supported for AWS Bedrock."
+                )
+            else:
+                context = AWSBedrockLLMContext.upgrade_to_bedrock(context)
+                messages = context.messages
+                system = getattr(context, "system", None) or system_instruction
+
+            # Determine if we're using Claude or Nova based on model ID
+            model_id = self.model_name
+
+            # Prepare request parameters
+            request_params = {
+                "modelId": model_id,
+                "messages": messages,
+                "inferenceConfig": {
+                    "maxTokens": 8192,
+                    "temperature": 0.7,
+                    "topP": 0.9,
+                },
+            }
+
+            if system:
+                request_params["system"] = [{"text": system}]
+
+            async with self._aws_session.client(
+                service_name="bedrock-runtime", **self._aws_params
+            ) as client:
+                # Call Bedrock without streaming
+                response = await client.converse(**request_params)
+
+                # Extract the response text
+                if (
+                    "output" in response
+                    and "message" in response["output"]
+                    and "content" in response["output"]["message"]
+                ):
+                    content = response["output"]["message"]["content"]
+                    if isinstance(content, list):
+                        for item in content:
+                            if item.get("text"):
+                                return item["text"]
+                    elif isinstance(content, str):
+                        return content
+
+                return None
+
+        except Exception as e:
+            logger.error(f"Bedrock summary generation failed: {e}", exc_info=True)
+            return None
+
     async def _create_converse_stream(self, client, request_params):
         """Create converse stream with optional timeout and retry.
 
@@ -1044,6 +1121,8 @@ class AWSBedrockLLMService(LLMService):
         context = None
         if isinstance(frame, OpenAILLMContextFrame):
             context = AWSBedrockLLMContext.upgrade_to_bedrock(frame.context)
+        if isinstance(frame, LLMContextFrame):
+            raise NotImplementedError("Universal LLMContext is not yet supported for AWS Bedrock.")
         elif isinstance(frame, LLMMessagesFrame):
             context = AWSBedrockLLMContext.from_messages(frame.messages)
         elif isinstance(frame, VisionImageRawFrame):

src/pipecat/services/aws_nova_sonic/aws.py

@@ -34,6 +34,7 @@ from pipecat.frames.frames import (
     FunctionCallFromLLM,
     InputAudioRawFrame,
     InterimTranscriptionFrame,
+    LLMContextFrame,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
     LLMTextFrame,
@@ -322,6 +323,10 @@ class AWSNovaSonicLLMService(LLMService):
 
         if isinstance(frame, OpenAILLMContextFrame):
             await self._handle_context(frame.context)
+        elif isinstance(frame, LLMContextFrame):
+            raise NotImplementedError(
+                "Universal LLMContext is not yet supported for AWS Nova Sonic."
+            )
         elif isinstance(frame, InputAudioRawFrame):
             await self._handle_input_audio_frame(frame)
         elif isinstance(frame, BotStoppedSpeakingFrame):

src/pipecat/services/azure/llm.py

@@ -60,3 +60,12 @@ class AzureLLMService(OpenAILLMService):
             azure_endpoint=self._endpoint,
             api_version=self._api_version,
         )
+
+    @property
+    def supports_universal_context(self) -> bool:
+        """Check if this service supports universal LLMContext.
+
+        Returns:
+            False, as Azure service does yet not support universal LLMContext.
+        """
+        return False

src/pipecat/services/cerebras/llm.py

@@ -9,9 +9,8 @@
 from typing import List
 
 from loguru import logger
-from openai.types.chat import ChatCompletionMessageParam
 
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.adapters.services.open_ai_adapter import OpenAILLMInvocationParams
 from pipecat.services.openai.llm import OpenAILLMService
 
 
@@ -54,25 +53,40 @@ class CerebrasLLMService(OpenAILLMService):
         logger.debug(f"Creating Cerebras client with api {base_url}")
         return super().create_client(api_key, base_url, **kwargs)
 
-    def build_chat_completion_params(
-        self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
-    ) -> dict:
+    def build_chat_completion_params(self, params_from_context: OpenAILLMInvocationParams) -> dict:
         """Build parameters for Cerebras chat completion request.
 
         Cerebras supports a subset of OpenAI parameters, focusing on core
         completion settings without advanced features like frequency/presence penalties.
+
+        Args:
+            params_from_context: Parameters, derived from the LLM context, to
+                use for the chat completion. Contains messages, tools, and tool
+                choice.
+
+        Returns:
+            Dictionary of parameters for the chat completion request.
         """
         params = {
             "model": self.model_name,
             "stream": True,
-            "messages": messages,
-            "tools": context.tools,
-            "tool_choice": context.tool_choice,
             "seed": self._settings["seed"],
             "temperature": self._settings["temperature"],
             "top_p": self._settings["top_p"],
             "max_completion_tokens": self._settings["max_completion_tokens"],
         }
 
+        # Messages, tools, tool_choice
+        params.update(params_from_context)
+
         params.update(self._settings["extra"])
         return params
+
+    @property
+    def supports_universal_context(self) -> bool:
+        """Check if this service supports universal LLMContext.
+
+        Returns:
+            False, as Cerebras service does not yet support universal LLMContext.
+        """
+        return False

src/pipecat/services/deepseek/llm.py

@@ -9,9 +9,8 @@
 from typing import List
 
 from loguru import logger
-from openai.types.chat import ChatCompletionMessageParam
 
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.adapters.services.open_ai_adapter import OpenAILLMInvocationParams
 from pipecat.services.openai.llm import OpenAILLMService
 
 
@@ -54,19 +53,22 @@ class DeepSeekLLMService(OpenAILLMService):
         logger.debug(f"Creating DeepSeek client with api {base_url}")
         return super().create_client(api_key, base_url, **kwargs)
 
-    def _build_chat_completion_params(
-        self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
-    ) -> dict:
+    def _build_chat_completion_params(self, params_from_context: OpenAILLMInvocationParams) -> dict:
         """Build parameters for DeepSeek chat completion request.
 
         DeepSeek doesn't support some OpenAI parameters like seed and max_completion_tokens.
+
+        Args:
+            params_from_context: Parameters, derived from the LLM context, to
+                use for the chat completion. Contains messages, tools, and tool
+                choice.
+
+        Returns:
+            Dictionary of parameters for the chat completion request.
         """
         params = {
             "model": self.model_name,
             "stream": True,
-            "messages": messages,
-            "tools": context.tools,
-            "tool_choice": context.tool_choice,
             "stream_options": {"include_usage": True},
             "frequency_penalty": self._settings["frequency_penalty"],
             "presence_penalty": self._settings["presence_penalty"],
@@ -75,5 +77,17 @@ class DeepSeekLLMService(OpenAILLMService):
             "max_tokens": self._settings["max_tokens"],
         }
 
+        # Messages, tools, tool_choice
+        params.update(params_from_context)
+
         params.update(self._settings["extra"])
         return params
+
+    @property
+    def supports_universal_context(self) -> bool:
+        """Check if this service supports universal LLMContext.
+
+        Returns:
+            False, as DeepSeekLLMService does not yet support universal LLMContext.
+        """
+        return False

src/pipecat/services/fireworks/llm.py

@@ -9,9 +9,8 @@
 from typing import List
 
 from loguru import logger
-from openai.types.chat import ChatCompletionMessageParam
 
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.adapters.services.open_ai_adapter import OpenAILLMInvocationParams
 from pipecat.services.openai.llm import OpenAILLMService
 
 
@@ -54,20 +53,23 @@ class FireworksLLMService(OpenAILLMService):
         logger.debug(f"Creating Fireworks client with api {base_url}")
         return super().create_client(api_key, base_url, **kwargs)
 
-    def build_chat_completion_params(
-        self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
-    ) -> dict:
+    def build_chat_completion_params(self, params_from_context: OpenAILLMInvocationParams) -> dict:
         """Build parameters for Fireworks chat completion request.
 
         Fireworks doesn't support some OpenAI parameters like seed, max_completion_tokens,
         and stream_options.
+
+        Args:
+            params_from_context: Parameters, derived from the LLM context, to
+                use for the chat completion. Contains messages, tools, and tool
+                choice.
+
+        Returns:
+            Dictionary of parameters for the chat completion request.
         """
         params = {
             "model": self.model_name,
             "stream": True,
-            "messages": messages,
-            "tools": context.tools,
-            "tool_choice": context.tool_choice,
             "frequency_penalty": self._settings["frequency_penalty"],
             "presence_penalty": self._settings["presence_penalty"],
             "temperature": self._settings["temperature"],
@@ -75,5 +77,17 @@ class FireworksLLMService(OpenAILLMService):
             "max_tokens": self._settings["max_tokens"],
         }
 
+        # Messages, tools, tool_choice
+        params.update(params_from_context)
+
         params.update(self._settings["extra"])
         return params
+
+    @property
+    def supports_universal_context(self) -> bool:
+        """Check if this service supports universal LLMContext.
+
+        Returns:
+            False, as FireworksLLMService does not yet support universal LLMContext.
+        """
+        return False

src/pipecat/services/google/llm.py

@@ -16,19 +16,20 @@ import json
 import os
 import uuid
 from dataclasses import dataclass
-from typing import Any, Dict, List, Optional
+from typing import Any, AsyncIterator, Dict, List, Optional
 
 from loguru import logger
 from PIL import Image
 from pydantic import BaseModel, Field
 
-from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter
+from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter, GeminiLLMInvocationParams
 from pipecat.frames.frames import (
     AudioRawFrame,
     Frame,
     FunctionCallCancelFrame,
     FunctionCallInProgressFrame,
     FunctionCallResultFrame,
+    LLMContextFrame,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
     LLMMessagesFrame,
@@ -38,6 +39,7 @@ from pipecat.frames.frames import (
     VisionImageRawFrame,
 )
 from pipecat.metrics.metrics import LLMTokenUsage
+from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.processors.aggregators.llm_response import (
     LLMAssistantAggregatorParams,
     LLMUserAggregatorParams,
@@ -67,6 +69,7 @@ try:
         FunctionCall,
         FunctionResponse,
         GenerateContentConfig,
+        GenerateContentResponse,
         HttpOptions,
         Part,
     )
@@ -418,7 +421,14 @@ class GoogleLLMContext(OpenAILLMContext):
         role = message["role"]
         content = message.get("content", [])
         if role == "system":
-            self.system_message = content
+            # System instructions are returned as plain text
+            if isinstance(content, str):
+                self.system_message = content
+            elif isinstance(content, list):
+                # If content is a list, we assume it's a list of text parts, per the standard
+                self.system_message = " ".join(
+                    part["text"] for part in content if part.get("type") == "text"
+                )
             return None
         elif role == "assistant":
             role = "model"
@@ -436,11 +446,20 @@ class GoogleLLMContext(OpenAILLMContext):
                 )
         elif role == "tool":
             role = "model"
+            try:
+                response = json.loads(message["content"])
+                if isinstance(response, dict):
+                    response_dict = response
+                else:
+                    response_dict = {"value": response}
+            except Exception as e:
+                # Response might not be JSON-deserializable (e.g. plain text).
+                response_dict = {"value": message["content"]}
             parts.append(
                 Part(
                     function_response=FunctionResponse(
                         name="tool_call_result",  # seems to work to hard-code the same name every time
-                        response=json.loads(message["content"]),
+                        response=response_dict,
                     )
                 )
             )
@@ -636,9 +655,8 @@ class GoogleLLMService(LLMService):
     """Google AI (Gemini) LLM service implementation.
 
     This class implements inference with Google's AI models, translating internally
-    from OpenAILLMContext to the messages format expected by the Google AI model.
-    We use OpenAILLMContext as a lingua franca for all LLM services to enable
-    easy switching between different LLMs.
+    from an OpenAILLMContext or a universal LLMContext to the messages format
+    expected by the Google AI model.
     """
 
     # Overriding the default adapter to use the Gemini one.
@@ -715,6 +733,50 @@ class GoogleLLMService(LLMService):
     def _create_client(self, api_key: str, http_options: Optional[HttpOptions] = None):
         self._client = genai.Client(api_key=api_key, http_options=http_options)
 
+    async def run_inference(
+        self, context: LLMContext | OpenAILLMContext, system_instruction: Optional[str] = None
+    ) -> Optional[str]:
+        """Run a one-shot, out-of-band (i.e. out-of-pipeline) inference with the given LLM context.
+
+        Args:
+            context: The LLM context containing conversation history.
+            system_instruction: Optional system instruction to guide the LLM's
+              behavior. You could also (again, optionally) provide a system
+              instruction directly in the context. If both are provided, the
+              one in the context takes precedence.
+
+        Returns:
+            The LLM's response as a string, or None if no response is generated.
+        """
+        messages = []
+        system = []
+        if isinstance(context, LLMContext):
+            adapter = self.get_llm_adapter()
+            params: GeminiLLMInvocationParams = adapter.get_llm_invocation_params(context)
+            messages = params["messages"]
+            system = params["system_instruction"]
+        else:
+            context = GoogleLLMContext.upgrade_to_google(context)
+            messages = context.messages
+            system = getattr(context, "system_message", None) or system_instruction
+
+        generation_config = GenerateContentConfig(system_instruction=system)
+
+        # Use the new google-genai client's async method
+        response = await self._client.aio.models.generate_content(
+            model=self._model_name,
+            contents=messages,
+            config=generation_config,
+        )
+
+        # Extract text from response
+        if response.candidates and response.candidates[0].content:
+            for part in response.candidates[0].content.parts:
+                if part.text:
+                    return part.text
+
+        return None
+
     def needs_mcp_alternate_schema(self) -> bool:
         """Check if this LLM service requires alternate MCP schema.
 
@@ -740,8 +802,89 @@ class GoogleLLMService(LLMService):
         except Exception as e:
             logger.exception(f"Failed to unset thinking budget: {e}")
 
+    async def _stream_content(
+        self, params_from_context: GeminiLLMInvocationParams
+    ) -> AsyncIterator[GenerateContentResponse]:
+        messages = params_from_context["messages"]
+        if (
+            params_from_context["system_instruction"]
+            and self._system_instruction != params_from_context["system_instruction"]
+        ):
+            logger.debug(f"System instruction changed: {params_from_context['system_instruction']}")
+            self._system_instruction = params_from_context["system_instruction"]
+
+        tools = []
+        if params_from_context["tools"]:
+            tools = params_from_context["tools"]
+        elif self._tools:
+            tools = self._tools
+        tool_config = None
+        if self._tool_config:
+            tool_config = self._tool_config
+
+        # Filter out None values and create GenerationContentConfig
+        generation_params = {
+            k: v
+            for k, v in {
+                "system_instruction": self._system_instruction,
+                "temperature": self._settings["temperature"],
+                "top_p": self._settings["top_p"],
+                "top_k": self._settings["top_k"],
+                "max_output_tokens": self._settings["max_tokens"],
+                "tools": tools,
+                "tool_config": tool_config,
+            }.items()
+            if v is not None
+        }
+
+        if self._settings["extra"]:
+            generation_params.update(self._settings["extra"])
+
+        # possibly modify generation_params (in place) to set thinking to off by default
+        self._maybe_unset_thinking_budget(generation_params)
+
+        generation_config = (
+            GenerateContentConfig(**generation_params) if generation_params else None
+        )
+
+        await self.start_ttfb_metrics()
+        return await self._client.aio.models.generate_content_stream(
+            model=self._model_name,
+            contents=messages,
+            config=generation_config,
+        )
+
+    async def _stream_content_specific_context(
+        self, context: OpenAILLMContext
+    ) -> AsyncIterator[GenerateContentResponse]:
+        logger.debug(
+            # f"{self}: Generating chat [{self._system_instruction}] | [{context.get_messages_for_logging()}]"
+            f"{self}: Generating chat from OpenAI context [{context.get_messages_for_logging()}]"
+        )
+
+        params = GeminiLLMInvocationParams(
+            messages=context.messages,
+            system_instruction=context.system_message,
+            tools=context.tools,
+        )
+
+        return await self._stream_content(params)
+
+    async def _stream_content_universal_context(
+        self, context: LLMContext
+    ) -> AsyncIterator[GenerateContentResponse]:
+        adapter = self.get_llm_adapter()
+        logger.debug(
+            # f"{self}: Generating chat [{self._system_instruction}] | [{context.get_messages_for_logging()}]"
+            f"{self}: Generating chat from universal context [{adapter.get_messages_for_logging(context)}]"
+        )
+
+        params: GeminiLLMInvocationParams = adapter.get_llm_invocation_params(context)
+
+        return await self._stream_content(params)
+
     @traced_llm
-    async def _process_context(self, context: OpenAILLMContext):
+    async def _process_context(self, context: OpenAILLMContext | LLMContext):
         await self.push_frame(LLMFullResponseStartFrame())
 
         prompt_tokens = 0
@@ -754,55 +897,11 @@ class GoogleLLMService(LLMService):
         search_result = ""
 
         try:
-            logger.debug(
-                # f"{self}: Generating chat [{self._system_instruction}] | [{context.get_messages_for_logging()}]"
-                f"{self}: Generating chat [{context.get_messages_for_logging()}]"
-            )
-
-            messages = context.messages
-            if context.system_message and self._system_instruction != context.system_message:
-                logger.debug(f"System instruction changed: {context.system_message}")
-                self._system_instruction = context.system_message
-
-            tools = []
-            if context.tools:
-                tools = context.tools
-            elif self._tools:
-                tools = self._tools
-            tool_config = None
-            if self._tool_config:
-                tool_config = self._tool_config
-
-            # Filter out None values and create GenerationContentConfig
-            generation_params = {
-                k: v
-                for k, v in {
-                    "system_instruction": self._system_instruction,
-                    "temperature": self._settings["temperature"],
-                    "top_p": self._settings["top_p"],
-                    "top_k": self._settings["top_k"],
-                    "max_output_tokens": self._settings["max_tokens"],
-                    "tools": tools,
-                    "tool_config": tool_config,
-                }.items()
-                if v is not None
-            }
-
-            if self._settings["extra"]:
-                generation_params.update(self._settings["extra"])
-
-            # possibly modify generation_params (in place) to set thinking to off by default
-            self._maybe_unset_thinking_budget(generation_params)
-
-            generation_config = (
-                GenerateContentConfig(**generation_params) if generation_params else None
-            )
-
-            await self.start_ttfb_metrics()
-            response = await self._client.aio.models.generate_content_stream(
-                model=self._model_name,
-                contents=messages,
-                config=generation_config,
+            # Generate content using either OpenAILLMContext or universal LLMContext
+            response = await (
+                self._stream_content_specific_context(context)
+                if isinstance(context, OpenAILLMContext)
+                else self._stream_content_universal_context(context)
             )
 
             function_calls = []
@@ -915,9 +1014,18 @@ class GoogleLLMService(LLMService):
 
         if isinstance(frame, OpenAILLMContextFrame):
             context = GoogleLLMContext.upgrade_to_google(frame.context)
+        elif isinstance(frame, LLMContextFrame):
+            # Handle universal (LLM-agnostic) LLM context frames
+            context = frame.context
         elif isinstance(frame, LLMMessagesFrame):
+            # NOTE: LLMMessagesFrame is deprecated, so we don't support the newer universal
+            # LLMContext with it
             context = GoogleLLMContext(frame.messages)
         elif isinstance(frame, VisionImageRawFrame):
+            # This is only useful in very simple pipelines because it creates
+            # a new context. Generally we want a context manager to catch
+            # UserImageRawFrames coming through the pipeline and add them
+            # to the context.
             context = GoogleLLMContext()
             context.add_image_frame_message(
                 format=frame.format, size=frame.size, image=frame.image, text=frame.text

src/pipecat/services/google/llm_openai.py

@@ -39,6 +39,10 @@ class GoogleLLMOpenAIBetaService(OpenAILLMService):
     Note: This service includes a workaround for a Google API bug where function
     call indices may be incorrectly set to None, resulting in empty function names.
 
+    .. deprecated:: 0.0.82
+        GoogleLLMOpenAIBetaService is deprecated and will be removed in a future version.
+        Use GoogleLLMService instead for better integration with Google's native API.
+
     Reference:
         https://ai.google.dev/gemini-api/docs/openai
     """
@@ -59,8 +63,26 @@ class GoogleLLMOpenAIBetaService(OpenAILLMService):
             model: Google model name to use (e.g., "gemini-2.0-flash").
             **kwargs: Additional arguments passed to the parent OpenAILLMService.
         """
+        import warnings
+
+        warnings.warn(
+            "GoogleLLMOpenAIBetaService is deprecated and will be removed in a future version. "
+            "Use GoogleLLMService instead for better integration with Google's native API.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
+
         super().__init__(api_key=api_key, base_url=base_url, model=model, **kwargs)
 
+    @property
+    def supports_universal_context(self) -> bool:
+        """Check if this service supports universal LLMContext.
+
+        Returns:
+            False, as GoogleLLMOpenAIBetaService does not yet support universal LLMContext.
+        """
+        return False
+
     async def _process_context(self, context: OpenAILLMContext):
         functions_list = []
         arguments_list = []
@@ -72,9 +94,9 @@ class GoogleLLMOpenAIBetaService(OpenAILLMService):
 
         await self.start_ttfb_metrics()
 
-        chunk_stream: AsyncStream[ChatCompletionChunk] = await self._stream_chat_completions(
-            context
-        )
+        chunk_stream: AsyncStream[
+            ChatCompletionChunk
+        ] = await self._stream_chat_completions_specific_context(context)
 
         async for chunk in chunk_stream:
             if chunk.usage:

src/pipecat/services/google/llm_vertex.py

@@ -139,3 +139,12 @@ class GoogleVertexLLMService(OpenAILLMService):
         creds.refresh(Request())  # Ensure token is up-to-date, lifetime is 1 hour.
 
         return creds.token
+
+    @property
+    def supports_universal_context(self) -> bool:
+        """Check if this service supports universal LLMContext.
+
+        Returns:
+            False, as GoogleVertexLLMService does not yet support universal LLMContext.
+        """
+        return False

src/pipecat/services/grok/llm.py

@@ -190,3 +190,12 @@ class GrokLLMService(OpenAILLMService):
         user = OpenAIUserContextAggregator(context, params=user_params)
         assistant = OpenAIAssistantContextAggregator(context, params=assistant_params)
         return GrokContextAggregatorPair(_user=user, _assistant=assistant)
+
+    @property
+    def supports_universal_context(self) -> bool:
+        """Check if this service supports universal LLMContext.
+
+        Returns:
+            False, as GrokLLMService does not yet support universal LLMContext.
+        """
+        return False

src/pipecat/services/groq/llm.py

@@ -49,3 +49,12 @@ class GroqLLMService(OpenAILLMService):
         """
         logger.debug(f"Creating Groq client with api {base_url}")
         return super().create_client(api_key, base_url, **kwargs)
+
+    @property
+    def supports_universal_context(self) -> bool:
+        """Check if this service supports universal LLMContext.
+
+        Returns:
+            False, as GroqLLMService does not yet support universal LLMContext.
+        """
+        return False

src/pipecat/services/llm_service.py

@@ -14,6 +14,7 @@ from typing import (
     Awaitable,
     Callable,
     Dict,
+    List,
     Mapping,
     Optional,
     Protocol,
@@ -40,6 +41,7 @@ from pipecat.frames.frames import (
     StartInterruptionFrame,
     UserImageRequestFrame,
 )
+from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.processors.aggregators.llm_response import (
     LLMAssistantAggregatorParams,
     LLMUserAggregatorParams,
@@ -88,7 +90,7 @@ class FunctionCallParams:
     tool_call_id: str
     arguments: Mapping[str, Any]
     llm: "LLMService"
-    context: OpenAILLMContext
+    context: OpenAILLMContext | LLMContext
     result_callback: FunctionCallResultCallback
 
 
@@ -129,7 +131,7 @@ class FunctionCallRunnerItem:
     function_name: str
     tool_call_id: str
     arguments: Mapping[str, Any]
-    context: OpenAILLMContext
+    context: OpenAILLMContext | LLMContext
     run_llm: Optional[bool] = None
 
 
@@ -189,6 +191,24 @@ class LLMService(AIService):
         """
         return self._adapter
 
+    async def run_inference(
+        self, context: LLMContext | OpenAILLMContext, system_instruction: Optional[str] = None
+    ) -> Optional[str]:
+        """Run a one-shot, out-of-band (i.e. out-of-pipeline) inference with the given LLM context.
+
+        Must be implemented by subclasses.
+
+        Args:
+            context: The LLM context containing conversation history.
+            system_instruction: Optional system instruction to guide the LLM's
+              behavior. You could also (again, optionally) provide a system
+              instruction directly in the context.
+
+        Returns:
+            The LLM's response as a string, or None if no response is generated.
+        """
+        raise NotImplementedError(f"run_inference() not supported by {self.__class__.__name__}")
+
     def create_context_aggregator(
         self,
         context: OpenAILLMContext,
@@ -432,7 +452,9 @@ class LLMService(AIService):
             else:
                 await self._sequential_runner_queue.put(runner_item)
 
-    async def _call_start_function(self, context: OpenAILLMContext, function_name: str):
+    async def _call_start_function(
+        self, context: OpenAILLMContext | LLMContext, function_name: str
+    ):
         if function_name in self._start_callbacks.keys():
             await self._start_callbacks[function_name](function_name, self, context)
         elif None in self._start_callbacks.keys():

src/pipecat/services/nim/llm.py

@@ -47,6 +47,15 @@ class NimLLMService(OpenAILLMService):
         self._has_reported_prompt_tokens = False
         self._is_processing = False
 
+    @property
+    def supports_universal_context(self) -> bool:
+        """Check if this service supports universal LLMContext.
+
+        Returns:
+            False, as NimLLMService does not yet support universal LLMContext.
+        """
+        return False
+
     async def _process_context(self, context: OpenAILLMContext):
         """Process a context through the LLM and accumulate token usage metrics.
 

src/pipecat/services/ollama/llm.py

@@ -43,3 +43,12 @@ class OLLamaLLMService(OpenAILLMService):
         """
         logger.debug(f"Creating Ollama client with api {base_url}")
         return super().create_client(base_url=base_url, **kwargs)
+
+    @property
+    def supports_universal_context(self) -> bool:
+        """Check if this service supports universal LLMContext.
+
+        Returns:
+            False, as OLLamaLLMService does not yet support universal LLMContext.
+        """
+        return False

src/pipecat/services/openai/base_llm.py

@@ -4,7 +4,7 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-"""Base OpenAI LLM service implementation."""
+"""Base LLM service implementation for services that use the AsyncOpenAI client."""
 
 import asyncio
 import base64
@@ -23,8 +23,10 @@ from openai import (
 from openai.types.chat import ChatCompletionChunk, ChatCompletionMessageParam
 from pydantic import BaseModel, Field
 
+from pipecat.adapters.services.open_ai_adapter import OpenAILLMInvocationParams
 from pipecat.frames.frames import (
     Frame,
+    LLMContextFrame,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
     LLMMessagesFrame,
@@ -33,6 +35,7 @@ from pipecat.frames.frames import (
     VisionImageRawFrame,
 )
 from pipecat.metrics.metrics import LLMTokenUsage
+from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.processors.aggregators.openai_llm_context import (
     OpenAILLMContext,
     OpenAILLMContextFrame,
@@ -45,10 +48,11 @@ from pipecat.utils.tracing.service_decorators import traced_llm
 class BaseOpenAILLMService(LLMService):
     """Base class for all services that use the AsyncOpenAI client.
 
-    This service consumes OpenAILLMContextFrame frames, which contain a reference
-    to an OpenAILLMContext object. The context defines what is sent to the LLM for
-    completion, including user, assistant, and system messages, as well as tool
-    choices and function call configurations.
+    This service consumes OpenAILLMContextFrame or LLMContextFrame frames,
+    which contain a reference to an OpenAILLMContext or LLMContext object. The
+    context defines what is sent to the LLM for completion, including user,
+    assistant, and system messages, as well as tool choices and function call
+    configurations.
     """
 
     class InputParams(BaseModel):
@@ -180,18 +184,19 @@ class BaseOpenAILLMService(LLMService):
         return True
 
     async def get_chat_completions(
-        self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
+        self, params_from_context: OpenAILLMInvocationParams
     ) -> AsyncStream[ChatCompletionChunk]:
         """Get streaming chat completions from OpenAI API with optional timeout and retry.
 
         Args:
-            context: The LLM context containing tools and configuration.
-            messages: List of chat completion messages to send.
+            params_from_context: Parameters, derived from the LLM context, to
+                use for the chat completion. Contains messages, tools, and tool
+                choice.
 
         Returns:
             Async stream of chat completion chunks.
         """
-        params = self.build_chat_completion_params(context, messages)
+        params = self.build_chat_completion_params(params_from_context)
 
         if self._retry_on_timeout:
             try:
@@ -208,16 +213,15 @@ class BaseOpenAILLMService(LLMService):
             chunks = await self._client.chat.completions.create(**params)
             return chunks
 
-    def build_chat_completion_params(
-        self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
-    ) -> dict:
+    def build_chat_completion_params(self, params_from_context: OpenAILLMInvocationParams) -> dict:
         """Build parameters for chat completion request.
 
         Subclasses can override this to customize parameters for different providers.
 
         Args:
-            context: The LLM context containing tools and configuration.
-            messages: List of chat completion messages to send.
+            params_from_context: Parameters, derived from the LLM context, to
+                use for the chat completion. Contains messages, tools, and tool
+                choice.
 
         Returns:
             Dictionary of parameters for the chat completion request.
@@ -225,9 +229,6 @@ class BaseOpenAILLMService(LLMService):
         params = {
             "model": self.model_name,
             "stream": True,
-            "messages": messages,
-            "tools": context.tools,
-            "tool_choice": context.tool_choice,
             "stream_options": {"include_usage": True},
             "frequency_penalty": self._settings["frequency_penalty"],
             "presence_penalty": self._settings["presence_penalty"],
@@ -238,13 +239,48 @@ class BaseOpenAILLMService(LLMService):
             "max_completion_tokens": self._settings["max_completion_tokens"],
         }
 
+        # Messages, tools, tool_choice
+        params.update(params_from_context)
+
         params.update(self._settings["extra"])
         return params
 
-    async def _stream_chat_completions(
+    async def run_inference(
+        self, context: LLMContext | OpenAILLMContext, system_instruction: Optional[str] = None
+    ) -> Optional[str]:
+        """Run a one-shot, out-of-band (i.e. out-of-pipeline) inference with the given LLM context.
+
+        Args:
+            context: The LLM context containing conversation history.
+            system_instruction: Optional system instruction to guide the LLM's
+              behavior. You could also (again, optionally) provide a system
+              instruction directly in the context.
+
+        Returns:
+            The LLM's response as a string, or None if no response is generated.
+        """
+        if isinstance(context, LLMContext):
+            adapter = self.get_llm_adapter()
+            params: OpenAILLMInvocationParams = adapter.get_llm_invocation_params(context)
+            messages = params["messages"]
+        else:
+            messages = context.messages
+
+        # LLM completion
+        response = await self._client.chat.completions.create(
+            model=self.model_name,
+            messages=messages,
+            stream=False,
+        )
+
+        return response.choices[0].message.content
+
+    async def _stream_chat_completions_specific_context(
         self, context: OpenAILLMContext
     ) -> AsyncStream[ChatCompletionChunk]:
-        logger.debug(f"{self}: Generating chat [{context.get_messages_for_logging()}]")
+        logger.debug(
+            f"{self}: Generating chat from OpenAI context [{context.get_messages_for_logging()}]"
+        )
 
         messages: List[ChatCompletionMessageParam] = context.get_messages()
 
@@ -263,12 +299,28 @@ class BaseOpenAILLMService(LLMService):
                 del message["data"]
                 del message["mime_type"]
 
-        chunks = await self.get_chat_completions(context, messages)
+        params = OpenAILLMInvocationParams(
+            messages=messages, tools=context.tools, tool_choice=context.tool_choice
+        )
+        chunks = await self.get_chat_completions(params)
+
+        return chunks
+
+    async def _stream_chat_completions_universal_context(
+        self, context: LLMContext
+    ) -> AsyncStream[ChatCompletionChunk]:
+        adapter = self.get_llm_adapter()
+        logger.debug(
+            f"{self}: Generating chat from universal context [{adapter.get_messages_for_logging(context)}]"
+        )
+
+        params: OpenAILLMInvocationParams = adapter.get_llm_invocation_params(context)
+        chunks = await self.get_chat_completions(params)
 
         return chunks
 
     @traced_llm
-    async def _process_context(self, context: OpenAILLMContext):
+    async def _process_context(self, context: OpenAILLMContext | LLMContext):
         functions_list = []
         arguments_list = []
         tool_id_list = []
@@ -279,8 +331,11 @@ class BaseOpenAILLMService(LLMService):
 
         await self.start_ttfb_metrics()
 
-        chunk_stream: AsyncStream[ChatCompletionChunk] = await self._stream_chat_completions(
-            context
+        # Generate chat completions using either OpenAILLMContext or universal LLMContext
+        chunk_stream = await (
+            self._stream_chat_completions_specific_context(context)
+            if isinstance(context, OpenAILLMContext)
+            else self._stream_chat_completions_universal_context(context)
         )
 
         async for chunk in chunk_stream:
@@ -364,11 +419,24 @@ class BaseOpenAILLMService(LLMService):
 
             await self.run_function_calls(function_calls)
 
+    @property
+    def supports_universal_context(self) -> bool:
+        """Check if this service supports universal LLMContext.
+
+        Returns:
+            Whether service supports universal LLMContext.
+        """
+        # Return True in subclasses that support universal LLMContext
+        # This property lets us gradually roll out support for universal
+        # LLMContext to OpenAI-like services in a controlled manner.
+        return False
+
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         """Process frames for LLM completion requests.
 
-        Handles OpenAILLMContextFrame, LLMMessagesFrame, VisionImageRawFrame,
-        and LLMUpdateSettingsFrame to trigger LLM completions and manage settings.
+        Handles OpenAILLMContextFrame, LLMContextFrame, LLMMessagesFrame,
+        VisionImageRawFrame, and LLMUpdateSettingsFrame to trigger LLM
+        completions and manage settings.
 
         Args:
             frame: The frame to process.
@@ -378,10 +446,26 @@ class BaseOpenAILLMService(LLMService):
 
         context = None
         if isinstance(frame, OpenAILLMContextFrame):
-            context: OpenAILLMContext = frame.context
+            # Handle OpenAI-specific context frames
+            context = frame.context
+        elif isinstance(frame, LLMContextFrame):
+            # Handle universal (LLM-agnostic) LLM context frames
+            if self.supports_universal_context:
+                context = frame.context
+            else:
+                raise NotImplementedError(
+                    f"Universal LLMContext is not yet supported for {self.__class__.__name__}."
+                )
         elif isinstance(frame, LLMMessagesFrame):
+            # NOTE: LLMMessagesFrame is deprecated, so we don't support the newer universal
+            # LLMContext with it
             context = OpenAILLMContext.from_messages(frame.messages)
         elif isinstance(frame, VisionImageRawFrame):
+            # This is only useful in very simple pipelines because it creates
+            # a new context. Generally we want a context manager to catch
+            # UserImageRawFrames coming through the pipeline and add them
+            # to the context.
+            # TODO: support the newer universal LLMContext with a VisionImageRawFrame equivalent?
             context = OpenAILLMContext()
             context.add_image_frame_message(
                 format=frame.format, size=frame.size, image=frame.image, text=frame.text

src/pipecat/services/openai/llm.py

@@ -107,6 +107,15 @@ class OpenAILLMService(BaseOpenAILLMService):
         assistant = OpenAIAssistantContextAggregator(context, params=assistant_params)
         return OpenAIContextAggregatorPair(_user=user, _assistant=assistant)
 
+    @property
+    def supports_universal_context(self) -> bool:
+        """Check if this service supports universal LLMContext.
+
+        Returns:
+            True, as OpenAI service supports universal LLMContext.
+        """
+        return True
+
 
 class OpenAIUserContextAggregator(LLMUserContextAggregator):
     """OpenAI-specific user context aggregator.

src/pipecat/services/openai_realtime_beta/openai.py

@@ -23,6 +23,7 @@ from pipecat.frames.frames import (
     Frame,
     InputAudioRawFrame,
     InterimTranscriptionFrame,
+    LLMContextFrame,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
     LLMMessagesAppendFrame,
@@ -343,6 +344,10 @@ class OpenAIRealtimeBetaLLMService(LLMService):
                 await self.reset_conversation()
             # Run the LLM at next opportunity
             await self._create_response()
+        elif isinstance(frame, LLMContextFrame):
+            raise NotImplementedError(
+                "Universal LLMContext is not yet supported for OpenAI Realtime."
+            )
         elif isinstance(frame, InputAudioRawFrame):
             if not self._audio_input_paused:
                 await self._send_user_audio(frame)

src/pipecat/services/openpipe/llm.py

@@ -13,9 +13,8 @@ enabling integration with OpenPipe's fine-tuning and monitoring capabilities.
 from typing import Dict, List, Optional
 
 from loguru import logger
-from openai.types.chat import ChatCompletionMessageParam
 
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.adapters.services.open_ai_adapter import OpenAILLMInvocationParams
 from pipecat.services.openai.llm import OpenAILLMService
 
 try:
@@ -86,22 +85,21 @@ class OpenPipeLLMService(OpenAILLMService):
         )
         return client
 
-    def build_chat_completion_params(
-        self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
-    ) -> dict:
+    def build_chat_completion_params(self, params_from_context: OpenAILLMInvocationParams) -> dict:
         """Build parameters for OpenPipe chat completion request.
 
         Adds OpenPipe-specific logging and tagging parameters.
 
         Args:
-            context: The LLM context containing tools and configuration.
-            messages: List of chat completion messages to send.
+            params_from_context: Parameters, derived from the LLM context, to
+                use for the chat completion. Contains messages, tools, and tool
+                choice.
 
         Returns:
             Dictionary of parameters for the chat completion request.
         """
         # Start with base parameters
-        params = super().build_chat_completion_params(context, messages)
+        params = super().build_chat_completion_params(params_from_context)
 
         # Add OpenPipe-specific parameters
         params["openpipe"] = {
@@ -110,3 +108,12 @@ class OpenPipeLLMService(OpenAILLMService):
         }
 
         return params
+
+    @property
+    def supports_universal_context(self) -> bool:
+        """Check if this service supports universal LLMContext.
+
+        Returns:
+            False, as OpenPipeLLMService does not yet support universal LLMContext.
+        """
+        return False

src/pipecat/services/openrouter/llm.py

@@ -61,3 +61,12 @@ class OpenRouterLLMService(OpenAILLMService):
         """
         logger.debug(f"Creating OpenRouter client with api {base_url}")
         return super().create_client(api_key, base_url, **kwargs)
+
+    @property
+    def supports_universal_context(self) -> bool:
+        """Check if this service supports universal LLMContext.
+
+        Returns:
+            False, as OpenRouterLLMService does not yet support universal LLMContext.
+        """
+        return False

src/pipecat/services/perplexity/llm.py

@@ -11,11 +11,9 @@ an OpenAI-compatible interface. It handles Perplexity's unique token usage
 reporting patterns while maintaining compatibility with the Pipecat framework.
 """
 
-from typing import List
-
 from openai import NOT_GIVEN
-from openai.types.chat import ChatCompletionMessageParam
 
+from pipecat.adapters.services.open_ai_adapter import OpenAILLMInvocationParams
 from pipecat.metrics.metrics import LLMTokenUsage
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.services.openai.llm import OpenAILLMService
@@ -53,17 +51,23 @@ class PerplexityLLMService(OpenAILLMService):
         self._has_reported_prompt_tokens = False
         self._is_processing = False
 
-    def build_chat_completion_params(
-        self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
-    ) -> dict:
+    def build_chat_completion_params(self, params_from_context: OpenAILLMInvocationParams) -> dict:
         """Build parameters for Perplexity chat completion request.
 
         Perplexity uses a subset of OpenAI parameters and doesn't support tools.
+
+        Args:
+            params_from_context: Parameters, derived from the LLM context, to
+                use for the chat completion. Contains messages, tools, and tool
+                choice.
+
+        Returns:
+            Dictionary of parameters for the chat completion request.
         """
         params = {
             "model": self.model_name,
             "stream": True,
-            "messages": messages,
+            "messages": params_from_context["messages"],
         }
 
         # Add OpenAI-compatible parameters if they're set
@@ -80,6 +84,15 @@ class PerplexityLLMService(OpenAILLMService):
 
         return params
 
+    @property
+    def supports_universal_context(self) -> bool:
+        """Check if this service supports universal LLMContext.
+
+        Returns:
+            False, as PerplexityLLMService does not yet support universal LLMContext.
+        """
+        return False
+
     async def _process_context(self, context: OpenAILLMContext):
         """Process a context through the LLM and accumulate token usage metrics.
 

src/pipecat/services/qwen/llm.py

@@ -50,3 +50,12 @@ class QwenLLMService(OpenAILLMService):
         """
         logger.debug(f"Creating Qwen client with base URL: {base_url}")
         return super().create_client(api_key, base_url, **kwargs)
+
+    @property
+    def supports_universal_context(self) -> bool:
+        """Check if this service supports universal LLMContext.
+
+        Returns:
+            False, as QwenLLMService does not yet support universal LLMContext.
+        """
+        return False

src/pipecat/services/sambanova/llm.py

@@ -7,12 +7,13 @@
 """SambaNova LLM service implementation using OpenAI-compatible interface."""
 
 import json
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, Optional
 
 from loguru import logger
 from openai import AsyncStream
-from openai.types.chat import ChatCompletionChunk, ChatCompletionMessageParam
+from openai.types.chat import ChatCompletionChunk
 
+from pipecat.adapters.services.open_ai_adapter import OpenAILLMInvocationParams
 from pipecat.frames.frames import (
     LLMTextFrame,
 )
@@ -67,17 +68,16 @@ class SambaNovaLLMService(OpenAILLMService):  # type: ignore
         logger.debug(f"Creating SambaNova client with API {base_url}")
         return super().create_client(api_key, base_url, **kwargs)
 
-    def build_chat_completion_params(
-        self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
-    ) -> dict:
+    def build_chat_completion_params(self, params_from_context: OpenAILLMInvocationParams) -> dict:
         """Build parameters for SambaNova chat completion request.
 
         SambaNova doesn't support some OpenAI parameters like frequency_penalty,
         presence_penalty, and seed.
 
         Args:
-            context: The LLM context containing tools and configuration.
-            messages: List of chat completion messages to send.
+            params_from_context: Parameters, derived from the LLM context, to
+                use for the chat completion. Contains messages, tools, and tool
+                choice.
 
         Returns:
             Dictionary of parameters for the chat completion request.
@@ -85,9 +85,6 @@ class SambaNovaLLMService(OpenAILLMService):  # type: ignore
         params = {
             "model": self.model_name,
             "stream": True,
-            "messages": messages,
-            "tools": context.tools,
-            "tool_choice": context.tool_choice,
             "stream_options": {"include_usage": True},
             "temperature": self._settings["temperature"],
             "top_p": self._settings["top_p"],
@@ -95,6 +92,9 @@ class SambaNovaLLMService(OpenAILLMService):  # type: ignore
             "max_completion_tokens": self._settings["max_completion_tokens"],
         }
 
+        # Messages, tools, tool_choice
+        params.update(params_from_context)
+
         params.update(self._settings["extra"])
         return params
 
@@ -122,9 +122,9 @@ class SambaNovaLLMService(OpenAILLMService):  # type: ignore
 
         await self.start_ttfb_metrics()
 
-        chunk_stream: AsyncStream[ChatCompletionChunk] = await self._stream_chat_completions(
-            context
-        )
+        chunk_stream: AsyncStream[
+            ChatCompletionChunk
+        ] = await self._stream_chat_completions_specific_context(context)
 
         async for chunk in chunk_stream:
             if chunk.usage:
@@ -210,3 +210,12 @@ class SambaNovaLLMService(OpenAILLMService):  # type: ignore
                 )
 
             await self.run_function_calls(function_calls)
+
+    @property
+    def supports_universal_context(self) -> bool:
+        """Check if this service supports universal LLMContext.
+
+        Returns:
+            False, as SambaNovaLLMService does not yet support universal LLMContext.
+        """
+        return False

src/pipecat/services/sarvam/tts.py

@@ -6,25 +6,41 @@
 
 """Sarvam AI text-to-speech service implementation."""
 
+import asyncio
 import base64
-from typing import AsyncGenerator, Optional
+import json
+import warnings
+from typing import Any, AsyncGenerator, Mapping, Optional
 
 import aiohttp
 from loguru import logger
 from pydantic import BaseModel, Field
 
 from pipecat.frames.frames import (
+    CancelFrame,
+    EndFrame,
     ErrorFrame,
     Frame,
+    LLMFullResponseEndFrame,
     StartFrame,
+    StartInterruptionFrame,
     TTSAudioRawFrame,
     TTSStartedFrame,
     TTSStoppedFrame,
 )
-from pipecat.services.tts_service import TTSService
+from pipecat.processors.frame_processor import FrameDirection
+from pipecat.services.tts_service import InterruptibleTTSService, TTSService
 from pipecat.transcriptions.language import Language
 from pipecat.utils.tracing.service_decorators import traced_tts
 
+try:
+    from websockets.asyncio.client import connect as websocket_connect
+    from websockets.protocol import State
+except ModuleNotFoundError as e:
+    logger.error(f"Exception: {e}")
+    logger.error("In order to use Sarvam, you need to `pip install pipecat-ai[sarvam]`.")
+    raise Exception(f"Missing module: {e}")
+
 
 def language_to_sarvam_language(language: Language) -> Optional[str]:
     """Convert Pipecat Language enum to Sarvam AI language codes.
@@ -52,7 +68,7 @@ def language_to_sarvam_language(language: Language) -> Optional[str]:
     return LANGUAGE_MAP.get(language)
 
 
-class SarvamTTSService(TTSService):
+class SarvamHttpTTSService(TTSService):
     """Text-to-Speech service using Sarvam AI's API.
 
     Converts text to speech using Sarvam AI's TTS models with support for multiple
@@ -95,9 +111,9 @@ class SarvamTTSService(TTSService):
         self,
         *,
         api_key: str,
+        aiohttp_session: aiohttp.ClientSession,
         voice_id: str = "anushka",
         model: str = "bulbul:v2",
-        aiohttp_session: aiohttp.ClientSession,
         base_url: str = "https://api.sarvam.ai",
         sample_rate: Optional[int] = None,
         params: Optional[InputParams] = None,
@@ -107,9 +123,9 @@ class SarvamTTSService(TTSService):
 
         Args:
             api_key: Sarvam AI API subscription key.
+            aiohttp_session: Shared aiohttp session for making requests.
             voice_id: Speaker voice ID (e.g., "anushka", "meera"). Defaults to "anushka".
             model: TTS model to use ("bulbul:v1" or "bulbul:v2"). Defaults to "bulbul:v2".
-            aiohttp_session: Shared aiohttp session for making requests.
             base_url: Sarvam AI API base URL. Defaults to "https://api.sarvam.ai".
             sample_rate: Audio sample rate in Hz (8000, 16000, 22050, 24000). If None, uses default.
             params: Additional voice and preprocessing parameters. If None, uses defaults.
@@ -117,16 +133,16 @@ class SarvamTTSService(TTSService):
         """
         super().__init__(sample_rate=sample_rate, **kwargs)
 
-        params = params or SarvamTTSService.InputParams()
+        params = params or SarvamHttpTTSService.InputParams()
 
         self._api_key = api_key
         self._base_url = base_url
         self._session = aiohttp_session
 
         self._settings = {
-            "language": self.language_to_service_language(params.language)
-            if params.language
-            else "en-IN",
+            "language": (
+                self.language_to_service_language(params.language) if params.language else "en-IN"
+            ),
             "pitch": params.pitch,
             "pace": params.pace,
             "loudness": params.loudness,
@@ -186,7 +202,7 @@ class SarvamTTSService(TTSService):
                 "pitch": self._settings["pitch"],
                 "pace": self._settings["pace"],
                 "loudness": self._settings["loudness"],
-                "speech_sample_rate": self.sample_rate,
+                "sample_rate": self.sample_rate,
                 "enable_preprocessing": self._settings["enable_preprocessing"],
                 "model": self._model_name,
             }
@@ -240,3 +256,387 @@ class SarvamTTSService(TTSService):
         finally:
             await self.stop_ttfb_metrics()
             yield TTSStoppedFrame()
+
+
+class SarvamTTSService(InterruptibleTTSService):
+    """WebSocket-based text-to-speech service using Sarvam AI.
+
+    Provides streaming TTS with real-time audio generation for multiple Indian languages.
+    Supports voice control parameters like pitch, pace, and loudness adjustment.
+
+    Example::
+
+        tts = SarvamTTSService(
+            api_key="your-api-key",
+            voice_id="anushka",
+            model="bulbul:v2",
+            params=SarvamTTSService.InputParams(
+                language=Language.HI,
+                pitch=0.1,
+                pace=1.2
+            )
+        )
+    """
+
+    class InputParams(BaseModel):
+        """Configuration parameters for Sarvam TTS.
+
+        Parameters:
+            pitch: Voice pitch adjustment (-0.75 to 0.75). Defaults to 0.0.
+            pace: Speech pace multiplier (0.3 to 3.0). Defaults to 1.0.
+            loudness: Volume multiplier (0.1 to 3.0). Defaults to 1.0.
+            enable_preprocessing: Enable text preprocessing. Defaults to False.
+            min_buffer_size: Minimum number of characters to buffer before generating audio.
+                Lower values reduce latency but may affect quality. Defaults to 50.
+            max_chunk_length: Maximum number of characters processed in a single chunk.
+                Controls memory usage and processing efficiency. Defaults to 200.
+            output_audio_codec: Audio codec format. Defaults to "linear16".
+            output_audio_bitrate: Audio bitrate. Defaults to "128k".
+            language: Target language for synthesis. Supports Bengali (bn-IN), English (en-IN),
+                Gujarati (gu-IN), Hindi (hi-IN), Kannada (kn-IN), Malayalam (ml-IN),
+                Marathi (mr-IN), Odia (od-IN), Punjabi (pa-IN), Tamil (ta-IN),
+                Telugu (te-IN). Defaults to en-IN.
+
+                Available Speakers:
+            Female: anushka, manisha, vidya, arya
+            Male: abhilash, karun, hitesh
+        """
+
+        pitch: Optional[float] = Field(default=0.0, ge=-0.75, le=0.75)
+        pace: Optional[float] = Field(default=1.0, ge=0.3, le=3.0)
+        loudness: Optional[float] = Field(default=1.0, ge=0.1, le=3.0)
+        enable_preprocessing: Optional[bool] = False
+        min_buffer_size: Optional[int] = 50
+        max_chunk_length: Optional[int] = 200
+        output_audio_codec: Optional[str] = "linear16"
+        output_audio_bitrate: Optional[str] = "128k"
+        language: Optional[Language] = Language.EN
+
+    def __init__(
+        self,
+        *,
+        api_key: str,
+        model: str = "bulbul:v2",
+        voice_id: str = "anushka",
+        url: str = "wss://api.sarvam.ai/text-to-speech/ws",
+        aiohttp_session: Optional[aiohttp.ClientSession] = None,
+        aggregate_sentences: Optional[bool] = True,
+        sample_rate: Optional[int] = None,
+        params: Optional[InputParams] = None,
+        **kwargs,
+    ):
+        """Initialize the Sarvam TTS service with voice and transport configuration.
+
+        Args:
+            api_key: Sarvam API key for authenticating TTS requests.
+            model: Identifier of the Sarvam speech model (default "bulbul:v2").
+            voice_id: Voice identifier for synthesis (default "anushka").
+            url: WebSocket URL for connecting to the TTS backend (default production URL).
+            aiohttp_session: Optional shared aiohttp session. To maintain backward compatibility.
+                .. deprecated:: 0.0.81
+                    aiohttp_session is no longer used. This parameter will be removed in a future version.
+            aggregate_sentences: Whether to merge multiple sentences into one audio chunk (default True).
+            sample_rate: Desired sample rate for the output audio in Hz (overrides default if set).
+            params: Optional input parameters to override global configuration.
+            **kwargs: Optional keyword arguments forwarded to InterruptibleTTSService (such as
+                `push_stop_frames`, `sample_rate`, task manager parameters, event hooks, etc.)
+                to customize transport behavior or enable metrics support.
+
+        This method sets up the internal TTS configuration mapping, constructs the WebSocket
+        URL based on the chosen model, and initializes state flags before connecting.
+        """
+        # Initialize parent class first
+        super().__init__(
+            aggregate_sentences=aggregate_sentences,
+            push_text_frames=True,
+            pause_frame_processing=True,
+            push_stop_frames=True,
+            sample_rate=sample_rate,
+            **kwargs,
+        )
+        params = params or SarvamTTSService.InputParams()
+        if aiohttp_session is not None:
+            warnings.warn(
+                "The 'aiohttp_session' parameter is deprecated and will be removed in a future version. ",
+                DeprecationWarning,
+                stacklevel=2,
+            )
+        # WebSocket endpoint URL
+        self._websocket_url = f"{url}?model={model}"
+        self._api_key = api_key
+        self.set_model_name(model)
+        self.set_voice(voice_id)
+        # Configuration parameters
+        self._settings = {
+            "target_language_code": (
+                self.language_to_service_language(params.language) if params.language else "en-IN"
+            ),
+            "pitch": params.pitch,
+            "pace": params.pace,
+            "speaker": voice_id,
+            "loudness": params.loudness,
+            "speech_sample_rate": 0,
+            "enable_preprocessing": params.enable_preprocessing,
+            "min_buffer_size": params.min_buffer_size,
+            "max_chunk_length": params.max_chunk_length,
+            "output_audio_codec": params.output_audio_codec,
+            "output_audio_bitrate": params.output_audio_bitrate,
+        }
+        self._started = False
+
+        self._receive_task = None
+        self._keepalive_task = None
+        self._disconnecting = False
+
+    def can_generate_metrics(self) -> bool:
+        """Check if this service can generate processing metrics.
+
+        Returns:
+            True, as Sarvam service supports metrics generation.
+        """
+        return True
+
+    def language_to_service_language(self, language: Language) -> Optional[str]:
+        """Convert a Language enum to Sarvam AI language format.
+
+        Args:
+            language: The language to convert.
+
+        Returns:
+            The Sarvam AI-specific language code, or None if not supported.
+        """
+        return language_to_sarvam_language(language)
+
+    async def start(self, frame: StartFrame):
+        """Start the Sarvam TTS service.
+
+        Args:
+            frame: The start frame containing initialization parameters.
+        """
+        await super().start(frame)
+
+        self._settings["speech_sample_rate"] = self.sample_rate
+        await self._connect()
+
+    async def stop(self, frame: EndFrame):
+        """Stop the Sarvam TTS service.
+
+        Args:
+            frame: The end frame.
+        """
+        await super().stop(frame)
+        await self._disconnect()
+
+    async def cancel(self, frame: CancelFrame):
+        """Cancel the Sarvam TTS service.
+
+        Args:
+            frame: The cancel frame.
+        """
+        await super().cancel(frame)
+        await self._disconnect()
+
+    async def flush_audio(self):
+        """Flush any pending audio synthesis by sending stop command."""
+        if self._websocket:
+            msg = {"type": "flush"}
+            await self._websocket.send(json.dumps(msg))
+
+    async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
+        """Push a frame downstream with special handling for stop conditions.
+
+        Args:
+            frame: The frame to push.
+            direction: The direction to push the frame.
+        """
+        await super().push_frame(frame, direction)
+        if isinstance(frame, (TTSStoppedFrame, StartInterruptionFrame)):
+            self._started = False
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        """Process a frame and flush audio if it's the end of a full response."""
+        if isinstance(frame, LLMFullResponseEndFrame):
+            await self.flush_audio()
+        return await super().process_frame(frame, direction)
+
+    async def _update_settings(self, settings: Mapping[str, Any]):
+        """Update service settings and reconnect if voice changed."""
+        prev_voice = self._voice_id
+        await super()._update_settings(settings)
+        if not prev_voice == self._voice_id:
+            logger.info(f"Switching TTS voice to: [{self._voice_id}]")
+            await self._send_config()
+
+    async def _connect(self):
+        """Connect to Sarvam WebSocket and start background tasks."""
+        await self._connect_websocket()
+
+        if self._websocket and not self._receive_task:
+            self._receive_task = self.create_task(self._receive_task_handler(self._report_error))
+
+        if self._websocket and not self._keepalive_task:
+            self._keepalive_task = self.create_task(
+                self._keepalive_task_handler(),
+            )
+
+    async def _disconnect(self):
+        """Disconnect from Sarvam WebSocket and clean up tasks."""
+        try:
+            # First, set a flag to prevent new operations
+            self._disconnecting = True
+
+            # Cancel background tasks BEFORE closing websocket
+            if self._receive_task:
+                await self.cancel_task(self._receive_task, timeout=2.0)
+                self._receive_task = None
+
+            if self._keepalive_task:
+                await self.cancel_task(self._keepalive_task, timeout=2.0)
+                self._keepalive_task = None
+
+            # Now close the websocket
+            await self._disconnect_websocket()
+
+        except Exception as e:
+            logger.error(f"Error during disconnect: {e}")
+        finally:
+            # Reset state only after everything is cleaned up
+            self._started = False
+            self._websocket = None
+            self._disconnecting = False
+
+    async def _connect_websocket(self):
+        """Establish WebSocket connection to Sarvam API."""
+        try:
+            if self._websocket and self._websocket.state is State.OPEN:
+                return
+
+            self._websocket = await websocket_connect(
+                self._websocket_url,
+                additional_headers={
+                    "api-subscription-key": self._api_key,
+                },
+            )
+            logger.debug("Connected to Sarvam TTS Websocket")
+            await self._send_config()
+
+        except Exception as e:
+            logger.error(f"{self} initialization error: {e}")
+            self._websocket = None
+            await self._call_event_handler("on_connection_error", f"{e}")
+
+    async def _send_config(self):
+        """Send initial configuration message."""
+        if not self._websocket:
+            raise Exception("WebSocket not connected")
+        self._settings["speaker"] = self._voice_id
+        logger.debug(f"Config being sent is {self._settings}")
+        config_message = {"type": "config", "data": self._settings}
+
+        try:
+            await self._websocket.send(json.dumps(config_message))
+            logger.debug("Configuration sent successfully")
+        except Exception as e:
+            logger.error(f"Failed to send config: {str(e)}")
+            await self.push_frame(ErrorFrame(f"Failed to send config: {str(e)}"))
+            raise
+
+    async def _disconnect_websocket(self):
+        """Close WebSocket connection and clean up state."""
+        try:
+            await self.stop_all_metrics()
+
+            if self._websocket:
+                logger.debug("Disconnecting from Sarvam")
+                await self._websocket.close()
+        except Exception as e:
+            logger.error(f"{self} error closing websocket: {e}")
+
+    def _get_websocket(self):
+        if self._websocket:
+            return self._websocket
+        raise Exception("Websocket not connected")
+
+    async def _receive_messages(self):
+        """Receive and process messages from Sarvam WebSocket."""
+        async for message in self._get_websocket():
+            if isinstance(message, str):
+                msg = json.loads(message)
+                if msg.get("type") == "audio":
+                    # Check for interruption before processing audio
+                    await self.stop_ttfb_metrics()
+                    audio = base64.b64decode(msg["data"]["audio"])
+                    frame = TTSAudioRawFrame(audio, self.sample_rate, 1)
+                    await self.push_frame(frame)
+                elif msg.get("type") == "error":
+                    error_msg = msg["data"]["message"]
+                    logger.error(f"TTS Error: {error_msg}")
+
+                    # If it's a timeout error, the connection might need to be reset
+                    if "too long" in error_msg.lower() or "timeout" in error_msg.lower():
+                        logger.warning("Connection timeout detected, service may need restart")
+
+                    await self.push_frame(ErrorFrame(f"TTS Error: {error_msg}"))
+
+    async def _keepalive_task_handler(self):
+        """Handle keepalive messages to maintain WebSocket connection."""
+        KEEPALIVE_SLEEP = 20
+        while True:
+            await asyncio.sleep(KEEPALIVE_SLEEP)
+            await self._send_keepalive()
+
+    async def _send_keepalive(self):
+        """Send keepalive message to maintain connection."""
+        if self._disconnecting:
+            return
+
+        if self._websocket and self._websocket.state == State.OPEN:
+            msg = {"type": "ping"}
+            await self._websocket.send(json.dumps(msg))
+
+    async def _send_text(self, text: str):
+        """Send text to Sarvam WebSocket for synthesis."""
+        if self._disconnecting:
+            logger.warning("Service is disconnecting, ignoring text send")
+            return
+
+        if self._websocket and self._websocket.state == State.OPEN:
+            msg = {"type": "text", "data": {"text": text}}
+            await self._websocket.send(json.dumps(msg))
+        else:
+            logger.warning("WebSocket not ready, cannot send text")
+
+    @traced_tts
+    async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
+        """Generate speech audio frames from input text using Sarvam TTS.
+
+        Sends text over WebSocket for synthesis and yields corresponding audio or status frames.
+
+        Args:
+            text: The text input to synthesize.
+
+        Yields:
+            Frame objects including TTSStartedFrame, TTSAudioRawFrame(s), or TTSStoppedFrame.
+        """
+        logger.debug(f"Generating TTS: [{text}]")
+
+        try:
+            if not self._websocket or self._websocket.state is State.CLOSED:
+                await self._connect()
+
+            try:
+                if not self._started:
+                    await self.start_ttfb_metrics()
+                    yield TTSStartedFrame()
+                    self._started = True
+                await self._send_text(text)
+                await self.start_tts_usage_metrics(text)
+            except Exception as e:
+                logger.error(f"{self} error sending message: {e}")
+                yield TTSStoppedFrame()
+                await self._disconnect()
+                await self._connect()
+                return
+            yield None
+        except Exception as e:
+            logger.error(f"{self} exception: {e}")

src/pipecat/services/together/llm.py

@@ -49,3 +49,12 @@ class TogetherLLMService(OpenAILLMService):
         """
         logger.debug(f"Creating Together.ai client with api {base_url}")
         return super().create_client(api_key, base_url, **kwargs)
+
+    @property
+    def supports_universal_context(self) -> bool:
+        """Check if this service supports universal LLMContext.
+
+        Returns:
+            False, as TogetherLLMService does not yet support universal LLMContext.
+        """
+        return False

tests/test_pipeline.py

@@ -9,7 +9,9 @@ import time
 import unittest
 
 from pipecat.frames.frames import (
+    CancelFrame,
     EndFrame,
+    Frame,
     HeartbeatFrame,
     InputAudioRawFrame,
     StartFrame,
@@ -418,3 +420,33 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase):
         diff_time = time.time() - start_time
 
         self.assertGreater(diff_time, sleep_time_secs * 3)
+
+    async def test_task_cancel_timeout(self):
+        class CancelFilter(FrameProcessor):
+            def __init__(self, **kwargs):
+                super().__init__(**kwargs)
+
+            async def process_frame(self, frame: Frame, direction: FrameDirection):
+                await super().process_frame(frame, direction)
+
+                if not isinstance(frame, CancelFrame):
+                    await self.push_frame(frame, direction)
+
+        pipeline = Pipeline([CancelFilter()])
+        task = PipelineTask(pipeline, cancel_timeout_secs=0.2)
+
+        cancelled = False
+
+        @task.event_handler("on_pipeline_started")
+        async def on_pipeline_started(task: PipelineTask, frame: StartFrame):
+            await task.cancel()
+
+        @task.event_handler("on_pipeline_cancelled")
+        async def on_pipeline_cancelled(task: PipelineTask, frame: CancelFrame):
+            nonlocal cancelled
+            cancelled = True
+
+        try:
+            await task.run(PipelineTaskParams(loop=asyncio.get_event_loop()))
+        except asyncio.CancelledError:
+            assert cancelled

uv.lock

@@ -4337,6 +4337,9 @@ runner = [
     { name = "python-dotenv" },
     { name = "uvicorn" },
 ]
+sarvam = [
+    { name = "websockets" },
+]
 sentry = [
     { name = "sentry-sdk" },
 ]
@@ -4491,10 +4494,11 @@ requires-dist = [
     { name = "websockets", marker = "extra == 'openai'", specifier = ">=13.1,<15.0" },
     { name = "websockets", marker = "extra == 'playht'", specifier = ">=13.1,<15.0" },
     { name = "websockets", marker = "extra == 'rime'", specifier = ">=13.1,<15.0" },
+    { name = "websockets", marker = "extra == 'sarvam'", specifier = ">=13.1,<15.0" },
     { name = "websockets", marker = "extra == 'soniox'", specifier = ">=13.1,<15.0" },
     { name = "websockets", marker = "extra == 'websocket'", specifier = ">=13.1,<15.0" },
 ]
-provides-extras = ["anthropic", "assemblyai", "asyncai", "aws", "aws-nova-sonic", "azure", "cartesia", "cerebras", "deepseek", "daily", "deepgram", "elevenlabs", "fal", "fireworks", "fish", "gladia", "google", "grok", "groq", "gstreamer", "heygen", "inworld", "krisp", "koala", "langchain", "livekit", "lmnt", "local", "mcp", "mem0", "mistral", "mlx-whisper", "moondream", "nim", "neuphonic", "noisereduce", "openai", "openpipe", "openrouter", "perplexity", "playht", "qwen", "rime", "riva", "runner", "sambanova", "sentry", "local-smart-turn", "remote-smart-turn", "silero", "simli", "soniox", "soundfile", "speechmatics", "tavus", "together", "tracing", "ultravox", "webrtc", "websocket", "whisper"]
+provides-extras = ["anthropic", "assemblyai", "asyncai", "aws", "aws-nova-sonic", "azure", "cartesia", "cerebras", "deepseek", "daily", "deepgram", "elevenlabs", "fal", "fireworks", "fish", "gladia", "google", "grok", "groq", "gstreamer", "heygen", "inworld", "krisp", "koala", "langchain", "livekit", "lmnt", "local", "mcp", "mem0", "mistral", "mlx-whisper", "moondream", "nim", "neuphonic", "noisereduce", "openai", "openpipe", "openrouter", "perplexity", "playht", "qwen", "rime", "riva", "runner", "sambanova", "sarvam", "sentry", "local-smart-turn", "remote-smart-turn", "silero", "simli", "soniox", "soundfile", "speechmatics", "tavus", "together", "tracing", "ultravox", "webrtc", "websocket", "whisper"]
 
 [package.metadata.requires-dev]
 dev = [