Pipecat

https://reference-server.pipecat.ai
Admin
Pipecat is a framework for building real-time, conversational AI applications with support for...
Tokens:266,450
Snippets:1,602
Trust Score:7.5
Update:6 months ago
Show doc for...
Tokens:
Raw
### Logging Examples

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/observers/loggers/metrics_log_observer.html

Examples demonstrating how to initialize and use the MetricsLogObserver.

```APIDOC
## Logging Examples

### Log all metrics types
```python
observers = [MetricsLogObserver()]
```

### Log only LLM and TTS metrics
```python
from pipecat.metrics.metrics import LLMUsageMetricsData, TTSUsageMetricsData
observers = [
    MetricsLogObserver(
        include_metrics={LLMUsageMetricsData, TTSUsageMetricsData}
    )
]
```
```

--------------------------------

### POST /client/start

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/transports/daily/transport.html

Starts the client and initializes audio/video components based on the provided start frame parameters.

```APIDOC
## POST /client/start

### Description
Configures audio and video tracks, initializes custom sources, and sets up callback tasks for media processing.

### Method
POST

### Endpoint
/client/start

### Parameters
#### Request Body
- **frame** (StartFrame) - Required - The start frame containing audio/video initialization parameters.

### Request Example
{
  "frame": { "audio_in_sample_rate": 16000, "video_in_enabled": true }
}

### Response
#### Success Response (200)
- **status** (string) - Returns 'started' upon successful initialization.
```

--------------------------------

### Create Start Metadata (Python)

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/pipeline/task.html

Builds and returns a dictionary containing start metadata. It checks for deprecated LLM context aggregators and merges user-provided start metadata from parameters.

```Python
def _create_start_metadata(self) -> Dict[str, Any]:
        """Build and return start metadata including user-provided values."""
        start_metadata = {}

        # NOTE(aleix): Remove when OpenAILLMContext/LLMUserContextAggregator is removed.
        if self._find_processor(self._pipeline, LLMUserContextAggregator):
            start_metadata["deprecated_openaillmcontext"] = True

        # Update with user provided metadata.
        start_metadata.update(self._params.start_metadata)

        return start_metadata
```

--------------------------------

### Initialize and Use SystemClock

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/clocks/system_clock.html

Demonstrates how to instantiate the SystemClock class, start the timer, and retrieve the elapsed time in nanoseconds. The clock must be started explicitly before it begins tracking time.

```python
from pipecat.clocks.system_clock import SystemClock

# Initialize the clock
clock = SystemClock()

# Start the clock
clock.start()

# Get elapsed time
elapsed = clock.get_time()
print(f"Elapsed time: {elapsed} ns")
```

--------------------------------

### Setup Pipeline and Processors (Python)

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/pipeline/pipeline.html

Asynchronously sets up the entire pipeline and all its constituent processors. It first calls the parent class's setup method and then proceeds to set up the individual processors using the provided FrameProcessorSetup configuration.

```python
async def setup(self, setup: FrameProcessorSetup):
        """Set up the pipeline and all contained processors.

        Args:
            setup: Configuration for frame processor setup.
        """
        await super().setup(setup)
        await self._setup_processors(setup)
```

--------------------------------

### Setup Input Transport

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/transports/daily/transport.html

Performs the setup for the input transport, including shared client setup.

```APIDOC
## DailyInputTransport - setup

### Description
Sets up the input transport with shared client setup configurations.

### Method
setup

### Parameters
#### Arguments
- **setup** (FrameProcessorSetup) - The frame processor setup configuration.

### Request Example
```python
await daily_input_transport.setup(frame_processor_setup_config)
```

### Response
#### Success Response (200)
Input transport is successfully set up.

#### Response Example
```python
# No direct response, setup is performed asynchronously.
```
```

--------------------------------

### Load Dynamic Setup Files

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/pipeline/task.html

Dynamically imports and executes setup functions from files specified in the PIPECAT_SETUP_FILES environment variable. This allows for modular configuration of pipeline tasks.

```python
async def _load_setup_files(self):
    setup_files = [f for f in os.environ.get("PIPECAT_SETUP_FILES", "").split(":") if f]
    for f in setup_files:
        try:
            path = Path(f).resolve()
            module_name = path.stem
            spec = importlib.util.spec_from_file_location(module_name, str(path))
            if spec and spec.loader:
                module = importlib.util.module_from_spec(spec)
                spec.loader.exec_module(module)
                if hasattr(module, "setup_pipeline_task"):
                    await module.setup_pipeline_task(self)
        except Exception as e:
            logger.error(f"{self} error running external setup from {f}: {e}")
```

--------------------------------

### Initialize and Configure ParallelPipeline

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/pipeline/parallel_pipeline.html

Demonstrates how to instantiate a ParallelPipeline with multiple branches and manage its lifecycle via setup and cleanup methods.

```python
from pipecat.pipeline.parallel_pipeline import ParallelPipeline

# Define processor lists for parallel branches
branch1 = [processor_a, processor_b]
branch2 = [processor_c, processor_d]

# Initialize the pipeline
parallel_pipe = ParallelPipeline(branch1, branch2)

# Setup and cleanup are handled asynchronously
await parallel_pipe.setup(setup_config)
# ... processing ...
await parallel_pipe.cleanup()
```

--------------------------------

### Internal Setup of Pipeline Processors (Python)

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/pipeline/pipeline.html

Internal asynchronous method to set up all individual processors within the pipeline. It iterates through the list of processors and calls the setup method for each one, passing the provided setup configuration.

