Voice Agent API — recommended
# Voice Agent API — single WebSocket, full healthcare pipeline
import asyncio, json, websockets
API_KEY = "YOUR_API_KEY"
async def run_agent():
async with websockets.connect(
"wss://agents.assemblyai.com/v1/ws",
additional_headers={"Authorization": f"Bearer {API_KEY}"},
) as ws:
await ws.send(json.dumps({
"type": "session.update",
"session": {
"system_prompt": (
"You are a triage agent for Acme Health. "
"Collect symptoms, medications, and insurance."
),
"greeting": "Hi, this is Acme Health — what brings you in today?",
"output": {"voice": "ivy"},
},
}))
async for msg in ws:
handle(json.loads(msg)) # session.ready, transcript.user, reply.audio, tool.call, ...
Universal-3.5 Pro Realtime + LiveKit — BYO stack
# LiveKit + AssemblyAI Universal-3.5 Pro Realtime — healthcare voice agent
from livekit.agents import Agent, AgentSession, TurnHandlingOptions
from livekit.plugins import assemblyai, cartesia, openai, silero
class IntakeAgent(Agent):
def __init__(self):
super().__init__(
instructions=(
"You are a patient intake agent for Acme Health. "
"Collect symptoms, current medications, allergies, and insurance."
),
)
async def entrypoint(ctx):
session = AgentSession(
stt=assemblyai.STT(
model="u3-rt-pro",
min_turn_silence=100,
max_turn_silence=1000, # patients pause longer; bump higher for entity dictation
vad_threshold=0.3,
keyterms_prompt=["amoxicillin", "metformin", "ICD-10", "PCP referral"],
),
llm=openai.LLM(model="gpt-4o"),
tts=cartesia.TTS(),
vad=silero.VAD.load(activation_threshold=0.3), # match AssemblyAI's vad_threshold
turn_handling=TurnHandlingOptions(
turn_detection="stt",
endpointing={"min_delay": 0}, # AssemblyAI handles endpointing; avoid additive delay
),
)
await session.start(room=ctx.room, agent=IntakeAgent())