<Card
title="OpenAI (LLM)"
icon="brain"
href="https://auth.openai.com/create-account"

Create an account and generate an API key for intelligent conversation
responses.

<Card
title="Cartesia (TTS)"
icon="volume-high"
href="https://play.cartesia.ai/sign-up"

Sign up and generate your API key for natural voice
synthesis.

<Card
title="Browse Examples"
icon="code"
href="https://github.com/pipecat-ai/pipecat-examples"

30+ production-ready examples for inspiration

<Card
title="Smart Turn Overview"
icon="brain"
href="/api-reference/server/utilities/turn-detection/smart-turn-overview"

Smart Turn model implementation guide

Interruptions Copy Section

When a user turn starts with interruptions enabled (the default, via the enable_interruptions parameter on start strategies), the bot immediately stops speaking, in-flight work is cancelled, and the pipeline is ready for the new user input.

Keep interruptions enabled for natural conversations. For how the cancellation works, what ends up in the LLM context, and how to tune, disable, or trigger interruptions yourself, see Interruptions.

Best Practices Copy Chapter

Optimal Configuration Copy Section

For most voice AI use cases, the defaults work well:

from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.processors.aggregators.llm_response_universal import (
    LLMContextAggregatorPair,
    LLMUserAggregatorParams,
)

transport = YourTransport(
    params=TransportParams(),
)

# Default configuration: Smart Turn detection + VAD with stop_secs=0.2
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
    context,
    user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()),
)

worker = PipelineWorker(pipeline)

The defaults use TurnAnalyzerUserTurnStopStrategy with LocalSmartTurnAnalyzerV3 for turn detection and SileroVADAnalyzer with stop_secs=0.2 for voice activity detection.

Performance Considerations Copy Section

• Use local VAD: 150-200ms faster than remote VAD services
• Tune for your use case: Test with real audio conditions
• Monitor CPU usage: VAD adds minimal overhead but monitor in production
• Consider turn detection: Improves conversation quality but adds complexity

Key Takeaways Copy Chapter

• VAD detects speech activity but turn detection understands conversation context
• Configuration affects user experience - tune parameters for your specific use case
• System frames coordinate behavior - enable interruptions and natural turn-taking
• Local processing is faster - Silero VAD provides low-latency speech detection
• Turn detection improves quality - but requires careful VAD configuration

What's Next Copy Chapter

Now that you understand how speech input is detected and processed, let's explore how that audio gets converted to text through speech recognition.

<Card
title="Speech to Text"
icon="arrow-right"
href="/pipecat/learn/speech-to-text"

Learn how to configure speech recognition in your voice AI pipeline

Speech to Text

Source: https://docs.pipecat.ai/pipecat/learn/speech-to-text.md

Learn how to configure speech recognition to convert user audio into text in your Pipecat pipeline

Speech to Text (STT) services are responsible for converting user audio into text transcriptions. They receive audio input from users and provide real-time transcriptions that your bot can process and respond to.

Pipeline Placement Copy Chapter

STT processors must be positioned correctly in your pipeline to receive and process audio frames:

pipeline = Pipeline([
    transport.input(),             # Creates InputAudioRawFrames
    stt,                           # Processes audio → creates TranscriptionFrames
    context_aggregator.user(),     # Uses transcriptions for context
    llm,
    tts,
    transport.output(),
])

Placement requirements:

• After transport.input(): STT needs InputAudioRawFrames from the transport
• Before context processing: Transcriptions must be available for context aggregation
• Before LLM processing: Text must be ready for language model input

STT Service Types Copy Chapter

Pipecat provides two types of STT services based on how they process audio:

1. STTService (Streaming) Copy Section

How it works:

• Establishes a WebSocket connection to the STT provider
• Continuously streams audio for real-time transcription
• Lower latency due to persistent connection

2. SegmentedSTTService (HTTP-based) Copy Section

How it works:

• Uses local VAD (Voice Activity Detection) to chunk speech
• Sends audio segments to STT service as wav files
• Higher latency due to segmentation and HTTP POST requests

Supported STT Services Copy Chapter

Pipecat supports a wide range of STT providers to fit different needs and budgets:

<Card
title="Supported STT Services"
icon="list"
href="/api-reference/server/services/supported-services#speech-to-text"

View the complete list of supported speech-to-text providers

Popular options include:

STT Configuration Copy Chapter

Service-Specific Configuration Copy Section

Each STT service has its own customization options. Refer to specific service documentation for details:

<Card
title="Individual STT Services"
icon="screwdriver-wrench"
href="/api-reference/server/services/supported-services#speech-to-text"

Explore configuration options for each supported STT provider

For example, let's look at configuring the DeepgramSTTService using the LiveOptions class:

from deepgram import LiveOptions
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.transcriptions.language import Language

# Configure using LiveOptions for full control
live_options = LiveOptions(
    model="nova-2",
    language=Language.EN_US,
    interim_results=True,        # Enable interim transcripts
    punctuate=True,              # Add punctuation
    profanity_filter=True,       # Filter profanity
    vad_events=False,            # Use pipeline VAD instead
)

stt = DeepgramSTTService(
    api_key=os.getenv("DEEPGRAM_API_KEY"),
    live_options=live_options,
)

STTService Base Class Configuration Copy Section

All STT services inherit from the STTService base class. The base class has base configuration options which are set with smart defaults:

stt = YourSTTService(
    # Service-specific options...
    audio_passthrough=True,      # Pass audio frames downstream (recommended)
    sample_rate=16000,           # Audio sample rate (better set in PipelineParams)
)

Key options:

• audio_passthrough=True: Allows audio frames to continue downstream to other processors (like audio recording)
• sample_rate: Audio sampling rate - best practice is to set the audio_in_sample_rate in PipelineParams for consistency

Pipeline-Level Audio Configuration Copy Section

Instead of setting sample rates on individual services, configure them pipeline-wide:

worker = PipelineWorker(
    pipeline,
    params=PipelineParams(
        audio_in_sample_rate=16000,   # All input processors use this rate
        audio_out_sample_rate=24000,  # All output processors use this rate
    ),
)

This ensures all audio processors use consistent sample rates without manual configuration.

Multilingual Transcription Copy Chapter

Many STT services in Pipecat default to Language.EN (English). If you need to transcribe speech in other languages or let the model auto-detect the spoken language, you can enable multilingual support. However, providers implement this differently:

language=None — Whisper-based services (Groq, OpenAI, local Whisper) and ElevenLabs support automatic language detection when no language is specified:

from pipecat.services.groq.stt import GroqSTTService

stt = GroqSTTService(
    api_key=os.getenv("GROQ_API_KEY"),
    settings=GroqSTTService.Settings(
        language=None,  # Auto-detect language
    ),
)

language="multi" — Deepgram uses a special "multi" language code to enable multilingual transcription:

from pipecat.services.deepgram.stt import DeepgramSTTService

stt = DeepgramSTTService(
    api_key=os.getenv("DEEPGRAM_API_KEY"),
    settings=DeepgramSTTService.Settings(
        language="multi",  # Enable multilingual mode
    ),
)

Language array — Google Cloud STT accepts a list of languages for multi-language recognition. See the Google STT docs for details.

Best Practices Copy Chapter

Enable Interim Results Copy Section

When available, enable interim transcripts for better user experience:

stt = DeepgramSTTService(
    api_key=os.getenv("DEEPGRAM_API_KEY"),
    live_options=LiveOptions(
      interim_results=True,
    )
)

Benefits:

• Notifies context aggregation that more text is coming
• Prevents premature LLM completions
• Enables interruption detection
• Improves conversation flow

Enable Punctuation and Formatting Copy Section

Use punctuation when available for better LLM comprehension:

stt = DeepgramSTTService(
    api_key=os.getenv("DEEPGRAM_API_KEY"),
    live_options=LiveOptions(
        punctuate=True,        # Adds punctuation
        profanity_filter=True, # Optional content filtering
    )
)

Benefits:

• Professional-looking transcripts
• Better LLM comprehension
• Eliminates post-processing needs
• Improved context understanding

Use Local VAD Copy Section

While many STT services provide Voice Activity Detection, use Pipecat's local Silero VAD for better performance:

from pipecat.audio.vad.silero import SileroVADAnalyzer

# Configure in context aggregator
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
    context,
    user_params=LLMUserAggregatorParams(
        vad_analyzer=SileroVADAnalyzer(),
    ),
)

Advantages:

• 150-200ms faster speech detection (no network round trip)
• More responsive conversation flow
• Better interruption handling
• Reduced latency overall

Tune STT Latency Copy Section

Each STT service has a measured P99 latency for delivering final transcripts after the user stops speaking. This value is used by turn stop strategies to decide how long to wait before ending the user's turn. If you notice the bot responding too early (cutting off the user) or too late (long pauses), tuning this value can help.

<Card
title="STT Latency Tuning"
icon="gauge-high"
href="/pipecat/fundamentals/stt-latency-tuning"

Learn about TTFS latency, see default values for every STT service, and how to
measure and override for your deployment

Key Takeaways Copy Chapter

• Pipeline placement matters - STT must come after transport input, before context processing
• Service types differ - streaming services have lower latency than segmented
• Services are modular - easily swap providers without code changes
• Best practices improve performance - use interim results, formatting, and local VAD
• Configuration affects quality - proper setup significantly impacts transcription accuracy

What's Next Copy Chapter

Now that you understand speech recognition, let's explore how to manage conversation context and memory in your voice AI bot.

<Card
title="Context Management"
icon="arrow-right"
href="/pipecat/learn/context-management"

Learn how to handle conversation history and context in your pipeline

Context Management

Source: https://docs.pipecat.ai/pipecat/learn/context-management.md

Work with Pipecat's LLM context and context aggregators: how conversation history is built and shared in a pipeline.

What is Context in Pipecat? Copy Chapter

In Pipecat, context refers to the conversation history that the LLM uses to generate responses. The context consists of user/assistant messages representing the conversation history, and can also include developer messages for task-specific instructions to the LLM.

# Example context structure
messages = [
    {"role": "developer", "content": "Keep responses under two sentences."},
    {"role": "user", "content": "Hello!"},
    {"role": "assistant", "content": "Hi there! How can I help you?"},
    # Context aggregators automatically add new messages here
]

The system prompt is typically set via system_instruction in the LLM service's Settings, not as a message in the context. See System Instruction, Developer Messages, and System Messages below for details.

Since Pipecat is a real-time voice AI framework, context management happens automatically as the conversation flows, but you can also control it manually when needed.

How Context Updates During Conversations Copy Chapter

Context updates happen automatically as frames flow through your pipeline:

User Messages:

1. User speaks → InputAudioRawFrame → STT Service → TranscriptionFrame
2. context_aggregator.user() receives TranscriptionFrame and adds user message to context

Assistant Messages:

1. LLM generates response → LLMTextFrame → TTS Service → TTSTextFrame
2. context_aggregator.assistant() receives TTSTextFrame and adds assistant message to context

Frame types that update context:

• TranscriptionFrame: Contains user speech converted to text by STT service
• LLMTextFrame: Contains LLM-generated responses
• TTSTextFrame: Contains bot responses converted to text by TTS service (represents what was actually spoken)

Setting Up Context Management Copy Chapter

Pipecat includes a context aggregator that creates and manages context for both user and assistant messages:

1. Create the Context and Context Aggregator Copy Section

# Create LLM service with system instruction
llm = OpenAILLMService(
    api_key=os.getenv("OPENAI_API_KEY"),
    settings=OpenAILLMService.Settings(
        model="gpt-4o",
        system_instruction="You are a helpful voice assistant.",
    ),
)

# Create context (no system message needed — system_instruction handles it)
context = LLMContext()

# Create context aggregator instance
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(context)

About LLMContext:

LLMContext is Pipecat's universal context container that stores conversation messages, tool definitions, and tool choice settings. It uses an OpenAI-compatible format that works across all LLM services through automatic adapter translation.

Key properties:

• messages: List of conversation messages (user, assistant, developer, tool)
• tools: Optional available functions — a list of direct functions and/or FunctionSchema objects (or a ToolsSchema)
• tool_choice: Optional strategy for tool selection

2. Context with Function Calling Copy Section

Context can also include tools (function definitions) that the LLM can call during conversations:

from pipecat.services.llm_service import FunctionCallParams

# A direct function: schema is auto-derived from the signature and docstring
async def get_current_weather(params: FunctionCallParams, location: str, format: str):
    """Get the current weather.

    Args:
        location: The city and state, e.g. "San Francisco, CA".
        format: The temperature unit to use. Must be either "celsius" or "fahrenheit".
    """
    await params.result_callback({"conditions": "sunny", "temperature": "75"})

# Create context with both messages and tools
context = LLMContext(messages, tools=[get_current_weather])
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(context)

Function call results are also automatically stored in the context, maintaining a complete conversation history including tool interactions.

3. Add Context Aggregators to Your Pipeline Copy Section

pipeline = Pipeline([
    transport.input(),
    stt,
    context_aggregator.user(),      # User context aggregator
    llm,
    tts,
    transport.output(),
    context_aggregator.assistant(), # Assistant context aggregator
])

System Instruction, Developer Messages, and System Messages Copy Chapter

Pipecat provides three ways to give instructions to your LLM, each suited for a different purpose.

Using `system_instruction` (recommended) Copy Section

Set the bot's personality and core behavior via the LLM service's Settings. The service automatically prepends it to the context messages on each request:

llm = OpenAILLMService(
    api_key=os.getenv("OPENAI_API_KEY"),
    settings=OpenAILLMService.Settings(
        model="gpt-4o",
        system_instruction="You are a helpful voice assistant.",
    ),
)

# Context only needs conversation messages
context = LLMContext()

This approach is recommended because:

• Survives context updates: The system prompt is always prepended, even after LLMMessagesUpdateFrame replaces the context or context summarization compresses older messages.
• Shared context: Multiple LLM services can share a single LLMContext while each provides its own system instruction.
• Runtime updates: You can change the system prompt mid-conversation via LLMUpdateSettingsFrame.

Using developer messages in context Copy Section

Developer messages ("role": "developer") are task-specific instructions placed directly in the context. Use them for supplementary guidance that applies to a particular phase of the conversation rather than defining the bot's overall personality.

messages = [
    {"role": "developer", "content": "Keep responses under two sentences. Use metric units."},
]
context = LLMContext(messages)

Key characteristics:

• Task-specific: Use for instructions like response format constraints, domain rules, or workflow steps. The bot's personality belongs in system_instruction.
• Part of normal context flow: Developer messages participate in context like any other message. They are included in the summarization range and may be compressed or dropped during context summarization.
• Cross-provider translation: Pipecat's adapters automatically convert developer messages for providers that don't support the role natively (for example, Anthropic receives them as user messages).

Using a context system message (legacy) Copy Section

You can also include a system message directly in the context messages:

messages = [
    {"role": "system", "content": "You are a helpful voice assistant."},
]
context = LLMContext(messages)

This works but has limitations: the system message can be lost during full context replacement, and it cannot be shared across multiple LLM services with different system prompts. Context summarization does preserve messages[0] when it's a system message, but system_instruction is still more reliable because it sits outside the context entirely.

For new projects, prefer system_instruction for personality and developer messages for task-specific instructions.

Context Aggregator Placement Copy Chapter

The placement of context aggregator instances in your pipeline is crucial for proper operation:

User Context Aggregator Copy Section

Place the user context aggregator downstream from the STT service. Since the user's speech results in TranscriptionFrame objects pushed by the STT service, the user aggregator needs to be positioned to collect these frames.

Assistant Context Aggregator Copy Section

Place the assistant context aggregator after transport.output(). This positioning is important because:

• The TTS service outputs TTSTextFrames in addition to audio
• The assistant aggregator must be downstream to collect those frames
• It ensures context updates happen word-by-word for specific services (e.g. Cartesia, ElevenLabs, and Rime)
• Your context stays updated at the word level in case an interruption occurs

Manual Context Control Copy Chapter

You can programmatically add new messages to the context by pushing or queueing specific frames:

Adding Messages Copy Section

• LLMMessagesAppendFrame: Appends a new message to the existing context
• LLMMessagesUpdateFrame: Completely replaces the existing context with new messages
• LLMMessagesTransformFrame: Edits the existing context in place using a transform function