```python
async def _setup_processors(self, setup: FrameProcessorSetup):
        """Set up all processors in the pipeline."""
        for p in self._processors:
            await p.setup(setup)
```

--------------------------------

### KrispVivaFilter Start

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/audio/filters/krisp_viva_filter.html

Starts the KrispVivaFilter session with the given sample rate.

```APIDOC
## KrispVivaFilter Start

### Description
Starts the KrispVivaFilter session, preparing it to process audio at the specified sample rate.

### Method
start

### Parameters
#### Path Parameters
- **sample_rate** (int) - Required - The sample rate of the audio stream.
```

--------------------------------

### Initialize and Manage TaskObserver

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/pipeline/task_observer.html

Demonstrates how to instantiate a TaskObserver with a task manager and manage the lifecycle of observers by adding, starting, and removing them asynchronously.

```python
from pipecat.observers.task_observer import TaskObserver

# Initialize the observer with a task manager
task_observer = TaskObserver(task_manager=my_task_manager)

# Add an observer
task_observer.add_observer(my_observer)

# Start the proxy tasks
await task_observer.start()

# Remove an observer when no longer needed
await task_observer.remove_observer(my_observer)

# Stop all tasks
await task_observer.stop()
```

--------------------------------

### Manage Pipeline Lifecycle

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/pipeline/sync_parallel_pipeline.html

Illustrates the asynchronous setup and cleanup methods used to initialize or tear down all processors within the parallel pipeline structure.

```python
# Setup all pipelines
await pipeline.setup(setup_config)

# Cleanup all pipelines
await pipeline.cleanup()
```

--------------------------------

### Start Input Transport

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/transports/daily/transport.html

Initiates the input transport and joins the Daily room.

```APIDOC
## DailyInputTransport - start

### Description
Starts the input transport and joins the Daily room. This method should be called after initialization and setup.

### Method
start

### Parameters
#### Arguments
- **frame** (StartFrame) - The start frame containing initialization parameters for joining the room.

### Request Example
```python
await daily_input_transport.start(start_frame_data)
```

### Response
#### Success Response (200)
Input transport is started and the client has joined the Daily room.

#### Response Example
```python
# No direct response, the state is updated internally.
```
```

--------------------------------

### POST /start_recording

Source: https://reference-server.pipecat.ai/en/latest/api/pipecat.transports.daily.transport.html

Starts a recording session for the current call.

```APIDOC
## POST /start_recording

### Description
Starts recording the call with specific streaming settings.

### Method
POST

### Endpoint
/start_recording

### Parameters
#### Request Body
- **streaming_settings** (object) - Required - Recording configuration.
- **stream_id** (string) - Required - Unique identifier for the stream.
- **force_new** (boolean) - Required - Whether to force a new session.

### Response
#### Success Response (200)
- **stream_id** (string) - The identifier for the recording stream.
- **error** (string|null) - Error description if applicable.
```

--------------------------------

### TTSGate Initialization and Setup

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/extensions/voicemail/voicemail_detector.html

Initializes the TTSGate processor with conversation and voicemail notifiers and sets up the internal tasks for monitoring classification events.

```APIDOC
## POST /processor/tts-gate/setup

### Description
Initializes the TTSGate processor. This component buffers TTS-related frames (TTSTextFrame, TTSAudioRawFrame) until a classification signal is received.

### Method
POST

### Endpoint
/processor/tts-gate/setup

### Parameters
#### Request Body
- **conversation_notifier** (Object) - Required - Notifier instance to signal conversation detection.
- **voicemail_notifier** (Object) - Required - Notifier instance to signal voicemail detection.

### Request Example
{
  "conversation_notifier": "<NotifierInstance>",
  "voicemail_notifier": "<NotifierInstance>"
}

### Response
#### Success Response (200)
- **status** (string) - Indicates successful initialization of the gating tasks.
```

--------------------------------

### Starting and Stopping Recording in Python

Source: https://reference-server.pipecat.ai/en/latest/api/pipecat.transports.daily.transport.html

This function allows you to start and stop call recording within the Daily system. The `start_recording` function initiates recording with optional streaming settings and returns a stream ID and potential error. Requires the 'daily-python' library.

```python
from daily.transport import Transport

# Assuming 'transport' is an initialized Transport instance

async def record_call(transport: Transport):
    # Optional streaming settings can be provided
    streaming_settings = {"format": "mp4"} # Example setting
    stream_id, error = await transport.start_recording(streaming_settings=streaming_settings)
    if error:
        print(f"Failed to start recording: {error}")
    else:
        print(f"Recording started with stream ID: {stream_id}")
        # To stop recording, you would typically use a different mechanism or event handler
        # For demonstration, let's assume we have a way to get the stream_id later
        # await stop_recording(transport, stream_id)

# Note: The provided text does not include a direct 'stop_recording' function signature, 
# but it's implied that recording can be stopped, possibly via an event or a separate method.
# The on_recording_stopped event handler is mentioned in the initial list.

```

--------------------------------

### Start VADController

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/audio/vad/vad_controller.html

Initializes the VAD analyzer with the audio sample rate from a `StartFrame` and broadcasts the initial VAD parameters to other services.

```python
async def _start(self, frame: StartFrame):
    self._vad_analyzer.set_sample_rate(frame.audio_in_sample_rate)
    # Broadcast initial VAD params so other services (e.g. STT) can use them
    await self.broadcast_frame(SpeechControlParamsFrame, vad_params=self._vad_analyzer.params)
```

