n4nAI

Build a voice assistant with Pipecat and GPT-4o Realtime

Step-by-step pipecat gpt-4o realtime voice assistant tutorial: build a low-latency voice agent with Pipecat and OpenAI Realtime, from install to verification.

n4n Team4 min read929 words

Audio narration

Coming soon — every post will get a voice note here.

This pipecat gpt-4o realtime voice assistant tutorial walks through building a low-latency voice agent using Pipecat’s pipeline and OpenAI’s GPT-4o Realtime model. We’ll stand up a working assistant that listens, reasons, and speaks without stitching together separate ASR, LLM, and TTS services.

GPT-4o Realtime handles speech-in, speech-out natively over a websocket. Pipecat wraps that connection as a single pipeline node, so your application code stays declarative instead of becoming a tangle of callbacks and buffers.

Step 1: Install dependencies

Use Python 3.10 or newer. Create a clean virtual environment, then install the Pipecat core plus the OpenAI and local-transport extras:

python -m venv venv
source venv/bin/activate
pip install "pipecat-ai[openai,local]" python-dotenv

The local extra pulls in sounddevice and numpy for microphone capture and speaker playback. If you plan to deploy on a server with a web frontend, swap local for daily and install daily-python instead—the pipeline code does not change.

Verify the install compiled native audio backends correctly:

python -c "import sounddevice; print(sounddevice.query_devices())"

You should see a list of input and output devices. If this errors, fix your system audio libraries before continuing; nothing downstream will work without a working sounddevice.

Step 2: Configure credentials and environment

GPT-4o Realtime requires an OpenAI API key with access to the preview model. Put it in a .env file at your project root:

OPENAI_API_KEY=sk-...

Pipecat can read this via python-dotenv, or you can pass the key explicitly to the service constructor. Do not hardcode secrets in source files—environment injection is the norm.

If you later switch to the Daily transport for browser clients, you will also need DAILY_API_KEY and a dynamically generated room URL. For local desktop development, that is unnecessary.

Step 3: Build the realtime pipeline

Choose a transport

The transport converts audio frames between your hardware (or a WebRTC room) and Pipecat’s frame pipeline. For a desktop quick-start, use LocalTransport:

from pipecat.transports.local import LocalTransport, LocalParams

transport = LocalTransport(
    LocalParams(
        audio_in_enabled=True,
        audio_out_enabled=True,
        vad_analyzer=None,  # Realtime model does its own VAD
    )
)

Setting vad_analyzer=None is not optional. GPT-4o Realtime performs server-side voice activity detection. Running a local VAD in front of it adds latency and causes double-triggered turns.

Wire the Realtime LLM

Pipecat’s OpenAIRealtimeLLM manages the websocket, session configuration, and audio frame conversion. Instantiate it with your key and the current preview model tag:

import os
from pipecat.services.openai.realtime import OpenAIRealtimeLLM

llm = OpenAIRealtimeLLM(
    api_key=os.getenv("OPENAI_API_KEY"),
    model="gpt-4o-realtime-preview-2024-10-01",
    instructions="You are a concise voice assistant. Answer in short spoken sentences.",
)

The instructions field sets the system prompt for the session. Keep it tight. The model speaks whatever it generates, so verbosity directly degrades the user experience.

Compose the pipeline

Pipecat pipelines are ordered lists of frame processors. The transport exposes input() (microphone to frames) and output() (frames to speaker). The LLM sits between them:

from pipecat.pipeline.pipeline import Pipeline

pipeline = Pipeline([
    transport.input(),
    llm,
    transport.output(),
])

That is the entire processing graph. There is no separate ASR node and no separate TTS node—the realtime model is both, and Pipecat hides the websocket framing behind a standard FrameProcessor interface.

Wrap in a task and runner

from pipecat.pipeline.task import PipelineTask
from pipecat.pipeline.runner import PipelineRunner
import asyncio

task = PipelineTask(pipeline)
runner = PipelineRunner()

asyncio.run(runner.run(task))

PipelineRunner owns the event loop, handles cancellation, and ensures the transport tears down cleanly on Ctrl-C. Do not wrap this in your own loop.run_until_complete unless you also forward shutdown signals.

Step 4: Run and verify the assistant

Save the snippets above as assistant.py. Run:

python assistant.py