# Add a new user message to context and trigger a response
new_message = {"role": "user", "content": "Tell me about your capabilities."}
await worker.queue_frames([
    LLMMessagesAppendFrame([new_message], run_llm=True), # Optionally trigger bot response, too
])

Adding a message silently

All three frames take a run_llm argument that controls whether the change also prompts a bot response. Pass run_llm=True to respond; the default (None, which behaves like False) updates the context silently. This is useful when you collect information in the background and don't want the bot to react every time:

# Add a message to context without triggering a bot response
note = {"role": "user", "content": "Caller's name is Maria. Account verified."}
await worker.queue_frames([
    LLMMessagesAppendFrame([note], run_llm=False), # Silent: no response
])

Editing or removing specific messages

To surgically edit or remove individual messages without rebuilding the whole list, push an LLMMessagesTransformFrame. It takes a function that receives the current list of messages and returns a modified list. Use it to drop stale instructions, remove an offensive turn, or rewrite content:

from pipecat.frames.frames import LLMMessagesTransformFrame

def remove_language_instructions(messages):
    # Drop any message that contains a per-turn language instruction
    return [m for m in messages if "LANGUAGE INSTRUCTION" not in str(m.get("content", ""))]

await worker.queue_frames([
    LLMMessagesTransformFrame(transform=remove_language_instructions, run_llm=False),
])

Retrieving Current Context Copy Section

The context aggregator provides a context property for getting the current context:

context = context_aggregator.user().context

Context Summarization Copy Chapter

In long-running conversations, context grows with every exchange, increasing token usage and potentially hitting context window limits. Pipecat includes built-in context summarization that automatically compresses older conversation history while preserving recent messages.

Enable it by setting enable_auto_context_summarization=True when creating your context aggregators (default: False):

user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
    context,
    assistant_params=LLMAssistantAggregatorParams(
        enable_auto_context_summarization=True,
    ),
)

<Card
title="Context Summarization Guide"
icon="arrow-right"
href="/pipecat/fundamentals/context-summarization"

Learn how to configure summarization triggers, customize behavior, and control
what gets preserved.

Triggering Bot Responses Copy Chapter

You may want to manually trigger the bot to speak in two scenarios:

1. Starting a pipeline where the bot should speak first
2. After editing the context using LLMMessagesAppendFrame or LLMMessagesUpdateFrame

# Example: Bot speaks first when pipeline starts
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
    # Trigger a response
    await worker.queue_frames([LLMRunFrame()])

# Example: Bot speaks after context is edited
new_message = {"role": "user", "content": "Tell me a fun fact."}
await worker.queue_frames([
    LLMMessagesAppendFrame([new_message], run_llm=True), # Trigger bot response
])

This gives you fine-grained control over when and how the bot responds during the conversation flow.

Key Takeaways Copy Chapter

• Context is conversation history - automatically maintained as users and bots exchange messages
• Use system_instruction for system prompts - set it in LLM Settings rather than as a context message for reliability across context updates and summarization
• Use developer messages for task instructions - place task-specific guidance in context with "role": "developer" rather than overloading the system prompt
• Frame types matter - TranscriptionFrame for users, TTSTextFrame for assistants
• Placement matters - user aggregator after STT, assistant aggregator after transport output
• Tools are included - function definitions and results are stored in context
• Manual control available - use frames to append messages or trigger responses when needed
• Word-level precision - proper placement ensures context accuracy during interruptions
• Automatic summarization - enable context summarization to manage long conversations efficiently and reduce token costs

What's Next Copy Chapter

Now that you understand context management, let's explore how to configure the LLM services that process this context to generate intelligent responses.

LLM Inference

Source: https://docs.pipecat.ai/pipecat/learn/llm.md

Learn how to configure language models to generate intelligent responses in your voice AI pipeline

LLM services are responsible for chat completions and tool calling based on the provided context (conversation history). The LLM responds by streaming tokens via LLMTextFrames, which are used by subsequent processors to create audio output for the bot.

Pipeline Placement Copy Chapter

The LLM instance should be placed after the user context aggregator and before any downstream services that depend on the LLM's output stream:

pipeline = Pipeline([
    transport.input(),
    stt,                           # Creates TranscriptionFrames
    context_aggregator.user(),     # Processes user context → creates LLMContextFrame
    llm,                           # Processes context → streams LLMTextFrames
    tts,                           # Processes LLMTextFrames → creates TTSAudioRawFrames
    transport.output(),
    context_aggregator.assistant(),
])

Frame flow:

• Input: Receives LLMContextFrame containing conversation history
• Processing:
  • Analyzes context and generates streaming response
  • Handles function calls if tools are available
  • Tracks token usage for metrics
• Output:
  • Denotes the start of the streaming response by pushing an LLMFullResponseStartFrame
  • Streams LLMTextFrames containing response tokens to downstream processors (enables real-time TTS processing)
  • Ends with an LLMFullResponseEndFrame to mark the completion of the response
  • Output frames can be configured to skip TTS via LLMConfigureOutputFrame(skip_tts=True), allowing text to flow through the pipeline without being spoken
• Function calls:
  • FunctionCallsStartedFrame: Indicates function execution beginning
  • FunctionCallInProgressFrame: Indicates a function is currently executing
  • FunctionCallResultFrame: Contains results from executed functions

Supported LLM Services Copy Chapter

Pipecat supports a wide range of LLM providers to fit different needs, performance requirements, and budgets:

Text-Based LLMs Copy Section

Most LLM services are built on the OpenAI chat completion specification for compatibility:

Compatible APIs: Any OpenAI-spec compatible service can be used via the base_url parameter.

Speech-to-Speech Models Copy Section

For lower latency, some providers offer direct speech-to-speech models:

• OpenAI Realtime: Direct speech input/output with GPT models
• Gemini Live: Real-time speech conversations with Gemini
• AWS Nova Sonic: Speech-optimized models on Bedrock

For a complete list of supported LLM services, see the Supported Services page:

<Card
title="Supported LLM Services"
icon="list"
href="/api-reference/server/services/supported-services#large-language-models"

View the complete list of supported language model providers

LLM Service Architecture Copy Chapter

BaseOpenAILLMService Copy Section

Many LLM services use the OpenAI chat completion specification. Pipecat provides a BaseOpenAILLMService that most providers extend, enabling easy switching between compatible services:

from pipecat.services.openai.llm import OpenAILLMService

# Native OpenAI
llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))

# OpenAI-compatible service via base_url
llm = OpenAILLMService(
    api_key=os.getenv("OTHER_API_KEY"),
    base_url="https://api.other-provider.com/v1"  # Custom endpoint
)

This architecture allows you to quickly plug in different LLM services without changing your pipeline code.

LLM Configuration Copy Chapter

Service-Specific Configuration Copy Section

Each LLM service has its own configuration options. For example, configuring OpenAI with various parameters:

from pipecat.services.openai.llm import OpenAILLMService

llm = OpenAILLMService(
    api_key=os.getenv("OPENAI_API_KEY"),
    settings=OpenAILLMService.Settings(
        model="gpt-4.1",
        system_instruction="You are a helpful voice assistant.",
        temperature=0.7,            # Response creativity (0.0-2.0)
        max_completion_tokens=150,  # Maximum response length
        frequency_penalty=0.5,      # Reduce repetition (0.0-2.0)
        presence_penalty=0.5,       # Encourage topic diversity (0.0-2.0)
    ),
)

system_instruction defines the bot's personality and core behavior. For task-specific instructions (response format constraints, domain rules, workflow steps), use developer messages in context instead. See Context Management for how these interact.

For detailed configuration options specific to each provider:

<Card
title="Individual LLM Services"
icon="settings"
href="/api-reference/server/services/supported-services#large-language-models"

Explore configuration options for each supported LLM provider

Base Class Configuration Copy Section

All LLM services inherit from the LLMService base class with shared configuration options:

llm = YourLLMService(
    # Service-specific options...
    run_in_parallel=True,  # Whether function calls run in parallel (default: True)
)

Key options:

• run_in_parallel: Controls whether function calls execute simultaneously or sequentially
  • True (default): Faster execution when multiple functions are called
  • False: Sequential execution for dependent function calls

Event Handlers Copy Chapter

LLM services provide event handlers for monitoring completion lifecycle:

@llm.event_handler("on_completion_timeout")
async def on_completion_timeout(service):
    logger.warning("LLM completion timed out")
    # Handle timeout (retry, fallback, etc.)

@llm.event_handler("on_function_calls_started")
async def on_function_calls_started(service, function_calls):
    logger.info(f"Starting {len(function_calls)} function calls")
    # Optionally notify user that bot is "thinking"
    await tts.queue_frame(TTSSpeakFrame("Let me check on that."))

Available events:

• on_completion_timeout: Triggered when LLM requests timeout
• on_function_calls_started: Triggered when function calls are initiated

These handlers enable you to provide user feedback and implement error recovery strategies.

Function Calling Copy Chapter

LLMs can call external functions to access real-time data and perform actions beyond their training data. This enables capabilities like checking weather, querying databases, or controlling external APIs.

Function calls and their results are automatically stored in the conversation context by the context aggregator.

<Card
title="Function Calling"
icon="code"
href="/pipecat/learn/function-calling"

Learn how to enable LLMs to interact with external services and APIs

Key Takeaways Copy Chapter

• Pipeline placement matters - LLM goes after user context, before TTS
• Token streaming enables real-time responses - no waiting for complete generation
• OpenAI compatibility enables easy provider switching
• Function calling extends capabilities beyond training data
• Configuration affects behavior - tune temperature, penalties, and limits
• Services are modular - swap providers without changing pipeline code

What's Next Copy Chapter

Now that you understand LLM configuration, let's explore how function calling enables your bot to interact with external services and real-time data.

<Card
title="Function Calling"
icon="arrow-right"
href="/pipecat/learn/function-calling"

Learn how to enable LLMs to interact with external services and APIs

Function Calling

Source: https://docs.pipecat.ai/pipecat/learn/function-calling.md

Enable LLMs to interact with external services and APIs in your voice AI pipeline

Function calling (also known as tool calling) allows LLMs to request information from external services and APIs during conversations. This extends your voice AI bot's capabilities beyond its training data to access real-time information and perform actions.

Pipeline Integration Copy Chapter

Function calling works seamlessly within your existing pipeline structure. The LLM service handles function calls automatically when they're needed:

pipeline = Pipeline([
    transport.input(),
    stt,
    context_aggregator.user(),     # Collects user transcriptions
    llm,                          # Processes context, calls functions when needed
    tts,
    transport.output(),
    context_aggregator.assistant(), # Collects function results and responses
])

Function call flow:

1. User asks a question requiring external data
2. LLM recognizes the need and calls appropriate function
3. Your function handler executes and returns results
4. LLM incorporates results into its response
5. Response flows to TTS and user as normal

Context integration: Function calls and their results are automatically stored in conversation context by the context aggregators, maintaining complete conversation history.

Understanding Function Calling Copy Chapter

Function calling allows your bot to access real-time data and perform actions that aren't part of its training data.

For example, you could give your bot the ability to:

• Check current weather conditions
• Look up stock prices
• Query a database
• Control smart home devices
• Schedule appointments

Here's how it works:

1. You define functions the LLM can use and make them available to the LLM service used in your pipeline
2. When needed, the LLM requests a function call
3. Your application executes any corresponding functions
4. The result is sent back to the LLM
5. The LLM uses this information in its response

Implementation Copy Chapter

1. Define a tool Copy Section

A tool needs two things: a handler — the code to run when the LLM calls the tool — and a schema that describes the tool to the LLM (its name, what it does, and its parameters) so the model knows it exists and how to call it.

The preferred way to define a tool is with a direct function: a single async function that is both the handler and the schema. Pipecat auto-derives the tool's metadata — name, description, parameter properties (with their descriptions), and which parameters are required — from the function's signature and docstring.

The first parameter is always params (a FunctionCallParams); the tool's own arguments follow. Document each argument in a Google-style docstring.

from pipecat.services.llm_service import FunctionCallParams

async def get_current_weather(params: FunctionCallParams, location: str, format: str):
    """Get the current weather.

    Args:
        location: The city and state, e.g. "San Francisco, CA".
        format: The temperature unit to use. Must be either "celsius" or "fahrenheit". Infer this from the user's location.
    """
    weather_data = {"conditions": "sunny", "temperature": "75"}
    await params.result_callback(weather_data)

2. Add the tool to the context Copy Section

List your direct functions in LLMContext(tools=[...]):

from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair

context = LLMContext(tools=[get_current_weather, get_restaurant_recommendation])
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(context)

The bot's personality (e.g. "You are a helpful assistant") is set via system_instruction in the LLM service's Settings, not as a context message. Tools are automatically converted to the correct format for your LLM provider through adapters.

3. Create the pipeline Copy Section

Include your LLM service in the pipeline:

# Create the pipeline
pipeline = Pipeline([
    transport.input(),     # Input from the transport
    stt,                   # STT processing
    user_aggregator,       # User context aggregation
    llm,                   # LLM processing
    tts,                   # TTS processing
    transport.output(),    # Output to the transport
    assistant_aggregator,  # Assistant context aggregation
])

Per-Tool Options with `@tool_options` Copy Chapter

By default, a direct function is cancelled if the user interrupts, and it uses the LLM service's global timeout. To override either, decorate the function with @tool_options. The decorator only attaches call options — the schema is still auto-derived — so decorated functions can stay at module level.

import asyncio

from pipecat.adapters.schemas.direct_function import tool_options
from pipecat.services.llm_service import FunctionCallParams

@tool_options(cancel_on_interruption=False, timeout_secs=30)
async def get_current_weather(params: FunctionCallParams, location: str, format: str):
    """Get the current weather.

    Args:
        location: The city and state, e.g. "San Francisco, CA".
        format: The temperature unit to use. Must be either "celsius" or "fahrenheit". Infer this from the user's location.
    """
    # Simulate a long-running API call.
    await asyncio.sleep(20)
    await params.result_callback({"conditions": "nice", "temperature": "75"})

Options:

• cancel_on_interruption (default True): When True, the call is cancelled if the user interrupts. When False, the call is treated as asynchronous — see below.
• timeout_secs (default None): Per-tool timeout in seconds. Overrides the global function_call_timeout_secs for this function. Use a longer timeout for slow operations (e.g. database queries) or a shorter one for quick lookups.

Synchronous vs. asynchronous calls Copy Section

With cancel_on_interruption=True (the default), the call is synchronous: the LLM waits for the result before generating its next response. This ensures the LLM has complete information before responding.

With cancel_on_interruption=False, the call is asynchronous: the LLM continues the conversation immediately without waiting. Once the result returns, it's injected back into the context as a developer message, triggering a new LLM inference at that point. This enables truly non-blocking calls where the conversation proceeds while the function runs in the background. Async calls can also send intermediate updates before their final result.

Async function call cancellation

For async functions (cancel_on_interruption=False), you can also enable model-directed cancellation:

llm = OpenAILLMService(
    api_key="your-api-key",
    enable_async_tool_cancellation=True,
)

When enable_async_tool_cancellation=True and at least one async function is available, Pipecat automatically adds the built-in cancel_async_tool_call tool and supporting system instructions. The LLM can call that tool to cancel a stale in-progress async function call — for example, when the user changes their request before a long-running lookup completes.

Parallel and Multiple Tool Calls Copy Chapter

When the LLM calls more than one tool in a single turn, two LLMService constructor options control how those calls run and when the LLM responds:

llm = OpenAILLMService(
    api_key="your-api-key",
    run_in_parallel=True,       # default
    group_parallel_tools=True,  # default
)

Whether multiple tool calls in one turn run in parallel or one after another.

When True, all tool calls in a batch are grouped so the LLM is triggered
exactly once after every call in the batch completes. When False, each
function call result triggers the LLM independently as it arrives.

Changing Tools Mid-Conversation Copy Chapter

To change the set of tools the LLM can use during a session, push an LLMSetToolsFrame. Its tools field takes the same things as LLMContext(tools=[...]) — a list of direct functions and/or FunctionSchema objects. Whatever you pass becomes the LLM's new tool set.

from pipecat.frames.frames import LLMSetToolsFrame
from pipecat.processors.aggregators.llm_context import NOT_GIVEN

# Make get_current_weather the only tool the LLM can call
await worker.queue_frame(LLMSetToolsFrame(tools=[get_current_weather]))

# Clear all tools
await worker.queue_frame(LLMSetToolsFrame(tools=NOT_GIVEN))

