End-to-end examples

1
import requests
2
import time
3
4
# ── Config ────────────────────────────────────────────────────
5
base_url = "https://api.assemblyai.com"
6
headers = {"authorization": "YOUR_API_KEY"}
7
8
audio_url = "https://assembly.ai/wildfires.mp3"
9
10
# ── Step 1: Transcribe with speaker labels + language detection ──
11
data = {
12
    "audio_url": audio_url,
13
    "speech_models": ["universal-3-pro", "universal-2"],
14
    "language_detection": True,
15
    "speaker_labels": True,
16
}
17
18
response = requests.post(base_url + "/v2/transcript", headers=headers, json=data)
19
response.raise_for_status()
20
transcript_id = response.json()["id"]
21
22
while True:
23
    result = requests.get(f"{base_url}/v2/transcript/{transcript_id}", headers=headers).json()
24
    if result["status"] == "completed":
25
        break
26
    elif result["status"] == "error":
27
        raise RuntimeError(f"Transcription failed: {result['error']}")
28
    time.sleep(3)
29
30
# ── Step 2: Identify speakers by name ──
31
understanding_response = requests.post(
32
    "https://llm-gateway.assemblyai.com/v1/understanding",
33
    headers=headers,
34
    json={
35
        "transcript_id": transcript_id,
36
        "speech_understanding": {
37
            "request": {
38
                "speaker_identification": {
39
                    "speaker_type": "name",
40
                    "known_values": ["Alice", "Bob"],  # Replace with actual participant names
41
                }
42
            }
43
        },
44
    },
45
)
46
understanding_response.raise_for_status()
47
identified = understanding_response.json()
48
49
# ── Step 3: Format identified transcript for the LLM ──
50
speaker_transcript = "\n".join(
51
    f"{u['speaker']}: {u['text']}" for u in identified["utterances"]
52
)
53
54
# ── Step 4: Generate meeting notes via LLM Gateway ──
55
llm_response = requests.post(
56
    "https://llm-gateway.assemblyai.com/v1/chat/completions",
57
    headers=headers,
58
    json={
59
        "model": "claude-sonnet-4-5-20250929",
60
        "messages": [
61
            {
62
                "role": "user",
63
                "content": (
64
                    "You are a meeting notes assistant. Given the transcript below, produce:\n"
65
                    "1. A concise summary (3-5 sentences)\n"
66
                    "2. Key decisions made\n"
67
                    "3. Action items with owners (use speaker labels)\n\n"
68
                    f"Transcript:\n{speaker_transcript}"
69
                ),
70
            }
71
        ],
72
        "max_tokens": 2000,
73
    },
74
)
75
llm_response.raise_for_status()
76
77
print("=== Meeting Notes ===\n")
78
print(llm_response.json()["choices"][0]["message"]["content"])

1
import requests
2
import time
3
from collections import Counter
4
5
# ── Config ────────────────────────────────────────────────────
6
base_url = "https://api.assemblyai.com"
7
headers = {"authorization": "YOUR_API_KEY"}
8
9
audio_url = "https://assembly.ai/wildfires.mp3"
10
11
# ── Step 1: Transcribe with speaker labels + sentiment analysis ──
12
data = {
13
    "audio_url": audio_url,
14
    "speech_models": ["universal-3-pro"],
15
    "speaker_labels": True,
16
    "sentiment_analysis": True,
17
}
18
19
response = requests.post(base_url + "/v2/transcript", headers=headers, json=data)
20
response.raise_for_status()
21
transcript_id = response.json()["id"]
22
23
while True:
24
    result = requests.get(f"{base_url}/v2/transcript/{transcript_id}", headers=headers).json()
25
    if result["status"] == "completed":
26
        break
27
    elif result["status"] == "error":
28
        raise RuntimeError(f"Transcription failed: {result['error']}")
29
    time.sleep(3)