--------------------------------

### Start Audio Input Streaming

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/transports/daily/transport.html

Configures and begins receiving audio streams from Daily call participants.

```APIDOC
## DailyInputTransport - start_audio_in_streaming

### Description
Starts receiving audio from participants if audio input is enabled in the parameters. It handles capturing individual participant tracks or setting up a general audio task.

### Method
start_audio_in_streaming

### Endpoint
N/A (This is a method within the DailyInputTransport class)

### Parameters
None

### Request Example
```python
await daily_input_transport.start_audio_in_streaming()
```

### Response
#### Success Response (200)
Audio input streaming is initiated.

#### Response Example
```python
# No direct response, streaming is initiated asynchronously.
```
```

--------------------------------

### StartupTimingObserver Overview

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/observers/startup_timing_observer.html

The StartupTimingObserver measures how long each processor's start() method takes during pipeline startup. It also measures transport timing.

```APIDOC
## StartupTimingObserver

### Description
Observer for tracking pipeline startup timing. This observer measures the duration of each processor's `start()` method and transport connection times.

### Usage
Instantiate the observer and add it to the pipeline. You can then register event handlers for `on_startup_timing_report` and `on_transport_timing_report` to receive timing data.

### Example
```python
from pipecat.observers.startup_timing_observer import StartupTimingObserver

observer = StartupTimingObserver()

@observer.event_handler("on_startup_timing_report")
async def on_report(observer, report):
    for t in report.processor_timings:
        print(f"{t.processor_name}: {t.duration_secs:.3f}s")

@observer.event_handler("on_transport_timing_report")
async def on_transport(observer, report):
    if report.bot_connected_secs is not None:
        print(f"Bot connected in {report.bot_connected_secs:.3f}s")
    print(f"Client connected in {report.client_connected_secs:.3f}s")

# Assuming 'pipeline' is an instance of BasePipeline
task = PipelineTask(pipeline, observers=[observer])
```

### Events
- **on_startup_timing_report**: Emitted when startup timing for all processors is complete. The event data includes a list of `ProcessorTiming` objects.
- **on_transport_timing_report**: Emitted when transport connection timings are available. The event data includes `bot_connected_secs` and `client_connected_secs`.
```

--------------------------------

### Pipeline Lifecycle Management

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/pipeline/task.html

Handles the setup and cleanup of pipeline tasks, including observer initialization, task manager configuration, and tracing termination.

```python
async def _setup(self, params: PipelineTaskParams):
    await self._load_setup_files()
    await self._load_observer_files()
    mgr_params = TaskManagerParams(loop=params.loop)
    self._task_manager.setup(mgr_params)
    setup = FrameProcessorSetup(clock=self._clock, task_manager=self._task_manager, observer=self._observer)
    await self._pipeline.setup(setup)

async def _cleanup(self, cleanup_pipeline: bool):
    await self.cleanup()
    if self._observer:
        await self._observer.cleanup()
    if self._enable_tracing and hasattr(self, "_turn_trace_observer"):
        self._turn_trace_observer.end_conversation_tracing()
    if cleanup_pipeline:
        await self._pipeline.cleanup()
```

--------------------------------

### POST /client/setup

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/transports/daily/transport.html

Initializes the client with the necessary task manager and event queues to handle incoming frames.

```APIDOC
## POST /client/setup

### Description
Initializes the client by setting up the task manager and creating an event queue for callback handling.

### Method
POST

### Endpoint
/client/setup

### Parameters
#### Request Body
- **setup** (FrameProcessorSetup) - Required - The configuration object containing the task manager and event configuration.

### Request Example
{
  "setup": { "task_manager": "<task_manager_instance>" }
}

### Response
#### Success Response (200)
- **status** (boolean) - Returns true if setup was successful.
```

--------------------------------

### InterruptionFrame

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/frames/frames.html

A frame pushed to interrupt the pipeline, for example, when a user starts speaking.

```APIDOC
## InterruptionFrame

### Description
Frame pushed to interrupt the pipeline. This frame is used to interrupt the pipeline. For example, when a user starts speaking to cancel any in-progress bot output. It can also be pushed by any processor.

### Parameters
None

### Request Example
```json
{}
```

### Response
#### Success Response (200)
None

#### Response Example
None
```

--------------------------------

### Get Pipecat Version (Python)

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat.html

Retrieves and returns the installed version of the pipecat-ai library. It uses importlib.metadata to get the version and loguru for logging information about the Pipecat and Python versions.

```python
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#

import sys
from importlib.metadata import version as lib_version

from loguru import logger

__version__ = lib_version("pipecat-ai")

logger.info(f"ᓚᘏᗢ Pipecat {__version__} (Python {sys.version}) ᓚᘏᗢ")




[docs]
def version() -> str:
    """Returns the Pipecat version."""
    return __version__
```

--------------------------------

### POST /pipeline/setup

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/pipeline/pipeline.html

Initializes the pipeline and all contained processors with the provided configuration.

```APIDOC
## POST /pipeline/setup

### Description
Initializes the pipeline and triggers the setup process for all nested frame processors.

### Method
POST

### Endpoint
/pipeline/setup

### Request Body
- **setup** (FrameProcessorSetup) - Required - Configuration object for the frame processor setup.

### Request Example
{
  "setup": { "config_id": "default_pipeline_config" }
}

### Response
#### Success Response (200)
- **status** (string) - Indicates successful initialization of all processors.

#### Response Example
{
  "status": "success"
}
```