Tools Across Service Switches Copy Chapter

When you use an LLMSwitcher to swap LLM providers mid-session, the tools you list in LLMContext(tools=[...]) are available on whichever provider is active. You define them once for the whole switcher.

from pipecat.pipeline.llm_switcher import LLMSwitcher

llm_switcher = LLMSwitcher(llms=[llm_openai, llm_google])

# Tools in the context are available on whichever provider is active
context = LLMContext(tools=[get_current_weather, get_restaurant_recommendation])
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(context)

Advanced: Defining Tools with `FunctionSchema` Copy Chapter

Direct functions cover most cases. Reach for the verbose FunctionSchema pattern when you need explicit control over the schema — for example, a strict enum constraint (which the direct-function generator doesn't yet emit) — or when the tool's handler isn't shaped like a direct function.

A FunctionSchema spells out the tool's name, description, and parameters by hand. Pass the handler that runs when the LLM calls the tool as the schema's handler, then list the schema in LLMContext(tools=[...]) — exactly as you would a direct function.

from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.services.llm_service import FunctionCallParams

async def fetch_weather_from_api(params: FunctionCallParams):
    weather_data = {"conditions": "sunny", "temperature": "75"}
    await params.result_callback(weather_data)

weather_function = FunctionSchema(
    name="get_current_weather",
    description="Get the current weather in a location",
    properties={
        "location": {
            "type": "string",
            "description": "The city and state, e.g. San Francisco, CA",
        },
        "format": {
            "type": "string",
            # A strict enum — the kind of explicit control a direct function can't yet express.
            "enum": ["celsius", "fahrenheit"],
            "description": "The temperature unit to use.",
        },
    },
    required=["location", "format"],
    handler=fetch_weather_from_api,  # bundle the handler on the schema
)

# List the schema in the context, just like a direct function.
context = LLMContext(tools=[weather_function])

To override the handler's default call options, decorate it with @tool_options — the same decorator direct functions use, with the same synchronous vs. asynchronous semantics:

from pipecat.adapters.schemas.direct_function import tool_options

@tool_options(cancel_on_interruption=False, timeout_secs=30)
async def fetch_weather_from_api(params: FunctionCallParams):
    ...

These schemas behave just like direct functions everywhere else in this guide — swap them mid-conversation with an LLMSetToolsFrame, and they keep working across an LLMSwitcher's providers.

Registering a handler manually Copy Section

Bundling the handler on the schema (above) is the recommended approach. If you'd rather keep the handler separate, list a handler-free FunctionSchema in the context as usual and register its handler by name:

# weather_function here is the same schema, just defined without handler=.
context = LLMContext(tools=[weather_function])
llm.register_function("get_current_weather", fetch_weather_from_api)

If the handler carries @tool_options, register_function honors it the same way bundling does — or pass cancel_on_interruption / timeout_secs to register_function directly to override.

To remove the tool, un-advertise it with an LLMSetToolsFrame; call llm.unregister_function(...) only afterward, since unregistering a still-advertised tool leaves the LLM able to call a handler that's no longer there.

This is uncommon — bundling keeps a tool and its handler together — but the option is there when you need to manage registration directly.

Provider-Specific Custom Tools Copy Section

For normal function calling, prefer standard_tools with FunctionSchema or direct functions so Pipecat can convert them to each provider's native format. When a provider has tools that don't fit Pipecat's standard function schema, add those provider-native definitions through ToolsSchema.custom_tools. These custom tools are passed only to the matching adapter and are appended to the converted standard tools.

from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema

# Standard function converted by Pipecat
weather_function = FunctionSchema(
    name="get_current_weather",
    description="Get the current weather",
    properties={"location": {"type": "string"}},
    required=["location"],
)

# Provider-native tool appended only for OpenAI-family adapters.
# This object must match the target OpenAI API you are using.
provider_tool = {"type": "tool_search"}

tools = ToolsSchema(
    standard_tools=[weather_function],
    custom_tools={AdapterType.OPENAI: [provider_tool]},
)

from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema

# Standard function converted by Pipecat
weather_function = FunctionSchema(
    name="get_current_weather",
    description="Get the current weather",
    properties={"location": {"type": "string"}},
    required=["location"],
)

# Provider-native tool appended only for Gemini adapters.
# This object must match the Gemini API you are using.
gemini_search_tool = {"google_search": {}}

tools = ToolsSchema(
    standard_tools=[weather_function],
    custom_tools={AdapterType.GEMINI: [gemini_search_tool]},
)

Function Handler Details Copy Chapter

FunctionCallParams Copy Section

Every function handler receives a FunctionCallParams object containing all the information needed for execution:

@dataclass
class FunctionCallParams:
    function_name: str                          # Name of the called function
    tool_call_id: str                           # Unique identifier for this call
    arguments: Mapping[str, Any]                # Arguments from the LLM
    llm: LLMService                             # Reference to the LLM service
    context: LLMContext                         # Current conversation context
    result_callback: FunctionCallResultCallback # Return results here
    app_resources: Any                          # Application-defined resources shared across tool calls

Using the parameters:

async def example_function_handler(params: FunctionCallParams):
    # Access function details
    print(f"Called function: {params.function_name}")
    print(f"Call ID: {params.tool_call_id}")

    # Extract arguments
    location = params.arguments["location"]

    # Access LLM context for conversation history
    messages = params.context.messages

    # Access shared resources (database, API clients, etc.)
    if params.app_resources:
        db = params.app_resources.database
        user_id = params.app_resources.current_user_id

    # Use LLM service for additional operations
    await params.llm.push_frame(TTSSpeakFrame("Looking up weather data..."))

    # Return results
    await params.result_callback({"conditions": "nice", "temperature": "75"})

See the API reference for complete details.

Handler Structure Copy Section

Your function handler should:

1. Receive necessary arguments, either:
   • From params.arguments
   • Directly from function arguments, if using direct functions
2. Process data or call external services
3. Return results via params.result_callback(result)

async def fetch_weather_from_api(params: FunctionCallParams):
    try:
        # Extract arguments
        location = params.arguments.get("location")
        format_type = params.arguments.get("format", "celsius")

        # Call external API
        api_result = await weather_api.get_weather(location, format_type)

        # Return formatted result
        await params.result_callback({
            "location": location,
            "temperature": api_result["temp"],
            "conditions": api_result["conditions"],
            "unit": format_type
        })
    except Exception as e:
        # Handle errors
        await params.result_callback({
            "error": f"Failed to get weather: {str(e)}"
        })

async def get_current_weather(params: FunctionCallParams, location: str, format: str):
    """Get the current weather.

    Args:
        location: The city and state, e.g. "San Francisco, CA".
        format: The temperature unit to use. Must be either "celsius" or "fahrenheit".
    """
    try:
        # Call external API
        api_result = await weather_api.get_weather(location, format)

        # Return formatted result
        await params.result_callback({
            "location": location,
            "temperature": api_result["temp"],
            "conditions": api_result["conditions"],
            "unit": format
        })
    except Exception as e:
        # Handle errors
        await params.result_callback({
            "error": f"Failed to get weather: {str(e)}"
        })

Sharing Resources with app_resources Copy Section

When function handlers need access to shared resources like database connections, API clients, or application state, you can pass them via app_resources when creating the PipelineWorker. These resources are then accessible in every function handler via params.app_resources.

from dataclasses import dataclass
from pipecat.pipeline.worker import PipelineWorker
from pipecat.services.llm_service import FunctionCallParams

# Define your application resources
@dataclass
class AppResources:
    database: DatabaseConnection
    api_client: WeatherAPIClient
    user_id: str

# Create your resources
resources = AppResources(
    database=db_connection,
    api_client=weather_client,
    user_id="user-123"
)

# Pass resources to the pipeline worker
worker = PipelineWorker(
    pipeline,
    app_resources=resources
)

# Access resources in function handlers
async def query_user_preferences(params: FunctionCallParams):
    # Access shared resources
    db = params.app_resources.database
    user_id = params.app_resources.user_id

    # Query database with shared connection
    prefs = await db.query("SELECT * FROM preferences WHERE user_id = ?", user_id)

    await params.result_callback(prefs)

Key points:

• Resources are passed by reference — the caller retains their handle and can read mutations after the task finishes
• The framework never copies or clears the app_resources object
• All function handlers in the pipeline share the same app_resources instance
• Useful for database connections, API clients, caches, or any shared state

Advanced: Controlling Function Call Behavior Copy Chapter

When returning results from a function handler, you can control how the LLM processes those results using a FunctionCallResultProperties object passed to the result callback.

Properties Copy Section

FunctionCallResultProperties provides fine-grained control over LLM execution:

@dataclass
class FunctionCallResultProperties:
    run_llm: bool | None = None                 # Whether to run LLM after this result
    on_context_updated: Callable | None = None  # Callback when context is updated
    is_final: bool = True                       # Whether this is the final result

Property options:

• run_llm=True: Run LLM after function call (default behavior)
• run_llm=False: Don't run LLM after function call (useful for chained calls)
• on_context_updated: Async callback executed after the function result is added to context
• is_final=False: Treat this as an intermediate result for an async function call. Only use this for async functions (cancel_on_interruption=False)

See the API reference for complete details.

Example Usage Copy Section

from pipecat.frames.frames import FunctionCallResultProperties
from pipecat.services.llm_service import FunctionCallParams

async def fetch_weather_from_api(params: FunctionCallParams):
    # Fetch weather data
    weather_data = {"conditions": "sunny", "temperature": "75"}

    # Don't run LLM after this function call
    properties = FunctionCallResultProperties(run_llm=False)

    await params.result_callback(weather_data, properties=properties)

async def query_database(params: FunctionCallParams):
    # Query database
    results = await db.query(params.arguments["query"])

    async def on_update():
        await notify_system("Database query complete")

    # Run LLM after function call and notify when context is updated
    properties = FunctionCallResultProperties(
        run_llm=True,
        on_context_updated=on_update
    )

    await params.result_callback(results, properties=properties)

Intermediate Results for Async Functions Copy Section

Async function calls can send progress updates before their final result. Make the function async with @tool_options(cancel_on_interruption=False), then call params.result_callback(..., properties=FunctionCallResultProperties(is_final=False)) for each intermediate update. Finish with a normal params.result_callback(...).

from pipecat.adapters.schemas.direct_function import tool_options
from pipecat.frames.frames import FunctionCallResultProperties
from pipecat.services.llm_service import FunctionCallParams

@tool_options(cancel_on_interruption=False)
async def track_delivery(params: FunctionCallParams):
    """Track a delivery, reporting each status update until it arrives."""
    await params.result_callback(
        {"status": "picked_up"},
        properties=FunctionCallResultProperties(is_final=False),
    )

    await params.result_callback(
        {"status": "nearby"},
        properties=FunctionCallResultProperties(is_final=False),
    )

    await params.result_callback({"status": "delivered"})

Intermediate results are injected into the LLM context as async-tool developer messages. They do not close the function call; the call remains in progress until the final result is sent.

Key Takeaways Copy Chapter

• Function calling extends LLM capabilities beyond training data to real-time information
• Context integration is automatic - function calls and results are stored in conversation history
• Direct functions are the preferred approach - one async function is both schema and handler; list it in LLMContext(tools=[...]) or add it via LLMSetToolsFrame to make it available. When you need explicit schema control, use a FunctionSchema with its handler bundled in
• Async function calls are opt-in - set cancel_on_interruption=False for deferred results, intermediate updates, and optional async-tool cancellation
• Pipeline integration is seamless - functions work within your existing voice AI architecture
• Advanced control available - fine-tune LLM execution and monitor function call lifecycle

What's Next Copy Chapter

Now that you understand function calling, let's explore how to configure text-to-speech services to convert your LLM's responses (including function call results) into natural-sounding speech.

<Card
title="Text to Speech"
icon="arrow-right"
href="/pipecat/learn/text-to-speech"

Learn how to configure speech synthesis in your voice AI pipeline

Text to Speech

Source: https://docs.pipecat.ai/pipecat/learn/text-to-speech.md

Learn how to configure speech synthesis to convert text into natural-sounding audio in your voice AI pipeline

Text to Speech (TTS) services are responsible for converting text into natural-sounding speech audio. They receive text input from LLMs and other sources, then generate audio output that users can hear through their connected devices.

Pipeline Placement Copy Chapter

TTS processors must be positioned correctly in your pipeline to receive text and generate audio frames:

pipeline = Pipeline([
    transport.input(),
    stt,
    context_aggregator.user(),
    llm,                           # Generates LLMTextFrames
    tts,                           # Processes text → creates TTSAudioRawFrames
    transport.output(),            # Sends audio to user
    context_aggregator.assistant(), # Processes TTSTextFrames for context
])

Placement requirements:

• After LLM processing: TTS needs LLMTextFrames from language model responses
• Before transport output: Audio must be generated before sending to user
• Before assistant context aggregator: Ensures spoken text is captured in conversation history

Frame Processing Flow Copy Section

TTS generates speech through two primary mechanisms:

1. Streamed LLM tokens via LLMTextFrames:

   • By default, TTS aggregates streaming tokens into complete sentences before synthesis (TextAggregationMode.SENTENCE)
   • Set text_aggregation_mode=TextAggregationMode.TOKEN to stream tokens directly for lower latency
   • Audio bytes stream back and play immediately
   • End-to-end latency often under 200ms

2. Direct speech requests via TTSSpeakFrames:

   • Bypasses LLM for immediate audio generation
   • Optionally appends text to conversation context via append_to_context parameter
   • Useful for developer messages, greetings, or injected speech

Frame output:

• TTSAudioRawFrames: Raw audio data for playback
• TTSTextFrames: Text that was actually spoken (for context updates)
• TTSStartedFrame/TTSStoppedFrame: Speech boundary markers

Supported TTS Services Copy Chapter

Pipecat supports a wide range of TTS providers with different capabilities and performance characteristics:

<Card
title="Supported TTS Services"
icon="list"
href="/api-reference/server/services/supported-services#text-to-speech"

View the complete list of supported text-to-speech providers

Service Categories Copy Section

WebSocket-Based Services (Recommended):

• Cartesia: Ultra-low latency with word timestamps
• ElevenLabs: High-quality voices with emotion control
• Rime: Ultra-realistic voices with advanced features

HTTP-Based Services:

• OpenAI TTS: High-quality synthesis with multiple voices
• Azure Speech: Enterprise-grade with extensive language support
• Google Text-to-Speech: Reliable with WaveNet voices

Advanced Features:

• Word timestamps: Enable word-level accuracy for context and subtitles
• Voice cloning: Custom voice creation from samples
• Emotion control: Dynamic emotional expression
• SSML support: Fine-grained pronunciation control

TTS Configuration Copy Chapter

Service-Specific Configuration Copy Section

Each TTS service has its own configuration options. Here's an example with Cartesia:

from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.transcriptions.language import Language

tts = CartesiaTTSService(
    api_key=os.getenv("CARTESIA_API_KEY"),
    settings=CartesiaTTSService.Settings(
        model="sonic-3.5",
        voice="voice-id-here",
        language=Language.EN,     # Speech language
    ),
    # Word timestamps automatically enabled for precise context updates
)

Word timestamps: Services like Cartesia, ElevenLabs, and Rime provide word-level timestamps that enable precise context updates during interruptions and better synchronization with other pipeline components. For example, if an interruption occurs while the bot is speaking, the word timestamps allow you to accurately capture which words were spoken up to that point, enabling better context management and user experience. Additionally, transcription events streamed from server to client can be done in sync with the audio output, allowing for real-time subtitles or captions.

<Card
title="Individual TTS Services"
icon="settings"
href="/api-reference/server/services/supported-services#text-to-speech"

Explore configuration options for each supported TTS provider

Pipeline-Level Audio Configuration Copy Section

Set consistent audio settings across your entire pipeline:

worker = PipelineWorker(
    pipeline,
    params=PipelineParams(
        audio_in_sample_rate=16000,   # Input audio quality
        audio_out_sample_rate=24000,  # Output audio quality (TTS)
    ),
)

Text Processing and Filtering Copy Chapter

Custom Text Aggregation Copy Section

By default, TTS services have a built-in text aggregator that collects streaming text into sentences before passing them to the underlying service. However, you can customize this behavior by inserting an LLMTextProcessor with a different text aggregator before the TTS in your pipeline. This allows the ability to categorize and structure text into logical units beyond simple sentences, such as code blocks, URLs, or custom tags. You can then configure the TTS to handle these different text types appropriately, such as skipping code blocks or transforming them in a just-in-time manner before speaking.

Skipping Text Aggregations Copy Section

To skip certain text aggregations (e.g., code snippets or URLs) and keep them from being spoken, use a custom text aggregator like PatternPairAggregator within an LLMTextProcessor, and configure it to identify and handle specific patterns in the text stream. With this, you can then pass any aggregated types you want to skip (like "code") to the TTS service's skip_aggregator_types parameter.

# Create pattern aggregator
pattern_aggregator = PatternPairAggregator()

# Add pattern for JSON data
pattern_aggregator.add_pattern(
    type="code",
    start_pattern="<code>",
    end_pattern="</code>",
    action=MatchAction.AGGREGATE
)

# Set the aggregator on an LLMTextProcessor
llm_text_processor = LLMTextProcessor(text_aggregator=pattern_aggregator)

# Initialize TTS service, and don't speak JSON data
tts = CartesiaTTSService(
    api_key=os.getenv("CARTESIA_API_KEY"),
    skip_aggregator_types=["code"], # The strings here should match the types defined in the PatternPairAggregator
)

# add the llm_text_processor to your pipeline after the llm and before the tts
# llm -> llm_text_processor -> tts

Text Transforms Copy Section

For TTS-specific text preprocessing, you can provide custom text transforms that modify text in a just-in-time manner before sending the text off to the TTS service. This is useful for handling special text segments that need to be altered for better pronunciation or clarity, such as spelling out phone numbers, removing URLs, or expanding abbreviations. These text transforms can be mapped to a specific text aggregation type, like with skip_aggregator_types, or applied globally to all text using '*' as the type.

Text transforms are registered directly on the TTS service instance via the add_text_transformer() method or during initialization using the text_transforms parameter.

# Create pattern aggregator
pattern_aggregator = PatternPairAggregator()

# Add patterns for different parts of an explanation
pattern_aggregator.add_pattern(
    type="phone_number",
    start_pattern="",
    end_pattern="",
    action=MatchAction.AGGREGATE
)

# Set the aggregator on an LLMTextProcessor
llm_text_processor = LLMTextProcessor(text_aggregator=pattern_aggregator)

# Text-to-Speech service
tts = CartesiaTTSService(
    api_key=os.getenv("CARTESIA_API_KEY"),
)

# Text transformers for TTS
# This will insert Cartesia's spell tags around the provided text.
async def spell_out_text(text: str, type: str) -> str:
    # CartesiaTTSService provides a helper for this along with other common transforms
    return CartesiaTTSService.SPELL(text)

async def replace_acronyms(text: str, type: str) -> str:
    # Replace "SEC" with "Southeastern Conference"
    return text.replace(" SEC ", " Southeastern Conference ")

# Setup the text transformers in TTS to spell out phone numbers and replace
# acronyms. The string below matches the type defined in the PatternPairAggregator
# above so that whenever those segments are encountered, this transform
# is applied
tts.add_text_transformer(spell_out_text, "phone_number")
tts.add_text_transformer(replace_acronyms, "*")  # Apply to all text

# add the llm_text_processor to your pipeline after the llm and before the tts
#   llm -> llm_text_processor -> tts

Text Filters Copy Section

Apply preprocessing to text before synthesis:

from pipecat.utils.text.markdown_text_filter import MarkdownTextFilter

tts = YourTTSService(
    # ... other options
    text_filters=[
        MarkdownTextFilter(),     # Remove markdown formatting
        CustomTextFilter(),       # Your custom processing
    ],
)

Common filters:

• MarkdownTextFilter: Strips markdown formatting from LLM responses
• Custom filters: Implement your own text preprocessing logic

Skipping TTS Output Copy Chapter

Sometimes you want text from the LLM to flow through the pipeline—updating the conversation context, reaching observers, or being processed by custom frame processors—without being spoken by the TTS service. Pipecat provides a skip_tts attribute on text and response frames for this purpose.

When skip_tts is True on a frame, the TTS service passes it through without generating audio, but the text still reaches downstream processors like the assistant context aggregator.

Configuring All LLM Output Copy Section

Use LLMConfigureOutputFrame to tell the LLM service to mark all subsequent output frames (LLMTextFrame, LLMFullResponseStartFrame, LLMFullResponseEndFrame) with skip_tts:

from pipecat.frames.frames import LLMConfigureOutputFrame

# Tell the LLM to skip TTS for all output
await worker.queue_frame(LLMConfigureOutputFrame(skip_tts=True))

# ... LLM responses will not be spoken ...

# Re-enable TTS
await worker.queue_frame(LLMConfigureOutputFrame(skip_tts=False))

This is useful when you want to toggle TTS on or off for an entire stretch of conversation, such as switching between voice and text input modes.

Setting skip_tts on Individual Frames Copy Section

For more granular control, set skip_tts=True directly on individual text frames. This is useful when building custom frame processors that selectively silence certain parts of the LLM output:

from pipecat.frames.frames import LLMTextFrame

# In a custom frame processor
frame = LLMTextFrame(text)
frame.skip_tts = True
await self.push_frame(frame)

Common Use Cases Copy Section

Encoding structured output from the LLM. You can instruct the LLM to include markers or metadata in its response that should be processed by pipeline logic but not spoken. For example, Pipecat's turn completion detection uses this approach — the LLM outputs completion markers (✓, ○, ◐) that are pushed with skip_tts=True so they update the context but aren't spoken.

Switching between voice and text input. When a client sends text input instead of speech, you may want the bot to respond with text only. The client SDKs support this via sendText() with audio_response: false, which uses LLMConfigureOutputFrame internally to temporarily disable TTS for that response.

Testing without audio. When building test pipelines, you can use LLMConfigureOutputFrame(skip_tts=True) to bypass audio generation entirely while still exercising the rest of the pipeline.

Advanced TTS Features Copy Chapter

Direct Speech Commands Copy Section

Use TTSSpeakFrame for immediate speech synthesis:

from pipecat.frames.frames import TTSSpeakFrame

# Make bot speak directly (added to context by default)
await tts.queue_frame(TTSSpeakFrame("Hello, how can I help you?"))

# Explicitly append spoken text to conversation context
await tts.queue_frame(
    TTSSpeakFrame("Welcome! Let's begin.", append_to_context=True)
)

# Speak without adding to context
await tts.queue_frame(
    TTSSpeakFrame("Processing...", append_to_context=False)
)

The append_to_context parameter controls whether the spoken text is added to the conversation history. When append_to_context=True, the text is automatically committed to the context after being spoken, making it useful for bot greetings and injected speech that should be part of the conversation flow.

Dynamic Settings Updates Copy Section

Update TTS settings during conversation using typed settings objects:

from pipecat.frames.frames import TTSUpdateSettingsFrame
from pipecat.services.cartesia.tts import CartesiaTTSSettings

# Change voice speed during conversation
await worker.queue_frames([
    TTSUpdateSettingsFrame(delta=CartesiaTTSSettings(speed="fast")),
    TTSSpeakFrame("I'm speaking faster now!")
])

Key Takeaways Copy Chapter

• Pipeline placement matters - TTS must come after LLM, before transport output
• Service types differ - WebSocket services provide lower latency than HTTP
• Text processing affects quality - use aggregation and filters for better results
• Word timestamps enable precision - better interruption handling and context accuracy
• Configuration impacts performance - balance quality, latency, and bandwidth needs
• Services are modular - easily swap providers without changing pipeline code

What's Next Copy Chapter

You've now learned how to build a complete voice AI pipeline! Let's explore some additional topics to enhance your implementation.

<Card
title="Pipeline Termination"
icon="arrow-right"
href="/pipecat/learn/pipeline-termination"

Learn how to terminate your voice AI pipeline at the end of a conversation

Pipeline Termination

Source: https://docs.pipecat.ai/pipecat/learn/pipeline-termination.md

Learn how to properly terminate Pipecat pipelines for clean shutdown and resource management

Pipeline termination ensures your voice AI applications shut down cleanly without resource leaks or hanging processes. Understanding the different termination methods helps you handle various scenarios from natural conversation endings to unexpected disconnections.

Pipeline Integration Copy Chapter

Pipeline termination works through the same frame-based system as other pipeline operations:

pipeline = Pipeline([
    transport.input(),
    stt,
    context_aggregator.user(),
    llm,
    tts,
    transport.output(),
    context_aggregator.assistant(),
])
# EndFrame or CancelFrame flows through entire pipeline for shutdown

Termination frames:

• EndFrame: A queued ControlFrame that triggers graceful shutdown after processing pending frames
• CancelFrame: A SystemFrame that triggers immediate shutdown, discarding pending frames

Both frames flow downstream through the pipeline, allowing each processor to clean up resources appropriately.

Termination frames at a glance Copy Section

Termination Methods Copy Chapter

Pipecat provides two primary approaches for pipeline termination, each designed for different scenarios:

1. Graceful Termination Copy Section

Graceful termination allows the bot to complete its current processing before shutting down. This is ideal when you want the bot to properly end a conversation. For example, after completing a specific task or reaching a natural conclusion.

When to use:

• Natural conversation endings
• Task completion scenarios
• When the bot should say goodbye

Implementation options:

Push an EndFrame from outside your pipeline:

# From outside the pipeline
from pipecat.frames.frames import EndFrame

await worker.queue_frame(EndFrame())

Push an EndWorkerFrame downstream from inside your pipeline. Here end_conversation is a direct function — an async function whose first parameter is params: FunctionCallParams, with a Google-style docstring that becomes the tool description. Register it by passing the function itself in your tools list:

from pipecat.frames.frames import EndWorkerFrame, TTSSpeakFrame
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.llm_service import FunctionCallParams


async def end_conversation(params: FunctionCallParams):
    """End the conversation and shut down the bot.

    Call this when the user says goodbye or the task is complete.
    """
    await params.llm.push_frame(TTSSpeakFrame("Have a nice day!"))

    # Resolve the function call so the LLM call doesn't hang
    await params.result_callback({"status": "ended"})

    # Signal that the worker should end after processing this frame
    await params.llm.push_frame(EndWorkerFrame(), FrameDirection.DOWNSTREAM)


# Pass the function directly in the tools list; it's registered automatically
context = LLMContext(tools=[end_conversation])

How graceful termination works:

1. EndFrame is queued and processes after any pending frames (like goodbye messages)
2. All processors shutdown when they receive the EndFrame
3. Once the EndFrame reaches the end of the pipeline, shutdown is complete
4. Resources are cleaned up and the process terminates

2. Immediate Termination Copy Section

Immediate termination cancels the pipeline without waiting for pending frames to complete. This is appropriate when the user is no longer active in the conversation.

When to use:

• User disconnections (browser closed, call ended)
• Error conditions requiring immediate shutdown
• When completing the conversation is no longer necessary

Implementation:

Use event handlers to detect disconnections and trigger cancellation:

@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
    logger.info("Client disconnected - terminating pipeline")
    await worker.cancel()

How immediate termination works:

1. An event triggers the cancellation (like client disconnection)
2. worker.cancel() pushes a CancelFrame downstream from the PipelineWorker
3. CancelFrames are SystemFrames, so they use the high-priority input queue and are processed before queued non-system frames
4. Processors handle the CancelFrame and shut down without waiting for pending non-system work to drain
5. Any pending frames are discarded during shutdown

Automatic Termination Copy Chapter

Pipeline Idle Detection Copy Section

Pipecat includes automatic idle detection to prevent hanging pipelines. This feature monitors activity and can automatically cancel tasks when no meaningful bot interactions occur for an extended period.

How it works:

• Monitors pipeline activity for meaningful bot interactions
• Automatically triggers termination after configured idle timeout
• Serves as a safety net for anomalous behavior or forgotten sessions

Configuration:

worker = PipelineWorker(
    pipeline,
    # Configure idle detection timeout
    cancel_on_idle_timeout=True, # Default is True
    idle_timeout_secs=600,  # Default is 300 seconds
    idle_timeout_frames=(BotSpeakingFrame,), # Default is (BotSpeakingFrame, UserSpeakingFrame)
)

You can further configure the idle detection behavior. To learn more, refer to the Pipeline Idle Detection documentation:

<Card
title="Pipeline Idle Detection"
icon="clock"
href="/api-reference/server/pipeline/pipeline-idle-detection"

Learn how to configure and customize idle detection for your use case

Maximum Call Duration Copy Section

Idle detection ends a call that goes quiet, but it does not cap the total length of an active call. To enforce a maximum call duration, run an asyncio timer that speaks a goodbye and then queues an EndFrame so the bot can sign off gracefully before shutdown:

import asyncio
from pipecat.frames.frames import EndFrame, TTSSpeakFrame

async def end_after(worker, timeout_secs: float):
    await asyncio.sleep(timeout_secs)
    await worker.queue_frame(TTSSpeakFrame("We've reached our time limit. Goodbye!"))
    await worker.queue_frame(EndFrame())  # Graceful: plays the goodbye, then shuts down

@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
    asyncio.create_task(end_after(worker, timeout_secs=300))  # 5-minute cap

Implementation Patterns Copy Chapter

Event-Driven Termination Copy Section

Connect termination to transport events for automatic cleanup:

@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
    logger.info("Client connected - starting conversation")
    await worker.queue_frames([LLMRunFrame()])

@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
    logger.info("Client disconnected - immediate termination")
    await worker.cancel()

# Run the pipeline
runner = WorkerRunner(handle_sigint=False)
await runner.run(worker)

Conditional Termination Copy Section

Use function calling or other logic to determine when conversations should end:

async def check_conversation_complete(params: FunctionCallParams):
    # Your logic to determine if conversation should end
    conversation_complete = await evaluate_completion_criteria()

    if conversation_complete:
        await params.llm.push_frame(TTSSpeakFrame("Thank you for using our service!"))
        await params.llm.push_frame(EndWorkerFrame(), FrameDirection.DOWNSTREAM)

    await params.result_callback({"status": "complete" if conversation_complete else "continuing"})

Error Handling Copy Section

Ensure pipelines can terminate properly even when exceptions occur:

try:
    runner = WorkerRunner(handle_sigint=False)
    await runner.run(worker)
except Exception as e:
    logger.error(f"Pipeline error: {e}")
    # Ensure cleanup happens even on errors
    await worker.cancel()

Running Cleanup Code on Shutdown Copy Section

To run cleanup or persist data when a call ends, use the on_pipeline_finished event handler. It fires after the pipeline reaches any terminal state, so it runs for both graceful (EndFrame) and cancelled (CancelFrame) shutdowns. This makes it the single write point for end-of-call work like saving a transcript or recording:

@worker.event_handler("on_pipeline_finished")
async def on_pipeline_finished(worker, frame):
    # Runs for both graceful and cancelled shutdown
    await save_transcript_to_db()

on_client_disconnected, by contrast, fires only when the client disconnects. Use it to tag the reason for the shutdown (for example, "user hung up"), and do the actual persistence in on_pipeline_finished so you write data exactly once regardless of how the call ended. See the PipelineWorker events reference for the full event signature.

Troubleshooting Copy Chapter

If your pipeline isn't shutting down properly, check these common issues:

Custom Processors Not Propagating Frames Copy Section

Problem: Custom processors that don't call push_frame() can block termination frames from reaching the end of the pipeline.

Solution: Ensure your custom processors propagate all frames downstream, including EndFrame and CancelFrame:

async def process_frame(self, frame: Frame, direction: FrameDirection):
    await super().process_frame(frame, direction)

    # Your custom processing logic here

    # Always push frames downstream (including termination frames)
    await self.push_frame(frame, direction)

Incorrect Termination Frame Direction Copy Section

Problem: Pushing EndFrame or CancelFrame from the middle of the pipeline may not reach the pipeline source properly.

Solution: Use the appropriate frame type and direction:

await self.push_frame(EndWorkerFrame(), FrameDirection.DOWNSTREAM)
await self.push_frame(CancelWorkerFrame(), FrameDirection.DOWNSTREAM)

# The pipeline source will then convert these to proper termination frames
# and push them downstream through the entire pipeline

"dangling tasks detected" Warning Copy Section

Problem: On shutdown you see a log warning like PipelineWorker#0 dangling tasks detected: [...].

Solution: This means asyncio tasks created during the session were never cancelled or awaited before the pipeline shut down. The usual causes are the two above: a custom processor that doesn't propagate termination frames, or a background task started inside a processor (for example, a timer or long-running coroutine) that isn't cleaned up. Create background tasks through the pipeline task manager so they are tracked and cancelled on shutdown, and make sure your processors push EndFrame/CancelFrame downstream.

Key Takeaways Copy Chapter

• Frame-based termination - shutdown uses the same frame system as processing
• Choose the right method - graceful for natural endings, immediate for disconnections
• Event handlers enable automatic termination - respond to user disconnections cleanly
• Idle detection provides safety net - prevents hanging processes and resource waste
• SystemFrames have priority - CancelFrames are processed before queued non-system frames for fast shutdown
• Resource cleanup is automatic - proper termination ensures clean resource disposal

What's Next Copy Chapter

You now understand how to build, run, and properly terminate voice AI pipelines! With the single-agent basics covered, let's see how Pipecat coordinates multiple agents, starting with giving an agent its own LLM and tools.

<Card
title="Multiple LLM Agents"
icon="arrow-right"
href="/pipecat/learn/multiple-llm-agents"

Give an agent its own LLM and register tools with the @tool decorator

Multiple LLM Agents

Source: https://docs.pipecat.ai/pipecat/learn/multiple-llm-agents.md

Run separate agents that each own their own LLM, tools, and conversation context.

So far you've built a single agent: one LLM, one context, one set of tools. But many problems are better served by several agents working together, each owning its own LLM. A greeter hands off to a support agent. A researcher runs in the background while the main agent keeps talking. A screen-driving agent acts on the UI while a voice agent converses.

Giving each agent its own LLM keeps every context small and focused. Instead of one model juggling every instruction and tool, each agent reasons over just its own job. That's cheaper, faster, and less prone to the model getting lost.

LLMWorker overview Copy Chapter

Multi-agent systems often run several LLM-backed agents -- a greeter, a support agent, a researcher -- each with its own instructions, tools, and (optionally) its own conversation context. LLMWorker is the building block for each one. It extends PipelineWorker with everything you need to run an LLM-powered agent:

• A pipeline with your LLM service, automatically built
• Tool registration via the @tool decorator
• Activation handling that injects messages and runs the LLM

To create an LLM agent, subclass LLMWorker so you can host @tool methods, then instantiate it with its own LLM service. Pass bridged=() so the agent receives frames from the bus:

import os

from pipecat.services.openai.llm import OpenAILLMService
from pipecat.workers.llm import LLMWorker


class MyAgent(LLMWorker):
    """An LLM agent. ``@tool`` methods go here."""


def build_agent() -> MyAgent:
    llm = OpenAILLMService(
        api_key=os.environ["OPENAI_API_KEY"],
        settings=OpenAILLMService.Settings(
            system_instruction="You are a helpful assistant.",
        ),
    )
    return MyAgent("assistant", llm=llm, bridged=())

You never pass a bus= argument to the constructor -- the worker gets its bus when you register it with runner.add_workers(...). The default pipeline is Pipeline([llm]), with tools from build_tools() automatically registered. When bridged=() is set, the framework wraps this pipeline with edge processors that connect it to the bus.

The @tool decorator Copy Chapter

The @tool decorator marks a method as an LLM-callable tool. The framework automatically collects all @tool-decorated methods and registers them with the LLM service.

from pipecat.services.llm_service import FunctionCallParams
from pipecat.workers.llm import tool


class MyAgent(LLMWorker):
    @tool
    async def get_weather(self, params: FunctionCallParams, city: str):
        """Get the current weather for a city.

        Args:
            city (str): The city name (e.g. 'San Francisco').
        """
        weather = await fetch_weather(city)
        await params.result_callback(weather)

The tool's name comes from the method name. The docstring becomes the tool description. Parameter types and descriptions are extracted from the type annotations and the Args section in the docstring.

Tool options Copy Section

The @tool decorator accepts options:

@tool(cancel_on_interruption=False, timeout=60)
async def long_running_tool(self, params: FunctionCallParams, query: str):
    """A tool that takes a while.

    Args:
        query (str): The search query.
    """
    result = await expensive_search(query)
    await params.result_callback(result)

Tool parameters Copy Section

Every tool method receives self and params: FunctionCallParams as the first two arguments. Additional arguments are the tool's parameters that the LLM fills in.

The params object gives you access to:

• params.result_callback(result) -- return the result to the LLM
• params.llm -- the LLM service instance, useful for queuing frames

Returning results Copy Section

Always call params.result_callback() to return the tool result to the LLM:

@tool
async def lookup(self, params: FunctionCallParams, item: str):
    """Look up an item.

    Args:
        item (str): The item to look up.
    """
    data = await database.get(item)
    await params.result_callback({"found": True, "data": data})

Activation with messages Copy Chapter

When an LLMWorker is activated, you can inject messages into its context. Pass an LLMWorkerActivationArgs via the args parameter:

from pipecat.workers.llm import LLMWorkerActivationArgs

await self.activate_worker(
    "support",
    args=LLMWorkerActivationArgs(
        messages=[{"role": "developer", "content": "The user asked about pricing."}],
        run_llm=True,  # Run the LLM immediately after injection
    ),
)

The default on_activated() implementation:

1. Sets the tools from build_tools()
2. Injects the provided messages into the LLM context
3. Runs the LLM if run_llm is True (the default when messages is set)

Managing context with LLMContextWorker Copy Chapter

A plain LLMWorker runs an LLM but doesn't manage conversation context on its own -- it relies on context coming from elsewhere (for example, the main agent's aggregators bridged in). When an agent needs to keep its own history, use LLMContextWorker. It extends LLMWorker with a built-in LLMContext and the user/assistant aggregator pair, building the pipeline as [user_aggregator, llm, assistant_aggregator] for you.