30
31
# ── Step 2: Identify speakers by role ──
32
understanding_response = requests.post(
33
    "https://llm-gateway.assemblyai.com/v1/understanding",
34
    headers=headers,
35
    json={
36
        "transcript_id": transcript_id,
37
        "speech_understanding": {
38
            "request": {
39
                "speaker_identification": {
40
                    "speaker_type": "role",
41
                    "known_values": ["Sales Rep", "Customer"],
42
                }
43
            }
44
        },
45
    },
46
)
47
understanding_response.raise_for_status()
48
identified = understanding_response.json()
49
50
# ── Step 3: Calculate talk/listen ratio per speaker ──
51
speaker_durations = Counter()
52
for utterance in identified["utterances"]:
53
    duration_ms = utterance["end"] - utterance["start"]
54
    speaker_durations[utterance["speaker"]] += duration_ms
55
56
total_ms = sum(speaker_durations.values())
57
talk_ratios = {
58
    speaker: round(dur / total_ms * 100, 1)
59
    for speaker, dur in speaker_durations.items()
60
}
61
62
# ── Step 4: Summarize sentiment shifts ──
63
sentiment_by_speaker = {}
64
for s in result["sentiment_analysis_results"]:
65
    speaker = s.get("speaker", "Unknown")
66
    sentiment_by_speaker.setdefault(speaker, []).append(s["sentiment"])
67
68
sentiment_summary = ""
69
for speaker, sentiments in sentiment_by_speaker.items():
70
    counts = Counter(sentiments)
71
    sentiment_summary += (
72
        f"{speaker}: "
73
        f"{counts.get('POSITIVE', 0)} positive, "
74
        f"{counts.get('NEUTRAL', 0)} neutral, "
75
        f"{counts.get('NEGATIVE', 0)} negative\n"
76
    )
77
78
# ── Step 5: Format transcript and generate coaching scorecard ──
79
speaker_transcript = "\n".join(
80
    f"{u['speaker']}: {u['text']}" for u in identified["utterances"]
81
)
82
83
llm_response = requests.post(
84
    "https://llm-gateway.assemblyai.com/v1/chat/completions",
85
    headers=headers,
86
    json={
87
        "model": "claude-sonnet-4-5-20250929",
88
        "messages": [
89
            {
90
                "role": "user",
91
                "content": (
92
                    "You are a sales coaching assistant. Analyze this sales call and produce a scorecard.\n\n"
93
                    f"Talk/listen ratios: {talk_ratios}\n\n"
94
                    f"Sentiment breakdown:\n{sentiment_summary}\n"
95
                    f"Transcript:\n{speaker_transcript}\n\n"
96
                    "Produce:\n"
97
                    "1. Call summary (2-3 sentences)\n"
98
                    "2. Talk/listen ratio analysis (ideal is 40/60 for the rep)\n"
99
                    "3. Customer sentiment shifts and what caused them\n"
100
                    "4. Top 3 coaching suggestions for the sales rep"
101
                ),
102
            }
103
        ],
104
        "max_tokens": 2000,
105
    },
106
)
107
llm_response.raise_for_status()
108
109
print("=== Sales Call Scorecard ===\n")
110
print(llm_response.json()["choices"][0]["message"]["content"])

1
import requests
2
import time
3
4
# ── Config ────────────────────────────────────────────────────
5
base_url = "https://api.assemblyai.com"
6
headers = {"authorization": "YOUR_API_KEY"}
7
8
audio_url = "https://assembly.ai/wildfires.mp3"  # Replace with your clinical audio
9
10
# ── Step 1: Transcribe with Medical Mode + speaker labels + entity detection ──
11
data = {
12
    "audio_url": audio_url,
13
    "speech_models": ["universal-3-pro-medical"],
14
    "speaker_labels": True,
15
    "entity_detection": True,
16
}
17
18
response = requests.post(base_url + "/v2/transcript", headers=headers, json=data)
19
response.raise_for_status()
20
transcript_id = response.json()["id"]
21
22
while True:
23
    result = requests.get(f"{base_url}/v2/transcript/{transcript_id}", headers=headers).json()
24
    if result["status"] == "completed":
25
        break
26
    elif result["status"] == "error":
27
        raise RuntimeError(f"Transcription failed: {result['error']}")
28
    time.sleep(3)
