Guidesv1

LiveKit Agents

Add voice biomarker screening to a LiveKit Agents worker in about a dozen lines. You subscribe to the patient's audio track, resample it to 16kHz mono, and forward the frames to Voxtar's streaming endpoint — your agent's conversation logic stays exactly as it is.

This follows the pattern in Add a diagnostic layer. Prerequisites: a running LiveKit Agents worker (Python), pip install websockets, and a VOXTAR_API_KEY.

1 · Attach Voxtar to a track

A small helper opens one Voxtar stream per call, pumps events in the background, and forwards resampled audio frames. LiveKit's rtc.AudioStream can resample for you — ask it for 16kHz mono to match the API.

import asyncio, json, os, websockets
from livekit import rtc
VOX_URL = "wss://api.voxtar.hallelx2.com/v1/analyze/stream"
async def attach_voxtar(track: rtc.Track, patient_id: str):
    vox = await websockets.connect(VOX_URL)
    await vox.send(json.dumps({
        "type": "start",
        "api_key": os.environ["VOXTAR_API_KEY"],
        "patient_id": patient_id,
        "encoding": "pcm_s16le",
        "sample_rate": 16000,
        "panels": ["respiratory", "neurological"],
        "write_to_ehr": True,
    }))
    async def pump_events():
        async for msg in vox:
            evt = json.loads(msg)
            if evt.get("type") == "complete" and evt.get("flags"):
                await notify_care_team(patient_id, evt["flags"])
    asyncio.create_task(pump_events())
    # Resample the patient's track to 16kHz mono and forward raw PCM frames
    stream = rtc.AudioStream(track, sample_rate=16000, num_channels=1)
    async for event in stream:
        await vox.send(bytes(event.frame.data))
    await vox.send(json.dumps({"type": "stop"}))

2 · Wire it into your worker

In your agent entrypoint, start the screening when the patient's audio track is subscribed. Everything else about your agent is unchanged.

from livekit import rtc
from livekit.agents import JobContext
async def entrypoint(ctx: JobContext):
    await ctx.connect()
    @ctx.room.on("track_subscribed")
    def on_track(track, publication, participant):
        is_audio = track.kind == rtc.TrackKind.KIND_AUDIO
        if is_audio and participant.identity.startswith("patient"):
            asyncio.create_task(
                attach_voxtar(track, patient_id=participant.identity)
            )
    # ... your existing AgentSession / dialogue logic runs as-is ...

3 · Act on the result

attach_voxtar already forwards flags to notify_care_team; you also get a persisted analysis you can fetch later via GET /analyses/{id}, and — with write_to_ehr: true — a FHIR Observation in the patient record. See FHIR write-back.

Notes

Only attach to the patient participant's track, never your agent's published audio. If a patient spans multiple calls, reuse the same patient_id so drift is scored against their baseline. The stream is best-effort and additive — if Voxtar is unreachable, your agent is unaffected.