from pipecat.workers.llm import LLMContextWorker


class AssistantAgent(LLMContextWorker):
    """An LLM agent that keeps its own conversation context."""


agent = AssistantAgent("assistant", llm=llm)  # gets its own context

Each LLMContextWorker gets its own context by default, so agents don't see each other's history. To give several agents a shared conversation, pass the same context= to each:

from pipecat.processors.aggregators.llm_context import LLMContext

shared = LLMContext()
agent_a = AssistantAgent("agent_a", llm=llm_a, context=shared)
agent_b = AssistantAgent("agent_b", llm=llm_b, context=shared)

Access the managed aggregators via self.user_aggregator and self.assistant_aggregator.

Custom pipelines Copy Chapter

If you need more control, you can pass a custom pipeline= to the LLMWorker constructor. For example, to add TTS to the agent's own pipeline:

from pipecat.pipeline.pipeline import Pipeline
from pipecat.services.cartesia.tts import CartesiaTTSService


class AgentWithTTS(LLMWorker):
    def __init__(self, name: str, *, llm, voice_id: str):
        tts = CartesiaTTSService(
            api_key=os.environ["CARTESIA_API_KEY"],
            settings=CartesiaTTSService.Settings(voice=voice_id),
        )
        super().__init__(name, llm=llm, pipeline=Pipeline([llm, tts]), bridged=())