--------------------------------

### Start Call Transcription - Python

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/transports/daily/transport.html

Begins the transcription process for an ongoing call. This function requires transcription configuration settings and a valid room token. It returns an error if transcription cannot be started, for example, due to a missing token.

```python
async def start_transcription(self, settings) -> Optional[CallClientError]:
    """Start transcription for the call.

    Args:
        settings: Transcription configuration settings.

    Returns:
        error: An error description or None.
    """
    if not self._token:
        return "Transcription can't be started without a room token"

    future = self._get_event_loop().create_future()
```

--------------------------------

### Initialize and Use SoundfileMixer

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/audio/mixers/soundfile_mixer.html

Demonstrates how to instantiate the SoundfileMixer with a set of sound files and integrate it into an asynchronous audio processing workflow.

```python
from pipecat.audio.mixers.soundfile_mixer import SoundfileMixer

# Define available sound files
sounds = {"alert": "/path/to/alert.wav", "bg": "/path/to/bg.wav"}

# Initialize the mixer
mixer = SoundfileMixer(
    sound_files=sounds,
    default_sound="alert",
    volume=0.5,
    loop=True
)

# Start the mixer with a specific sample rate
await mixer.start(sample_rate=16000)

# Mix incoming audio bytes
mixed_audio = await mixer.mix(incoming_audio_bytes)
```

--------------------------------

### Start Client Audio and Video Components

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/transports/daily/transport.html

Initializes audio and video tracks based on provided parameters. It creates custom sources and tracks for media streaming and sets up task handlers for callback processing.

```python
async def start(self, frame: StartFrame):
    self._in_sample_rate = self._params.audio_in_sample_rate or frame.audio_in_sample_rate
    self._out_sample_rate = self._params.audio_out_sample_rate or frame.audio_out_sample_rate
    if self._params.audio_in_enabled:
        if self._params.audio_in_user_tracks and not self._audio_task and self._task_manager:
            self._audio_queue = asyncio.Queue()
            self._audio_task = self._task_manager.create_task(
                self._callback_task_handler(self._audio_queue),
                f"{self}::audio_callback_task",
            )
        elif not self._speaker:
            self._speaker = Daily.create_speaker_device(
                self._speaker_name(),
                sample_rate=self._in_sample_rate,
                channels=self._params.audio_in_channels,
                non_blocking=True,
            )
            Daily.select_speaker_device(self._speaker_name())
    if self._params.video_in_enabled and not self._video_task and self._task_manager:
        self._video_queue = asyncio.Queue()
        self._video_task = self._task_manager.create_task(
            self._callback_task_handler(self._video_queue),
            f"{self}::video_callback_task",
        )
    if self._params.video_out_enabled and not self._camera_track:
        video_source = CustomVideoSource(self._params.video_out_width, self._params.video_out_height, self._params.video_out_color_format)
        video_track = CustomVideoTrack(video_source)
        self._camera_track = DailyVideoTrack(source=video_source, track=video_track)
```

--------------------------------

### Initialize and Setup Client Resources

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/transports/daily/transport.html

Configures the task manager and initializes event queues for the client. This ensures that background tasks are tracked and managed properly during the client's lifecycle.

```python
async def setup(self, setup: FrameProcessorSetup):
    if self._task_manager:
        return
    self._task_manager = setup.task_manager
    self._event_queue = asyncio.Queue()
    self._event_task = self._task_manager.create_task(
        self._callback_task_handler(self._event_queue),
        f"{self}::event_callback_task",
    )
```

--------------------------------

### Handle VAD Speech Started Event

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/processors/aggregators/llm_response_universal.html

Responds to the VAD (Voice Activity Detection) speech started event by broadcasting a `VADUserStartedSpeakingFrame` upstream. It includes timing parameters from the VAD analyzer.

```python
async def _on_vad_speech_started(self, controller):
        await self._queued_broadcast_frame(
            VADUserStartedSpeakingFrame,
            start_secs=controller._vad_analyzer.params.start_secs,
        )
```

--------------------------------

### Initialize Pipeline with Observers and RTVI

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/pipeline/task.html

Demonstrates how to configure pipeline parameters, attach observers for turn tracking and tracing, and validate the presence of RTVI components.

```python
if self._enable_turn_tracking:
    self._turn_tracking_observer = TurnTrackingObserver()
    observers.append(self._turn_tracking_observer)

if self._enable_tracing and self._turn_tracking_observer:
    self._tracing_context = TracingContext()
    self._user_bot_latency_observer = UserBotLatencyObserver()
    observers.append(self._user_bot_latency_observer)
    self._turn_trace_observer = TurnTraceObserver(
        self._turn_tracking_observer,
        latency_tracker=self._user_bot_latency_observer,
        conversation_id=self._conversation_id,
        additional_span_attributes=self._additional_span_attributes,
        tracing_context=self._tracing_context,
    )
    observers.append(self._turn_trace_observer)
```

--------------------------------

### Handle LLM Response Start for Summarization Trigger (Python)

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/processors/aggregators/llm_context_summarizer.html

Handles the start of an LLM response to determine if context summarization should be triggered automatically. It checks the summarization conditions and initiates a request if necessary.

```python
async def _handle_llm_response_start(self, frame: LLMFullResponseStartFrame):
        """Handle LLM response start to check if summarization is needed.

        Args:
            frame: The LLM response start frame.
        """
        if self._should_summarize():
            await self._request_summarization()
```