29
30
# ── Step 2: Identify speakers by role ──
31
understanding_response = requests.post(
32
    "https://llm-gateway.assemblyai.com/v1/understanding",
33
    headers=headers,
34
    json={
35
        "transcript_id": transcript_id,
36
        "speech_understanding": {
37
            "request": {
38
                "speaker_identification": {
39
                    "speaker_type": "role",
40
                    "known_values": ["Provider", "Patient"],
41
                }
42
            }
43
        },
44
    },
45
)
46
understanding_response.raise_for_status()
47
identified = understanding_response.json()
48
49
# ── Step 3: Extract detected entities ──
50
entities = result.get("entities", [])
51
entity_summary = "\n".join(
52
    f"- {e['entity_type']}: {e['text']}" for e in entities
53
)
54
55
# ── Step 4: Format identified transcript ──
56
speaker_transcript = "\n".join(
57
    f"{u['speaker']}: {u['text']}" for u in identified["utterances"]
58
)
59
60
# ── Step 5: Generate SOAP note via LLM Gateway ──
61
llm_response = requests.post(
62
    "https://llm-gateway.assemblyai.com/v1/chat/completions",
63
    headers=headers,
64
    json={
65
        "model": "claude-sonnet-4-5-20250929",
66
        "messages": [
67
            {
68
                "role": "user",
69
                "content": (
70
                    "You are a medical scribe. Given the clinical encounter transcript and "
71
                    "detected entities below, generate a structured SOAP note.\n\n"
72
                    "Format the note with these sections:\n"
73
                    "- **Subjective**: Patient's reported symptoms and history\n"
74
                    "- **Objective**: Clinical observations and measurements\n"
75
                    "- **Assessment**: Diagnosis or clinical impression\n"
76
                    "- **Plan**: Treatment plan, prescriptions, and follow-up\n\n"
77
                    f"Detected entities:\n{entity_summary}\n\n"
78
                    f"Transcript:\n{speaker_transcript}"
79
                ),
80
            }
81
        ],
82
        "max_tokens": 2000,
83
    },
84
)
85
llm_response.raise_for_status()
86
87
print("=== SOAP Note ===\n")
88
print(llm_response.json()["choices"][0]["message"]["content"])

1
import requests
2
import time
3
4
# ── Config ────────────────────────────────────────────────────
5
base_url = "https://api.assemblyai.com"
6
headers = {"authorization": "YOUR_API_KEY"}
7
8
audio_url = "https://assembly.ai/wildfires.mp3"
9
10
# ── Step 1: Transcribe with key phrases enabled ──
11
data = {
12
    "audio_url": audio_url,
13
    "speech_models": ["universal-3-pro", "universal-2"],
14
    "language_detection": True,
15
    "auto_highlights": True,
16
}
17
18
response = requests.post(base_url + "/v2/transcript", headers=headers, json=data)
19
response.raise_for_status()
20
transcript_id = response.json()["id"]
21
22
while True:
23
    result = requests.get(f"{base_url}/v2/transcript/{transcript_id}", headers=headers).json()
24
    if result["status"] == "completed":
25
        break
26
    elif result["status"] == "error":
27
        raise RuntimeError(f"Transcription failed: {result['error']}")
28
    time.sleep(3)
29
30
# ── Step 2: Get paragraph-level content for structure ──
31
paragraphs_response = requests.get(
32
    f"{base_url}/v2/transcript/{transcript_id}/paragraphs", headers=headers
33
)
34
paragraphs = paragraphs_response.json()["paragraphs"]
35
36
# ── Step 3: Extract top key phrases ──
37
highlights = result.get("auto_highlights_result", {}).get("results", [])
38
top_phrases = sorted(highlights, key=lambda x: x["rank"], reverse=True)[:10]
39
phrases_list = ", ".join(p["text"] for p in top_phrases)
40
41
# ── Step 4: Generate blog post via LLM Gateway ──
42
transcript_text = result["text"]
43
44
llm_response = requests.post(
45
    "https://llm-gateway.assemblyai.com/v1/chat/completions",
46
    headers=headers,
47
    json={
48
        "model": "claude-sonnet-4-5-20250929",
49
        "messages": [
50
            {
51
                "role": "user",
52
                "content": (
53
                    "You are a content writer. Transform this transcript into an engaging "
54
                    "blog post.\n\n"
55
                    "Requirements:\n"
56
                    "- Write a compelling title and subtitle\n"
57
                    "- Break the content into 3-5 sections with headers\n"
58
                    "- Weave in the key phrases naturally\n"
59
                    "- Add a TL;DR at the top\n"
60
                    "- End with a call-to-action\n\n"
61
                    f"Key phrases: {phrases_list}\n\n"
62
                    f"Transcript:\n{transcript_text}"
63
                ),
64
            }
65
        ],
66
        "max_tokens": 3000,
67
    },
68
)
69
llm_response.raise_for_status()
70
71
print("=== Blog Post Draft ===\n")
72
print(llm_response.json()["choices"][0]["message"]["content"])