This is how you give each agent its own voice -- each agent adds its own TTS after the LLM, so a handoff sounds like a real transfer between distinct speakers.

What's next Copy Chapter

Now that your agents can run LLMs and call tools, here's a powerful one: an agent that sees and drives the user's screen.

Controlling the UI

Source: https://docs.pipecat.ai/pipecat/learn/ui-worker.md

Bridge a voice agent and a client GUI with a UIWorker over a two-way RTVI interface.

What is a UIWorker? Copy Chapter

When you put a voice agent in front of an app, talking isn't enough — the agent needs to see what the user sees and act on the screen: read the page, point at things, fill in fields, click buttons. A UIWorker is the server-side agent that makes this possible. It voice-enables a client UI by connecting an LLM to whatever the user is looking at.

The connection is two-way, over the RTVI UI channel:

• Client → server. The client streams the screen to the worker as accessibility snapshots, and forwards the user's UI interactions as events.
• Server → client. The worker drives the page back — scrolling, highlighting, selecting text, filling inputs, clicking, or running app-defined commands — and can surface long-running work as progress cards.

A UIWorker is the screen half of a voice/UI split: a voice agent owns the conversation, and the UIWorker owns the screen. Each is a separate LLM with its own focused context. The worker auto-injects the latest screen state into its context before every turn, so the conversational voice LLM never has to carry a giant accessibility tree:

• The voice agent converses and decides what's worth saying.
• The UIWorker reasons over the current page and acts on it.

The result is two small, fast contexts instead of one bloated one — cheaper, and less prone to the model getting lost.

The two-way interface Copy Chapter

A UIWorker gives an LLM a handful of capabilities, split across the two directions of the interface. Everything below works out of the box on any UIWorker subclass.

What the worker sees (client → server) Copy Section

The screen, as a snapshot. The client sends an accessibility snapshot of the page whenever it changes. The worker renders the latest one as a `` block and — with auto_inject_ui_state on (the default) — injects it into the LLM context before every turn, so the model always reasons over what's currently on screen. Each element carries a stable ref the worker uses to act on it:

- heading "Shopping list" [level=1] [ref=e3]
- list:
  - checkbox "milk" [checked] [ref=e5]
  - checkbox "eggs" [ref=e6]

When the user has text selected, the snapshot includes a <selection> block so the LLM can resolve deictic references like "this paragraph" or "what I selected".

User interactions, as events. The client dispatches app-defined events (a button click, a custom gesture) with sendUIEvent(name, payload). Route them to handlers with @ui_event(name); each runs in its own task:

from pipecat.workers.ui import UIWorker, ui_event

class MyUIWorker(UIWorker):
    @ui_event("note_click")
    async def on_note_click(self, message):
        ref = (message.payload or {}).get("ref")
        await self.scroll_to(ref)
        await self.select_text(ref)

What the worker does (server → client) Copy Section

Drives the page. The worker acts on the screen by sending UI commands. The built-in helpers cover the common actions, and send_command(name, payload) sends any app-defined command:

The standard client handlers ship in @pipecat-ai/client-react; apps can override them or define their own command names.

Answers back. A UIWorker answers via a built-in single-flight respond job: a requester dispatches job("ui", name="respond", payload={"query": ...}), the worker runs one screen-grounded LLM turn, and a @tool ends it by calling respond_to_job(). That call chooses how the answer reaches the user:

• respond_to_job(text, tts_speak=True) — speak text verbatim through the requester's TTS.
• respond_to_job(text) — return {"answer": text} for the requester's voice LLM to phrase.
• respond_to_job() — complete the turn silently (the worker acted, but said nothing).

Surfaces long work. When a turn kicks off background work, ui_job_group / start_ui_job_group fan it out to peer workers and surface it to the client as a cancellable progress card, streaming each worker's updates as they arrive:

await self.start_ui_job_group(
    "wikipedia", "news", "scholar",
    payload={"query": research_query},
    label=f"Research: {research_query}",
)

Hello world Copy Chapter

The smallest UIWorker ties the interface together: a delegate that answers questions about the page. The voice agent forwards screen-relevant utterances to it; the worker reads the screen (client → server) and speaks the reply (server → client).

The worker needs only an LLM and one @tool that ends the turn with respond_to_job():

from pipecat.workers.ui import UIWorker
from pipecat.workers.llm import tool

class HelloWorker(UIWorker):
    @tool
    async def answer(self, params, text: str):
        """Speak `text` back to the user."""
        await self.respond_to_job(text, tts_speak=True)
        await params.result_callback(None)

The voice agent exposes a tool that dispatches a respond job to the worker and speaks back whatever it returns:

async def answer_about_screen(params, query: str):
    """Ask the screen-aware UI layer to answer about the current page."""
    async with params.pipeline_worker.job(
        "hello", name="respond", payload={"query": query}, timeout=30
    ) as t:
        pass
    await params.result_callback(t.response)

Register both with the runner — the UIWorker comes online to receive snapshots and jobs as soon as its pipeline starts:

await runner.add_workers(HelloWorker(), worker)

Here's the full round trip for one utterance:

Patterns Copy Chapter

How the worker gets triggered — and who speaks — falls into two patterns.

Delegation Copy Section

In the hello-world example, the voice agent is the gatekeeper: it decides which turns involve the screen and routes those to the UIWorker, then voices the result. This is the delegation pattern, and it's the common one. The voice LLM stays small and screen-unaware; the worker owns all screen reasoning.

Most apps don't need a custom tool per action. ReplyToolMixin provides a single bundled reply tool — a required spoken answer plus optional scroll_to, highlight, select_text, fills, and click — covering pointing, reading, and form apps:

from pipecat.workers.ui import ReplyToolMixin, UIWorker

class FormWorker(ReplyToolMixin, UIWorker):
    def __init__(self):
        super().__init__("ui", llm=OpenAILLMService(api_key="..."))

The LLM uses whichever fields fit the turn — select_text to point at "this paragraph", fills + click to complete a form — and unused fields stay null.

Delegation also scales to background work. A @tool can fan out to peer workers with start_ui_job_group, which surfaces a cancellable progress card on the client and returns immediately so the voice agent isn't blocked:

class ResearchWorker(UIWorker):
    @tool
    async def reply(self, params, answer: str, research_query: str | None = None):
        if research_query:
            await self.start_ui_job_group(
                "wikipedia", "news", "scholar",
                payload={"query": research_query},
                label=f"Research: {research_query}",
            )
        await self.respond_to_job(answer)
        await params.result_callback(None)

Parallel handling Copy Section

Sometimes the screen should react to every user turn, not only the ones the voice agent chooses to delegate. In the parallel handling pattern, both agents receive each turn and act in parallel: the voice agent converses while the UIWorker updates the screen, independently.

The key difference: there's no tool call routing work to the worker. Instead, the voice pipeline's user aggregator fires on_user_turn_stopped once per turn, and that handler dispatches the transcript to the worker as a respond job. Because it runs in its own task, the voice LLM (running from the same turn) and the worker act concurrently:

@user_aggregator.event_handler("on_user_turn_stopped")
async def on_user_turn_stopped(aggregator, strategy, message):
    transcript = (message.content or "").strip()
    if not transcript:
        return
    async with worker.job("ui", name="respond", payload={"query": transcript}, timeout=15):
        pass  # fire-and-forget; the worker acts on its own

The worker acts silently — its @tool completes the job with respond_to_job() and no answer, so nothing it does reaches TTS. The separate voice layer owns speech:

class ListWorker(UIWorker):
    @tool
    async def update_list(self, params, add=None, check=None, remove=None):
        for text in add or []:
            await self.send_command("add_item", {"text": text})
        for ref in check or []:
            await self.send_command("set_checked", {"ref": ref, "checked": True})
        for ref in remove or []:
            await self.send_command("remove_item", {"ref": ref})
        await self.respond_to_job()  # no answer — acts silently
        await params.result_callback(None)

Here the snapshot is the shared source of truth. The worker acts on it, and the voice agent reads it through a read-only tool — so the voice agent can answer "what's left on my list?" from what's actually on screen (including items the user checked off by hand), not from conversation memory:

async def check_list(params):
    """Look up what's currently on the list."""
    await params.result_callback(list_worker.list_summary())  # reads the live snapshot

Choosing a pattern Copy Section

Both keep the voice LLM's context small. Delegation gives the voice agent control over when the screen is involved; parallel handling makes the screen a first-class output of every turn.

What's next Copy Chapter

You've built agents that converse, call tools, and drive the screen. Next, learn how to transfer control between them.

Full reference for the `UIWorker` class, UI commands, job groups, and
`ReplyToolMixin`.

Agent Handoff

Source: https://docs.pipecat.ai/pipecat/learn/agent-handoff.md

Transfer control between agents with activation, deactivation, and handoff.

The activation model Copy Chapter

In a multi-agent system, only one agent is active at a time (per bridge). The active agent receives frames from the bus. Inactive agents exist but don't process frames.

Every worker has an active property. LLMWorker defaults to active=False, so in a handoff setup the LLM agents start inactive and the main agent activates the first one explicitly.

You control which agent is active with two methods:

Passing deactivate_self=True to activate_worker() is the most common form -- it's a single call that transfers control from the current agent to another.

Building a handoff system Copy Chapter

Let's walk through how the two-agent handoff works. You need three pieces.

1. A main agent with a bus bridge Copy Section