--------------------------------

### Initialize Daily Transport Client

Source: https://reference-server.pipecat.ai/en/latest/api/pipecat.transports.daily.transport.html

Demonstrates how to instantiate the DailyTransportClient with required room credentials and configuration parameters.

```python
from pipecat.transports.daily.transport import DailyTransportClient

client = DailyTransportClient(
    room_url="https://your-domain.daily.co/room",
    token="your-token",
    bot_name="my-bot",
    params=daily_params,
    callbacks=daily_callbacks,
    transport_name="daily-transport"
)
```

--------------------------------

### Initialize KrispVivaTurn Analyzer

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/audio/turn/krisp_viva_turn.html

Demonstrates how to instantiate the KrispVivaTurn analyzer. It requires a path to a .kef model file and optionally accepts custom configuration parameters for sensitivity and frame duration.

```python
from pipecat.audio.krisp.krisp_turn import KrispVivaTurn, KrispTurnParams

params = KrispTurnParams(threshold=0.6, frame_duration_ms=20)
turn_analyzer = KrispVivaTurn(
    model_path="/path/to/model.kef",
    params=params,
    api_key="your_krisp_api_key"
)
```

--------------------------------

### Handle User Turn Started Event

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/processors/aggregators/llm_response_universal.html

Manages the start of a user's turn, logging the event, updating timestamps, and optionally broadcasting `UserStartedSpeakingFrame` and handling interruptions. It also processes the frame through the user idle controller.

```python
async def _on_user_turn_started(
        self,
        controller: UserTurnController,
        strategy: BaseUserTurnStartStrategy,
        params: UserTurnStartedParams,
    ):
        logger.debug(f"{self}: User started speaking (strategy: {strategy})")

        self._user_turn_start_timestamp = time_now_iso8601()

        if params.enable_user_speaking_frames:
            await self.broadcast_frame(UserStartedSpeakingFrame)

        await self._user_idle_controller.process_frame(UserStartedSpeakingFrame())

        if params.enable_interruptions and self._allow_interruptions:
            await self.broadcast_interruption()

        await self._call_event_handler("on_user_turn_started", strategy)
```

--------------------------------

### Start and End Turn Logic - Python

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/observers/turn_tracking_observer.html

Provides the core logic for starting and ending a turn in the conversation. `_start_turn` initializes turn variables and emits an event. `_end_turn` calculates duration, updates turn status, and emits an event.

```python
async def _start_turn(self, data: FramePushed):
        """Start a new turn."""
        self._is_turn_active = True
        self._has_bot_spoken = False
        self._turn_count += 1
        self._turn_start_time = data.timestamp
        logger.trace(f"Turn {self._turn_count} started")
        await self._call_event_handler("on_turn_started", self._turn_count)

    async def _end_turn(self, data: FramePushed, was_interrupted: bool):
        """End the current turn."""
        if not self._is_turn_active:
            return

        duration = (data.timestamp - self._turn_start_time) / 1_000_000_000  # Convert to seconds
        self._is_turn_active = False

        status = "interrupted" if was_interrupted else "completed"
        logger.trace(f"Turn {self._turn_count} {status} after {duration:.2f}s")
        await self._call_event_handler("on_turn_ended", self._turn_count, duration, was_interrupted)
```

--------------------------------

### Initialize SyncParallelPipeline

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/pipeline/sync_parallel_pipeline.html

Demonstrates how to instantiate the SyncParallelPipeline with multiple processor lists and define the frame output order strategy.

```python
from pipecat.pipeline.parallel import SyncParallelPipeline, FrameOrder

# Define parallel paths
path1 = [ProcessorA(), ProcessorB()]
path2 = [ProcessorC(), ProcessorD()]

# Initialize with pipeline order
pipeline = SyncParallelPipeline(path1, path2, frame_order=FrameOrder.PIPELINE)
```

--------------------------------

### Update VAD Start Seconds in Turn Analyzer

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/audio/turn/smart_turn/base_smart_turn.html

Updates the internal `_vad_start_secs` attribute of the turn analyzer. This value is used to adjust the start time for processing speech segments, influencing the audio context provided to the prediction model.

```python
def update_vad_start_secs(self, vad_start_secs: float):
    """Store the new vad_start_secs value."""
    self._vad_start_secs = vad_start_secs
```

--------------------------------

### Initialize DailyTransportClient

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/transports/daily/transport.html

Demonstrates the initialization of the DailyTransportClient, which ensures the Daily SDK is initialized and sets up internal state for managing room connections and media renderers.

```python
client = DailyTransportClient(
    room_url="https://your-room.daily.co/name",
    token="your-token",
    bot_name="my-bot",
    params=daily_params,
    callbacks=daily_callbacks,
    transport_name="daily-transport"
)
```

--------------------------------

### Define LLM Thought Start Frame (Python)

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/frames/frames.html

Defines a control frame indicating the start of an LLM thought process. It includes parameters to control context appending and specify the LLM provider, with validation to ensure the LLM is set if context appending is enabled.