Speak a prompt like “What’s the capital of Japan?” into your microphone. Within roughly 300–600 ms you should hear a spoken answer through your speakers.

Verification checklist:

  • No ValueError: audio device not found (set sounddevice default device if needed).
  • WebSocket connects without a 401 (key is correct and has preview access).
  • Speech stops echoing when you stop talking (server VAD is working).
  • Response audio plays exactly once, not in a loop.

If you get silence, enable debug logging:

import logging
logging.basicConfig(level=logging.DEBUG)

You will see frame flow: UserStartedSpeakingInboundAudioLLMOutboundAudioBotStartedSpeaking. Absence of InboundAudio means your mic isn’t captured; absence of OutboundAudio means the session isn’t generating.

Step 5: Add session configuration and guardrails

The raw preview model will chat freely. For anything user-facing, set turn-taking limits and explicit audio formats. Pass session_config to the LLM:

llm = OpenAIRealtimeLLM(
    api_key=os.getenv("OPENAI_API_KEY"),
    model="gpt-4o-realtime-preview-2024-10-01",
    instructions="You are a support bot for Acme Corp.",
    session_config={
        "turn_detection": {"type": "server_vad", "silence_duration_ms": 500},
        "input_audio_format": "pcm16",
        "output_audio_format": "pcm16",
    },
)

Tightening silence_duration_ms reduces dead air. Keep pcm16 to match the local transport’s default sample format.

To add a tool—say, fetching order status—register a function handler before running the pipeline:

async def get_order(parameters):
    order_id = parameters.get("order_id")
    # call your backend here
    return {"status": "shipped", "order_id": order_id}

llm.register_function("get_order", get_order)

The model will emit a function call frame when appropriate, Pipecat invokes your coroutine, and the return value is forwarded back into the realtime session as a response. No JSON parsing on your side.

How Pipecat frames map to Realtime events

Understanding the mapping helps when you extend the graph. AudioRawFrame from the transport becomes inbound PCM sent over the websocket. The LLM emits AudioRawFrame for model speech and TranscriptionFrame if you request transcripts. Because everything is a frame, you can insert a logging node, a PII redactor, or a context-retrieval node before the LLM without touching the transport or the model client.

from pipecat.processors.frame_processor import FrameProcessor

class LogFrames(FrameProcessor):
    async def process_frame(self, frame, direction):
        print(f"frame: {frame.__class__.__name__}")
        await self.push_frame(frame, direction)

pipeline = Pipeline([transport.input(), LogFrames(), llm, transport.output()])

This is the architectural win: the pipecat gpt-4o realtime voice assistant tutorial code you wrote stays linear even as requirements grow.

Common failure modes and debugging

Echo and feedback loops. If your microphone picks up speaker output, use headphones. LocalTransport has no acoustic echo cancellation.

Model deprecation. OpenAI rotates preview tags quarterly. If you see model_not_found, check the current gpt-4o-realtime-preview date suffix on the OpenAI docs.

High first-turn latency. Cold websocket handshake plus model warm-up dominates the first interaction. Subsequent turns are noticeably faster.

Token metering. Realtime sessions bill on audio tokens, not word count. Watch your provider usage dashboard; there is no local counter in Pipecat.

Crashing on exit. If you see tracebacks on Ctrl-C, ensure you are using PipelineRunner rather than manually cancelling tasks. It installs the signal handlers for you.

Deploying beyond localhost

For a web frontend, replace LocalTransport with DailyTransport:

from pipecat.transports.services.daily import DailyTransport, DailyParams

transport = DailyTransport(
    room_url,   # from Daily REST API
    token,      # scoped to that room
    "Assistant",
    DailyParams(audio_in_enabled=True, audio_out_enabled=True),
)

You generate the room and token from your Daily account. The pipeline list, the LLM config, and the runner stay identical. That transport abstraction is the reason Pipecat is worth adopting for voice work.

This pipecat gpt-4o realtime voice assistant tutorial gave you a runnable core. From here, add retrieval-augmented context, user authentication, or multi-agent routing by inserting nodes before the LLM. The frame model scales; the boilerplate does not.

Tagspipecatgpt-4ovoice-airealtime

Written by

n4n Team

The team building n4n — a single OpenAI-compatible API in front of 240+ models, with automatic fallback, load balancing and pay-per-token metering.

More from n4n Team →

All multimodal & voice apps with ai frameworks posts →