The main agent owns the transport (audio I/O) and places a BusBridgeProcessor in its pipeline instead of an LLM. The bridge routes frames to whichever LLM agent is active. The main agent is a PipelineWorker wrapping that pipeline:

from pipecat.bus import BusBridgeProcessor
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.worker import PipelineParams, PipelineWorker

MAIN_NAME = "main"

stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
tts = CartesiaTTSService(api_key=os.getenv("CARTESIA_API_KEY"))

context = LLMContext()
aggregators = LLMContextAggregatorPair(
    context,
    user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()),
)

bridge = BusBridgeProcessor(bus=runner.bus, worker_name=MAIN_NAME)

pipeline = Pipeline([
    transport.input(),
    stt,
    aggregators.user(),
    bridge,                  # Where the LLM would go
    tts,
    transport.output(),
    aggregators.assistant(),
])

main = PipelineWorker(pipeline, name=MAIN_NAME, params=PipelineParams())

2. LLM agents with bridged=() Copy Section

Each LLM agent is an LLMWorker created with bridged=() so it receives frames from the bus. Subclass LLMWorker to add tools, then instantiate it with its own LLM:

from pipecat.workers.llm import LLMWorker, tool

class AcmeAgent(LLMWorker):
    # tools defined below

def build_greeter() -> AcmeAgent:
    llm = OpenAILLMService(
        api_key=os.getenv("OPENAI_API_KEY"),
        settings=OpenAILLMService.Settings(
            system_instruction="You are a friendly greeter. Route product questions to support.",
        ),
    )
    return AcmeAgent("greeter", llm=llm, bridged=())

3. Handoff via tools Copy Section

The LLM decides when to transfer by calling a tool. The tool calls activate_worker() with deactivate_self=True:

from pipecat.workers.llm import LLMWorker, LLMWorkerActivationArgs, tool

class AcmeAgent(LLMWorker):
    @tool(cancel_on_interruption=False)
    async def transfer_to_agent(self, params: FunctionCallParams, agent: str, reason: str):
        """Transfer the user to another agent.

        Args:
            agent (str): The agent to transfer to (e.g. 'support').
            reason (str): Why the user is being transferred.
        """
        await self.activate_worker(
            agent,
            args=LLMWorkerActivationArgs(
                messages=[{"role": "developer", "content": reason}],
            ),
            deactivate_self=True,
            result_callback=params.result_callback,
        )

For LLMWorker, activate_worker() also accepts a messages parameter. These messages are injected and spoken by the current agent before the transfer happens -- useful for announcing the handoff:

await self.activate_worker(
    agent,
    messages=[{"role": "developer", "content": f"Tell the user about the transfer ({reason})."}],
    args=LLMWorkerActivationArgs(
        messages=[{"role": "developer", "content": reason}],
    ),
    deactivate_self=True,
    result_callback=params.result_callback,
)

When the greeter calls activate_worker("support", deactivate_self=True, ...):

1. The greeter is deactivated -- it stops receiving frames from the bus
2. The support agent is activated with the provided arguments
3. The support agent's on_activated() fires, injecting the reason message into its LLM context
4. The support agent starts responding to the user

The transition is seamless -- the user experiences it as a natural conversation flow.

Activation arguments Copy Chapter

When activating an LLMWorker, pass LLMWorkerActivationArgs via args= to give the target agent context about why it was activated:

from pipecat.workers.llm import LLMWorkerActivationArgs

await self.activate_worker(
    "support",
    args=LLMWorkerActivationArgs(
        messages=[{"role": "developer", "content": "The user asked about Rocket Boots."}],
    ),
)

The target agent receives these messages in its LLM context and immediately runs the LLM to respond.

Activating the first agent Copy Chapter

Before you can activate an agent, it needs to be ready (its pipeline must be started). The simplest place to kick off the conversation is the transport's on_client_connected handler, where you activate the first agent:

@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
    await main.activate_worker(
        "greeter",
        args=LLMWorkerActivationArgs(
            messages=[{"role": "developer", "content": "Welcome the user."}],
        ),
    )

If you need to react to a specific agent registering (for example in a distributed setup), use the @worker_ready decorator instead. See Agent Registry and Discovery.

Putting it all together Copy Chapter

Here's the full flow:

What's next Copy Chapter

Handoff transfers a conversation between agents. But sometimes you need agents to do work in parallel. Next, let's look at job coordination.

<Card
title="Job Coordination"
icon="arrow-right"
href="/pipecat/learn/job-coordination"

Dispatch work to multiple agents in parallel

Job Coordination

Source: https://docs.pipecat.ai/pipecat/learn/job-coordination.md

Dispatch work from one Pipecat agent to others and collect results with job coordination in the multi-agent framework.

What are jobs? Copy Chapter

Jobs let one agent dispatch work to other agents. While handoff transfers the conversation, jobs are for background work -- an agent asks other agents to do something and collects their results.

Use cases for jobs:

• Researching a topic from multiple perspectives in parallel
• Running a code analysis while talking to the user
• Fetching data from multiple sources simultaneously

Single job Copy Chapter

The simplest form is self.job(), which sends a job to one agent and waits for the response:

async with self.job("worker", payload={"question": "What is Pipecat?"}, timeout=30) as j:
    pass

print(j.response)

self.job() is a context manager that handles the full lifecycle: it waits for the agent to be ready, sends the request, and collects the response. Under the hood, this is equivalent to calling request_job() (fire-and-forget) and then handling on_job_response() and on_job_completed() yourself.

Job group Copy Chapter

When you need to dispatch work to multiple agents in parallel, use self.job_group():

async with self.job_group("worker1", "worker2", "worker3", payload={"topic": "AI safety"}, timeout=30) as jg:
    pass

for worker_name, response in jg.responses.items():
    print(f"{worker_name}: {response}")

self.job_group() works the same way but for multiple agents: it waits for all of them to be ready, sends a request to each, and collects all responses. Under the hood, this is equivalent to calling request_job_group() and then handling on_job_response() for each agent and a final on_job_completed() yourself.

The same payload is sent to every agent. If you need different arguments per agent, you can structure the payload so each one reads its own key:

async with self.job_group("researcher", "fact_checker", payload={
    "researcher": {"topic": "AI safety", "depth": "detailed"},
    "fact_checker": {"claims": ["AI can self-improve", "AGI is near"]},
}, timeout=30) as jg:
    pass

Handling job requests Copy Chapter

Agents handle incoming jobs in two ways.

The @job decorator Copy Section

The @job decorator marks a method as a job handler. The framework automatically dispatches matching requests to it. The decorator requires a name argument. The handler responds with message.job_id:

from pipecat.pipeline.job_decorator import job

class MyWorker(BaseWorker):
    @job(name="process")
    async def on_process(self, message: BusJobRequestMessage):
        result = await self._do_work(message.payload)
        await self.send_job_response(message.job_id, result)

You can use different job names to route work to different handlers:

class MyWorker(BaseWorker):
    @job(name="research")
    async def on_research(self, message: BusJobRequestMessage):
        await self.send_job_response(message.job_id, {"answer": "..."})

    @job(name="summarize")
    async def on_summarize(self, message: BusJobRequestMessage):
        await self.send_job_response(message.job_id, {"summary": "..."})

The requester specifies the job name when dispatching:

async with self.job("worker", name="research", payload={"topic": "AI"}) as j:
    pass

Each request runs in its own asyncio task, so multiple requests to the same handler are executed concurrently without blocking the bus message loop.

Overriding on_job_request Copy Section

Alternatively, you can override on_job_request() directly without the @job decorator:

class MyWorker(BaseWorker):
    async def on_job_request(self, message: BusJobRequestMessage) -> None:
        await super().on_job_request(message)
        result = await self._do_work(message.payload)
        await self.send_job_response(message.job_id, result)

This is useful when you need custom routing logic or want to integrate with an existing pipeline, as shown in the example below.

Building a job system Copy Chapter

Let's build a debate system where a moderator dispatches a topic to three agents, each arguing from a different perspective.

Worker agents Copy Section

Each debate agent runs its own LLM pipeline. The LLM response arrives asynchronously through an event handler, so the agent tracks the current job ID until it has something to respond with:

from pipecat.workers.llm import LLMContextWorker
from pipecat.bus.messages import BusJobRequestMessage
from pipecat.frames.frames import LLMMessagesAppendFrame

class DebateWorker(LLMContextWorker):
    def __init__(self, role: str):
        llm = OpenAILLMService(
            api_key=os.getenv("OPENAI_API_KEY"),
            settings=OpenAILLMService.Settings(system_instruction=f"You argue as a {role}."),
        )
        super().__init__(role, llm=llm)
        self._role = role
        self._current_job_id: str | None = None

        @self.assistant_aggregator.event_handler("on_assistant_turn_stopped")
        async def on_assistant_turn_stopped(aggregator, message):
            if self._current_job_id:
                job_id = self._current_job_id
                self._current_job_id = None
                await self.send_job_response(job_id, {"role": self._role, "text": message.content})

    async def on_job_request(self, message: BusJobRequestMessage) -> None:
        await super().on_job_request(message)
        self._current_job_id = message.job_id
        await self.queue_frame(
            LLMMessagesAppendFrame(
                messages=[{"role": "developer", "content": f"Topic: {message.payload['topic']}"}],
                run_llm=True,
            )
        )

The agent:

1. Receives a job request and stores the job_id
2. Injects the topic into its LLM context and runs the LLM
3. When the LLM finishes its turn, the event handler sends the response with the stored job_id

Coordinator agent Copy Section

The moderator triggers jobs via a tool and synthesizes the results:

from pipecat.workers.llm import LLMWorker, tool

class ModeratorAgent(LLMWorker):
    @tool(cancel_on_interruption=False)
    async def debate(self, params: FunctionCallParams, topic: str):
        """Analyze a topic from multiple perspectives.

        Args:
            topic (str): The topic to debate.
        """
        async with self.job_group("advocate", "critic", "analyst", payload={"topic": topic}, timeout=30) as jg:
            pass

        result = "\n\n".join(
            f"{r['role'].upper()}: {r['text']}" for r in jg.responses.values()
        )
        await params.result_callback(result)

Create the debate agents and the moderator, then register them all with runner.add_workers(ModeratorAgent("moderator", llm=...), DebateWorker("advocate"), DebateWorker("critic"), DebateWorker("analyst")).

When the user says "debate whether AI should be regulated," the moderator:

1. The LLM calls the debate tool
2. job_group() sends the topic to all three agents in parallel
3. Each agent runs its own LLM and responds with its perspective
4. The moderator collects all responses and returns them to the LLM
5. The LLM synthesizes a balanced summary and speaks it to the user

Job lifecycle Copy Chapter

The framework also supports fire-and-forget jobs, progress updates, and streaming.

Job cancellation Copy Chapter

Jobs can be cancelled in several ways. The framework handles cleanup automatically and notifies agents so they can stop in-progress work.

Automatic cancellation with context managers Copy Section

When using job() or job_group(), cancellation happens automatically if the context block raises an exception. This includes tool interruptions (CancelledError) when using cancel_on_interruption=True:

# If an exception or CancelledError is raised inside the block,
# all agents are cancelled automatically
async with self.job_group("w1", "w2", payload=data) as jg:
    # ... if this raises, the agents get cancelled
    pass

Worker errors Copy Section

By default (cancel_on_error=True), if any agent responds with an error status, the remaining agents in the group are cancelled and JobGroupError is raised:

try:
    async with self.job_group("w1", "w2", payload=data) as jg:
        pass
except JobGroupError as e:
    # An agent errored -- remaining agents were cancelled
    pass

Manual cancellation Copy Section

For fire-and-forget jobs (using request_job()), you manage cancellation yourself by tracking job IDs:

job_ids = []
try:
    job_ids.append(await self.request_job("w1", payload={"job": 1}))
    job_ids.append(await self.request_job("w2", payload={"job": 2}))
    # ...
except asyncio.CancelledError:
    for jid in job_ids:
        await self.cancel_job_group(jid, reason="tool cancelled")

Handling cancellation on the agent side Copy Section

When a job is cancelled, the agent's on_job_cancelled hook fires. The framework automatically sends a CANCELLED response back to the requester, so you only need to override this hook if you have resources to clean up:

class MyWorker(BaseWorker):
    async def on_job_cancelled(self, message: BusJobCancelMessage) -> None:
        # Optional: clean up resources, stop in-progress work
        logger.info(f"Job {message.job_id} cancelled: {message.reason}")

Agent shutdown Copy Section

When an agent stops with jobs still in flight, it automatically sends a CANCELLED response for each active job so requesters aren't left waiting on a timeout.

What's next Copy Chapter

You can now build agents, hand off between them, and dispatch jobs. So far everything has run in a single process. Next, let's scale across processes and machines.

<Card
title="Distributed Agents"
icon="arrow-right"
href="/pipecat/learn/distributed-agents"

Run agents across processes and machines on a shared bus

Distributed Agents

Source: https://docs.pipecat.ai/pipecat/learn/distributed-agents.md

Run Pipecat agents across processes and machines connected to the same bus for distributed multi-agent deployments.

Overview Copy Chapter

Distributed agents are agents connected to the same bus but running in different processes or on different machines. By default, all agents run in a single process using a local bus. For distributed setups, swap to a network bus -- all agents share the same channel and discover each other automatically. Your agent code stays the same.

Setting up a distributed bus Copy Chapter

Pipecat provides two distributed bus implementations: RedisBus and PgmqBus. Choose based on your infrastructure.

RedisBus Copy Section

Each process creates its own WorkerRunner with a RedisBus connected to the same Redis channel:

from redis.asyncio import Redis
from pipecat.bus.network.redis import RedisBus
from pipecat.workers.runner import WorkerRunner

redis = Redis.from_url("redis://localhost:6379")
bus = RedisBus(redis=redis, channel="pipecat:my-app")
runner = WorkerRunner(bus=bus, handle_sigint=True)

All runners sharing the same channel can exchange messages. Agents discover each other automatically through registry snapshots.

Install the Redis extra: uv add "pipecat-ai[redis]"

PgmqBus Copy Section

Alternatively, use PgmqBus backed by PostgreSQL Message Queue:

from pgmq.async_queue import PGMQueue
from pipecat.bus.network.pgmq import PgmqBus
from pipecat.workers.runner import WorkerRunner

pgmq = PGMQueue(
    host="localhost",
    port="5432",
    database="postgres",
    username="postgres",
    password="...",
    pool_size=4,
)
await pgmq.init()
bus = PgmqBus(pgmq=pgmq, channel="pipecat:my-app")
runner = WorkerRunner(bus=bus, handle_sigint=True)

Install the PGMQ extra: uv add "pipecat-ai[pgmq]"

Example: distributed handoff Copy Chapter

This example splits the two-agent handoff across separate processes. The main agent handles transport on one machine, and LLM agents run independently on other machines.

Process 1: Main transport agent Copy Section

The main agent owns the transport and bridges frames to the bus. It has no LLM -- it waits for the remote greeter agent to register, then activates it.

/* Detailed source-code truncated for AI context efficiency. */

python main.py --redis-url redis://localhost:6379

Process 2 & 3: LLM agents Copy Section

Each LLM agent runs as a standalone process. It connects to the same Redis channel and waits for activation. Subclass LLMWorker to host @tool methods and pass the LLM service in the constructor:

/* Detailed source-code truncated for AI context efficiency. */

# Run on Machine B
python llm.py greeter --redis-url redis://your-redis-host:6379

# Run on Machine C
python llm.py support --redis-url redis://your-redis-host:6379

Each LLM agent watches its sibling so that, when the user is transferred, the target is already known to be available. You can do the same with the @worker_ready decorator on a worker subclass to react automatically when a specific agent registers.

How discovery works Copy Chapter

Runners exchange registry information automatically over the shared bus. To get notified when an agent is ready, watch it with runner.registry.watch(...), call watch_workers() from inside a worker, or use the @worker_ready decorator -- they all work the same way locally and distributed.

Considerations Copy Chapter

• Latency: Network buses add overhead. For latency-sensitive voice applications, keep the main transport agent and its active LLM agent geographically close to each other and the bus server (Redis or PostgreSQL).
• Serialization: Both RedisBus and PgmqBus serialize messages to JSON. Custom frame types need to be registered with the serializer.
• Single channel: All agents on the same channel see all messages. Use different channels for different sessions or applications.

What's next Copy Chapter

Distributed agents share one bus. When you need to connect agents across separate buses or networks, bridge them with a proxy.

