Voice agents for real-time agent assist

Voice Agent API — tier-1 caller agent with human escalation

# Voice Agent API: tier-1 caller agent that escalates to a human
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 tier-1 support agent. Triage the caller's issue. "
                    "If they ask for a supervisor, say 'cancel', or the issue is "
                    "complex, call escalate_to_human with a one-sentence summary."
                ),
                "greeting": "Hi, thanks for calling Acme — how can I help today?",
                "input": {"keyterms": ["cancel", "supervisor", "refund", "lawyer"]},
                "output": {"voice": "ivy"},
                "tools": [{
                    "type": "function",
                    "name": "escalate_to_human",
                    "description": "Warm-transfer to a human agent with full context.",
                    "parameters": {
                        "type": "object",
                        "properties": {"summary": {"type": "string"}},
                        "required": ["summary"],
                    },
                }],
            },
        }))
        async for msg in ws:
            handle(json.loads(msg))  # transcript.user, reply.audio, tool.call, ...

Universal-3 Pro Streaming — live transcript for agent desktop

# Universal-3 Pro Streaming: live transcript for an agent assist desktop
import asyncio, json, websockets
from urllib.parse import urlencode

API_KEY = "YOUR_API_KEY"

params = urlencode({
    "sample_rate": 16000,
    "speech_model": "u3-rt-pro",
    "keyterms_prompt": json.dumps([
        "cancel", "supervisor", "refund",
        "order #", "account number",
    ]),
    "format_turns": "true",
    "speaker_labels": "true",                  # diarize agent vs. caller
    "redact_pii": "true",                      # mask PII before it hits the UI
    "redact_pii_policies": json.dumps([
        "credit_card_number", "us_social_security_number",
        "date_of_birth", "phone_number", "email_address",
    ]),
    "redact_pii_sub": "entity_name",           # e.g. [CREDIT_CARD_NUMBER]
})

async def stream_agent_assist(audio_iter):
    url = f"wss://streaming.assemblyai.com/v3/ws?{params}"
    async with websockets.connect(
        url, additional_headers={"Authorization": API_KEY},
    ) as ws:
        async def send_audio():
            async for chunk in audio_iter:
                await ws.send(chunk)
        asyncio.create_task(send_audio())
        async for raw in ws:
            evt = json.loads(raw)
            if evt.get("type") == "Turn" and evt.get("end_of_turn"):
                # finalized, PII-redacted turn — pipe to LLM Gateway for coaching
                push_to_coach(evt["transcript"], evt.get("speaker_label"))