### 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": "", "voicemail_notifier": "" } ### 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": "" } } ### 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"**: `01151990` - **Account Number "12345"**: `12345` - **Phone Number Last 4 Digits "6789"**: `6789` ### Note Each `` 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 } ```