<Card
title="Proxy Agents"
icon="arrow-right"
href="/pipecat/learn/proxy-agents"

Bridge agents on different buses point-to-point over WebSocket

Proxy Agents

Source: https://docs.pipecat.ai/pipecat/learn/proxy-agents.md

Connect Pipecat agents running on different buses with proxy agents, extending multi-agent pipelines across processes.

Overview Copy Chapter

Proxy agents connect agents running on different buses. Unlike distributed agents (which all share the same bus), proxy agents bridge two isolated bus instances point-to-point. This is useful when:

• You want to run an LLM agent on a separate server without shared infrastructure
• You need fine-grained control over which messages cross the network
• You're connecting to a third-party service that hosts agents

Architecture Copy Chapter

Each side has its own WorkerRunner and bus. The proxy agents relay messages between the two buses over a WebSocket connection. Like any worker, a proxy gets its bus from the runner when you register it -- you never pass bus= to its constructor.

Client side: WebSocketProxyClient Copy Chapter

The client connects to a remote server and forwards specific message types:

from pipecat.bus import BusFrameMessage
from pipecat.workers.proxy.websocket import WebSocketProxyClient

proxy = WebSocketProxyClient(
    "proxy",
    url="ws://remote-server:8765/ws",
    local_worker_name="acme",         # Agent on this bus
    remote_worker_name="assistant",   # Agent on remote bus
    forward_messages=(BusFrameMessage,),
)
await runner.add_workers(proxy)

The proxy connects when activated. Activate it when the client connects, by calling activate_worker on the main agent:

@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
    await main.activate_worker("proxy")

Server side: WebSocketProxyServer Copy Chapter

The server accepts WebSocket connections and creates a proxy for each session:

from fastapi import FastAPI, WebSocket

from pipecat.bus import BusFrameMessage
from pipecat.workers.runner import WorkerRunner
from pipecat.workers.proxy.websocket import WebSocketProxyServer

app = FastAPI()


@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()

    runner = WorkerRunner(handle_sigint=False)

    proxy = WebSocketProxyServer(
        "gateway",
        websocket=websocket,
        worker_name="assistant",        # Agent on this bus
        remote_worker_name="acme",      # Agent on remote bus
        forward_messages=(BusFrameMessage,),
    )

    assistant = build_assistant()  # an LLMWorker named "assistant"

    await runner.add_workers(proxy, assistant)
    await runner.run()

Each WebSocket connection gets its own WorkerRunner, bus, and set of agents. This isolates sessions from each other.

Message filtering Copy Chapter

Proxy agents provide security through message filtering:

• Only messages targeted at the configured agent names cross the connection
• Broadcast messages (no target) are not forwarded
• Local-only messages (BusLocalMessage) never cross
• forward_messages controls which message types are allowed

This means internal bus traffic stays local. Only the specific message types you opt into are relayed.

Full example Copy Chapter

Client (main.py) Copy Section

async def run_bot(transport, runner_args):
    runner = WorkerRunner(handle_sigint=runner_args.handle_sigint)

    # ... build the main transport pipeline with a BusBridgeProcessor ...
    main = PipelineWorker(pipeline, name="acme", params=PipelineParams(...))

    proxy = WebSocketProxyClient(
        "proxy",
        url=runner_args.cli_args.remote_url,
        local_worker_name="acme",
        remote_worker_name="assistant",
        forward_messages=(BusFrameMessage,),
    )

    async def on_assistant_ready(_data):
        await main.activate_worker(
            "assistant",
            args=LLMWorkerActivationArgs(
                messages=[{"role": "developer", "content": "Welcome the user."}],
            ),
        )

    await runner.registry.watch("assistant", on_assistant_ready)

    @transport.event_handler("on_client_connected")
    async def on_client_connected(transport, client):
        await main.activate_worker("proxy")

    @transport.event_handler("on_client_disconnected")
    async def on_client_disconnected(transport, client):
        await runner.cancel()

    await runner.add_workers(proxy, main)
    await runner.run()

Server (assistant.py) Copy Section

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    runner = WorkerRunner(handle_sigint=False)

    proxy = WebSocketProxyServer(
        "gateway",
        websocket=websocket,
        worker_name="assistant",
        remote_worker_name="acme",
        forward_messages=(BusFrameMessage,),
    )

    @proxy.event_handler("on_client_connected")
    async def on_client_connected(proxy, client):
        logger.info("WebSocket client connected")

    @proxy.event_handler("on_client_disconnected")
    async def on_client_disconnected(proxy, client):
        await runner.cancel()

    assistant = AcmeAssistant()  # an LLMWorker named "assistant"
    await runner.add_workers(proxy, assistant)
    await runner.run()

Running Copy Section

# Terminal 1: Start the LLM server
python assistant.py --port 8765

# Terminal 2: Start the transport client
python main.py --remote-url ws://localhost:8765/ws

Install the WebSocket extra: uv add "pipecat-ai[websocket]"

Proxy agents vs distributed agents Copy Chapter

What's next Copy Chapter

You've built a multi-agent system and scaled it across processes, machines, and networks. Here's where to go from here.

What's Next

Source: https://docs.pipecat.ai/pipecat/learn/whats-next.md

Continue your Pipecat journey with advanced features, examples, and production deployment

You've Mastered Voice AI Pipelines! 🎉 Copy Chapter

Congratulations! You've learned how to build complete voice AI applications with Pipecat. You're now well on your way to understanding pipelines, processors, transports, and all the components needed to create sophisticated conversational AI.

Choose Your Path Copy Chapter

💬 Join the Community Copy Chapter

Get involved with the Pipecat community to share your projects, get support, and contribute to the ecosystem:

• Discord Community: Get support, share projects, and connect with other developers
• GitHub: Contribute to the project and report issues
• Examples Repository: Community-contributed applications

Service Settings

Source: https://docs.pipecat.ai/pipecat/fundamentals/service-settings.md

Learn how to configure and update AI service settings at initialization and runtime.

Settings are the runtime-configurable properties of AI services, including things like the model, voice, language, temperature, and other provider-specific options. Every service exposes a Settings class that you can use in two ways:

1. Configure at initialization: pass a settings= argument when constructing a service.
2. Update at runtime: push an *UpdateSettingsFrame through the pipeline to change settings mid-conversation.

Configuring settings at initialization Copy Chapter

Pass a settings= argument to any service constructor using that service's Settings class:

from pipecat.services.openai.llm import OpenAILLMService

llm = OpenAILLMService(
    api_key=os.getenv("OPENAI_API_KEY"),
    settings=OpenAILLMService.Settings(
        model="gpt-4o",
        temperature=0.7,
        system_instruction="You are a helpful assistant.",
    ),
)

from pipecat.services.cartesia.tts import CartesiaTTSService

tts = CartesiaTTSService(
    api_key=os.getenv("CARTESIA_API_KEY"),
    settings=CartesiaTTSService.Settings(
        voice="71a7ad14-091c-4e8e-a314-022ece01c121",
    ),
)

from pipecat.services.deepgram.stt import DeepgramSTTService

stt = DeepgramSTTService(
    api_key=os.getenv("DEEPGRAM_API_KEY"),
    settings=DeepgramSTTService.Settings(
        model="nova-3",
        language="en",
        smart_format=True,
    ),
)

You only need to specify the settings you want to override. Everything else uses the service's defaults.

Common settings by service type Copy Chapter

Each service type has a base set of settings. Individual services may extend these with provider-specific fields.

LLM settings Copy Section

TTS settings Copy Section

STT settings Copy Section

Updating settings at runtime Copy Chapter

You can change service settings while the pipeline is running by pushing an update settings frame. This is useful for scenarios like switching languages, changing voices, or adjusting LLM parameters mid-conversation.

Use these frame types:

Only include the fields you want to change. Unspecified fields are left as-is.

from pipecat.frames.frames import TTSUpdateSettingsFrame

# Change TTS voice mid-conversation
await worker.queue_frame(
    TTSUpdateSettingsFrame(
        delta=CartesiaTTSService.Settings(voice="new-voice-id")
    )
)

from pipecat.frames.frames import LLMUpdateSettingsFrame

# Lower the temperature for more deterministic responses
await worker.queue_frame(
    LLMUpdateSettingsFrame(
        delta=OpenAILLMService.Settings(temperature=0.2)
    )
)

from pipecat.frames.frames import STTUpdateSettingsFrame
from pipecat.transcriptions.language import Language

# Switch STT language to Spanish
await worker.queue_frame(
    STTUpdateSettingsFrame(
        delta=DeepgramSTTService.Settings(language=Language.ES)
    )
)

Service-specific settings Copy Chapter

Services can extend the base settings with provider-specific fields. For example:

• Cartesia TTS adds generation_config (volume, speed, emotion) and pronunciation_dict_id
• OpenAI TTS adds instructions and speed
• Deepgram STT adds endpointing, smart_format, diarize, punctuate, and many more
• OpenAI LLM adds max_completion_tokens

These service-specific fields work the same way — set them at initialization or update them at runtime:

from pipecat.services.cartesia.tts import CartesiaTTSService, GenerationConfig

tts = CartesiaTTSService(
    api_key=os.getenv("CARTESIA_API_KEY"),
    settings=CartesiaTTSService.Settings(
        voice="71a7ad14-091c-4e8e-a314-022ece01c121",
        generation_config=GenerationConfig(speed=1.2, emotion="excited"),
    ),
)

See the individual service documentation for a complete list of available settings:

Passing extra parameters Copy Chapter

Every Settings class includes an extra dict for passing provider-specific parameters that Pipecat doesn't have a dedicated field for. This is useful when a provider supports options that haven't been explicitly added to the Settings dataclass yet:

llm = OpenAILLMService(
    api_key=os.getenv("OPENAI_API_KEY"),
    settings=OpenAILLMService.Settings(
        model="gpt-4o",
        extra={"logprobs": True, "top_logprobs": 5},
    ),
)

Values in extra are passed through to the underlying API call. You can also update extra at runtime:

await worker.queue_frame(
    LLMUpdateSettingsFrame(
        delta=OpenAILLMService.Settings(
            extra={"logprobs": False},
        )
    )
)

Migration from InputParams Copy Chapter

llm = OpenAILLMService(
    model="gpt-4o",
    params=OpenAILLMService.InputParams(
        temperature=0.7,
    ),
)

llm = OpenAILLMService(
    settings=OpenAILLMService.Settings(
        model="gpt-4o",
        temperature=0.7,
    ),
)

Interruptions

Source: https://docs.pipecat.ai/pipecat/fundamentals/interruptions.md

How Pipecat stops the bot when the user speaks, what happens to in-flight LLM and TTS output, and how to control it.

Interruptions (also called barge-in) let the user talk over the bot. When the user starts speaking while the bot is talking, the bot stops immediately, in-flight work is cancelled, and the pipeline is ready for the new user input.

Interruptions are enabled by default. This page explains how they work under the hood, what ends up in the conversation context, and how to configure or trigger them yourself.

What happens when the user interrupts Copy Chapter

When a user turn start strategy triggers with enable_interruptions=True (the default), the user aggregator broadcasts an InterruptionFrame both upstream and downstream through the pipeline. From there:

The result: the bot goes silent within roughly one audio write, and the pipeline is clean and ready for the user's new turn.

What ends up in the context Copy Chapter

A common question: if the bot is cut off mid-sentence, what does the LLM context contain?

Only the words that were actually spoken. As the bot speaks, the output transport pushes TTSTextFrames downstream in sync with audio playback. Text that never played never reaches the assistant context aggregator. On interruption, the aggregator commits the partial, spoken-so-far text to the context as the assistant message.

This means the LLM's next completion sees an accurate transcript of the conversation: the bot's message ends where the user cut it off, not where the LLM's generation ended.

You can observe this with the on_assistant_turn_stopped event, which reports the committed text and whether the turn was interrupted:

@assistant_aggregator.event_handler("on_assistant_turn_stopped")
async def on_assistant_turn_stopped(aggregator, message):
    if message.interrupted:
        print(f"Bot was cut off after saying: {message.content}")

Controlling interruptions Copy Chapter

Let the bot finish speaking with user input muting. Mute strategies block user audio, transcriptions, and interruption signals while active, so speech over the bot is discarded rather than answered later. Use AlwaysUserMuteStrategy to mute whenever the bot is speaking, or FirstSpeechUserMuteStrategy to protect just the introduction:

from pipecat.processors.aggregators.llm_response_universal import (
    LLMContextAggregatorPair,
    LLMUserAggregatorParams,
)
from pipecat.turns.user_mute import AlwaysUserMuteStrategy

user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
    context,
    user_params=LLMUserAggregatorParams(
        user_mute_strategies=[AlwaysUserMuteStrategy()],
    ),
)

Require a minimum number of words so short utterances like "okay" or "yeah" don't interrupt the bot. This is configured on the user turn start strategy:

from pipecat.turns.user_start import MinWordsUserTurnStartStrategy

start_strategy = MinWordsUserTurnStartStrategy(min_words=3)

Filter out backchannels with a model. The Krisp VIVA Interruption Prediction strategy distinguishes genuine interruptions from acknowledgments like "uh-huh".

Disable interruptions so in-flight work is never cancelled:

from pipecat.turns.user_start import VADUserTurnStartStrategy

start_strategy = VADUserTurnStartStrategy(enable_interruptions=False)

When an STT service with built-in turn detection drives turns, it routes its own should_interrupt setting to the strategies it requests, so pass should_interrupt=False to the service instead. See External Turn Management.

Triggering an interruption yourself Copy Chapter

Sometimes the bot should stop itself: a timeout fires, an external event arrives, or your own logic decides the current response is no longer relevant.

From inside a custom FrameProcessor, broadcast the interruption directly:

await self.broadcast_interruption()

From code that has a reference to a processor or the worker, you can also push an InterruptionWorkerFrame. The pipeline worker converts it into an InterruptionFrame and sends it through the whole pipeline:

from pipecat.frames.frames import InterruptionWorkerFrame

await worker.queue_frame(InterruptionWorkerFrame())

Both approaches run the same interruption flow described above: the bot stops speaking, in-flight work is cancelled, and the spoken-so-far text is committed to context.

Speech-to-speech services Copy Chapter

Realtime speech-to-speech services (like Gemini Live and OpenAI Realtime) handle interruption detection on the provider side. In these pipelines, the service emits the speaking events and Pipecat's aggregators follow along using external turn strategies. The provider decides when the user barged in; Pipecat still flushes local audio output so the bot goes silent right away.

Related Copy Chapter

• Speech Input & Turn Detection - how user turns are detected
• User Turn Strategies - full strategy reference
• User Input Muting - suppress user input while the bot speaks
• System Frames - InterruptionFrame reference

User Input Muting

Source: https://docs.pipecat.ai/pipecat/fundamentals/user-input-muting.md

Learn how to control when user speech is processed in your conversational bot

Overview Copy Chapter

In conversational applications, there are moments when you don't want to process user speech, such as during bot introductions or while executing function calls. Pipecat's user mute strategies let you selectively "mute" user input based on different conversation states.

When to Use Mute Strategies Copy Chapter

Common scenarios for muting user input include:

• During introductions: Prevent the bot from being interrupted during its initial greeting
• While processing functions: Block input while the bot is retrieving external data
• During bot speech: Reduce false transcriptions while the bot is speaking
• For guided conversations: Create more structured interactions with clear turn-taking

How It Works Copy Chapter

User mute strategies work by blocking specific user-related frames from flowing through your pipeline. When muted, the following frames are filtered:

• Voice activity detection (VAD) events
• Interruption signals
• Raw audio input frames
• Transcription frames (both interim and final)

This prevents user speech from being processed during muted periods.

Mute Strategies Copy Chapter

Pipecat provides several built-in strategies for determining when to mute user input:

Basic Implementation Copy Chapter

Import and configure the mute strategies you need:

from pipecat.processors.aggregators.llm_response_universal import (
    LLMContextAggregatorPair,
    LLMUserAggregatorParams,
)
from pipecat.turns.user_mute import AlwaysUserMuteStrategy

# Configure with one or more strategies
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
    context,
    user_params=LLMUserAggregatorParams(
        user_mute_strategies=[AlwaysUserMuteStrategy()],
    ),
)

Combining Multiple Strategies Copy Chapter

Multiple strategies can be combined. They use OR logic—if any strategy indicates the user should be muted, input is suppressed:

