Python SDK for Attention Labs real-time selective auditory attention.
Every voice pipeline has the same problem: the microphone hears everything, but your ASR should only process speech directed at the device. Wake words solve this with a rigid trigger phrase. SAA solves it without one, classifying every audio frame as silent, human-directed, or device-directed and routing only what matters.
attenlabs-saa is a thin Apache-2.0 client: it captures and encodes your mic and webcam locally and streams them to the hosted SAA inference server over WebSocket, where the addressee model runs. It emits typed events back, attention predictions, voice activity, conversation state, and ready-to-forward speech audio. The pipeline is audio in, addressee gate, only addressed audio out. It is model-agnostic and drop-in: LLM routing is left to you.
Get your API key at attentionlabs.ai.
You need your API key for this project to work
pip install attenlabs-saaRequires Python 3.10+. sounddevice and opencv-python are pulled in automatically for mic and camera access.
import time
from saa import AttentionClient
client = AttentionClient(token="your-token")
@client.on_prediction
def _(event):
label = {0: "silent", 1: "human", 2: "device"}.get(event.cls, "?")
print(f"{label} {event.confidence:.0%} faces={event.num_faces} src={event.source}")
@client.on_turn_ready
def _(turn):
# turn.audio_base64, base64 PCM16 @ 16 kHz mono, ready for OpenAI Realtime / any LLM
# turn.audio_pcm16, same audio as np.int16 array
print(f"turn ready ({turn.duration_sec:.2f}s)")
@client.on_error
def _(event):
print(f"ERROR: {event.title}: {event.message}")
client.start()
try:
while True:
time.sleep(0.1)
except KeyboardInterrupt:
client.stop()Run client.start() and you'll see live silent / human / device predictions stream from your own mic in seconds — all you need is your API key (the token= argument above). A full CLI demo wiring SAA + OpenAI Realtime lives at saa-py-demo.
from saa import AttentionClient, CameraConfig, MicConfig
client = AttentionClient(
token="...", # API key, sent as WS subprotocol
url=None, # Server URL (default: https://broker.attentionlabs.ai)
video=CameraConfig(), # Webcam config
audio=MicConfig(), # Mic config
initial_threshold=0.7, # Device-class confidence threshold (0..1)
enable_audio=True, # Set False to skip mic capture
enable_video=True, # Set False to skip webcam capture
server_profile=None, # Override server profile; auto "audio_only" when enable_video=False
auto_reconnect=True, # Auto-reconnect with backoff after a retriable drop
)| field | type | default | notes |
|---|---|---|---|
device |
int | str | None |
None |
Device index, name, or None for system default |
channels |
int |
1 |
Number of input channels |
| field | type | default | notes |
|---|---|---|---|
device_index |
int |
0 |
Webcam device index |
width |
int |
1920 |
Capture width |
height |
int |
1080 |
Capture height |
jpeg_quality |
int |
50 |
JPEG compression quality 0-100 |
| method | description |
|---|---|
start() |
Probes the camera, opens the WebSocket, acquires mic + camera, starts capture threads. Non-blocking. Raises on handshake failure. |
stop() |
Tears down capture, joins threads, closes WebSocket. |
mute() |
Signals the server to stop feeding your speech to the turn/VAD pipeline |
unmute() |
Resumes server-side turn/VAD processing. |
mark_responding(bool) |
Tell the server an LLM response is in flight. Server stops emitting predictions while True. |
set_threshold(value: float) |
Update device-class confidence threshold (0..1). Server acks via config event. |
feed_audio(audio, *, sample_rate=16000) |
Stream audio captured by another stack instead of the SDK's own mic. Requires enable_audio=False. See Feeding external audio and video. |
feed_video(frame) |
Push an externally-captured frame instead of the SDK's own camera. Requires enable_video=False. Accepts pre-encoded JPEG bytes or a raw np.ndarray (BGR, JPEG-encoded internally). See Feeding external audio and video. |
When enable_video=True but the camera can't be opened (missing device, or one held by another app), start() continues audio-only if enable_audio=True. The original enable_video request is restored on the next start(), so a later session retries video.
When another stack already owns the microphone, an ElevenLabs / OpenAI Realtime AudioInterface tap, a Twilio media stream, a game engine, construct the client with enable_audio=False and push frames in with feed_audio() instead of letting the SDK open its own mic:
client = AttentionClient(token="...", enable_audio=False, enable_video=False)
client.start() # opens the WebSocket; captures nothing itself
# in your existing audio callback (any chunk size, mono):
client.feed_audio(pcm_chunk) # bytes (int16 LE), np.int16, or np.float32 [-1, 1]feed_audio accepts arbitrary chunk sizes and re-chunks internally to the wire's 100 ms blocks; pass sample_rate= if your audio isn't already 16 kHz and it'll resample. Calling it while enable_audio=True raises (that would double the audio source). A runnable ElevenLabs Conversational AI example lives in saa-sdk/examples/elevenlabs.
Frames work the same way, construct with enable_video=False and push with feed_video():
client = AttentionClient(token="...", enable_video=False)
client.start()
# in your frame callback, pre-encoded JPEG bytes, or a raw BGR np.ndarray:
client.feed_video(jpeg_bytes) # bytes / bytearray / memoryview, sent as-is
client.feed_video(bgr_frame) # np.ndarray, JPEG-encoded at CameraConfig.jpeg_qualityCalling feed_video while enable_video=True raises. Mix freely: enable_video=False with the internal mic gives audio-only with external frames, or enable_audio=False + feed_audio() while the SDK still grabs camera frames.
Register handlers with decorators. All callbacks fire on internal threads, keep them fast or hand work off to your own thread.
@client.on_prediction
def handle(event):
...| decorator | payload | fires when |
|---|---|---|
@on_connected |
none | WebSocket opens |
@on_started |
none | Server has loaded the model |
@on_warmup_complete |
none | Model warmed up and producing predictions |
@on_prediction |
PredictionEvent |
Each attention prediction |
@on_vad |
VadEvent |
Voice activity update |
@on_state |
StateEvent |
Conversation state transition |
@on_turn_ready |
TurnReadyEvent |
Complete user turn ready to forward |
@on_config |
ConfigEvent |
Server acks a threshold change |
@on_stats |
StatsEvent |
Every ~10s with connection health |
@on_interrupt |
InterruptEvent |
User is barging in mid-LLM-response |
@on_interjection |
InterjectionEvent |
Proactive AI volunteer after humans go quiet |
@on_error |
AttentionErrorEvent |
Connection, auth, or server error |
@on_disconnected |
DisconnectedEvent |
WebSocket closes |
@on_reconnecting |
ReconnectingEvent |
Before each auto-reconnect attempt |
@on_reconnected |
ReconnectedEvent |
Auto-reconnect succeeded |
cls: int # 0 = silent, 1 = human-directed, 2 = device-directed
confidence: float # 0..1
source: str # "model" | "rules" | "ai_responding"
num_faces: int # faces detected in frame
responding: bool # True while the AI is mid-playbackprobability: float # VAD probability 0..1
is_speech: bool # whether speech was detectedstate: ConversationState # "listening" | "sending" | "cancelled" | "idle"audio_pcm16: np.ndarray # int16 array @ 16 kHz mono
audio_base64: str # same audio as base64, ready for OpenAI Realtime, etc.
duration_sec: float # duration in seconds
frames: list[TurnFrame] # JPEG stills, empty unless the server has frames_per_turn > 0
context: str | None # e.g. "interjection_follow_up"; None for normal turnsmodel_class2_threshold: float # server-confirmed thresholdrtt_ms: float | None # round-trip latency in ms
sent_video: int # total video frames sent
skipped_video: int # total video frames skipped
sent_audio: int # total audio chunks sent
uptime_s: float # connection uptime in secondsfade_ms: int # suggested fade duration (ms) before stopping playback
confidence: float # raw model confidence of the class-2 prediction that firedFires when the server detects the user trying to take the turn back while
the LLM is mid-response. The server has already moved its state machine to
listening and pre-rolled the user's recent audio into the next turn, the
following turn_ready event will carry the actual barge-in question. The
consumer's job is to (a) fade and stop its local LLM playback over
fade_ms, (b) cancel any in-flight LLM response, and (c) re-open the mic
immediately (do not wait for the fade to finish, or the user's continued
speech is dropped for the duration of the fade).
reason: str # why the volunteer fired
audio_pcm16: np.ndarray # int16 array @ 16 kHz mono (recent conversation audio)
audio_base64: str # same audio as base64
duration_sec: float # duration in secondsFires when humans chat and then go quiet, so the agent can volunteer a brief
check-in. Hand audio_base64 to your LLM as context for the volunteer prompt.
title: str # error category ("Auth Failed", "Connection Stalled", etc.)
message: str # human-readable message
detail: str | None = None # technical detail
code: int | None = None # WebSocket close code, if applicable
kind: str | None = None # transport | auth | rate_limit | audio | server | environment
retriable: bool = False # True if the SDK will (or you could) retrycode: int # WebSocket close code
reason: str # close reason
was_clean: bool # True if code == 1000attempt: int # 1-based attempt counter
delay_s: float # backoff delay before this attempt
last_code: int # close code that triggered the reconnectattempts: int # attempts it took to reconnectLLM routing is intentionally not part of the SDK. The turn_ready event hands you PCM16 audio, both as a NumPy array and as base64, forward it wherever you like.
When your LLM starts generating, call mute() + mark_responding(True) to suppress predictions during playback. When it finishes, unmute() + mark_responding(False).
from saa import AttentionClient
client = AttentionClient(token="...")
@client.on_turn_ready
def _(turn):
# Forward to your LLM of choice
your_llm.send(turn.audio_base64)
def on_llm_speaking():
client.mute()
client.mark_responding(True)
def on_llm_done():
client.unmute()
client.mark_responding(False)When the server detects the user trying to take the turn back while the
LLM is speaking, it fires interrupt. Wire it to a fade-and-cancel on
your LLM playback layer, then re-open the mic immediately:
@client.on_interrupt
def _(event):
# Fade your local LLM audio and cancel its in-flight response.
your_llm.interrupt(event.fade_ms)
# Re-open the mic immediately, do NOT wait for the fade to finish,
# or the user's continued speech is dropped for the fade duration.
client.unmute()
client.mark_responding(False)The server has already moved its state machine to listening and
pre-rolled the user's recent audio into the chunk accumulator by the time
this event arrives. The next turn_ready event will carry the user's
actual barge-in question.
See saa-py-demo for a full working example with OpenAI Realtime.
The SDK manages these threads internally:
| thread | purpose |
|---|---|
saa-ws |
WebSocket send/receive |
saa-heartbeat |
JSON pings every 5s, stats every 10s |
saa-camera-read |
Drains frames off the webcam into a latest-frame buffer |
saa-camera-send |
JPEG-encodes the latest frame and sends it at 4 fps (250 ms) |
| (sounddevice) | Audio callback at native sample rate, resampled to 16 kHz |
Camera capture (when enable_video=True) runs as two threads — a reader and a sender — so a slow encode never stalls frame acquisition. Two more threads spawn transiently: saa-reconnect runs the backoff loop while auto-reconnecting, and saa-clientlog posts the HTTP-beacon fallback for send_client_log().
All event callbacks fire on saa-ws, saa-heartbeat, or saa-reconnect (reconnect events). Don't block them, offload heavy work to your own thread.
Apache-2.0