1
# pip install pyaudio websocket-client
2
import pyaudio
3
import websocket
4
import json
5
import threading
6
import time
7
from urllib.parse import urlencode
8
9
# ── Config ────────────────────────────────────────────────────
10
YOUR_API_KEY = "YOUR_API_KEY"
11
12
PROMPT = (
13
    "Summarize this speaker turn in one sentence, then list any "
14
    "action items mentioned.\n\nTranscript: {{turn}}"
15
)
16
17
LLM_GATEWAY_CONFIG = {
18
    "model": "claude-sonnet-4-5-20250929",
19
    "messages": [{"role": "user", "content": PROMPT}],
20
    "max_tokens": 500,
21
}
22
23
CONNECTION_PARAMS = {
24
    "sample_rate": 16000,
25
    "speech_model": "u3-rt-pro",
26
    "format_turns": True,
27
    "min_turn_silence": 560,   # Wait longer for natural meeting pauses
28
    "max_turn_silence": 2000,
29
    "llm_gateway": json.dumps(LLM_GATEWAY_CONFIG),
30
}
31
32
API_ENDPOINT = f"wss://streaming.assemblyai.com/v3/ws?{urlencode(CONNECTION_PARAMS)}"
33
34
# Audio settings
35
FRAMES_PER_BUFFER = 800
36
SAMPLE_RATE = 16000
37
stop_event = threading.Event()
38
39
def on_open(ws):
40
    print("Connected — speak into your microphone. Press Ctrl+C to stop.\n")
41
42
    def stream_audio():
43
        audio = pyaudio.PyAudio()
44
        stream = audio.open(
45
            input=True, frames_per_buffer=FRAMES_PER_BUFFER,
46
            channels=1, format=pyaudio.paInt16, rate=SAMPLE_RATE,
47
        )
48
        while not stop_event.is_set():
49
            try:
50
                data = stream.read(FRAMES_PER_BUFFER, exception_on_overflow=False)
51
                ws.send(data, websocket.ABNF.OPCODE_BINARY)
52
            except Exception:
53
                break
54
        stream.stop_stream()
55
        stream.close()
56
        audio.terminate()
57
58
    threading.Thread(target=stream_audio, daemon=True).start()
59
60
def on_message(ws, message):
61
    data = json.loads(message)
62
    msg_type = data.get("type")
63
64
    if msg_type == "Turn":
65
        transcript = data.get("transcript", "")
66
        if data.get("end_of_turn") and transcript:
67
            print(f"[Turn] {transcript}\n")
68
        elif transcript:
69
            print(f"\r  ... {transcript[-80:]}", end="", flush=True)
70
71
    elif msg_type == "LLMGatewayResponse":
72
        content = data.get("data", {}).get("choices", [{}])[0].get("message", {}).get("content", "")
73
        print(f"[Assistant] {content}\n")
74
75
    elif msg_type == "Termination":
76
        print(f"\nSession ended — {data.get('audio_duration_seconds', 0)}s of audio processed.")
77
78
def on_error(ws, error):
79
    print(f"Error: {error}")
80
    stop_event.set()
81
82
def on_close(ws, code, msg):
83
    stop_event.set()
84
85
ws_app = websocket.WebSocketApp(
86
    API_ENDPOINT,
87
    header={"Authorization": YOUR_API_KEY},
88
    on_open=on_open, on_message=on_message,
89
    on_error=on_error, on_close=on_close,
90
)
91
92
ws_thread = threading.Thread(target=ws_app.run_forever, daemon=True)
93
ws_thread.start()
94
95
try:
96
    while ws_thread.is_alive():