from pipecat.processors.aggregators.llm_response_universal import (
    LLMContextAggregatorPair,
    LLMUserAggregatorParams,
)

from pipecat.turns.user_mute import (
    MuteUntilFirstBotCompleteUserMuteStrategy,
    FunctionCallUserMuteStrategy,
)

user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
    context,
    user_params=LLMUserAggregatorParams(
        user_mute_strategies=[
            MuteUntilFirstBotCompleteUserMuteStrategy(),  # Mute until first response
            FunctionCallUserMuteStrategy(),               # Mute during function calls
        ],
    ),
)

Building Custom Strategies Copy Chapter

Subclass BaseUserMuteStrategy (in pipecat.turns.user_mute) when none of the built-in strategies fit. A strategy only needs to answer one question per frame: should the user be muted right now? Override process_frame(self, frame: Frame) -> bool to update internal state and return the current mute decision.

Which frames reach a strategy Copy Section

Each strategy's process_frame is called for every frame that passes through the user aggregator, except StartFrame, EndFrame, and CancelFrame. This includes:

• User-direction frames from the input transport and STT: TranscriptionFrame, InterimTranscriptionFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame, ProposedUserStartedSpeakingFrame, ProposedUserStoppedSpeakingFrame, VADUserStartedSpeakingFrame, VADUserStoppedSpeakingFrame, InputAudioRawFrame, InterruptionFrame
• Bot and function-calling lifecycle frames from elsewhere in the pipeline: BotStartedSpeakingFrame, BotStoppedSpeakingFrame, FunctionCallsStartedFrame, FunctionCallResultFrame, FunctionCallCancelFrame

Which frames get suppressed when muted Copy Section

Returning True from your strategy sets the aggregator's mute state. While muted, only these frame types are actually dropped:

• InterruptionFrame
• VADUserStartedSpeakingFrame, VADUserStoppedSpeakingFrame
• UserStartedSpeakingFrame, UserStoppedSpeakingFrame
• ProposedUserStartedSpeakingFrame, ProposedUserStoppedSpeakingFrame
• InputAudioRawFrame
• InterimTranscriptionFrame, TranscriptionFrame

All other frames continue to flow so the rest of the pipeline keeps functioning.

Toggling a strategy at runtime Copy Section

Strategies are plain Python objects. Anything that holds a reference to one can flip its state between frames, which means a companion processor placed elsewhere in the pipeline can drive the mute decision based on signals the strategy can't observe directly (LLM text, tool results, external events).

This example strategy adds its own enable/disable methods (not part of the base contract) and returns their state from process_frame:

from pipecat.frames.frames import Frame
from pipecat.turns.user_mute import BaseUserMuteStrategy


class ToggleableUserMuteStrategy(BaseUserMuteStrategy):
    def __init__(self):
        super().__init__()
        self._muted = False

    def enable(self):
        self._muted = True

    def disable(self):
        self._muted = False

    async def process_frame(self, frame: Frame) -> bool:
        await super().process_frame(frame)
        return self._muted

A companion processor watches for the trigger and toggles the strategy:

from pipecat.frames.frames import (
    BotStartedSpeakingFrame,
    BotStoppedSpeakingFrame,
    Frame,
    LLMTextFrame,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor


class DisclaimerGuardProcessor(FrameProcessor):
    def __init__(self, strategy: ToggleableUserMuteStrategy, trigger_phrase: str, **kwargs):
        super().__init__(**kwargs)
        self._strategy = strategy
        self._trigger = trigger_phrase
        # Keep a small sliding window so cross-frame matches work without
        # the buffer growing unbounded if the trigger never appears.
        self._max_buffer = max(len(trigger_phrase) * 4, 512)
        self._buffer = ""
        self._active = False

    async def process_frame(self, frame: Frame, direction: FrameDirection):
        await super().process_frame(frame, direction)

        if isinstance(frame, BotStartedSpeakingFrame):
            # Start each bot turn with a fresh buffer.
            self._buffer = ""
        elif isinstance(frame, LLMTextFrame) and direction == FrameDirection.DOWNSTREAM:
            self._buffer = (self._buffer + frame.text)[-self._max_buffer :]
            if not self._active and self._trigger in self._buffer:
                self._active = True
                self._strategy.enable()
        elif isinstance(frame, BotStoppedSpeakingFrame) and self._active:
            self._active = False
            self._buffer = ""
            self._strategy.disable()

        await self.push_frame(frame, direction)

Wire them together by passing the same strategy instance to both the aggregator and the processor:

mute_strategy = ToggleableUserMuteStrategy()

user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
    context,
    user_params=LLMUserAggregatorParams(user_mute_strategies=[mute_strategy]),
)

disclaimer_guard = DisclaimerGuardProcessor(
    strategy=mute_strategy,
    trigger_phrase="Please read the following disclosure",
)

pipeline = Pipeline([
    transport.input(),
    stt,
    user_aggregator,
    llm,
    disclaimer_guard,   # positioned where LLMTextFrame flows downstream
    tts,
    transport.output(),
    assistant_aggregator,
])

Responding to Mute Events Copy Chapter

You can register event handlers to be notified when muting starts or stops. This is particularly useful for providing visual feedback to users:

@user_aggregator.event_handler("on_user_mute_started")
async def on_user_mute_started(aggregator):
    logger.info("User mute started")
    # Send a visual indicator to your client
    # e.g., show a "Bot is speaking" indicator

@user_aggregator.event_handler("on_user_mute_stopped")
async def on_user_mute_stopped(aggregator):
    logger.info("User mute stopped")
    # Update your client UI
    # e.g., show a "You can speak now" indicator

These events fire whenever the mute state changes, allowing you to keep your UI synchronized with the bot's state.

RTVI Events Copy Section

When mute strategies activate or deactivate, the server automatically sends RTVI messages (user-mute-started and user-mute-stopped) to the client. You can listen for these in the JavaScript client to update your UI:

import { PipecatClient, RTVIEvent } from "@pipecat-ai/client-js";

const pcClient = new PipecatClient({
  callbacks: {
    onUserMuteStarted: () => {
      // Show a visual indicator that the bot is not listening
      // e.g., disable a microphone button or show "Bot is speaking..."
    },
    onUserMuteStopped: () => {
      // Remove the indicator, show the user they can speak
    },
  },
});

// Or using event listeners
pcClient.on(RTVIEvent.UserMuteStarted, () => {
  console.log("Server is ignoring user audio");
});
pcClient.on(RTVIEvent.UserMuteStopped, () => {
  console.log("Server is listening to user audio again");
});

Best Practices Copy Chapter

• Choose strategies wisely: Select the minimal set of strategies needed for your use case
• Test user experience: Excessive muting can frustrate users; balance control with usability
• Provide feedback: Use the mute event handlers to show visual cues when the user is muted to improve the experience

Next Steps Copy Chapter

Learn how to configure turn detection behavior for more control over
conversation flow.

Experiment with different muting strategies to find the right balance for your application.

Detecting Idle Users

Source: https://docs.pipecat.ai/pipecat/fundamentals/detecting-user-idle.md

Learn how to detect and respond when users are inactive in conversations

Overview Copy Chapter

In conversational applications, it's important to handle situations where users go silent or inactive. Pipecat provides built-in idle detection through LLMUserAggregator and UserTurnProcessor, allowing your bot to respond appropriately when users haven't spoken for a defined period.

How It Works Copy Chapter

Idle detection monitors user activity and:

1. Starts a timer when the bot finishes speaking (BotStoppedSpeakingFrame)
2. Cancels the timer when the user or bot starts speaking
3. Suppresses the timer during function calls and active user turns (to avoid false triggers during interruptions)
4. Emits an on_user_turn_idle event when the timer expires
5. Allows you to implement escalating responses or gracefully end the conversation in your application code

Basic Implementation Copy Chapter

Step 1: Enable Idle Detection Copy Section

Enable idle detection by setting the user_idle_timeout parameter when creating your aggregator:

from pipecat.processors.aggregators.llm_response_universal import (
    LLMContextAggregatorPair,
    LLMUserAggregatorParams,
)

user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
    context,
    user_params=LLMUserAggregatorParams(
        user_idle_timeout=5.0,  # Detect idle after 5 seconds
    ),
)

Step 2: Handle Idle Events Copy Section

Create an event handler to respond when the user becomes idle:

@user_aggregator.event_handler("on_user_turn_idle")
async def on_user_turn_idle(aggregator):
    # Send a reminder to the user
    message = {
        "role": "developer",
        "content": "The user has been quiet. Politely ask if they're still there.",
    }
    await aggregator.push_frame(LLMMessagesAppendFrame([message], run_llm=True))

Step 3: Implement Retry Logic (Optional) Copy Section

For escalating responses, track retry count in your application:

class IdleHandler:
    def __init__(self):
        self._retry_count = 0

    def reset(self):
        self._retry_count = 0

    async def handle_idle(self, aggregator):
        self._retry_count += 1

        if self._retry_count == 1:
            # First attempt - gentle reminder
            message = {
                "role": "developer",
                "content": "The user has been quiet. Politely ask if they're still there.",
            }
            await aggregator.push_frame(LLMMessagesAppendFrame([message], run_llm=True))
        elif self._retry_count == 2:
            # Second attempt - more direct
            message = {
                "role": "developer",
                "content": "The user is still inactive. Ask if they'd like to continue.",
            }
            await aggregator.push_frame(LLMMessagesAppendFrame([message], run_llm=True))
        else:
            # Third attempt - end conversation
            await aggregator.push_frame(
                TTSSpeakFrame("It seems like you're busy. Have a nice day!")
            )
            await aggregator.push_frame(EndWorkerFrame(), FrameDirection.UPSTREAM)

# Use the handler
idle_handler = IdleHandler()

@user_aggregator.event_handler("on_user_turn_idle")
async def on_user_turn_idle(aggregator):
    await idle_handler.handle_idle(aggregator)

@user_aggregator.event_handler("on_user_turn_started")
async def on_user_turn_started(aggregator, strategy):
    idle_handler.reset()  # Reset retry count when user speaks

Updating Timeout at Runtime Copy Chapter

You can enable, disable, or change the idle timeout at runtime by pushing a UserIdleTimeoutUpdateFrame:

from pipecat.frames.frames import UserIdleTimeoutUpdateFrame

# Enable idle detection (or change timeout)
await worker.queue_frame(UserIdleTimeoutUpdateFrame(timeout=10.0))

# Disable idle detection
await worker.queue_frame(UserIdleTimeoutUpdateFrame(timeout=0))

Immediate application: Timeout updates take effect immediately. If an idle timer is currently running, it restarts with the new duration. If the bot is waiting for the user to speak and you enable a positive timeout (e.g., upgrading from 0 to 10.0), the timer arms right away without waiting for the next bot turn.

This is useful when you want to enable idle detection only at certain points in the conversation, or adjust the timeout based on context.

Best Practices Copy Chapter

• Set appropriate timeouts: Shorter timeouts (5-10 seconds) work well for voice conversations
• Use escalating responses: Start with gentle reminders and gradually become more direct
• Limit retry attempts: After 2-3 unsuccessful attempts, consider ending the conversation gracefully by pushing an EndWorkerFrame
• Reset on user activity: Use the on_user_turn_started event to reset your retry counter when the user speaks
• Let the LLM respond naturally: Use developer messages to prompt the LLM rather than hardcoded TTS responses for more natural interactions

Next Steps Copy Chapter

Learn about all available turn events and their parameters.

Implementing idle user detection improves the conversational experience by ensuring your bot can handle periods of user inactivity gracefully, either by prompting for re-engagement or politely ending the conversation when appropriate.

STT Latency Tuning

Source: https://docs.pipecat.ai/pipecat/fundamentals/stt-latency-tuning.md

Measure and tune STT latency in Pipecat to improve turn detection timing and end-of-turn responsiveness.

What is TTFS? Copy Chapter

Time To Final Segment (TTFS) measures how long it takes from the moment a user stops speaking until the STT service delivers the final transcript. This latency directly affects how long your bot waits before it starts responding.

User stops → [TTFS latency] → Final transcript arrives → Bot starts

Every STT service has a different TTFS profile based on its architecture, model complexity, and infrastructure. Pipecat ships with measured P99 latency values for each supported service so that turn detection can account for this delay automatically. Measured values were benchmarked using the stt-benchmark tool.

Why TTFS matters Copy Chapter

TTFS feeds directly into turn stop strategies, which decide when the user has finished speaking and the bot should respond.

• Value too low: The turn stop strategy gives up waiting before the final transcript arrives. The bot responds based on incomplete text, or misses the user's input entirely.
• Value too high: The bot waits longer than necessary after the user stops speaking, creating awkward pauses in the conversation.
• Value just right: The bot waits long enough for the transcript to arrive, then responds immediately.

Getting TTFS right is one of the most impactful tuning knobs for perceived conversation responsiveness.

Default P99 latency values Copy Chapter

Pipecat includes measured P99 TTFS values for every supported STT service. These are used automatically when you create a service — no configuration required.

For the current P99 TTFS value for each service, see stt_latency.py. These values are refined and added to often, so the source file is the source of truth.

Measuring latency for your deployment Copy Chapter

The default values are measured under standard conditions, but your actual latency depends on:

• Network distance to the STT provider
• Region where the service is hosted
• Service configuration (model size, language, features enabled)
• Audio quality and encoding settings

Use the stt-benchmark tool to measure TTFS for your specific setup. The tool sends standardized audio samples to your STT service and reports P50, P90, and P99 latency values.

To run it, clone the repo and install with uv:

git clone https://github.com/pipecat-ai/stt-benchmark
cd stt-benchmark
uv sync

# Add your provider API keys
cp env.example .env

# Download standardized audio samples
uv run stt-benchmark download --num-samples 100

# Run the benchmark for one or more services
uv run stt-benchmark run --services deepgram,openai

# View the P50/P90/P99 report
uv run stt-benchmark report --service deepgram

If you run your bot with a non-default VAD setting, match the benchmark to it with --vad-stop-secs so the measured value reflects your configuration:

uv run stt-benchmark run --services deepgram --vad-stop-secs 0.3

See the stt-benchmark README for the full command reference.

Overriding the default value Copy Chapter

Pass the ttfs_p99_latency parameter to any STT service constructor to override the built-in default:

from pipecat.services.deepgram.stt import DeepgramSTTService

# Use a measured value from your deployment
stt = DeepgramSTTService(
    api_key=os.getenv("DEEPGRAM_API_KEY"),
    ttfs_p99_latency=0.45,  # Override with your measured P99
)

This value is broadcast to the pipeline via an STTMetadataFrame at startup, so turn stop strategies automatically adjust their timing.

Context Summarization

Source: https://docs.pipecat.ai/pipecat/fundamentals/context-summarization.md

Automatically compress older conversation history in long-running Pipecat conversations to manage LLM token usage and cost.

Overview Copy Chapter

In long-running voice AI conversations, context grows with every exchange. This increases token usage, raises costs, and can eventually hit context window limits. Pipecat includes built-in context summarization that automatically compresses older conversation history while preserving recent messages and important context.

How It Works Copy Chapter

Context summarization automatically triggers when either condition is met:

• Token limit reached: Context size exceeds max_context_tokens (estimated using ~4 characters per token)
• Message count reached: Number of new messages exceeds max_unsummarized_messages

You can disable either threshold by setting it to None, but at least one must remain active. Summarization always generates a summary and cannot be reduced to pure truncation.

When triggered, the system:

1. Sends a LLMContextSummaryRequestFrame to the LLM service
2. The LLM generates a concise summary of older messages
3. Context is reconstructed as: [system_message (if present)] + [summary] + [recent_messages]
4. Incomplete function call sequences and recent messages are preserved

Enabling Context Summarization Copy Chapter

Enable summarization by setting enable_auto_context_summarization=True in LLMAssistantAggregatorParams:

from pipecat.processors.aggregators.llm_response_universal import (
    LLMAssistantAggregatorParams,
    LLMContextAggregatorPair,
)

# Create aggregators with summarization enabled
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
    context,
    assistant_params=LLMAssistantAggregatorParams(
        enable_auto_context_summarization=True,
    ),
)

Automatic summarization is disabled by default (enable_auto_context_summarization=False). When enabled with the default configuration, summarization triggers at 8000 estimated tokens or after 20 new messages, whichever comes first.

Customizing Behavior Copy Chapter