Building an AWS Serverless Pipeline for Speaker Diarization and Transcription at ~$1 per Video

┌─────────────────────────────────────────────────────────────────┐
│                          AWS Cloud                              │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  S3 (Input)  ──▶  EventBridge  ──▶  Lambda (StartPipeline)     │
│  uploads/         (Object Created)        │                     │
│                                           ▼                     │
│                                    DynamoDB (InterviewsTable)   │
│                                           │                     │
│                                           ▼                     │
│  ┌────────────────── Step Functions ─────────────────────────┐  │
│  │                                                           │  │
│  │  ExtractAudio ──▶ ChunkAudio ──▶ DiarizeChunks (Map x5) │  │
│  │       │                                    │              │  │
│  │       ▼                                    ▼              │  │
│  │  S3 (Output)    MergeSpeakers ◀────────────┘              │  │
│  │       ▲              │                                    │  │
│  │       │              ▼                                    │  │
│  │       │         SplitBySpeaker ──▶ Transcribe (Map x10)  │  │
│  │       │                                    │              │  │
│  │       │              AggregateResults ◀────┘              │  │
│  │       │                    │                              │  │
│  │       └────────── LLMAnalysis ◀────┘                     │  │
│  │                   (gpt-4o-mini)                           │  │
│  └───────────────────────────────────────────────────────────┘  │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

Secrets Manager:    $0.80/mo (OpenAI + HuggingFace, 2 secrets)
ECR:               $0.50–$1.00/mo (Docker images with ML models, 8 images)
CloudWatch Logs:    ~$0.05/mo (Step Functions + Lambda logs)
───────────────────────────────────────────
Total:              ~$1.50–$2.00/mo

[Video] → ExtractAudio → Diarize → SplitBySpeaker → Transcribe → LLMAnalysis
                           │
                    (single Lambda processing all audio)

                    ┌─ DiarizeChunk_0 ─┐
[Video] → Chunk →   ├─ DiarizeChunk_1 ─┤ → Merge → Split → Transcribe(parallel) → LLM
                    ├─ DiarizeChunk_2 ─┤
                    └─      ...       ─┘

def extract_audio(input_path: str, output_path: str) -> None:
    """Extract 16kHz mono WAV from video"""
    cmd = [
        "ffmpeg", "-i", input_path,
        "-vn",                    # No video
        "-acodec", "pcm_s16le",   # 16-bit PCM
        "-ar", "16000",           # 16kHz
        "-ac", "1",               # Mono
        "-y", output_path,
    ]
    subprocess.run(cmd, check=True)

CHUNK_DURATION = 480      # 8 minutes
OVERLAP_DURATION = 30     # 30-second overlap
 
# chunk_0: 0–510s (effective: 0–480)
# chunk_1: 450–960s (effective: 480–960)
# chunk_2: 900–1410s (effective: 960–1440)

from pyannote.audio import Pipeline
 
pipeline = Pipeline.from_pretrained(
    "pyannote/speaker-diarization-3.1",
    token=hf_token,
)
 
if torch.cuda.is_available():
    pipeline.to(torch.device("cuda"))
 
diarization = pipeline({"waveform": audio_tensor, "sample_rate": sample_rate})
 
speaker_embeddings = extract_speaker_embeddings(audio_path, segments)

from sklearn.cluster import AgglomerativeClustering
from sklearn.metrics.pairwise import cosine_similarity
 
similarity_matrix = cosine_similarity(all_embeddings)
distance_matrix = 1 - similarity_matrix
 
clustering = AgglomerativeClustering(
    n_clusters=None,
    distance_threshold=1 - 0.75,  # 75%+ similarity = same speaker
    metric="precomputed",
    linkage="average",
)
labels = clustering.fit_predict(distance_matrix)

from faster_whisper import WhisperModel
 
model = WhisperModel("medium", device="cpu", compute_type="int8")
segments, info = model.transcribe(audio_path, language="ja", beam_size=5)
text = "".join([seg.text for seg in segments])

from openai import OpenAI
 
completion = client.beta.chat.completions.parse(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": ANALYSIS_PROMPT},
        {"role": "user", "content": f"Analyze this:\n{transcript}"},
    ],
    response_format=AnalysisResult,
)

States.DataLimitExceeded - The state/task returned a result with a size
exceeding the maximum number of bytes service limit.

// CDK: Discard Map state results
const transcribeSegments = new sfn.Map(this, "TranscribeSegments", {
  itemsPath: "$.segment_files",
  maxConcurrency: 10,
  resultPath: sfn.JsonPath.DISCARD,  // ← This is key
});

# Lambda side: Save results to S3
s3.put_object(
    Bucket=bucket,
    Key=f"transcribe_results/{segment_name}.json",
    Body=json.dumps(result_data, ensure_ascii=False),
)
# Return only metadata to Step Functions
return {"bucket": bucket, "result_key": result_key}

FutureWarning: You are using `torch.load` with `weights_only=False`

import torch
 
_orig_torch_load = torch.load
 
def _torch_load_legacy(*args, **kwargs):
    """Always call torch.load with weights_only=False"""
    kwargs["weights_only"] = False
    return _orig_torch_load(*args, **kwargs)
 
torch.load = _torch_load_legacy  # Apply BEFORE pyannote import

FROM public.ecr.aws/lambda/python:3.11
 
ENV HF_HOME=/var/task/models
RUN pip install huggingface_hub
RUN python -c "from huggingface_hub import snapshot_download; \
    snapshot_download('pyannote/speaker-diarization-3.1', token='${HF_TOKEN}')"

ARG HF_TOKEN
RUN --mount=type=secret,id=hf_token \
    HF_TOKEN=$(cat /run/secrets/hf_token) python download_models.py

Google Calendar (meeting schedule)
       │
       ▼ Cloud Functions (Calendar Webhook)
Google Meet Space (Auto-Recording enabled)
       │
       ▼ Recording complete
Google Drive (recording storage)
       │
       ▼ Workspace Events API + Pub/Sub
EventBridge (Cross-Cloud)
       │
       ▼
Lambda (DownloadRecording)
       │
       ▼
S3 → Step Functions (existing pipeline)
       │
       ▼
DynamoDB + AppSync → Dashboard