97
        time.sleep(0.1)
98
except KeyboardInterrupt:
99
    print("\nStopping...")
100
    stop_event.set()
101
    if ws_app.sock and ws_app.sock.connected:
102
        ws_app.send(json.dumps({"type": "Terminate"}))
103
        time.sleep(2)
104
    ws_app.close()

1
# pip install pyaudio websocket-client
2
import pyaudio
3
import websocket
4
import json
5
import threading
6
import time
7
from urllib.parse import urlencode
8
9
# ── Config ────────────────────────────────────────────────────
10
YOUR_API_KEY = "YOUR_API_KEY"
11
12
# Add domain-specific terms to boost recognition accuracy
13
KEYTERMS = ["AssemblyAI", "Universal-3 Pro", "LLM Gateway", "speech-to-text"]
14
15
CONNECTION_PARAMS = {
16
    "sample_rate": 16000,
17
    "speech_model": "u3-rt-pro",
18
    "format_turns": True,
19
    "keyterms_prompt": KEYTERMS,
20
}
21
22
API_ENDPOINT = (
23
    f"wss://streaming.assemblyai.com/v3/ws?{urlencode(CONNECTION_PARAMS, doseq=True)}"
24
)
25
26
# Audio settings
27
FRAMES_PER_BUFFER = 800
28
SAMPLE_RATE = 16000
29
stop_event = threading.Event()
30
caption_count = 0
31
32
def on_open(ws):
33
    print(f"Live captioning started — keyterms: {', '.join(KEYTERMS)}")
34
    print("Speak into your microphone. Press Ctrl+C to stop.\n")
35
    print("-" * 60)
36
37
    def stream_audio():
38
        audio = pyaudio.PyAudio()
39
        stream = audio.open(
40
            input=True, frames_per_buffer=FRAMES_PER_BUFFER,
41
            channels=1, format=pyaudio.paInt16, rate=SAMPLE_RATE,
42
        )
43
        while not stop_event.is_set():
44
            try:
45
                data = stream.read(FRAMES_PER_BUFFER, exception_on_overflow=False)
46
                ws.send(data, websocket.ABNF.OPCODE_BINARY)
47
            except Exception:
48
                break
49
        stream.stop_stream()
50
        stream.close()
51
        audio.terminate()
52
53
    threading.Thread(target=stream_audio, daemon=True).start()
54
55
def on_message(ws, message):
56
    global caption_count
57
    data = json.loads(message)
58
59
    if data.get("type") == "Turn":
60
        transcript = data.get("transcript", "")
61
        if data.get("end_of_turn") and transcript:
62
            caption_count += 1
63
            print(f"\r[{caption_count:03d}] {transcript}")
64
        elif transcript:
65
            # Show partial (live) caption
66
            print(f"\r  >> {transcript[-70:]}", end="", flush=True)
67
68
    elif data.get("type") == "Termination":
69
        duration = data.get("audio_duration_seconds", 0)
70
        print(f"\n{'=' * 60}")
71
        print(f"Session ended — {caption_count} captions, {duration}s of audio")
72
73
def on_error(ws, error):
74
    print(f"\nError: {error}")
75
    stop_event.set()
76
77
def on_close(ws, code, msg):
78
    stop_event.set()
79
80
ws_app = websocket.WebSocketApp(
81
    API_ENDPOINT,
82
    header={"Authorization": YOUR_API_KEY},
83
    on_open=on_open, on_message=on_message,
84
    on_error=on_error, on_close=on_close,
85
)
86
87
ws_thread = threading.Thread(target=ws_app.run_forever, daemon=True)
88
ws_thread.start()
89
90
try:
91
    while ws_thread.is_alive():
92
        time.sleep(0.1)
93
except KeyboardInterrupt:
94
    print("\n\nStopping...")
95
    stop_event.set()
96
    if ws_app.sock and ws_app.sock.connected:
97
        ws_app.send(json.dumps({"type": "Terminate"}))
98
        time.sleep(2)
99
    ws_app.close()