```python
from typing import Optional
from dataclasses import dataclass

# Assuming ControlFrame and format_pts are defined elsewhere
# class ControlFrame:
#     def __init__(self, name, pts):
#         self.name = name
#         self.pts = pts
#     def __post_init__(self):
#         pass
# def format_pts(pts):
#     return pts

@dataclass
class LLMThoughtStartFrame(ControlFrame):
    """Frame indicating the start of an LLM thought.

    Parameters:
        append_to_context: Whether the thought should be appended to the LLM context.
            If it is appended, the `llm` field is required, since it will be
            appended as an `LLMSpecificMessage`.
        llm: Optional identifier of the LLM provider for LLM-specific handling.
            Only required if `append_to_context` is True, as the thought is
            appended to context as an `LLMSpecificMessage`.
    """

    append_to_context: bool = False
    llm: Optional[str] = None

    def __post_init__(self):
        super().__post_init__()
        if self.append_to_context and self.llm is None:
            raise ValueError("When append_to_context is True, llm must be set")

    def __str__(self):
        pts = format_pts(self.pts)
        return (
            f"{self.name}(pts: {pts}, append_to_context: {self.append_to_context}, llm: {self.llm})"
        )
```

--------------------------------

### POST /transcription/start

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/transports/daily/transport.html

Starts the transcription service for the active call.

```APIDOC
## POST /transcription/start

### Description
Enables real-time transcription for the current call session.

### Method
POST

### Endpoint
/transcription/start

### Request Body
- **settings** (object) - Optional - Transcription configuration.

### Response
#### Success Response (200)
- **error** (null) - Returns null on success.
```

--------------------------------

### Initialize Krisp Audio Module

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/audio/krisp_instance.html

This snippet demonstrates the import and dependency check for the krisp_audio library. It ensures the environment is correctly configured before attempting to use Krisp audio processing features.

```python
import atexit
import os
from threading import Lock
from loguru import logger

try:
    import krisp_audio
except ModuleNotFoundError as e:
    logger.error(f"Exception: {e}")
    logger.error("In order to use the Krisp instance, you need to install krisp_audio.")
    raise Exception(f"Missing module: {e}")
```

--------------------------------

### StartInterruptionFrame

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/frames/frames.html

Indicates that a user started speaking, signaling an interruption.

```APIDOC
## StartInterruptionFrame

### Description
Frame indicating user started speaking (interruption detected). This frame is deprecated and will be removed in a future version. Instead, use `InterruptionFrame`. Emitted by the BaseInputTransport to indicate that a user has started speaking (i.e. is interrupting). This is similar to `UserStartedSpeakingFrame` except that it should be pushed concurrently with other frames (so the order is not guaranteed).

### Parameters
None

### Request Example
```json
{}
```

### Response
#### Success Response (200)
None

#### Response Example
None
```

--------------------------------

### Gemini LLM Adapter Initialization and Parameter Handling

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/adapters/services/gemini_adapter.html

Demonstrates the initialization of the GeminiLLMAdapter and the process of extracting LLM invocation parameters from a universal context. It handles system instructions and converts standard Pipecat messages and tools into a format compatible with the Gemini API.

```python
"""Gemini LLM adapter for Pipecat."""

import base64
import json
from dataclasses import dataclass, field
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, FileData, 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}")




[docs]
class GeminiLLMInvocationParams(TypedDict):
    """Context-based parameters for invoking Gemini LLM."""

    system_instruction: Optional[str]
    messages: List[Content]
    tools: List[Any] | NotGiven





[docs]
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.
    """

    @property
    def id_for_llm_specific_messages(self) -> str:
        """Get the identifier used in LLMSpecificMessage instances for Google."""
        return "google"




    def get_llm_invocation_params(
        self,
        context: LLMContext,
        *, 
        system_instruction: Optional[str] = None
    ) -> GeminiLLMInvocationParams:
        """Get Gemini-specific LLM invocation parameters from a universal LLM context.

        Args:
            context: The LLM context containing messages, tools, etc.
            system_instruction: Optional system instruction from service settings
                or ``run_inference``.

        Returns:
            Dictionary of parameters for Gemini's API.
        """
        converted = self._from_universal_context_messages(
            self.get_messages(context), system_instruction=system_instruction
        )
        effective_system = self._resolve_system_instruction(
            converted.system_instruction,
            system_instruction,
            discard_context_system=True,
        )
        return {
            "system_instruction": effective_system,
            "messages": converted.messages,
            # NOTE: LLMContext's tools are guaranteed to be a ToolsSchema (or NOT_GIVEN)
            "tools": self.from_standard_tools(context.tools),
        }

```

--------------------------------

### Emit Startup Timing Report (Python)

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/observers/startup_timing_observer.html

Builds and emits the startup timing report if it has not been reported yet. It calculates the total duration of all tracked processor timings.

```Python
async def _emit_report(self):
        """Build and emit the startup timing report."""
        if self._startup_timing_reported:
            return
        self._startup_timing_reported = True

        total = sum(t.duration_secs for t in self._timings)
```

--------------------------------

### Recording Management

Source: https://reference-server.pipecat.ai/en/latest/api/pipecat.transports.daily.transport.html

Functions for starting and managing call recordings.

```APIDOC
## Start Recording

### Description
Starts recording the ongoing Daily call.

### Method
`start_recording(_streaming_settings=None_, _stream_id=None_, _force_new=None_)`

### Parameters
- **streaming_settings** (`dict`, optional) - Settings for streaming the recording.
- **stream_id** (`str`, optional) - Identifier for the recording stream.
- **force_new** (`bool`, optional) - If true, forces a new recording session even if one is active.

### Returns
- `Tuple[str, str | None]` - A tuple containing the session ID and an error description (or None if successful).
```

--------------------------------

### POST /start

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/audio/filters/aic_filter.html

