Phase 7: Voice Agent Frameworks · 60 min · Python · Pipecat · Deepgram SDK

Pipecat: Frame-Based Streaming Pipelines

Pipecat gives you full pipeline control. Every frame is yours to inspect, modify, or cancel.

What you will learn

Build voice agents with Pipecat's frame-based streaming architecture
Compose pipelines: transport → ASR → LLM → TTS → transport
Use pluggable services: DeepgramService, ElevenLabsService, OpenAILLMService
Implement interruption handling via upstream frame propagation

The Problem

Building a voice agent from scratch means wiring ASR → LLM → TTS with custom streaming logic, handling interruptions, managing conversation state, and dealing with edge cases. Pipecat provides a framework with pre-built services and a frame-based pipeline architecture so you focus on conversation logic, not plumbing.

The Concept

Frame-Based Architecture

Pipecat processes everything as frames — discrete units of data that flow through a pipeline:

Frame Type	Contains	Source
AudioRawFrame	PCM audio bytes	Microphone, TTS
TextFrame	Text string	ASR, LLM
LLMMessagesFrame	Chat messages	User, system
TTSSpeakFrame	Text to synthesize	LLM
TranscriptionFrame	ASR result	ASR service
EndFrame	Pipeline stop signal	System
CancelFrame	Cancel current operation	Barge-in
BotInterruptionFrame	Agent interrupting	Barge-in handler

What is a "frame" in Pipecat's architecture?

A frame is the fundamental unit of data in Pipecat's pipeline. Each frame carries specific data: AudioRawFrame carries PCM audio bytes, TextFrame carries text from ASR or LLM, LLMMessagesFrame carries chat messages, CancelFrame signals barge-in. Frames flow through the pipeline from source (microphone) to sink (speaker), passing through services (ASR, LLM, TTS). This architecture makes the pipeline composable — you can insert filters, aggregators, and handlers between services. It also makes debugging easier — you can log every frame as it passes through.

Pipecat Pipeline Structure

from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.task import PipelineTask
from pipecat.services.openai import OpenAILLMService
from pipecat.services.deepgram import DeepgramTTSService
from pipecat.services.cartesia import CartesiaTTSService
from pipecat.transports.network.fastapi import FastAPIWebsocketTransport

async def build_voice_agent():
    """Build a voice agent pipeline with Pipecat."""
    # Transport (WebSocket audio)
    transport = FastAPIWebsocketTransport(
        websocket=websocket,
        params=FastAPIWebsocketParams(
            audio_out_enabled=True,
            handler=handle_client_audio,
        )
    )

    # Services
    llm = OpenAILLMService(model="gpt-4o-mini", api_key=OPENAI_API_KEY)
    tts = CartesiaTTSService(
        api_key=CARTESIA_API_KEY,
        voice_id="79125125-1e22-47a6-9e8e-f4e1c4b2b914",
    )

    # Pipeline: Transport → LLM → TTS → Transport
    pipeline = Pipeline([
        transport.input(),      # Audio from client
        stt,                    # ASR (Deepgram)
        context_aggregator,     # Manage conversation context
        llm,                    # LLM (GPT-4o-mini)
        tts,                    # TTS (Cartesia)
        transport.output(),     # Audio to client
    ])

    task = PipelineTask(pipeline)
    await task.queue_frame(StartFrame())

Key Pipecat Services

Service	Providers	Purpose
STT	Deepgram, Whisper, AssemblyAI	Speech-to-text
LLM	OpenAI, Anthropic, Google, Groq	Text generation
TTS	Cartesia, ElevenLabs, PlayHT, Azure	Text-to-speech
Transport	FastAPI, LiveKit, Twilio	Audio transport
Context	Aggregator	Conversation history management

Interruption Handling in Pipecat

from pipecat.processors.filters.frame_filter import FrameFilter

class BargeInHandler:
    """Handle barge-in using Pipecat frames."""
    async def on_user_speaking(self, frame):
        """User started speaking — cancel TTS."""
        await self.task.queue_frame(CancelFrame())
        # Pipecat automatically stops TTS and clears the audio queue

Context Management

from pipecat.processors.aggregators.llm_context import LLMContextAggregator

context_aggregator = LLMContextAggregator(
    user_messages=[{"role": "system", "content": SYSTEM_PROMPT}],
    assistant_messages=[],
)

Unlock the full lesson

You've read the first 2 sections. The rest of this lesson covers Build, Check Yourself, Key Terms & Next — plus a hands-on lab, quiz, and project artifact.

Create a free account — the opening phases of 29 of 35 courses are free, no credit card.