saa-livekit-client

LiveKit client for the SAA.

Adds attention-aware gating, barge-in, and proactive interjection to any LiveKit voice agent.

SAA is the addressee layer for voice agents: one decision per utterance about whether speech was meant for your agent, before STT, LLM, or TTS. No wake word, model-agnostic, drop-in for the LiveKit pipeline you already run. The mental model is simple: audio in -> addressee gate (the SAA decision) -> only addressed audio reaches your agent.

The attention model runs on attention labs' hosted service, so this is a thin Apache-2.0 client: it summons a hidden hosted agent into your room and fans out its typed events. There is no ML dependency and no inference in this package.

Install

pip install saa-livekit-client

Quickstart: existing voice agent

import asyncio
import os

from livekit.agents import Agent, AgentServer, AgentSession, JobContext, cli
from livekit.plugins import openai
from saa_livekit_client import (
    AttentionEngine, attention_agent_token, start_attention_session,
)


class MyAssistant(Agent):
    def __init__(self):
        super().__init__(instructions="You are a helpful assistant")


server = AgentServer()


@server.rtc_session()
async def entrypoint(ctx: JobContext):
    await ctx.connect()
    user = await ctx.wait_for_participant()

    # summon the saa hosted agent into the room
    saa = await start_attention_session(
        api_key=os.environ["SAA_API_KEY"],
        livekit_url=os.environ["LIVEKIT_URL"],
        agent_token=attention_agent_token(
            api_key=os.environ["LIVEKIT_API_KEY"],
            api_secret=os.environ["LIVEKIT_API_SECRET"],
            room_name=ctx.room.name,
        ),
        room_name=ctx.room.name,
        participant_identity=user.identity,
    )
    ctx.add_shutdown_callback(saa.stop)

    # stand up the voice agent, then wire saa on top of the running session
    # (session.input / session.interrupt() are only valid once started)
    voice = AgentSession(llm=openai.realtime.RealtimeModel(voice="alloy"))
    await voice.start(agent=MyAssistant(), room=ctx.room)

    engine = AttentionEngine(ctx.room, agent_identity=saa.agent_identity)
    ctx.add_shutdown_callback(engine.stop)

    @engine.on_prediction
    def _(p):
        # the addressee gate: only class 2 (talking-to-device) reaches the agent
        voice.input.set_audio_enabled(p.aligned_class == 2)

    @engine.on_interrupt
    def _(ev):
        voice.interrupt()

    @engine.on_interjection
    async def _(ev):
        await voice.generate_reply(instructions="Briefly offer to help")

    # tell saa when the agent is speaking, arms interrupt, suppresses interjection
    @voice.on("agent_state_changed")
    def _(ev):
        if ev.new_state == "speaking":
            asyncio.create_task(engine.responding_start())
        elif ev.old_state == "speaking":
            asyncio.create_task(engine.responding_stop())

    await engine.start()


if __name__ == "__main__":
    cli.run_app(server)

That's the whole integration. Works with any LiveKit pipeline, including RealtimeModel speech-to-speech. Runnable variants are in the examples/livekit/ samples. These need livekit-agents>=1.0, verified on the current 1.5.x/1.6.x line. (WorkerOptions(entrypoint_fnc=...) also works and is the older idiom.)

Greenfield, build_attention_entrypoint

For new voice agents that don't have an existing pipeline:

from livekit.agents import AgentServer, JobContext, cli
from saa_livekit_client import build_attention_entrypoint, TurnReadyEvent

async def handle_turn(event: TurnReadyEvent, ctx: JobContext):
    # event.audio_pcm16 = int16 mono 16 kHz; event.frames = list[TurnFrame]
    response_pcm = await my_llm.respond(event.audio_pcm16, frames=event.frames)
    await publish_response_audio(ctx.room, response_pcm)

entrypoint = build_attention_entrypoint(on_turn=handle_turn)

server = AgentServer()
server.rtc_session()(entrypoint)
# or the older idiom: cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint))

if __name__ == "__main__":
    cli.run_app(server)

Environment: SAA_API_KEY, LIVEKIT_API_KEY, LIVEKIT_API_SECRET, LIVEKIT_URL.

Event types

Classes: 0=silent, 1=human-to-human, 2=human-to-device. responding is True while the AI is mid-playback.

Each is delivered through an @engine.on_* callback: on_prediction, on_vad, on_warmup, on_listening_start, on_listening_cancelled, on_turn_ready, on_interrupt, on_interjection, on_error, on_utterance_ended, on_utterance_config.

Upstream actions

await engine.mute()                       # stop feeding mic to processor
await engine.unmute()
await engine.responding_start()           # AI is now speaking
await engine.responding_stop()
await engine.set_threshold(0.65)          # model class-2 confidence threshold
await engine.add_assistant_turn(text)     # utterance handling: what the assistant said, as spoken
await engine.set_utterance_threshold(0.9) # utterance handling: one-sided class-1 decision threshold
await engine.clear_utterance_history()    # utterance handling: forget the dialogue history

These are routed only to the SAA agent (destination_identities=[...]) so they never leak to other room participants.

Utterance handling

Opt in with start_attention_session(..., utterance_handling=True). The hosted agent then segments speech into utterances on voice activity, transcribes each one when it ends, and scores whether it was aimed at the device (prediction == 2) or at a person (prediction == 1). One UtteranceEndedEvent per utterance carries the text, the verdict and the utterance audio; UtteranceConfigEvent arrives after started and says whether the feature is on for this session.

The classifier is conditioned on the preceding dialogue and is unreliable without it, so feed back every assistant response, as spoken, with await engine.add_assistant_turn(text). assistant_turns on the event shows how many assistant lines were in the scored history; 0 means you have not fed any.

saa = await start_attention_session(..., utterance_handling=True)
engine = AttentionEngine(ctx.room, agent_identity=saa.agent_identity)

@engine.on_utterance_ended
def _(u):
    print(u.text, u.prediction, u.confidence, u.decision, u.preview)

# after each assistant response has been spoken
await engine.add_assistant_turn(assistant_transcript)

This stream is independent of TurnReadyEvent: the two will not line up one to one. Drive your LLM from one or the other. While the feature is in preview, preview is True and the verdict is preview grade; the server delivers transcripts and does not retain them.

Requirements

• Python 3.10+
• LiveKit URL must be publicly reachable from our cloud (no private VPC)
• Audio + video tracks must both be available (the model is multimodal)
• Customer voice agent and hosted attention agent share the same LiveKit room

License

Apache-2.0. © Socero Inc.