Initializes the AIC processor with the specified sample rate and allocates necessary processing buffers.

```APIDOC
## POST /start

### Description
Initializes the filter, acquires the shared model, calculates optimal block sizes, and allocates processing buffers.

### Method
POST

### Parameters
#### Request Body
- **sample_rate** (int) - Required - The sample rate of the input transport in Hz.

### Response
#### Success Response (200)
- **status** (string) - Indicates successful initialization.
```

--------------------------------

### Handle Bot Speaking Events - Python

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/observers/turn_tracking_observer.html

Manages events when the bot starts or stops speaking. When the bot starts speaking, it cancels any pending turn end timers. When the bot stops speaking, it schedules a turn end timer to handle potential speech resumption.

```python
async def _handle_bot_started_speaking(self, data: FramePushed):
        """Handle bot speaking events."""
        self._is_bot_speaking = True
        self._has_bot_spoken = True
        # Cancel any pending turn end timer when bot starts speaking again
        self._cancel_turn_end_timer()

    async def _handle_bot_stopped_speaking(self, data: FramePushed):
        """Handle bot stopped speaking events."""
        self._is_bot_speaking = False
        # Schedule turn end with timeout
        # This is needed to handle cases where the bot's speech ends and then resumes
        # This can happen with HTTP TTS services or function calls
        self._schedule_turn_end(data)
```

--------------------------------

### GET /messages

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/processors/aggregators/llm_context.html

Retrieves the current list of messages in the conversation context.

```APIDOC
## GET /messages

### Description
Retrieves the current conversation message history, with optional filtering for LLM-specific messages.

### Method
GET

### Endpoint
/messages

### Parameters
#### Query Parameters
- **llm_specific_filter** (string) - Optional - Filter to return only messages compatible with a specific LLM.

### Response
#### Success Response (200)
- **messages** (list) - A list of conversation message objects.

#### Response Example
[
  {"role": "user", "content": "Hello"},
  {"role": "assistant", "content": "Hi there!"}
]
```

--------------------------------

### Initialize and Use LocalCoreMLSmartTurnAnalyzer

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/audio/turn/smart_turn/local_coreml_smart_turn.html

Demonstrates how to initialize the analyzer with a model path and perform an endpoint prediction on an audio array. The analyzer processes audio input to determine if a turn is complete based on a probability threshold.

```python
from pipecat.audio.turn.smart_turn.local_coreml_smart_turn import LocalCoreMLSmartTurnAnalyzer

# Initialize the analyzer
# Requires a path to the directory containing the model files
analyzer = LocalCoreMLSmartTurnAnalyzer(smart_turn_model_path="/path/to/model")

# Perform prediction on an audio numpy array
# Returns a dictionary containing the prediction (0 or 1) and probability
result = await analyzer._predict_endpoint(audio_array)
print(f"Prediction: {result['prediction']}, Probability: {result['probability']}")
```

--------------------------------

### IVR Verbal Response Examples

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/extensions/ivr/ivr_navigator.html

Illustrates how to respond verbally to IVR prompts.

```APIDOC
## IVR Verbal Response Examples

### Description
Provides examples of natural language responses to common IVR prompts, covering confirmations, information provision, and clarifications.

### Examples
- **Confirmation**: "Yes."
- **Information Provision**: "John Smith."
- **Department Inquiry**: "Billing."
- **Yes/No Question**: "No."
- **Correction/Confirmation**: "Yes."

### Usage
These verbal responses are used when the IVR system requests spoken input.
```

--------------------------------

### Implement StartupTimingObserver in Pipecat

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/observers/startup_timing_observer.html

This snippet demonstrates how to instantiate the StartupTimingObserver and register event handlers to receive timing reports for processors and transport connections. It shows how to integrate the observer into a PipelineTask.

```python
observer = StartupTimingObserver()

@observer.event_handler("on_startup_timing_report")
async def on_report(observer, report):
    for t in report.processor_timings:
        print(f"{t.processor_name}: {t.duration_secs:.3f}s")

@observer.event_handler("on_transport_timing_report")
async def on_transport(observer, report):
    if report.bot_connected_secs is not None:
        print(f"Bot connected in {report.bot_connected_secs:.3f}s")
    print(f"Client connected in {report.client_connected_secs:.3f}s")

task = PipelineTask(pipeline, observers=[observer])
```

--------------------------------

### Initialize UserBotLatencyObserver

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/observers/user_bot_latency_observer.html

Demonstrates the initialization of the UserBotLatencyObserver class, setting up event handlers and internal state for tracking frame processing and speech timing.

```python
class UserBotLatencyObserver(BaseObserver):
    def __init__(self, *, max_frames=100, **kwargs):
        super().__init__(**kwargs)
        self._user_stopped_time: Optional[float] = None
        self._processed_frames: set = set()
        self._frame_history: deque = deque(maxlen=max_frames)
        self._register_event_handler("on_latency_measured")
        self._register_event_handler("on_latency_breakdown")
        self._register_event_handler("on_first_bot_speech_latency")
```

--------------------------------

### GET /vad-context

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/audio/filters/aic_filter.html

Retrieves the current VAD context instance associated with the initialized processor.

```APIDOC
## GET /vad-context

### Description
Retrieves the VAD context instance bound to the underlying processor. This method requires the processor to be initialized via the start() method.

### Method
GET

### Endpoint
/vad-context

### Response
#### Success Response (200)
- **vad_ctx** (VadContext) - The active VAD context instance.

#### Error Response (500)
- **error** (RuntimeError) - Raised if the processor has not been initialized.
```

