introduce pipeline nodes

Make FrameProcessors async iterators and decouple them from pipeline. A new
PipelineNode now handles routing frames between processors.

CHANGELOG.md

@@ -5,10 +5,39 @@ All notable changes to **Pipecat** will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Added
+
+- Refactored pipeline architecture by introducing a new `PipelineNode`
+  abstraction.  Frame processors are now standalone async iterators, and
+  `PipelineNode` is responsible for routing frames upstream or downstream. This
+  decouples frame processors from direct linking, simplifies processor reuse,
+  and provides a clearer separation between processing logic and pipeline
+  wiring. This is an internal, transparent improvement and does not require any
+  changes to existing frame processor code.
+
+- `EndFrame` and `EndTaskFrame` have an optional `reason` field to indicate why
+  the pipeline is being ended.
+
+- `CancelFrame` and `CancelTaskFrame` have an optional `reason` field to
+  indicate why the pipeline is being canceled. This can be also specified when
+  you cancel a task with `PipelineTask.cancel(reason="cancellation your
+  reason")`.
+
+### Fixed
+
+- `GeminiLiveLLMService` now properly supports context-provided system
+  instruction and tools
+
 ## [0.0.92] - 2025-10-31 🎃 "The Haunted Edition" 👻
 
 ### Added
 
+- Added supprt for Sarvam Speech-to-Text service (`SarvamSTTService`) with
+  streaming WebSocket support for `saarika` (STT) and `saaras` (STT-translate)
+  models.
+
 - Added a new `DeepgramHttpTTSService`, which delivers a meaningful reduction
   in latency when compared to the `DeepgramTTSService`.
 

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

@@ -22,8 +22,8 @@ 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.deepgram.stt import DeepgramSTTService
 from pipecat.services.openai.llm import OpenAILLMService
+from pipecat.services.sarvam.stt import SarvamSTTService
 from pipecat.services.sarvam.tts import SarvamHttpTTSService
 from pipecat.transcriptions.language import Language
 from pipecat.transports.base_transport import BaseTransport, TransportParams
@@ -63,7 +63,10 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
 
     # Create an HTTP session
     async with aiohttp.ClientSession() as session:
-        stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+        stt = SarvamSTTService(
+            api_key=os.getenv("SARVAM_API_KEY"),
+            model="saarika:v2.5",
+        )
 
         tts = SarvamHttpTTSService(
             api_key=os.getenv("SARVAM_API_KEY"),

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

@@ -24,8 +24,8 @@ 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.deepgram.stt import DeepgramSTTService
 from pipecat.services.openai.llm import OpenAILLMService
+from pipecat.services.sarvam.stt import SarvamSTTService
 from pipecat.services.sarvam.tts import SarvamTTSService
 from pipecat.transports.base_transport import BaseTransport, TransportParams
 from pipecat.transports.daily.transport import DailyParams
@@ -62,7 +62,10 @@ transport_params = {
 async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     logger.info(f"Starting bot")
 
-    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+    stt = SarvamSTTService(
+        api_key=os.getenv("SARVAM_API_KEY"),
+        model="saarika:v2.5",
+    )
 
     tts = SarvamTTSService(
         api_key=os.getenv("SARVAM_API_KEY"),

examples/foundational/26b-gemini-live-function-calling.py

@@ -141,8 +141,16 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     llm.register_function("get_current_weather", fetch_weather_from_api)
     llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation)
 
+    # You can provide the system instructions and tools in the context rather
+    # than as arguments to GeminiLiveLLMService, but note that doing so will
+    # trigger a (fast) reconnection when the GeminiLiveLLMService first
+    # receives the context (i.e. when we send the LLMRunFrame below).
     context = LLMContext(
-        [{"role": "user", "content": "Say hello."}],
+        [
+            # {"role": "system", "content": system_instruction},
+            {"role": "user", "content": "Say hello."},
+        ],
+        # tools,
     )
     context_aggregator = LLMContextAggregatorPair(context)
 

pyproject.toml

@@ -93,7 +93,7 @@ rime = [ "pipecat-ai[websockets-base]" ]
 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 = [ "pipecat-ai[websockets-base]" ]
+sarvam = [ "sarvamai==0.1.21", "pipecat-ai[websockets-base]" ]
 sentry = [ "sentry-sdk>=2.28.0,<3" ]
 local-smart-turn = [ "coremltools>=8.0", "transformers", "torch>=2.5.0,<3", "torchaudio>=2.5.0,<3" ]
 local-smart-turn-v3 = [ "transformers", "onnxruntime>=1.20.1,<2" ]

src/pipecat/frames/frames.py

@@ -773,9 +773,15 @@ class CancelFrame(SystemFrame):
 
     Indicates that a pipeline needs to stop right away without
     processing remaining queued frames.
+
+    Parameters:
+        reason: Optional reason for pushing a cancel frame.
     """
 
-    pass
+    reason: Optional[str] = None
+
+    def __str__(self):
+        return f"{self.name}(reason: {self.reason})"
 
 
 @dataclass
@@ -1366,9 +1372,15 @@ class EndTaskFrame(TaskFrame):
     This is used to notify the pipeline task that the pipeline should be
     closed nicely (flushing all the queued frames) by pushing an EndFrame
     downstream. This frame should be pushed upstream.
+
+    Parameters:
+        reason: Optional reason for pushing an end frame.
     """
 
-    pass
+    reason: Optional[str] = None
+
+    def __str__(self):
+        return f"{self.name}(reason: {self.reason})"
 
 
 @dataclass
@@ -1378,9 +1390,15 @@ class CancelTaskFrame(TaskFrame):
     This is used to notify the pipeline task that the pipeline should be
     stopped immediately by pushing a CancelFrame downstream. This frame
     should be pushed upstream.
+
+    Parameters:
+        reason: Optional reason for pushing a cancel frame.
     """
 
-    pass
+    reason: Optional[str] = None
+
+    def __str__(self):
+        return f"{self.name}(reason: {self.reason})"
 
 
 @dataclass
@@ -1451,9 +1469,15 @@ class EndFrame(ControlFrame):
     sending frames to its output channel(s) and close all its threads. Note,
     that this is a control frame, which means it will be received in the order it
     was sent.
+
+    Parameters:
+        reason: Optional reason for pushing an end frame.
     """
 
-    pass
+    reason: Optional[str] = None
+
+    def __str__(self):
+        return f"{self.name}(reason: {self.reason})"
 
 
 @dataclass

src/pipecat/pipeline/pipeline.py

@@ -15,6 +15,7 @@ from typing import Callable, Coroutine, List, Optional
 
 from pipecat.frames.frames import Frame
 from pipecat.pipeline.base_pipeline import BasePipeline
+from pipecat.pipeline.pipeline_node import PipelineNode
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup
 
 
@@ -117,8 +118,7 @@ class Pipeline(BasePipeline):
         self._source = source or PipelineSource(self.push_frame, name=f"{self}::Source")
         self._sink = sink or PipelineSink(self.push_frame, name=f"{self}::Sink")
         self._processors: List[FrameProcessor] = [self._source] + processors + [self._sink]
-
-        self._link_processors()
+        self._nodes = self._link_processors()
 
     #
     # Frame processor
@@ -196,17 +196,22 @@ class Pipeline(BasePipeline):
 
     async def _setup_processors(self, setup: FrameProcessorSetup):
         """Set up all processors in the pipeline."""
-        for p in self._processors:
-            await p.setup(setup)
+        for n in self._nodes:
+            await n.setup(setup)
 
     async def _cleanup_processors(self):
         """Clean up all processors in the pipeline."""
-        for p in self._processors:
-            await p.cleanup()
+        for n in self._nodes:
+            await n.cleanup()
 
-    def _link_processors(self):
-        """Link all processors in sequence and set their parent."""
-        prev = self._processors[0]
+    def _link_processors(self) -> List[PipelineNode]:
+        """Link all processors in sequence."""
+        nodes = []
+        prev_node = PipelineNode(self._processors[0])
+        nodes.append(prev_node)
         for curr in self._processors[1:]:
-            prev.link(curr)
-            prev = curr
+            curr_node = PipelineNode(curr)
+            nodes.append(curr_node)
+            prev_node.link(curr_node)
+            prev_node = curr_node
+        return nodes

src/pipecat/pipeline/pipeline_node.py

@@ -0,0 +1,140 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""This module defines pipeline nodes.
+
+A pipeline node (`PipelineNode`) wraps a frame processor (`FrameProcessor`) and
+can link to previous and next nodes in the pipeline. Pipeline nodes allow
+linking frame processors together with the benefit that stateless frame
+processors can be re-used in different pipelines, since what is linked is the
+actual pipeline node, not the frame processor itself.
+
+"""
+
+import asyncio
+from typing import Optional
+
+from loguru import logger
+
+from pipecat.observers.base_observer import FramePushed
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup
+from pipecat.utils.base_object import BaseObject
+
+
+class PipelineNode(BaseObject):
+    """A node in a pipeline that hosts a frame processor.
+
+    A `PipelineNode` wraps a single `FrameProcessor` and is responsible for
+    connecting it to previous and next nodes in a pipeline. It pushes frames
+    emitted by its processor to the appropriate neighbor based on frame
+    direction (UPSTREAM or DOWNSTREAM).
+    """
+
+    def __init__(self, processor: FrameProcessor):
+        """Initialize the pipeline node with a given FrameProcessor.
+
+        Args:
+            processor: The FrameProcessor instance that this node will host.
+        """
+        super().__init__()
+        self._processor = processor
+
+        self._prev: Optional["PipelineNode"] = None
+        self._next: Optional["PipelineNode"] = None
+
+        self.__push_task: Optional[asyncio.Task] = None
+
+    @property
+    def processor(self) -> FrameProcessor:
+        """Returns the frame processor of this pipeline node."""
+        return self._processor
+
+    @property
+    def next(self) -> Optional["PipelineNode"]:
+        """Get the next pipeline node.
+
+        Returns:
+            The next node, or None if there's no next node.
+        """
+        return self._next
+
+    @property
+    def previous(self) -> Optional["PipelineNode"]:
+        """Get the previous pipeline node.
+
+        Returns:
+            The previous node, or None if there's no previous node.
+        """
+        return self._prev
+
+    async def setup(self, setup: FrameProcessorSetup):
+        """Set up this pipeline node.
+
+        This sets up the wrapped frame processor with required components.
+
+        Args:
+            setup: Configuration object containing setup parameters.
+        """
+        await self.processor.setup(setup)
+        self._clock = setup.clock
+        self._task_manager = setup.task_manager
+        self._observer = setup.observer
+
+        self.__create_push_task()
+
+    async def cleanup(self):
+        """Clean up this pipeline node."""
+        await super().cleanup()
+        await self.processor.cleanup()
+        if self.__push_task:
+            await self.__push_task
+            self.__push_task = None
+
+    def link(self, node: "PipelineNode"):
+        """Link this node to the next node in the pipeline.
+
+        Args:
+            node: The node to link to.
+        """
+        self._next = node
+        node._prev = self
+        logger.debug(f"Linking {self.processor} -> {node.processor}")
+
+    def __create_push_task(self):
+        """Create the frame push task."""
+        if not self.__push_task:
+            self.__push_task = self._task_manager.create_task(
+                self.__push_task_handler(), f"{self.processor}::_push_task"
+            )
+
+    async def __push_task_handler(self):
+        """Push task handler.
+
+        Receive frames from the wrapped frame processor and push them to the
+        next or previous node depending on the direction.
+        """
+        async for frame, direction in self.processor:
+            destination = None
+            if direction == FrameDirection.DOWNSTREAM and self.next:
+                logger.trace(f"Pushing {frame} from {self.processor} to {self.next.processor}")
+                destination = self.next.processor
+            elif direction == FrameDirection.UPSTREAM and self.previous:
+                logger.trace(f"Pushing {frame} upstream from {self} to {self._prev}")
+                destination = self.previous.processor
+
+            if destination:
+                await destination.queue_frame(frame, direction)
+
+            if self._observer and destination:
+                timestamp = self._clock.get_time() if self._clock else 0
+                data = FramePushed(
+                    source=self.processor,
+                    destination=destination,
+                    frame=frame,
+                    direction=direction,
+                    timestamp=timestamp,
+                )
+                await self._observer.on_push_frame(data)

src/pipecat/pipeline/task.py

@@ -446,10 +446,14 @@ class PipelineTask(BasePipelineTask):
         logger.debug(f"Task {self} scheduled to stop when done")
         await self.queue_frame(EndFrame())
 
-    async def cancel(self):
-        """Request the running pipeline to cancel."""
+    async def cancel(self, *, reason: Optional[str] = None):
+        """Request the running pipeline to cancel.
+
+        Args:
+            reason: Optional reason to indicate why the pipeline is being cancelled.
+        """
         if not self._finished:
-            await self._cancel()
+            await self._cancel(reason=reason)
 
     async def run(self, params: PipelineTaskParams):
         """Start and manage the pipeline execution until completion or cancellation.
@@ -513,12 +517,16 @@ class PipelineTask(BasePipelineTask):
             for frame in frames:
                 await self.queue_frame(frame)
 
-    async def _cancel(self):
-        """Internal cancellation logic for the pipeline task."""
+    async def _cancel(self, *, reason: Optional[str] = None):
+        """Internal cancellation logic for the pipeline task.
+
+        Args:
+            reason: Optional reason to indicate why the pipeline is being cancelled.
+        """
         if not self._cancelled:
             logger.debug(f"Cancelling pipeline task {self}")
             self._cancelled = True
-            await self.queue_frame(CancelFrame())
+            await self.queue_frame(CancelFrame(reason=reason))
 
     async def _create_tasks(self):
         """Create and start all pipeline processing tasks."""
@@ -716,11 +724,11 @@ class PipelineTask(BasePipelineTask):
         if isinstance(frame, EndTaskFrame):
             # Tell the task we should end nicely.
             logger.debug(f"{self}: received end task frame {frame}")
-            await self.queue_frame(EndFrame())
+            await self.queue_frame(EndFrame(reason=frame.reason))
         elif isinstance(frame, CancelTaskFrame):
             # Tell the task we should end right away.
             logger.debug(f"{self}: received cancel task frame {frame}")
-            await self.queue_frame(CancelFrame())
+            await self.queue_frame(CancelFrame(reason=frame.reason))
         elif isinstance(frame, StopTaskFrame):
             # Tell the task we should stop nicely.
             logger.debug(f"{self}: received stop task frame {frame}")

src/pipecat/processors/frame_processor.py

@@ -132,14 +132,17 @@ INPUT_TASK_CANCEL_TIMEOUT_SECS = 3
 
 
 class FrameProcessor(BaseObject):
-    """Base class for all frame processors in the pipeline.
+    """Base class for all frame processors in Pipecat.
 
-    Frame processors are the building blocks of Pipecat pipelines, they can be
-    linked to form complex processing pipelines. They receive frames, process
-    them, and pass them to the next or previous processor in the chain.  Each
-    frame processor guarantees frame ordering and processes frames in its own
-    task. System frames are also processed in a separate task which guarantees
-    frame priority.
+    A FrameProcessor is an independent, asynchronous component that consumes
+    input frames and produces zero or more output frames. Frames are delivered
+    to the processor via the `queue_frame(frame, direction)` method. The
+    processor internally manages queues and background tasks to handle incoming
+    frames and generate output frames.
+
+    Output frames are made available through the processor's asynchronous
+    iterator interface, allowing consumers to iterate over processed frames
+    using `async for frame in processor`. Frame ordering is guaranteed.
 
     Event handlers available:
 
@@ -147,6 +150,7 @@ class FrameProcessor(BaseObject):
     - on_after_process_frame: Called after a frame is processed
     - on_before_push_frame: Called before a frame is pushed
     - on_after_push_frame: Called after a frame is pushed
+
     """
 
     def __init__(
@@ -166,8 +170,6 @@ class FrameProcessor(BaseObject):
             **kwargs: Additional arguments passed to parent class.
         """
         super().__init__(name=name, **kwargs)
-        self._prev: Optional["FrameProcessor"] = None
-        self._next: Optional["FrameProcessor"] = None
 
         # Enable direct mode to skip queues and process frames right away.
         self._enable_direct_mode = enable_direct_mode
@@ -234,6 +236,9 @@ class FrameProcessor(BaseObject):
         self._wait_for_interruption = False
         self._wait_interruption_event = asyncio.Event()
 
+        # Push queue
+        self.__push_queue = asyncio.Queue()
+
         # Frame processor events.
         self._register_event_handler("on_before_process_frame", sync=True)
         self._register_event_handler("on_after_process_frame", sync=True)
@@ -284,24 +289,6 @@ class FrameProcessor(BaseObject):
         """
         return []
 
-    @property
-    def next(self) -> Optional["FrameProcessor"]:
-        """Get the next processor.
-
-        Returns:
-            The next processor, or None if there's no next processor.
-        """
-        return self._next
-
-    @property
-    def previous(self) -> Optional["FrameProcessor"]:
-        """Get the previous processor.
-
-        Returns:
-            The previous processor, or None if there's no previous processor.
-        """
-        return self._prev
-
     @property
     def interruptions_allowed(self):
         """Check if interruptions are allowed for this processor.
@@ -518,16 +505,7 @@ class FrameProcessor(BaseObject):
         await self.__cancel_process_task()
         if self._metrics is not None:
             await self._metrics.cleanup()
-
-    def link(self, processor: "FrameProcessor"):
-        """Link this processor to the next processor in the pipeline.
-
-        Args:
-            processor: The processor to link to.
-        """
-        self._next = processor
-        processor._prev = self
-        logger.debug(f"Linking {self} -> {self._next}")
+        await self.__push_queue.put(None)
 
     def get_clock(self) -> BaseClock:
         """Get the clock used by this processor.
@@ -761,36 +739,7 @@ class FrameProcessor(BaseObject):
             frame: The frame to push.
             direction: The direction to push the frame.
         """
-        try:
-            timestamp = self._clock.get_time() if self._clock else 0
-            if direction == FrameDirection.DOWNSTREAM and self._next:
-                logger.trace(f"Pushing {frame} from {self} to {self._next}")
-
-                if self._observer:
-                    data = FramePushed(
-                        source=self,
-                        destination=self._next,
-                        frame=frame,
-                        direction=direction,
-                        timestamp=timestamp,
-                    )
-                    await self._observer.on_push_frame(data)
-                await self._next.queue_frame(frame, direction)
-            elif direction == FrameDirection.UPSTREAM and self._prev:
-                logger.trace(f"Pushing {frame} upstream from {self} to {self._prev}")
-                if self._observer:
-                    data = FramePushed(
-                        source=self,
-                        destination=self._prev,
-                        frame=frame,
-                        direction=direction,
-                        timestamp=timestamp,
-                    )
-                    await self._observer.on_push_frame(data)
-                await self._prev.queue_frame(frame, direction)
-        except Exception as e:
-            logger.exception(f"Uncaught exception in {self}: {e}")
-            await self.push_error(ErrorFrame(str(e)))
+        await self.__push_queue.put((frame, direction))
 
     def _check_started(self, frame: Frame):
         """Check if the processor has been started.
@@ -912,3 +861,18 @@ class FrameProcessor(BaseObject):
             await self.__process_frame(frame, direction, callback)
 
             self.__process_queue.task_done()
+
+    def __aiter__(self):
+        """A frame processor is an asynchronous iterator itself."""
+        return self
+
+    async def __anext__(self):
+        """Retrieve the next frame to push from this processor.
+
+        Returns:
+            The next (frame, direction) item to push form this processor.
+        """
+        data = await self.__push_queue.get()
+        if data is None:
+            raise StopAsyncIteration
+        return data

src/pipecat/services/google/gemini_live/llm.py

@@ -672,8 +672,8 @@ class GeminiLiveLLMService(LLMService):
         self._voice_id = voice_id
         self._language_code = params.language
 
-        self._system_instruction = system_instruction
-        self._tools = tools
+        self._system_instruction_from_init = system_instruction
+        self._tools_from_init = tools
         self._inference_on_context_initialization = inference_on_context_initialization
         self._needs_turn_complete_message = False
 
@@ -964,16 +964,51 @@ class GeminiLiveLLMService(LLMService):
         if not self._context:
             # We got our initial context
             self._context = context
-            if context.tools:
-                self._tools = context.tools
+
+            # If context contains system instruction or tools, reconnect in
+            # order to apply them.
+            # (Context-provided system instruction and tools take precedence
+            # over the ones provided at initialization time. Note that we could
+            # do more sophisticated comparisons here, but for now this is
+            # sufficient: we'll assume folks won't mean to provide these
+            # settings both in the context and at initialization time. In a
+            # future change, we could/should implement the ability to swap
+            # these settings at any point).
+            adapter: GeminiLLMAdapter = self.get_llm_adapter()
+            params = adapter.get_llm_invocation_params(self._context)
+            system_instruction = params["system_instruction"]
+            tools = params["tools"]
+            if system_instruction and self._system_instruction_from_init:
+                logger.warning(
+                    "System instruction provided both at init time and in context; using context-provided value."
+                )
+            if tools and self._tools_from_init:
+                logger.warning(
+                    "Tools provided both at init time and in context; using context-provided value."
+                )
+            if system_instruction or tools:
+                await self._reconnect()
+
             # Initialize our bookkeeping of already-completed tool calls in
             # the context
             await self._process_completed_function_calls(send_new_results=False)
+
+            # Create initial response if needed, based on conversation history
+            # in context
             await self._create_initial_response()
         else:
             # We got an updated context.
-            # This may contain a new user message or tool call result.
             self._context = context
+
+            # Here we assume that the updated context will contain either:
+            # - new messages (that the Gemini Live service, with its own
+            #   context management, is already aware of), or
+            # - tool call results (that we need to tell the remote service
+            #   about).
+            # (In the future, we could do more sophisticated diffing here,
+            # which would enable the user to programmatically manipulate the
+            # context).
+
             # Send results for newly-completed function calls, if any.
             await self._process_completed_function_calls(send_new_results=True)
 
@@ -1103,18 +1138,25 @@ class GeminiLiveLLMService(LLMService):
                         automatic_activity_detection=vad_config
                     )
 
-            # Add system instruction to configuration, if provided
-            system_instruction = self._system_instruction or ""
-            if self._context and hasattr(self._context, "extract_system_instructions"):
-                system_instruction += "\n" + self._context.extract_system_instructions()
+            # Add system instruction and tools to configuration, if provided.
+            # These settings from the context take precedence over the ones
+            # provided at initialization time.
+            adapter: GeminiLLMAdapter = self.get_llm_adapter()
+            system_instruction = None
+            tools = None
+            if self._context:
+                params = adapter.get_llm_invocation_params(self._context)
+                system_instruction = params["system_instruction"]
+                tools = params["tools"]
+            else:
+                system_instruction = self._system_instruction_from_init
+                tools = adapter.from_standard_tools(self._tools_from_init)
             if system_instruction:
                 logger.debug(f"Setting system instruction: {system_instruction}")
                 config.system_instruction = system_instruction
-
-            # Add tools to configuration, if provided
-            if self._tools:
-                logger.debug(f"Setting tools: {self._tools}")
-                config.tools = self.get_llm_adapter().from_standard_tools(self._tools)
+            if tools:
+                logger.debug(f"Setting tools: {tools}")
+                config.tools = tools
 
             # Start the connection
             self._connection_task = self.create_task(self._connection_task_handler(config=config))

src/pipecat/services/sarvam/stt.py

@@ -0,0 +1,468 @@
+"""Sarvam AI Speech-to-Text service implementation.
+
+This module provides a streaming Speech-to-Text service using Sarvam AI's WebSocket-based
+API. It supports real-time transcription with Voice Activity Detection (VAD) and
+can handle multiple audio formats for Indian language speech recognition.
+"""
+
+import base64
+from typing import Optional
+
+from loguru import logger
+from pydantic import BaseModel
+
+from pipecat.frames.frames import (
+    CancelFrame,
+    EndFrame,
+    ErrorFrame,
+    StartFrame,
+    TranscriptionFrame,
+)
+from pipecat.services.stt_service import STTService
+from pipecat.transcriptions.language import Language
+from pipecat.utils.time import time_now_iso8601
+from pipecat.utils.tracing.service_decorators import traced_stt
+
+try:
+    from sarvamai import AsyncSarvamAI
+    from sarvamai.core.api_error import ApiError
+    from sarvamai.core.events import EventType
+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) -> str:
+    """Convert a Language enum to Sarvam's language code format.
+
+    Args:
+        language: The Language enum value to convert.
+
+    Returns:
+        The Sarvam language code string.
+    """
+    # Mapping of pipecat Language enum to Sarvam language codes
+    SARVAM_LANGUAGES = {
+        Language.BN_IN: "bn-IN",
+        Language.GU_IN: "gu-IN",
+        Language.HI_IN: "hi-IN",
+        Language.KN_IN: "kn-IN",
+        Language.ML_IN: "ml-IN",
+        Language.MR_IN: "mr-IN",
+        Language.TA_IN: "ta-IN",
+        Language.TE_IN: "te-IN",
+        Language.PA_IN: "pa-IN",
+        Language.OR_IN: "od-IN",
+        Language.EN_IN: "en-IN",
+        Language.AS_IN: "as-IN",
+    }
+
+    return SARVAM_LANGUAGES.get(
+        language, "unknown"
+    )  # Default to unknown (Sarvam models auto-detect the language)
+
+
+class SarvamSTTService(STTService):
+    """Sarvam speech-to-text service.
+
+    Provides real-time speech recognition using Sarvam's WebSocket API.
+    """
+
+    class InputParams(BaseModel):
+        """Configuration parameters for Sarvam STT service.
+
+        Parameters:
+            language: Target language for transcription. Defaults to None (required for saarika models).
+            prompt: Optional prompt to guide translation style/context for STT-Translate models.
+                   Only applicable to saaras (STT-Translate) models. Defaults to None.
+            vad_signals: Enable VAD signals in response. Defaults to True.
+            high_vad_sensitivity: Enable high VAD (Voice Activity Detection) sensitivity. Defaults to False.
+        """
+
+        language: Optional[Language] = None
+        prompt: Optional[str] = None
+        vad_signals: bool = True
+        high_vad_sensitivity: bool = False
+
+    def __init__(
+        self,
+        *,
+        api_key: str,
+        model: str = "saarika:v2.5",
+        sample_rate: Optional[int] = None,
+        input_audio_codec: str = "wav",
+        params: Optional[InputParams] = None,
+        **kwargs,
+    ):
+        """Initialize the Sarvam STT service.
+
+        Args:
+            api_key: Sarvam API key for authentication.
+            model: Sarvam model to use for transcription.
+            sample_rate: Audio sample rate. Defaults to 16000 if not specified.
+            input_audio_codec: Audio codec/format of the input file. Defaults to "wav".
+            params: Configuration parameters for Sarvam STT service.
+            **kwargs: Additional arguments passed to the parent STTService.
+        """
+        params = params or SarvamSTTService.InputParams()
+
+        # Validate that saaras models don't accept language parameter
+        if "saaras" in model.lower():
+            if params.language is not None:
+                raise ValueError(
+                    f"Model '{model}' does not accept language parameter. "
+                    "STT-Translate models auto-detect language."
+                )
+
+        # Validate that saarika models don't accept prompt parameter
+        if "saarika" in model.lower():
+            if params.prompt is not None:
+                raise ValueError(
+                    f"Model '{model}' does not accept prompt parameter. "
+                    "Prompts are only supported for STT-Translate models"
+                )
+
+        super().__init__(sample_rate=sample_rate, **kwargs)
+
+        self.set_model_name(model)
+        self._api_key = api_key
+        self._language_code = params.language
+        # For saarika models, default to "unknown" if language is not provided
+        if params.language:
+            self._language_string = language_to_sarvam_language(params.language)
+        elif "saarika" in model.lower():
+            self._language_string = "unknown"
+        else:
+            self._language_string = None
+        self._prompt = params.prompt
+
+        # Store connection parameters
+        self._vad_signals = params.vad_signals
+        self._high_vad_sensitivity = params.high_vad_sensitivity
+        self._input_audio_codec = input_audio_codec
+
+        # Initialize Sarvam SDK client
+        self._sarvam_client = AsyncSarvamAI(api_subscription_key=api_key)
+        self._websocket_context = None
+        self._socket_client = None
+        self._receive_task = None
+
+    def language_to_service_language(self, language: Language) -> str:
+        """Convert pipecat Language enum to Sarvam's language code.
+
+        Args:
+            language: The Language enum value to convert.
+
+        Returns:
+            The Sarvam language code string.
+        """
+        return language_to_sarvam_language(language)
+
+    def can_generate_metrics(self) -> bool:
+        """Check if this service can generate processing metrics.
+
+        Returns:
+            True, as Sarvam service supports metrics generation.
+        """
+        return True
+
+    async def set_language(self, language: Language):
+        """Set the recognition language and reconnect.
+
+        Args:
+            language: The language to use for speech recognition.
+        """
+        # saaras models do not accept a language parameter
+        if "saaras" in self.model_name.lower():
+            raise ValueError(
+                f"Model '{self.model_name}' (saaras) does not accept language parameter. "
+                "saaras models auto-detect language."
+            )
+
+        logger.info(f"Switching STT language to: [{language}]")
+        self._language_code = language
+        self._language_string = language_to_sarvam_language(language)
+        await self._disconnect()
+        await self._connect()
+
+    async def set_prompt(self, prompt: Optional[str]):
+        """Set the translation prompt and reconnect.
+
+        Args:
+            prompt: Prompt text to guide translation style/context.
+                   Pass None to clear/disable prompt.
+                   Only applicable to STT-Translate models, not STT models.
+        """
+        # saarika models do not accept prompt parameter
+        if "saarika" in self.model_name.lower():
+            if prompt is not None:
+                raise ValueError(
+                    f"Model '{self.model_name}' does not accept prompt parameter. "
+                    "Prompts are only supported for STT-Translate models."
+                )
+            # If prompt is None and it's saarika, just silently return (no-op)
+            return
+
+        logger.info("Updating STT-Translate prompt.")
+        self._prompt = prompt
+        await self._disconnect()
+        await self._connect()
+
+    async def start(self, frame: StartFrame):
+        """Start the Sarvam STT service.
+
+        Args:
+            frame: The start frame containing initialization parameters.
+        """
+        await super().start(frame)
+        await self._connect()
+
+    async def stop(self, frame: EndFrame):
+        """Stop the Sarvam STT service.
+
+        Args:
+            frame: The end frame.
+        """
+        await super().stop(frame)
+        await self._disconnect()
+
+    async def cancel(self, frame: CancelFrame):
+        """Cancel the Sarvam STT service.
+
+        Args:
+            frame: The cancel frame.
+        """
+        await super().cancel(frame)
+        await self._disconnect()
+
+    async def run_stt(self, audio: bytes):
+        """Send audio data to Sarvam for transcription.
+
+        Args:
+            audio: Raw audio bytes to transcribe.
+
+        Yields:
+            Frame: None (transcription results come via WebSocket callbacks).
+        """
+        if not self._socket_client:
+            logger.warning("WebSocket not connected, cannot process audio")
+            yield None
+            return
+
+        try:
+            # Convert audio bytes to base64 for Sarvam API
+            audio_base64 = base64.b64encode(audio).decode("utf-8")
+
+            # Convert input_audio_codec to encoding format (prepend "audio/" if needed)
+            encoding = (
+                self._input_audio_codec
+                if self._input_audio_codec.startswith("audio/")
+                else f"audio/{self._input_audio_codec}"
+            )
+
+            # Build method arguments
+            method_kwargs = {
+                "audio": audio_base64,
+                "encoding": encoding,
+                "sample_rate": self.sample_rate,
+            }
+
+            # Use appropriate method based on service type
+            if "saarika" in self.model_name.lower():
+                # STT service
+                await self._socket_client.transcribe(**method_kwargs)
+            else:
+                # STT-Translate service - auto-detects input language and returns translated text
+                await self._socket_client.translate(**method_kwargs)
+
+        except Exception as e:
+            logger.error(f"Error sending audio to Sarvam: {e}")
+            await self.push_error(ErrorFrame(f"Failed to send audio: {e}"))
+
+        yield None
+
+    async def _connect(self):
+        """Connect to Sarvam WebSocket API using the SDK."""
+        logger.debug("Connecting to Sarvam")
+
+        try:
+            # Convert boolean parameters to string for SDK
+            vad_signals_str = "true" if self._vad_signals else "false"
+            high_vad_sensitivity_str = "true" if self._high_vad_sensitivity else "false"
+
+            # Build common connection parameters
+            connect_kwargs = {
+                "model": self.model_name,
+                "vad_signals": vad_signals_str,
+                "high_vad_sensitivity": high_vad_sensitivity_str,
+                "input_audio_codec": self._input_audio_codec,
+                "sample_rate": str(self.sample_rate),
+            }
+
+            # Choose the appropriate service based on model
+            if "saarika" in self.model_name.lower():
+                # STT service - requires language_code
+                connect_kwargs["language_code"] = self._language_string
+                self._websocket_context = self._sarvam_client.speech_to_text_streaming.connect(
+                    **connect_kwargs
+                )
+            else:
+                # STT-Translate service - auto-detects input language and returns translated text
+                self._websocket_context = (
+                    self._sarvam_client.speech_to_text_translate_streaming.connect(**connect_kwargs)
+                )
+
+            # Enter the async context manager
+            self._socket_client = await self._websocket_context.__aenter__()
+
+            # Set prompt if provided (only for STT-Translate models, after connection)
+            if self._prompt is not None and "saaras" in self.model_name.lower():
+                await self._socket_client.set_prompt(self._prompt)
+
+            # Register event handler for incoming messages
+            def _message_handler(message):
+                """Wrapper to handle async response handler."""
+                # Use Pipecat's built-in task management
+                self.create_task(self._handle_message(message))
+
+            self._socket_client.on(EventType.MESSAGE, _message_handler)
+
+            # Start receive task using Pipecat's task management
+            self._receive_task = self.create_task(self._receive_task_handler())
+
+            logger.info("Connected to Sarvam successfully")
+
+        except ApiError as e:
+            logger.error(f"Sarvam API error: {e}")
+            await self.push_error(ErrorFrame(f"Sarvam API error: {e}"))
+        except Exception as e:
+            logger.error(f"Failed to connect to Sarvam: {e}")
+            self._socket_client = None
+            self._websocket_context = None
+            await self.push_error(ErrorFrame(f"Failed to connect to Sarvam: {e}"))
+
+    async def _disconnect(self):
+        """Disconnect from Sarvam WebSocket API using SDK."""
+        if self._receive_task:
+            await self.cancel_task(self._receive_task)
+            self._receive_task = None
+
+        if self._websocket_context and self._socket_client:
+            try:
+                # Exit the async context manager
+                await self._websocket_context.__aexit__(None, None, None)
+            except Exception as e:
+                logger.error(f"Error closing WebSocket connection: {e}")
+            finally:
+                logger.debug("Disconnected from Sarvam WebSocket")
+                self._socket_client = None
+                self._websocket_context = None
+
+    async def _receive_task_handler(self):
+        """Handle incoming messages from Sarvam WebSocket.
+
+        This task wraps the SDK's start_listening() method which processes
+        messages via the registered event handler callback.
+        """
+        if not self._socket_client:
+            return
+
+        try:
+            # Start listening for messages from the Sarvam SDK
+            # Messages will be handled via the _message_handler callback
+            await self._socket_client.start_listening()
+        except Exception as e:
+            logger.error(f"Error in Sarvam receive task: {e}")
+            await self.push_error(ErrorFrame(f"Sarvam receive task error: {e}"))
+
+    async def _handle_message(self, message):
+        """Handle incoming WebSocket message from Sarvam SDK.
+
+        Processes transcription data and VAD events from the Sarvam service.
+
+        Args:
+            message: The parsed response object from Sarvam WebSocket.
+        """
+        logger.debug(f"Received response: {message}")
+
+        try:
+            if message.type == "events":
+                # VAD event
+                signal = message.data.signal_type
+                timestamp = message.data.occured_at
+                logger.debug(f"VAD Signal: {signal}, Occurred at: {timestamp}")
+
+                if signal == "START_SPEECH":
+                    await self.start_metrics()
+                    logger.debug("User started speaking")
+                    await self._call_event_handler("on_speech_started")
+
+            elif message.type == "data":
+                await self.stop_ttfb_metrics()
+                transcript = message.data.transcript
+                language_code = message.data.language_code
+                # Prefer language from message (auto-detected for translate models). Fallback to configured.
+                if language_code:
+                    language = self._map_language_code_to_enum(language_code)
+                elif self._language_string:
+                    language = self._map_language_code_to_enum(self._language_string)
+                else:
+                    language = Language.HI_IN
+
+                # Emit utterance end event
+                await self._call_event_handler("on_utterance_end")
+
+                if transcript and transcript.strip():
+                    # Record tracing for this transcription event
+                    await self._handle_transcription(transcript, True, language)
+                    await self.push_frame(
+                        TranscriptionFrame(
+                            transcript,
+                            self._user_id,
+                            time_now_iso8601(),
+                            language,
+                            result=(message.dict() if hasattr(message, "dict") else str(message)),
+                        )
+                    )
+
+                await self.stop_processing_metrics()
+
+        except Exception as e:
+            logger.error(f"Error handling Sarvam message: {e}")
+            await self.push_error(ErrorFrame(f"Failed to handle message: {e}"))
+            await self.stop_all_metrics()
+
+    @traced_stt
+    async def _handle_transcription(
+        self, transcript: str, is_final: bool, language: Optional[Language] = None
+    ):
+        """Handle a transcription result with tracing.
+
+        This method is decorated with @traced_stt for observability.
+        """
+        pass
+
+    def _map_language_code_to_enum(self, language_code: str) -> Language:
+        """Map Sarvam language code to pipecat Language enum."""
+        mapping = {
+            "bn-IN": Language.BN_IN,
+            "gu-IN": Language.GU_IN,
+            "hi-IN": Language.HI_IN,
+            "kn-IN": Language.KN_IN,
+            "ml-IN": Language.ML_IN,
+            "mr-IN": Language.MR_IN,
+            "ta-IN": Language.TA_IN,
+            "te-IN": Language.TE_IN,
+            "pa-IN": Language.PA_IN,
+            "od-IN": Language.OR_IN,
+            "en-US": Language.EN_US,
+            "en-IN": Language.EN_IN,
+            "as-IN": Language.AS_IN,
+        }
+        return mapping.get(language_code, Language.HI_IN)
+
+    async def start_metrics(self):
+        """Start TTFB and processing metrics collection."""
+        await self.start_ttfb_metrics()
+        await self.start_processing_metrics()

uv.lock

@@ -4550,6 +4550,7 @@ runner = [
     { name = "uvicorn" },
 ]
 sarvam = [
+    { name = "sarvamai" },
     { name = "websockets" },
 ]
 sentry = [
@@ -4704,6 +4705,7 @@ requires-dist = [
     { name = "python-dotenv", marker = "extra == 'runner'", specifier = ">=1.0.0,<2.0.0" },
     { name = "pyvips", extras = ["binary"], marker = "extra == 'moondream'", specifier = "~=3.0.0" },
     { name = "resampy", specifier = "~=0.4.3" },
+    { name = "sarvamai", marker = "extra == 'sarvam'", specifier = "==0.1.21" },
     { name = "sentry-sdk", marker = "extra == 'sentry'", specifier = ">=2.28.0,<3" },
     { name = "simli-ai", marker = "extra == 'simli'", specifier = "~=0.1.10" },
     { name = "soundfile", marker = "extra == 'soundfile'", specifier = "~=0.13.0" },
@@ -6212,6 +6214,22 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/2c/c3/c0be1135726618dc1e28d181b8c442403d8dbb9e273fd791de2d4384bcdd/safetensors-0.6.2-cp38-abi3-win_amd64.whl", hash = "sha256:c7b214870df923cbc1593c3faee16bec59ea462758699bd3fee399d00aac072c", size = 320192, upload-time = "2025-08-08T13:13:59.467Z" },
 ]
 
+[[package]]
+name = "sarvamai"
+version = "0.1.21"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "httpx" },
+    { name = "pydantic" },
+    { name = "pydantic-core" },
+    { name = "typing-extensions" },
+    { name = "websockets" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/e9/08/e5efcb30818ed220b818319255c22fd91e379489ebaa93efd6f444fb4987/sarvamai-0.1.21.tar.gz", hash = "sha256:865065635b2b99d40f5519308832954015627938e06a6333b5f62ae9c36278bb", size = 87386, upload-time = "2025-10-07T07:37:47.085Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/2e/4e/b9933f72681b7aed91b86913337dd3981fad97027881fbc66c3c5eb03568/sarvamai-0.1.21-py3-none-any.whl", hash = "sha256:daa4e5d16635fe434f5f270cee416849249285369141d77132a17f0bf670f120", size = 175204, upload-time = "2025-10-07T07:37:46.024Z" },
+]
+
 [[package]]
 name = "scipy"
 version = "1.15.3"