How to add addressee detection to a Pipecat voice agent

L5 audio stack

The missing layer in your voice pipeline: addressee gating.

Noise suppression, VAD, diarization, and turn detection all pass audio through. None asks is this speech addressed to the agent?

SAA (Selective Auditory Attention) adds addressee gating as a fifth layer, upstream of ASR. The gate fails closed: a missed command beats a ghost response.

Evaluation results on held-out multi-party sessions are published at arXiv:2604.08412.

Integration architecture

SAA attaches alongside your DailyTransport. Your pipeline code stays the same.

start_attention_session() provisions a hosted bot that joins your Daily room as a hidden participant. AttentionEngine hooks DailyTransport's on_app_message handler and fires typed Python callbacks on the 'saa' topic.

Getting started

Install saa-pipecat-client and wire AttentionEngine in five calls.

Requires Python 3.11+ alongside pipecat-ai[daily]. New pipelines can skip to step 4, build_attention_runner, which wraps all five calls into one coroutine.

pip install saa-pipecat-client pipecat-ai[daily]

export SAA_API_KEY="your-saa-api-key"
export DAILY_API_KEY="your-daily-api-key"

import asyncio
from saa_pipecat_client import (
    attention_agent_token,
    start_attention_session,
    AttentionEngine,
    AttentionStartupError,
)

async def main(transport, task):
    # Step 1: mint a Daily meeting token for the hosted bot.
    token = await attention_agent_token()

    # Step 2: provision a session; the hosted bot joins the room.
    session = await start_attention_session(room_url=DAILY_ROOM_URL)

    # Step 3: create the engine, passing the transport and the
    # agent identity returned by start_attention_session().
    engine = AttentionEngine(
        transport,
        agent_identity=session.agent_identity,
    )

    # Step 4: bind the PipelineTask so upstream actions can be
    # queued via DailyOutputTransportMessageUrgentFrame.
    engine.bind_task(task)

    # Step 5: start. Raises AttentionStartupError if the bot
    # publishes an error, or asyncio.TimeoutError if it never
    # publishes 'started' within ready_timeout (default 30 s).
    await engine.start()

from saa_pipecat_client import build_attention_runner

# build_attention_runner composes all three provisioning steps
# (attention_agent_token + start_attention_session +
# AttentionEngine.start) into a single run() coroutine.
runner = build_attention_runner(room_url=DAILY_ROOM_URL)
await runner.run()

Event API

PredictionEvent fires at 4 Hz. TurnReadyEvent delivers each utterance as PCM.

Suppress audio forwarding on aligned_class, not raw_class: a 5-tick correction cancels flips right after AI playback ends.

from saa_pipecat_client import (
    PredictionEvent,
    TurnReadyEvent,
    InterruptEvent,
    InterjectionEvent,
)

@engine.on("prediction")
async def on_prediction(event: PredictionEvent) -> None:
    # Use aligned_class (5-tick AI-aware correction), not raw_class.
    # 0 = silent, 1 = human-to-human, 2 = human-to-device.
    if event.aligned_class != 2:
        # Suppress audio: do not forward to STT or LLM.
        return
    # Forward audio to your STT pipeline here.

@engine.on("turn_ready")
async def on_turn_ready(event: TurnReadyEvent) -> None:
    # event.audio_pcm16: int16 mono PCM at 16 kHz, assembled utterance.
    # event.context == 'interjection_follow_up' flags a proactive turn;
    # consider using a different LLM prompt for these.
    await send_to_stt(event.audio_pcm16)

@engine.on("interrupt")
async def on_interrupt(event: InterruptEvent) -> None:
    # Class-2 streak detected during AI playback.
    # Ring-buffer audio is pre-rolled; next TurnReadyEvent has the barge-in.
    await cancel_tts_playback()

@engine.on("interjection")
async def on_interjection(event: InterjectionEvent) -> None:
    # event.audio_pcm16: the preceding conversation audio, not silence.
    await handle_interjection(event.audio_pcm16)

Upstream actions

Signal AI playback to arm the interrupt detector. Adjust the threshold anytime.

responding_start / responding_stop arm the interrupt detector. set_threshold changes the gate live, no restart needed.

from pipecat.processors.frame_processor import FrameProcessor
from pipecat.frames.frames import (
    TTSStartedFrame,
    TTSStoppedFrame,
)

class BotSpeakingObserver(FrameProcessor):
    """Thin FrameProcessor that signals AI playback state to SAA.

    Passes all frames through unchanged. Watches TTSStartedFrame
    and TTSStoppedFrame as side effects and calls
    engine.responding_start() / engine.responding_stop().

    Without this signal, InterruptEvent and InterjectionEvent
    cannot fire correctly.
    """

    def __init__(self, engine: AttentionEngine):
        super().__init__()
        self._engine = engine

    async def process_frame(self, frame, direction):
        await super().process_frame(frame, direction)
        if isinstance(frame, TTSStartedFrame):
            await self._engine.responding_start()
        elif isinstance(frame, TTSStoppedFrame):
            await self._engine.responding_stop()

# Adjust the class-2 confidence threshold at runtime.
# Accepts a float in [0.0, 1.0]. Takes effect immediately;
# no session restart required.
#
# Suggested comparison points:
#   0.6:  passes borderline utterances; higher recall
#   0.77: balanced default for most shared-space deployments
#   0.88: high-confidence only; fewer false triggers in noisy rooms
#
# Run each under realistic room conditions and keep the value
# with the best command-catch to false-trigger ratio.
await engine.set_threshold(0.77)

OpenAI Realtime

Disable OpenAI's server-side VAD and hand turn ownership to SAA.

Set turn_detection=False and start_audio_paused=True. In on_turn_ready, resample SAA's 16 kHz PCM to 24 kHz and inject it via ConversationItemCreateEvent plus ResponseCreateEvent.

import base64
from pipecat.services.openai_realtime_beta import (
    OpenAIRealtimeLLMService,
    SessionProperties,
    ConversationItemCreateEvent,
    ResponseCreateEvent,
)

# Disable OpenAI server-side VAD; SAA owns turn boundaries.
llm = OpenAIRealtimeLLMService(
    session_properties=SessionProperties(
        audio={"input": {"turn_detection": False}}
    ),
    # No mic audio reaches OpenAI until SAA opens the gate.
    start_audio_paused=True,
)

@engine.on("turn_ready")
async def on_turn_ready(event: TurnReadyEvent) -> None:
    # SAA delivers int16 mono PCM at 16 kHz.
    # OpenAI Realtime expects 24 kHz; resample before injecting.
    pcm_24k = resample_16k_to_24k(event.audio_pcm16)
    encoded = base64.b64encode(pcm_24k.tobytes()).decode()

    # Inject the assembled utterance as a conversation item,
    # then request a response.
    await llm.push_frame(ConversationItemCreateEvent(audio=encoded))
    await llm.push_frame(ResponseCreateEvent())

Deployment

Any host that runs pipecat-ai[daily] runs saa-pipecat-client identically.

It's a pure Python package with no platform-specific extension modules, so behavior is identical across hosts.

For the LiveKit equivalent, see the LiveKit integration guide. For OpenAI Realtime, see the OpenAI Realtime integration guide. For Twilio and ElevenLabs transport adapters, contact us.

Common questions

What Pipecat developers ask before integrating.

Get access

Sign up for API keys and start gating addressee audio today.

Self-serve, no approval step. Includes saa-pipecat-client and full docs. Production licensing is per device.

Last updated June 29, 2026