--------------------------------

### Initialize FalSmartTurnAnalyzer

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/audio/turn/smart_turn/fal_smart_turn.html

Demonstrates how to instantiate the FalSmartTurnAnalyzer class. It requires an aiohttp session and an optional API key for authentication.

```python
from pipecat.audio.turn.smart_turn.fal_smart_turn import FalSmartTurnAnalyzer
import aiohttp

async def setup_analyzer(session):
    analyzer = FalSmartTurnAnalyzer(
        aiohttp_session=session,
        api_key="your-fal-api-key"
    )
    return analyzer
```

--------------------------------

### IVR DTMF Sequence Input

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/extensions/ivr/ivr_navigator.html

Provides examples of DTMF sequences for inputting numerical data.

```APIDOC
## IVR DTMF Sequence Input

### Description
Examples of how to represent numerical inputs using DTMF (Dual-Tone Multi-Frequency) signals, typically used for entering dates, account numbers, or phone digits.

### Examples
- **Date of Birth "01/15/1990"**: `<dtmf>0</dtmf><dtmf>1</dtmf><dtmf>1</dtmf><dtmf>5</dtmf><dtmf>1</dtmf><dtmf>9</dtmf><dtmf>9</dtmf><dtmf>0</dtmf>`
- **Account Number "12345"**: `<dtmf>1</dtmf><dtmf>2</dtmf><dtmf>3</dtmf><dtmf>4</dtmf><dtmf>5</dtmf>`
- **Phone Number Last 4 Digits "6789"**: `<dtmf>6</dtmf><dtmf>7</dtmf><dtmf>8</dtmf><dtmf>9</dtmf>`

### Note
Each `<dtmf>` tag represents a single digit press.
```

--------------------------------

### StartFrame

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/frames/frames.html

Represents the initial frame to start pipeline processing. It initializes processors with configuration parameters.

```APIDOC
## StartFrame

### Description
Initial frame to start pipeline processing. This is the first frame that should be pushed down a pipeline to initialize all processors with their configuration parameters.

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
- **audio_in_sample_rate** (int) - Optional - Input audio sample rate in Hz. Defaults to 16000.
- **audio_out_sample_rate** (int) - Optional - Output audio sample rate in Hz. Defaults to 24000.
- **allow_interruptions** (bool) - Optional - Whether to allow user interruptions. Deprecated: Use `LLMUserAggregator`'s new `user_mute_strategies` parameter instead.
- **enable_metrics** (bool) - Optional - Whether to enable performance metrics collection. Defaults to False.
- **enable_tracing** (bool) - Optional - Whether to enable OpenTelemetry tracing. Defaults to False.
- **enable_usage_metrics** (bool) - Optional - Whether to enable usage metrics collection. Defaults to False.
- **interruption_strategies** (List[BaseInterruptionStrategy]) - Optional - List of interruption handling strategies. Deprecated: Use `LLMUserAggregator`'s new `user_turn_strategies` parameter instead.
- **report_only_initial_ttfb** (bool) - Optional - Whether to report only initial time-to-first-byte. Defaults to False.
- **tracing_context** (Optional[TracingContext]) - Optional - Pipeline-scoped tracing context for span hierarchy.

### Request Example
```json
{
  "audio_in_sample_rate": 16000,
  "audio_out_sample_rate": 24000,
  "allow_interruptions": false,
  "enable_metrics": false,
  "enable_tracing": false,
  "enable_usage_metrics": false,
  "interruption_strategies": [],
  "report_only_initial_ttfb": false,
  "tracing_context": null
}
```

### Response
#### Success Response (200)
None

#### Response Example
None
```

--------------------------------

### Generate and Handle Startup Timing Report - Python

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/observers/startup_timing_observer.html

This snippet demonstrates the creation of a StartupTimingReport object with timing information and its subsequent handling via an event.

```Python
report = StartupTimingReport(
            start_time=self._start_wall_clock or 0.0,
            total_duration_secs=total,
            processor_timings=self._timings,
        )

        await self._call_event_handler("on_startup_timing_report", report)
```

--------------------------------

### GET /vad/frames-required

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/audio/vad/krisp_viva_vad.html

Retrieves the number of audio frames required for analysis based on the current configuration.

```APIDOC
## GET /vad/frames-required

### Description
Calculates the number of samples required for a single VAD processing frame based on the current sample rate and frame duration.

### Method
GET

### Endpoint
/vad/frames-required

### Response
#### Success Response (200)
- **num_frames** (int) - The number of samples required.

#### Response Example
{
  "num_frames": 160
}
```

--------------------------------

### POST /sdk/acquire

Source: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/audio/krisp_instance.html

Initializes the Krisp SDK if not already active and increments the reference count for the current session.

```APIDOC
## POST /sdk/acquire

### Description
Acquires a reference to the Krisp SDK. If this is the first call, it performs global initialization using the provided API key or the KRISP_VIVA_API_KEY environment variable.

### Method
POST

### Endpoint
/sdk/acquire

### Parameters
#### Request Body
- **api_key** (string) - Optional - The Krisp SDK license key. If omitted, the system checks the environment variable.

### Request Example
{
  "api_key": "your-license-key-here"
}

### Response
#### Success Response (200)
- **status** (string) - Indicates successful acquisition and current reference count.

#### Response Example
{
  "status": "success",
  "reference_count": 1
}
```
Tokens:
Raw