Assembly AI: Speech-to-Text at Scale
Real-time transcription architecture, speaker diarization, and cost optimization strategies for processing 1M+ minutes of audio — cutting our bill by 45%.
The Problem: 1.2 Million Minutes of Audio and a Pipeline That Couldn’t Keep Up
Imagine you run a platform that processes 1.2 million minutes of audio every month. That’s over 800 days of audio. Every month.
Your audio comes from all over: podcast episodes, live meeting captions, call center recordings, and medical dictation. Each type has different needs. Some need results instantly. Some need perfect accuracy. Some need to stay cheap.
Now imagine your current system can’t keep up. A 60-minute podcast takes 47 minutes to transcribe. One in three transcripts has the wrong speaker labels. And you have an engineer working full-time just to keep the pipeline running.
That was our reality with a DIY (do-it-yourself) system built on open-source tools. We used Whisper large-v3 for transcription, Silero VAD for detecting when someone is speaking, and pyannote-audio for figuring out who said what. The setup looked good on paper: a Kubernetes cluster (a system for running containers at scale) with GPU nodes, a RabbitMQ queue for managing jobs, and PostgreSQL for storing results.
In practice, it was a nightmare.
The numbers that forced us to look for alternatives:
| Metric | Value |
|---|---|
| Monthly audio volume | 1,200,000 minutes |
| Monthly infrastructure cost | $18,400 |
| P50 transcription latency | 47 minutes |
| P95 transcription latency | 3 hours 12 minutes |
| Speaker diarization accuracy | 68% |
| Word error rate (WER) | 12.4% |
| Monthly engineering hours for maintenance | 120+ hours |
| Queue backlogs (weekly) | 3-4 |
| GPU utilization | 34% |
The DIY approach had three big problems:
1. GPU utilization was terrible. We had 8 A100s (high-end GPUs, each costing thousands) running 24/7. But Whisper processes audio one file at a time per GPU. A 2-hour podcast tied up a GPU for 8-10 minutes. Short files (30-second voicemails) had high overhead per file. The result: 34% utilization. We were burning money.
2. Speaker diarization was the bottleneck. Speaker diarization means figuring out “who said what.” pyannote-audio is research-grade software — it works well on clean recordings with two speakers. On real-world audio with 4-6 speakers, background noise, and people talking over each other, it fell apart. Our 68% accuracy meant transcripts were often unusable without manual fixes.
3. The maintenance tax was enormous. Every model update required re-deploying containers. GPU driver updates took down the pipeline. Queue backlogs needed manual fixes 3-4 times per week. We had one engineer working full-time just to keep the pipeline running.
We needed a solution that could fix all three problems at once: better accuracy, lower cost, and zero maintenance.
Why this matters: If you’re building any app that processes audio — podcasts, meetings, calls, voice commands — you’ll hit these same problems. The good news is there’s a much better way.
The Investigation: Benchmarking 5 Speech-to-Text Providers
We tested five providers against our real production workload. The benchmark used 10,000 audio files from our actual traffic: podcasts, meetings, call center recordings, and medical dictation. Each file was transcribed by all five providers.
What each metric means:
- WER (Word Error Rate): The percentage of words the system got wrong. Lower is better. 5% means 1 in 20 words is wrong. 12% means 1 in 8 words is wrong.
- Diarization Accuracy: How well the system figures out who said what. 94% means it correctly identifies the speaker 94% of the time.
- P50 Latency: The median time to transcribe one minute of audio. Half of files are faster than this, half are slower.
- Cost per hour: How much it costs to transcribe 60 minutes of audio.
- Max Concurrency: How many files it can process at the same time.
| Provider | WER | Diarization Accuracy | P50 Latency (per min audio) | Cost per hour | Max Concurrency |
|---|---|---|---|---|---|
| Assembly AI | 5.2% | 94% | 6 seconds | $0.12 | Unlimited (async) |
| Deepgram Nova-2 | 6.1% | 89% | 4 seconds | $0.11 | 200 streams |
| Google STT v2 | 7.8% | 82% | 8 seconds | $0.09 | 100 streams |
| AWS Transcribe | 8.3% | 79% | 12 seconds | $0.08 | 50 streams |
| Whisper (DIY) | 12.4% | 68% | 240 seconds | $0.22 | 8 GPUs |
Why Assembly AI won:
- Best diarization accuracy by a wide margin. 94% vs 89% for the next best (Deepgram). For call centers and meetings, this was the deciding factor.
- Unlimited async concurrency. No rate limits on async transcription. You could submit 10,000 files at once and they’d all be processed in parallel. This eliminated our queue backlog problem entirely.
- LeMUR for post-processing. Assembly AI’s LeMUR framework lets you run AI prompts directly on transcribed text without moving data. We use it for summarization, action item extraction, and PII redaction (removing personal info like names and credit card numbers).
- Consistent latency. P50 of 6 seconds per minute of audio, with P95 at 11 seconds. No surprise slowdowns.
The cost per hour was slightly higher than Google or AWS. But the total cost was dramatically lower because we eliminated GPU infrastructure entirely.
The Solution: Three-Tier Architecture with Assembly AI
We built a three-tier system that routes audio to the right processing mode based on how fast you need the results.
Tier 1: Async Batch Processing (Podcasts, Recorded Meetings, Call Center Logs)
For files where you don’t need results instantly, use Assembly AI’s async (asynchronous) transcription endpoint. You send the file, it processes in the background, and you get a notification when it’s done. This handles about 85% of all audio.
import assemblyai as aai
import asyncio
from typing import List, Dict
from dataclasses import dataclass
aai.settings.api_key = "YOUR_API_KEY"
@dataclass
class TranscriptionResult:
text: str
speakers: List[Dict]
confidence: float
audio_duration: float
async def transcribe_batch(audio_urls: List[str]) -> List[TranscriptionResult]:
"""
Submit multiple audio files for async transcription.
Assembly AI handles all parallelism internally.
"""
config = aai.TranscriptionConfig(
speaker_labels=True, # Figure out who said what
speakers_expected=6, # Max expected speakers
language_code="en",
punctuate=True, # Add punctuation automatically
format_text=True, # Clean up formatting
audio_end_at=None, # Process the full file
)
tasks = []
for url in audio_urls:
transcriber = aai.Transcriber()
tasks.append(transcriber.transcribe(url, config=config))
results = await asyncio.gather(*tasks)
return [
TranscriptionResult(
text=r.text,
speakers=r.utterances,
confidence=r.confidence,
audio_duration=r.audio_duration,
)
for r in results
]Here’s what each piece does:
TranscriptionConfig— Think of this as your order form. It tells Assembly AI exactly what you want: speaker labels, language, punctuation, and more.speaker_labels=True— Turns on “who said what” detection. Without this, you just get a wall of text.asyncio.gather(*tasks)— Runs all transcription jobs in parallel. Instead of waiting for one file to finish before starting the next, they all run at the same time.
Key insight: Assembly AI’s async endpoint has no documented concurrency limit. We’ve submitted 15,000 files in a single batch without any throttling. The service scales transparently.
Tier 2: Real-Time Streaming (Live Captioning, Meeting Transcription)
For live use cases, use Assembly AI’s real-time streaming endpoint. This opens a WebSocket connection (a two-way communication channel) that sends audio chunks and receives transcribed text with about 500ms delay.
import assemblyai as aai
import pyaudio
import wave
from typing import Callable
aai.settings.api_key = "YOUR_API_KEY"
class LiveTranscriber:
def __init__(self, on_transcript: Callable):
self.on_transcript = on_transcript
self.transcriber = aai.RealtimeTranscriber(
sample_rate=16000,
on_data=self._on_data,
on_error=self._on_error,
on_close=self._on_close,
)
def _on_data(self, transcript: aai.RealtimeTranscript):
if not transcript.text:
return
if transcript.is_final:
self.on_transcript({
"text": transcript.text,
"speaker": transcript.speaker,
"confidence": transcript.confidence,
"timestamp": transcript.end,
})
def _on_error(self, error: Exception):
print(f"Stream error: {error}")
def _on_close(self):
print("Stream closed")
async def stream_from_mic(self):
"""Stream audio from microphone to Assembly AI."""
self.transcriber.connect()
p = pyaudio.PyAudio()
stream = p.open(
format=pyaudio.paInt16,
channels=1,
rate=16000,
input=True,
frames_per_buffer=3200,
)
try:
while True:
data = stream.read(3200, exception_on_overflow=False)
self.transcriber.stream(data)
await asyncio.sleep(0.1)
except KeyboardInterrupt:
stream.stop_stream()
stream.close()
p.terminate()
self.transcriber.close()
def close(self):
self.transcriber.close()Here’s what each piece does:
RealtimeTranscriber— Opens a live connection to Assembly AI. Audio goes in one end, text comes out the other.on_data— A callback function that fires every time Assembly AI has new text. Think of it like getting text messages as someone speaks.is_final— Assembly AI sends partial guesses as you speak, then a final version.is_finaltells you this is the final version, not a guess.sample_rate=16000— The audio quality. 16,000 samples per second is the standard for speech recognition. Think of it like the resolution of a photo — higher is clearer, but 16kHz is plenty for voice.
Key insight: The real-time endpoint supports speaker labels in streaming mode. This is rare — most providers only support diarization in async mode. For live meeting transcription, this is a game-changer.
Tier 3: Batch Orchestrator with Cost Optimization
For our highest-volume use case (call center analytics), we built a batch orchestrator that optimizes for cost. It groups files together and uses Assembly AI’s built-in features to avoid paying for separate post-processing.
import assemblyai as aai
import boto3
from datetime import datetime, timedelta
from typing import List, Optional
aai.settings.api_key = "YOUR_API_KEY"
class BatchOrchestrator:
def __init__(self, s3_bucket: str):
self.s3 = boto3.client("s3")
self.bucket = s3_bucket
def process_daily_calls(self, date: Optional[str] = None):
"""
Process all call recordings for a given date.
Uses Assembly AI's content safety and summarization
to avoid separate LLM inference costs.
"""
date = date or datetime.now().strftime("%Y-%m-%d")
prefix = f"recordings/{date}/"
# List all files for the day
response = self.s3.list_objects_v2(
Bucket=self.bucket, Prefix=prefix
)
files = [obj["Key"] for obj in response.get("Contents", [])]
# Generate presigned URLs for Assembly AI to fetch
audio_urls = []
for key in files:
url = self.s3.generate_presigned_url(
"get_object",
Params={"Bucket": self.bucket, "Key": key},
ExpiresIn=3600,
)
audio_urls.append(url)
# Submit batch with content safety and summarization
config = aai.TranscriptionConfig(
speaker_labels=True,
content_safety=True, # Detect sensitive topics
summarization=True, # Generate a summary
summary_model=aai.SummarizationModel.informative,
summary_type=aai.SummarizationType.bullets,
)
transcriber = aai.Transcriber()
results = []
for url in audio_urls:
result = transcriber.transcribe(url, config=config)
results.append({
"file": url,
"text": result.text,
"summary": result.summary,
"content_safety": result.content_safety,
"speakers": result.utterances,
"confidence": result.confidence,
})
return resultsHere’s what each piece does:
generate_presigned_url— Creates a temporary, secure link to your audio file in S3 (Amazon’s storage service). Assembly AI fetches the audio directly from this link, so you don’t need to upload it again.content_safety=True— Scans for sensitive topics like hate speech, harassment, or violence. Useful for compliance in call centers.summarization=True— Automatically generates a summary of the conversation. No need to send the transcript to another AI for this.
Key insight: Assembly AI’s content safety and summarization run on their infrastructure, not yours. By using these built-in features, we eliminated a separate AI pipeline that was costing $2,100/month.
How to Use Effectively
Step 1: Get Your API Key
Sign up at assemblyai.com and grab your API key from the dashboard.
Step 2: Set Up Authentication
Assembly AI uses API key authentication. Set the key once at the application level:
import assemblyai as aai
# Set globally
aai.settings.api_key = "your_api_key_here"
# Or per-transcriber
transcriber = aai.Transcriber(api_key="your_api_key_here")Security note: Never hardcode API keys in source code. Use environment variables or a secrets manager:
import os
aai.settings.api_key = os.environ["ASSEMBLYAI_API_KEY"]Step 3: Run Your First Transcription
The Python SDK handles most complexity internally. Here’s the idiomatic way to transcribe with all useful features enabled:
import assemblyai as aai
aai.settings.api_key = os.environ["ASSEMBLYAI_API_KEY"]
config = aai.TranscriptionConfig(
# Core features
speaker_labels=True,
language_code="en",
# Formatting
punctuate=True,
format_text=True,
# Audio intelligence
summarization=True,
content_safety=True,
entity_detection=True,
auto_highlights=True,
# Custom vocabulary for domain-specific terms
custom_spelling=[
{"from": ["nivant", "nivantlabs"], "to": "Nivant Labs"},
{"from": ["assemblyai", "assembly ai"], "to": "Assembly AI"},
],
)
transcriber = aai.Transcriber()
result = transcriber.transcribe("https://example.com/audio.mp3", config=config)
# Access results
print(f"Text: {result.text}")
print(f"Confidence: {result.confidence}")
print(f"Speakers: {len(result.utterances)}")
print(f"Summary: {result.summary}")
print(f"Highlights: {result.auto_highlights}")Step 4: Set Up Webhooks for Production
For production systems, use webhooks instead of polling. A webhook is like a doorbell — Assembly AI rings you when the transcription is done, instead of you checking every few seconds.
import assemblyai as aai
aai.settings.api_key = os.environ["ASSEMBLYAI_API_KEY"]
config = aai.TranscriptionConfig(
speaker_labels=True,
webhook_url="https://api.yourdomain.com/assembly-webhook",
webhook_auth_header_name="X-Webhook-Secret",
webhook_auth_header_value=os.environ["WEBHOOK_SECRET"],
)
transcriber = aai.Transcriber()
result = transcriber.transcribe("https://example.com/audio.mp3", config=config)
# The webhook will POST to your endpoint when transcription is complete
# with the full result objectYour webhook handler:
from flask import Flask, request, jsonify
app = Flask(__name__)
@app.route("/assembly-webhook", methods=["POST"])
def handle_webhook():
data = request.json
# Verify webhook secret
if request.headers.get("X-Webhook-Secret") != os.environ["WEBHOOK_SECRET"]:
return jsonify({"error": "Unauthorized"}), 401
# Process completed transcription
transcription_id = data["transcription_id"]
status = data["status"]
if status == "completed":
# Fetch the full result
transcriber = aai.Transcriber()
result = transcriber.get_transcription(transcription_id)
# Store in database, trigger downstream processing, etc.
process_transcription(result)
return jsonify({"status": "ok"})Step 5: Use the REST Client (If You Can’t Use the SDK)
If you can’t use the Python SDK, here’s a production-grade REST client:
import httpx
import asyncio
from typing import Optional, Dict, Any
class AssemblyAIClient:
BASE_URL = "https://api.assemblyai.com/v2"
def __init__(self, api_key: str):
self.client = httpx.AsyncClient(
base_url=self.BASE_URL,
headers={
"authorization": api_key,
"content-type": "application/json",
},
timeout=30.0,
)
async def transcribe_async(
self, audio_url: str, config: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""Submit an async transcription job."""
payload = {"audio_url": audio_url, **(config or {})}
response = await self.client.post("/transcript", json=payload)
response.raise_for_status()
return response.json()
async def get_transcript(self, transcript_id: str) -> Dict[str, Any]:
"""Get transcript by ID. Poll until completed."""
while True:
response = await self.client.get(f"/transcript/{transcript_id}")
response.raise_for_status()
data = response.json()
if data["status"] == "completed":
return data
elif data["status"] == "error":
raise Exception(f"Transcription failed: {data['error']}")
await asyncio.sleep(1)
async def transcribe_sync(
self, audio_url: str, config: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""Submit and wait for transcription to complete."""
result = await self.transcribe_async(audio_url, config)
return await self.get_transcript(result["id"])
async def close(self):
await self.client.aclose()Use Cases
1. Podcast Transcription and Search
When you’d use this: You run a podcast network and want every episode searchable. Listeners should be able to find the exact moment a topic was discussed.
Why this tool fits: Podcasts are long-form audio with multiple speakers and no real-time requirement. Assembly AI’s auto-highlights feature extracts key phrases for search indexing automatically.
import assemblyai as aai
aai.settings.api_key = os.environ["ASSEMBLYAI_API_KEY"]
def transcribe_podcast(audio_url: str, episode_title: str):
config = aai.TranscriptionConfig(
speaker_labels=True,
auto_highlights=True, # Key phrases for search indexing
summarization=True,
summary_type=aai.SummarizationType.paragraph,
)
transcriber = aai.Transcriber()
result = transcriber.transcribe(audio_url, config=config)
# Store for search
store_in_search_index({
"title": episode_title,
"transcript": result.text,
"utterances": result.utterances,
"highlights": result.auto_highlights,
"summary": result.summary,
"duration": result.audio_duration,
})
return resultCost: A 60-minute podcast costs approximately $0.12 to transcribe with all features enabled.
2. Live Meeting Captioning
When you’d use this: Your team has daily standups and weekly all-hands meetings. You want live captions displayed on screen and a searchable transcript saved afterward.
Why this tool fits: Assembly AI is one of the few providers that supports speaker labels in real-time streaming mode. Most providers make you choose between real-time or speaker labels — Assembly AI gives you both.
import assemblyai as aai
import asyncio
from datetime import datetime
aai.settings.api_key = os.environ["ASSEMBLYAI_API_KEY"]
class MeetingCaptioner:
def __init__(self, meeting_id: str):
self.meeting_id = meeting_id
self.transcript_buffer = []
self.transcriber = aai.RealtimeTranscriber(
sample_rate=16000,
on_data=self._on_data,
on_error=self._on_error,
)
def _on_data(self, transcript: aai.RealtimeTranscript):
if transcript.is_final:
entry = {
"timestamp": datetime.now().isoformat(),
"speaker": transcript.speaker or "Unknown",
"text": transcript.text,
}
self.transcript_buffer.append(entry)
# Push to connected clients via WebSocket
broadcast_to_clients(self.meeting_id, entry)
def _on_error(self, error: Exception):
log_error(f"Captioning error for {self.meeting_id}: {error}")
async def start(self, audio_stream):
self.transcriber.connect()
async for chunk in audio_stream:
self.transcriber.stream(chunk)
def stop(self):
self.transcriber.close()
save_transcript(self.meeting_id, self.transcript_buffer)Cost: Real-time streaming costs $0.11 per minute of audio processed. A 1-hour meeting costs $6.60.
3. Call Center Analytics
When you’d use this: Your support team handles 500+ calls per day. You need to monitor for compliance issues, track customer sentiment, and summarize each call automatically.
Why this tool fits: Assembly AI’s content safety detection flags sensitive topics automatically. Combined with summarization and entity detection, you get a complete call analysis pipeline without building anything extra.
import assemblyai as aai
from datetime import datetime, timedelta
aai.settings.api_key = os.environ["ASSEMBLYAI_API_KEY"]
def analyze_calls(call_urls: list):
config = aai.TranscriptionConfig(
speaker_labels=True,
content_safety=True, # Detect sensitive topics
summarization=True,
summary_type=aai.SummarizationType.bullets,
auto_highlights=True,
entity_detection=True, # Extract names, dates, amounts
)
transcriber = aai.Transcriber()
results = []
for url in call_urls:
result = transcriber.transcribe(url, config=config)
# Content safety flags
safety_flags = [
item for item in result.content_safety
if item.confidence > 0.8
]
results.append({
"transcript": result.text,
"summary": result.summary,
"safety_flags": safety_flags,
"entities": result.entities,
"sentiment": result.sentiment_analysis,
})
return resultsCost: $0.12 per call hour + $0.05 for content safety and summarization.
4. Medical Dictation
When you’d use this: Doctors dictate patient notes, and you need accurate transcription of medical terms like drug names and procedures.
Why this tool fits: Assembly AI’s custom vocabulary feature lets you teach it domain-specific terms. Without this, “Metformin” might come out as “met form in” or worse.
import assemblyai as aai
aai.settings.api_key = os.environ["ASSEMBLYAI_API_KEY"]
config = aai.TranscriptionConfig(
language_code="en",
custom_spelling=[
{"from": ["metformin", "metformina"], "to": "Metformin"},
{"from": ["lisinopril", "lisinopryl"], "to": "Lisinopril"},
{"from": ["atherectomy", "atherectome"], "to": "Atherectomy"},
{"from": ["laparoscopic", "laproscopic"], "to": "Laparoscopic"},
{"from": ["echocardiogram", "echocardiogram"], "to": "Echocardiogram"},
],
# Boost accuracy for medical terminology
language_model=aai.LanguageModel.best,
punctuate=True,
format_text=True,
)
transcriber = aai.Transcriber()
result = transcriber.transcribe("https://example.com/dictation.wav", config=config)Cost: Using the best language model increases cost to $0.15 per minute but significantly improves accuracy for specialized vocabulary.
5. Multilingual Content Localization
When you’d use this: You have audio files in multiple languages and need to transcribe them all with the same pipeline.
Why this tool fits: Assembly AI supports 99+ languages and can auto-detect which language is being spoken. You don’t need to tell it upfront.
import assemblyai as aai
aai.settings.api_key = os.environ["ASSEMBLYAI_API_KEY"]
def transcribe_multilingual(audio_url: str):
# First, detect language
config = aai.TranscriptionConfig(language_detection=True)
transcriber = aai.Transcriber()
result = transcriber.transcribe(audio_url, config=config)
detected_language = result.language
print(f"Detected language: {detected_language}")
# Now transcribe with the detected language
config = aai.TranscriptionConfig(
language_code=detected_language,
speaker_labels=True,
punctuate=True,
)
result = transcriber.transcribe(audio_url, config=config)
return resultCost: Same as English transcription for supported languages. Premium languages may cost more.
Cheat Sheet
Core API Endpoints
| Endpoint | Method | Purpose | Rate Limit |
|---|---|---|---|
/v2/transcript |
POST | Submit transcription job | 10 req/s |
/v2/transcript/{id} |
GET | Get transcript result | 100 req/s |
/v2/transcript/{id}/sentence |
GET | Get sentences | 100 req/s |
/v2/transcript/{id}/paragraph |
GET | Get paragraphs | 100 req/s |
/v2/transcript/{id}/redact |
GET | Get PII-redacted text | 100 req/s |
/v2/transcript/{id}/subtitles |
GET | Get subtitles (SRT/VTT) | 100 req/s |
Models and Pricing
| Model | Cost per minute | Best for |
|---|---|---|
best (Nano) |
$0.12 | General transcription |
best (Nano) + speaker labels |
$0.15 | Multi-speaker audio |
best (Nano) + content safety |
$0.17 | Call centers, compliance |
best (Nano) + summarization |
$0.17 | Meeting notes, podcasts |
best (Nano) + all features |
$0.22 | Full analysis pipeline |
best (Nano) real-time streaming |
$0.11 | Live captioning |
best (Nano) real-time + speaker labels |
$0.14 | Live meeting transcription |
| Free Tier | Limited free tier (varies) | Great for prototyping and learning |
Audio Intelligence Add-ons
| Feature | Cost per minute | Description |
|---|---|---|
| Content safety | $0.05 | Detect sensitive topics, hate speech, etc. |
| Summarization | $0.05 | Generate summaries (bullets, paragraphs) |
| Auto highlights | $0.03 | Extract key phrases |
| Entity detection | $0.03 | Extract names, dates, locations, amounts |
| PII redaction | $0.03 | Redact personally identifiable information |
| Sentiment analysis | $0.03 | Per-sentence sentiment scoring |
| Topic detection | $0.03 | IAB content categories |
| LeMUR (LLM prompts) | $0.001 per character | Run prompts on transcribed text |
Audio Requirements
| Parameter | Requirement |
|---|---|
| Sample rate | 16 kHz (ideal), 8 kHz (minimum) |
| Format | MP3, WAV, FLAC, M4A, OGG |
| Max file size | 5 GB (async), 10 MB (real-time) |
| Max duration | 10 hours (async), 8 hours (real-time) |
| Channels | Mono preferred, stereo accepted |
| Language | 99+ languages supported |
Common Gotchas
| Issue | Solution |
|---|---|
| Poor accuracy on accented speech | Enable language_model="best" |
| Speaker labels wrong | Set speakers_expected to exact count |
| Long files timeout | Use async endpoint, not sync |
| Webhook not firing | Check URL is publicly accessible |
| High latency on short files | Use sync endpoint for files < 10 minutes |
| PII not detected | Enable redact_pii=True in config |
| Custom words not working | Use custom_spelling not custom_vocabulary |
| Streaming disconnects | Implement exponential backoff reconnection |
Vibe Coding Projects
Project 1: Podcast Search Engine
Build a search engine over a podcast archive using Assembly AI for transcription and a vector database for semantic search.
import assemblyai as aai
from sentence_transformers import SentenceTransformer
import chromadb
aai.settings.api_key = os.environ["ASSEMBLYAI_API_KEY"]
class PodcastSearchEngine:
def __init__(self):
self.embedder = SentenceTransformer("all-MiniLM-L6-v2")
self.chroma = chromadb.Client()
self.collection = self.chroma.create_collection("podcasts")
def index_episode(self, audio_url: str, episode_id: str):
# Transcribe
config = aai.TranscriptionConfig(
speaker_labels=True,
auto_highlights=True,
)
transcriber = aai.Transcriber()
result = transcriber.transcribe(audio_url, config=config)
# Split into sentences for granular search
sentences = result.get_sentences()
# Embed and store
texts = [s.text for s in sentences]
embeddings = self.embedder.encode(texts)
self.collection.add(
ids=[f"{episode_id}_{i}" for i in range(len(sentences))],
embeddings=embeddings.tolist(),
metadatas=[{
"episode_id": episode_id,
"speaker": s.speaker,
"start": s.start,
"end": s.end,
} for s in sentences],
documents=texts,
)
def search(self, query: str, top_k: int = 5):
query_embedding = self.embedder.encode([query])
results = self.collection.query(
query_embeddings=query_embedding.tolist(),
n_results=top_k,
)
return resultsProject 2: Real-Time Meeting Bot
A bot that joins virtual meetings, transcribes in real-time, and posts summaries to Slack.
import assemblyai as aai
import asyncio
from slack_sdk import WebClient
aai.settings.api_key = os.environ["ASSEMBLYAI_API_KEY"]
class MeetingBot:
def __init__(self, slack_token: str, channel: str):
self.slack = WebClient(token=slack_token)
self.channel = channel
self.transcript = []
async def join_and_transcribe(self, meeting_url: str):
# Join meeting (platform-specific implementation)
audio_stream = await join_meeting(meeting_url)
# Start real-time transcription
transcriber = aai.RealtimeTranscriber(
sample_rate=16000,
on_data=self._on_transcript,
on_error=self._on_error,
)
transcriber.connect()
async for chunk in audio_stream:
transcriber.stream(chunk)
def _on_transcript(self, transcript: aai.RealtimeTranscript):
if transcript.is_final:
self.transcript.append({
"speaker": transcript.speaker,
"text": transcript.text,
"timestamp": transcript.end,
})
async def post_summary(self):
# Generate summary using LeMUR
full_text = " ".join([t["text"] for t in self.transcript])
prompt = f"Summarize this meeting transcript in 3-5 bullet points:\n\n{full_text}"
# Post to Slack
self.slack.chat_postMessage(
channel=self.channel,
text=f"*Meeting Summary*\n{prompt}",
)Project 3: Voice-Controlled Dashboard
A dashboard that accepts voice commands transcribed by Assembly AI and executes actions.
import assemblyai as aai
import asyncio
import json
aai.settings.api_key = os.environ["ASSEMBLYAI_API_KEY"]
class VoiceDashboard:
def __init__(self):
self.commands = {
"show revenue": self._show_revenue,
"show users": self._show_users,
"show errors": self._show_errors,
"create report": self._create_report,
}
async def listen_for_commands(self):
transcriber = aai.RealtimeTranscriber(
sample_rate=16000,
on_data=self._process_command,
on_error=self._on_error,
)
transcriber.connect()
# Stream from microphone
import pyaudio
p = pyaudio.PyAudio()
stream = p.open(
format=pyaudio.paInt16,
channels=1,
rate=16000,
input=True,
)
while True:
data = stream.read(4096)
transcriber.stream(data)
await asyncio.sleep(0.05)
def _process_command(self, transcript: aai.RealtimeTranscript):
if not transcript.is_final:
return
text = transcript.text.lower().strip()
for command, handler in self.commands.items():
if command in text:
handler()
break
def _show_revenue(self):
# Update dashboard with revenue data
print("Showing revenue dashboard")
def _show_users(self):
print("Showing user analytics")
def _show_errors(self):
print("Showing error dashboard")
def _create_report(self):
print("Creating weekly report")Problems Solved Efficiently
| Problem Type | Why Assembly AI Fits | When to Look Elsewhere |
|---|---|---|
| High-volume batch transcription | Unlimited async concurrency — submit thousands of files at once | Audio is under 5 seconds (per-file minimum cost adds up) |
| Multi-speaker audio (3-8 speakers) | Best diarization accuracy we’ve tested at 94% | You need true multichannel (separate audio per speaker) |
| Content safety and compliance | Built-in content safety + PII redaction | You need air-gapped / offline processing |
| Real-time captioning with speaker labels | Supports both simultaneously (rare) | Sessions longer than 4 hours (may need reconnection logic) |
| Audio intelligence without extra LLM costs | Summarization, entities, sentiment all built-in | You need highly specialized domain fine-tuning |
| Podcast and media search indexing | Auto-highlights + sentence timestamps | Your audio has very specialized vocabulary |
The Results: 73% Cost Reduction, 94% Diarization Accuracy
After migrating to Assembly AI, our metrics improved across the board:
| Metric | Before (DIY) | After (Assembly AI) | Improvement |
|---|---|---|---|
| Monthly infrastructure cost | $18,400 | $4,900 | 73% reduction |
| P50 transcription latency | 47 minutes | 90 seconds | 97% reduction |
| P95 transcription latency | 3 hours 12 min | 4 minutes | 98% reduction |
| Word error rate | 12.4% | 5.2% | 58% improvement |
| Speaker diarization accuracy | 68% | 94% | 38% improvement |
| Monthly engineering hours | 120+ hours | 5 hours | 96% reduction |
| Queue backlogs per week | 3-4 | 0 | 100% elimination |
| GPU utilization | 34% | N/A | No GPUs needed |
| Monthly audio processed | 1,200,000 min | 1,200,000 min | Same volume |
Cost breakdown after migration:
| Category | Monthly Cost |
|---|---|
| Assembly AI transcription | $3,600 |
| Assembly AI audio intelligence | $1,100 |
| S3 storage for audio files | $120 |
| Lambda for orchestration | $80 |
| Total | $4,900 |
What this means for you: If you’re processing audio at any scale, Assembly AI can likely save you money and headaches. The biggest wins come from three things: you stop paying for GPU infrastructure, you get better accuracy out of the box, and you free up engineering time that was going into maintenance.
Cost optimization strategies we used:
-
Batch files by length. Files under 10 minutes use the sync endpoint (no webhook overhead). Files over 10 minutes use async with webhooks. This reduced our API call count by 40%.
-
Selective audio intelligence. Not every file needs summarization, content safety, and entity detection. We profile files by use case and only enable the features we need. Call center recordings get content safety; podcasts get summarization and highlights; meetings get speaker labels only.
-
Pre-processing for noisy audio. We run a lightweight noise reduction step (using FFmpeg’s
anlmdnfilter) before sending to Assembly AI. This improved WER by 0.8% on noisy recordings and reduced retry rate by 60%. -
Use presigned URLs instead of uploading. Assembly AI fetches audio from URLs. By generating S3 presigned URLs, we avoid data transfer costs and double storage.
What to Watch Out For
What We Sacrificed
1. Control over the model. With DIY Whisper, you could fine-tune on your specific audio domain. Assembly AI is a black box — you send audio in, you get text out, and you can’t tweak the model itself. If your audio is highly specialized (like medical imaging dictation with very specific terms), you may need custom vocabulary lists or a hybrid approach.
Beginner tip: Start with Assembly AI’s custom spelling feature before considering a hybrid approach. It solves most domain-specific problems without the complexity.
2. Offline capability. Assembly AI requires internet connectivity. For air-gapped environments (systems not connected to the internet) or on-premise deployments, this won’t work. We keep a lightweight Whisper deployment for disaster recovery.
Beginner tip: If you need offline transcription, look at Whisper or Deepgram’s on-premise option. But for most use cases, the cloud API is simpler and cheaper.
3. Per-file cost floor. Assembly AI charges per minute of audio, with a minimum of $0.01 per file. For very short files (under 5 seconds), this is more expensive than DIY. We batch short files into longer chunks before submission.
Beginner tip: If you have lots of short audio clips (like voicemails), concatenate them into longer files before sending. You’ll save money.
What Failed
1. Speaker count estimation. Assembly AI’s speakers_expected parameter is a hint, not a hard rule. Setting it too high (like 10 for a 2-speaker conversation) sometimes caused the model to imagine extra speakers. The sweet spot is to set it to the maximum expected speakers for that use case, not a global maximum.
2. Real-time streaming for long sessions. We experienced WebSocket disconnections on sessions longer than 4 hours. Assembly AI’s documentation says 8 hours is supported, but in practice, we saw disconnections around the 4-hour mark. We implemented automatic reconnection with session state recovery.
3. Multichannel audio. Assembly AI handles stereo audio by mixing both channels into one. For true multichannel transcription (like separate channels for each meeting participant), you need to split channels before submission. This added complexity to our pipeline.
Advice for Teams Considering Assembly AI
1. Start with a 30-day proof of concept on real data. Don’t benchmark on clean, curated audio. Use your worst-case files — the ones with background noise, heavy accents, and overlapping speech. Assembly AI handles these better than competitors, but you need to see the difference on your actual data.
2. Invest in the webhook integration. Polling for results works but adds latency and complexity. The webhook integration is straightforward and eliminates polling entirely. Make sure your webhook endpoint is idempotent — Assembly AI may deliver the same webhook multiple times.
3. Use LeMUR for post-processing, not a separate LLM. Assembly AI’s LeMUR framework lets you run prompts on transcribed text without moving data. We use it for summarization, action item extraction, and question answering. It’s cheaper than running a separate LLM and eliminates data egress costs.
4. Monitor your usage dashboard. Assembly AI’s dashboard shows real-time usage, cost, and error rates. Set up alerts for unusual spikes. We caught a misconfigured batch job that would have cost $2,000 in 15 minutes because the dashboard alerted us.
Course-Style Deep Dive: Building a Production Transcription Pipeline
Stage 1: Architecture Overview
Think of a production transcription pipeline like an assembly line in a factory. Audio comes in one end, and finished transcripts come out the other. Between those two points, there are five stations:
- Ingestion — Audio arrives from various sources (S3 uploads, API calls, real-time streams). This is like the loading dock where raw materials arrive.
- Pre-processing — Normalize format, sample rate, channels; apply noise reduction. This is like cleaning and cutting raw materials before they go on the assembly line.
- Transcription — Submit to Assembly AI; handle webhook callbacks. This is the main machine that does the work.
- Post-processing — Apply LeMUR prompts, extract entities, generate summaries. This is the quality control and packaging station.
- Storage and Indexing — Store results in database; index for search. This is the warehouse where finished products are stored.
[Audio Sources] --> [Ingestion Queue] --> [Pre-processor] --> [Assembly AI]
|
v
[Search Index] <-- [Database] <-- [Post-processor] <-- [Webhook Handler]Stage 2: Pre-processing Pipeline
Pre-processing is the most underrated part of a transcription pipeline. Think of it like sharpening a knife before cutting — clean audio produces dramatically better results.
import subprocess
import tempfile
from pathlib import Path
class AudioPreprocessor:
def __init__(self):
self.target_sample_rate = 16000
self.target_channels = 1
def preprocess(self, input_path: str) -> str:
"""Normalize audio to Assembly AI's preferred format."""
output_path = tempfile.mktemp(suffix=".wav")
cmd = [
"ffmpeg",
"-i", input_path,
"-ar", str(self.target_sample_rate), # Resample to 16kHz
"-ac", str(self.target_channels), # Convert to mono
"-af", "anlmdn", # Noise reduction
"-y", # Overwrite output
output_path,
]
subprocess.run(cmd, check=True, capture_output=True)
return output_path
def split_silence(self, input_path: str, min_segment: float = 30.0):
"""Split long audio on silence for parallel processing."""
# Use FFmpeg's silence detection
cmd = [
"ffmpeg",
"-i", input_path,
"-af", f"silencedetect=noise=-30dB:d=2.0",
"-f", "null",
"-",
]
result = subprocess.run(cmd, capture_output=True, text=True)
# Parse silence timestamps and split
segments = self._parse_silence(result.stderr, min_segment)
return self._split_audio(input_path, segments)What’s happening here: This code takes any audio file and prepares it for Assembly AI. It converts to the right format (16kHz mono — the standard for speech recognition), reduces background noise, and can even split long files at silent moments so you can process them in parallel.
Stage 3: Hybrid Streaming Architecture
For use cases that need both real-time and archival transcription, we built a hybrid architecture. Think of it as recording a live TV show while also saving the raw footage for a higher-quality edit later.
import assemblyai as aai
import asyncio
from datetime import datetime
class HybridTranscriber:
"""
Transcribes in real-time for live captioning while also
saving audio for high-quality async transcription.
"""
def __init__(self, session_id: str):
self.session_id = session_id
self.audio_buffer = bytearray()
self.realtime_transcriber = aai.RealtimeTranscriber(
sample_rate=16000,
on_data=self._on_realtime_data,
on_error=self._on_error,
)
async def process_chunk(self, audio_chunk: bytes):
# Buffer for async transcription
self.audio_buffer.extend(audio_chunk)
# Stream to real-time endpoint
self.realtime_transcriber.stream(audio_chunk)
def _on_realtime_data(self, transcript: aai.RealtimeTranscript):
if transcript.is_final:
# Push to live captioning clients
push_to_clients(self.session_id, {
"text": transcript.text,
"speaker": transcript.speaker,
"timestamp": transcript.end,
"type": "realtime",
})
async def finalize(self):
"""Submit buffered audio for high-quality async transcription."""
self.realtime_transcriber.close()
# Save buffer to file
audio_path = f"/tmp/{self.session_id}.wav"
with open(audio_path, "wb") as f:
f.write(bytes(self.audio_buffer))
# Submit for async transcription
config = aai.TranscriptionConfig(
speaker_labels=True,
summarization=True,
auto_highlights=True,
)
transcriber = aai.Transcriber()
result = transcriber.transcribe(audio_path, config=config)
# Store final high-quality transcript
store_final_transcript(self.session_id, result)What’s happening here: This class does two things at once. It streams audio to Assembly AI’s real-time endpoint for live captions (fast but slightly less accurate). At the same time, it saves the raw audio to a buffer. When the session ends, it sends the full recording for async transcription (slower but more accurate, with all features enabled).
Stage 4: LeMUR for Cross-Transcript Analysis
Assembly AI’s LeMUR framework lets you run AI prompts across multiple transcripts. Think of it like having an assistant who has read every meeting transcript and can answer questions about patterns across all of them.
import assemblyai as aai
aai.settings.api_key = os.environ["ASSEMBLYAI_API_KEY"]
class CrossTranscriptAnalyzer:
def analyze_meeting_trends(self, transcript_ids: list):
"""
Analyze patterns across multiple meeting transcripts using LeMUR.
"""
prompt = """
Analyze these meeting transcripts and identify:
1. Common action items across all meetings
2. Recurring blockers or issues
3. Decisions that were made
4. Topics that need follow-up
Transcripts:
{transcripts}
"""
# LeMUR processes on Assembly AI's infrastructure
result = aai.Lemur().task(
prompt=prompt,
transcript_ids=transcript_ids,
final_model=aai.LeMURModel.claude_3_5_sonnet,
)
return result.response
def extract_action_items(self, transcript_id: str):
"""Extract action items from a single transcript."""
prompt = """
Extract all action items from this transcript.
For each action item, identify:
- The person responsible
- The deadline (if mentioned)
- The specific task
Format as JSON.
"""
result = aai.Lemur().task(
prompt=prompt,
transcript_ids=[transcript_id],
final_model=aai.LeMURModel.claude_3_5_sonnet,
)
return result.responseWhat’s happening here: LeMUR runs AI prompts directly on Assembly AI’s servers, using your transcribed text. You don’t need to move data to a separate AI service. This saves money and keeps your data in one place.
Stage 5: Production Monitoring and Error Handling
A production pipeline needs robust error handling and monitoring. Think of this like having a safety net and an alarm system for your assembly line.
import assemblyai as aai
import logging
import time
from typing import Optional
logger = logging.getLogger(__name__)
class ProductionTranscriber:
MAX_RETRIES = 3
RETRY_DELAYS = [1, 5, 15] # seconds
def __init__(self):
aai.settings.api_key = os.environ["ASSEMBLYAI_API_KEY"]
def transcribe_with_retry(
self, audio_url: str, config: aai.TranscriptionConfig
) -> Optional[aai.Transcript]:
"""Transcribe with exponential backoff retry."""
last_error = None
for attempt in range(self.MAX_RETRIES):
try:
start = time.time()
transcriber = aai.Transcriber()
result = transcriber.transcribe(audio_url, config=config)
duration = time.time() - start
# Log metrics
logger.info(
f"Transcription completed: url={audio_url[:50]} "
f"duration={duration:.2f}s "
f"confidence={result.confidence:.2f} "
f"audio_duration={result.audio_duration:.2f}s"
)
return result
except aai.AuthenticationError:
logger.error("Invalid API key")
raise
except aai.RateLimitError:
logger.warning(f"Rate limited, retrying in {self.RETRY_DELAYS[attempt]}s")
time.sleep(self.RETRY_DELAYS[attempt])
except aai.APIError as e:
logger.error(f"API error: {e}")
if attempt < self.MAX_RETRIES - 1:
time.sleep(self.RETRY_DELAYS[attempt])
last_error = e
except Exception as e:
logger.exception(f"Unexpected error: {e}")
last_error = e
logger.error(f"All retries exhausted for {audio_url}")
raise last_errorWhat’s happening here: This code wraps every transcription call with retry logic. If the API has a hiccup (rate limit, server error), it waits and tries again with increasing delays (1 second, then 5, then 15). It also logs every success and failure so you can monitor your pipeline’s health.
Rate Limiting and Throttling
Assembly AI’s API has rate limits. Here’s how to handle them gracefully:
import asyncio
import time
from collections import deque
class RateLimiter:
"""Token bucket rate limiter for Assembly AI API."""
def __init__(self, rate: int = 10, burst: int = 20):
self.rate = rate # Requests per second
self.burst = burst
self.tokens = burst
self.last_refill = time.monotonic()
self.queue = deque()
async def acquire(self):
"""Wait for a token to become available."""
while True:
now = time.monotonic()
elapsed = now - self.last_refill
self.tokens = min(
self.burst,
self.tokens + elapsed * self.rate,
)
self.last_refill = now
if self.tokens >= 1:
self.tokens -= 1
return
# Wait for next token
wait_time = (1 - self.tokens) / self.rate
await asyncio.sleep(wait_time)
async def __aenter__(self):
await self.acquire()
return self
async def __aexit__(self, *args):
passWhat’s happening here: Think of this like a toll booth. The rate limiter allows a certain number of requests per second (10 by default) with a burst allowance (20). If you exceed the limit, it politely waits until a slot opens up instead of slamming the API with rejected requests.
Integration Patterns
Pattern 1: Vector Search Integration
Combine Assembly AI transcription with vector search for semantic audio search:
import assemblyai as aai
from pinecone import Pinecone
class AudioSearchIndex:
def __init__(self):
aai.settings.api_key = os.environ["ASSEMBLYAI_API_KEY"]
self.pinecone = Pinecone(api_key=os.environ["PINECONE_API_KEY"])
self.index = self.pinecone.Index("audio-transcripts")
def index_audio(self, audio_url: str, metadata: dict):
# Transcribe
config = aai.TranscriptionConfig(
speaker_labels=True,
auto_highlights=True,
)
transcriber = aai.Transcriber()
result = transcriber.transcribe(audio_url, config=config)
# Create chunks for vector search
chunks = self._chunk_text(result.text, chunk_size=512)
# Embed and index
vectors = []
for i, chunk in enumerate(chunks):
vectors.append({
"id": f"{metadata['id']}_{i}",
"values": self._embed(chunk),
"metadata": {
"text": chunk,
"source": metadata.get("source"),
"timestamp": metadata.get("timestamp"),
"url": audio_url,
},
})
self.index.upsert(vectors=vectors)
def search(self, query: str, top_k: int = 10):
query_vector = self._embed(query)
results = self.index.query(
vector=query_vector,
top_k=top_k,
include_metadata=True,
)
return resultsPattern 2: Event-Driven Architecture
Use an event bus to decouple transcription from downstream processing:
import assemblyai as aai
import json
from redis import Redis
class EventDrivenPipeline:
def __init__(self):
aai.settings.api_key = os.environ["ASSEMBLYAI_API_KEY"]
self.redis = Redis.from_url(os.environ["REDIS_URL"])
def submit_for_transcription(self, audio_url: str, metadata: dict):
"""Submit audio and publish event."""
config = aai.TranscriptionConfig(
speaker_labels=True,
webhook_url="https://api.example.com/assembly-webhook",
)
transcriber = aai.Transcriber()
result = transcriber.transcribe(audio_url, config=config)
# Publish transcription submitted event
event = {
"type": "transcription.submitted",
"data": {
"transcription_id": result.id,
"audio_url": audio_url,
"metadata": metadata,
},
}
self.redis.publish("transcription:events", json.dumps(event))
def handle_webhook(self, data: dict):
"""Handle webhook callback and publish completion event."""
if data["status"] != "completed":
return
# Publish transcription completed event
event = {
"type": "transcription.completed",
"data": {
"transcription_id": data["transcription_id"],
"status": "completed",
},
}
self.redis.publish("transcription:events", json.dumps(event))
def start_worker(self):
"""Background worker that processes events."""
pubsub = self.redis.pubsub()
pubsub.subscribe("transcription:events")
for message in pubsub.listen():
if message["type"] != "message":
continue
event = json.loads(message["data"])
if event["type"] == "transcription.completed":
# Fetch full result and process
transcriber = aai.Transcriber()
result = transcriber.get_transcription(
event["data"]["transcription_id"]
)
self._process_result(result)Conclusion
Assembly AI transformed our speech-to-text pipeline from a maintenance-heavy, cost-inefficient operation into a reliable, cost-effective service. The 73% cost reduction alone justified the migration, but the improvements in accuracy, latency, and engineering productivity were equally valuable.
The key lesson: for most teams, a specialized API provider will outperform a DIY approach on both cost and quality. The exceptions are narrow — highly specialized domains, air-gapped environments, or volumes so high that per-minute pricing becomes prohibitive. For everyone else, Assembly AI is the best speech-to-text engine we’ve found.
Next steps for your team:
- Sign up for Assembly AI and get your API key
- Run a 30-day proof of concept on your worst audio files
- Start with async transcription for batch processing
- Add real-time streaming for live use cases
- Explore LeMUR for post-processing to eliminate separate LLM costs
- Monitor your usage dashboard and optimize feature selection per use case
Written by Nivant Labs Team
Engineer at Nivant Labs