You need to turn recorded calls into searchable transcripts and actionable summaries. Whisper handles the speech-to-text; LangChain orchestrates the summarization with prompt templates, chunking strategies, and model routing. This guide walks through a complete, runnable pipeline — from raw audio to structured JSON output — with verification checkpoints at each stage.
Step 1: Set up the environment and dependencies
Create a virtual environment and install the core packages. You’ll need openai-whisper for transcription, langchain and langchain-openai for the summarization chain, and ffmpeg for audio preprocessing.
python -m venv .venv
source .venv/bin/activate
pip install --upgrade pip
pip install openai-whisper langchain langchain-openai pydantic python-dotenvInstall ffmpeg if it’s not already on your system:
# macOS
brew install ffmpeg
# Ubuntu/Debian
sudo apt-get update && sudo apt-get install -y ffmpeg
# Windows (via chocolatey)
choco install ffmpegVerify the installation:
python -c "import whisper; import langchain; print('whisper', whisper.__version__, 'langchain', langchain.__version__)"You should see version numbers printed without errors.
Step 2: Prepare audio — normalize to 16 kHz mono WAV
Whisper expects 16 kHz mono audio. Real-world call recordings come in various formats (MP3, M4A, stereo, 44.1 kHz, etc.). Normalize everything upfront to avoid silent failures.
# audio_prep.py
import subprocess
from pathlib import Path
def normalize_audio(input_path: str, output_path: str) -> Path:
"""
Convert any audio file to 16 kHz mono WAV using ffmpeg.
Overwrites output_path if it exists.
"""
cmd = [
"ffmpeg", "-y",
"-i", input_path,
"-ar", "16000",
"-ac", "1",
"-c:a", "pcm_s16le",
output_path
]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
raise RuntimeError(f"ffmpeg failed: {result.stderr}")
return Path(output_path)
if __name__ == "__main__":
import sys
if len(sys.argv) != 3:
print("Usage: python audio_prep.py <input> <output.wav>")
sys.exit(1)
normalize_audio(sys.argv[1], sys.argv[2])
print(f"Normalized audio written to {sys.argv[2]}")Test it:
python audio_prep.py sample_call.mp3 sample_call.wavVerify the output:
ffprobe -v error -select_streams a:0 -show_entries stream=sample_rate,channels -of csv=p=0 sample_call.wav
# Expected: 16000,1Step 3: Transcribe with Whisper — choose the right model size
Whisper offers five model sizes: tiny, base, small, medium, large-v3. For call transcription, small or medium hits the sweet spot — large-v3 is overkill for telephony bandwidth (8 kHz effective) and runs 5-10x slower.
# transcribe.py
import whisper
import json
from pathlib import Path
from typing import Dict, Any
def transcribe_audio(
audio_path: str,
model_size: str = "small",
language: str = "en",
**kwargs
) -> Dict[str, Any]:
"""
Transcribe audio file with Whisper.
Returns the full result dict including segments with timestamps.
"""
model = whisper.load_model(model_size)
result = model.transcribe(
audio_path,
language=language,
word_timestamps=True,
**kwargs
)
return result
def save_transcript(result: Dict[str, Any], output_path: str) -> None:
"""Save transcript as JSON for downstream processing."""
with open(output_path, "w") as f:
json.dump(result, f, indent=2)
if __name__ == "__main__":
import sys
if len(sys.argv) < 2:
print("Usage: python transcribe.py <audio.wav> [model_size]")
sys.exit(1)
audio_path = sys.argv[1]
model_size = sys.argv[2] if len(sys.argv) > 2 else "small"
print(f"Loading {model_size} model...")
result = transcribe_audio(audio_path, model_size=model_size)
output_path = Path(audio_path).with_suffix(".transcript.json")
save_transcript(result, output_path)
print(f"Transcript saved to {output_path}")
print(f"Detected language: {result.get('language')}")
print(f"Duration: {result.get('duration', 'N/A')}s")
print(f"Segments: {len(result.get('segments', []))}")Run it:
python transcribe.py sample_call.wav smallVerify success — check the JSON output:
head -50 sample_call.transcript.jsonYou should see a segments array with start, end, text, and words (if word_timestamps=True). The text field at the root contains the full concatenated transcript.
Step 4: Build the summarization chain with LangChain
Now turn the raw transcript into a structured summary. A good call summary extracts: participants, key topics, action items, decisions, and follow-ups. Use a prompt template that enforces JSON output for programmatic consumption.
# summarize.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.runnables import RunnablePassthrough
from pydantic import BaseModel, Field
from typing import List, Optional
import json
class CallSummary(BaseModel):
"""Structured output schema for call summaries."""
participants: List[str] = Field(description="Names or roles of speakers identified")
duration_seconds: float = Field(description="Total call duration")
key_topics: List[str] = Field(description="Main discussion topics")
decisions: List[str] = Field(description="Decisions made during the call")
action_items: List[dict] = Field(description="Action items with owner and deadline")
follow_ups: List[str] = Field(description="Items requiring follow-up")
sentiment: str = Field(description="Overall sentiment: positive, neutral, negative")
summary: str = Field(description="2-3 paragraph narrative summary")
SUMMARIZATION_PROMPT = ChatPromptTemplate.from_messages([
("system", """You are an expert call analyst. Analyze the transcript and produce a structured summary.
Guidelines:
- Identify speakers by name when mentioned, otherwise use roles (e.g., "Agent", "Customer")
- Extract specific, verifiable action items with owners and deadlines when stated
- Decisions must be explicit agreements, not suggestions
- Sentiment reflects the overall tone and outcome
- Keep the narrative summary concise but complete
Output ONLY valid JSON matching the schema."""),
("human", """Transcript:
{transcript}
Call metadata:
- Duration: {duration_seconds}s
- Language: {language}
Produce the structured summary."""),
])
def build_summarization_chain(model: str = "gpt-4o-mini", temperature: float = 0.1):
"""Create the LangChain runnable for call summarization."""
llm = ChatOpenAI(model=model, temperature=temperature)
parser = JsonOutputParser(pydantic_object=CallSummary)
chain = (
RunnablePassthrough.assign(
transcript=lambda x: x["transcript"][:120000] # truncate to ~30k tokens
)
| SUMMARIZATION_PROMPT
| llm
| parser
)
return chain
def summarize_transcript(
transcript_path: str,
output_path: str,
model: str = "gpt-4o-mini"
) -> CallSummary:
"""Load transcript, run summarization, save structured output."""
with open(transcript_path) as f:
transcript_data = json.load(f)
full_text = transcript_data.get("text", "")
duration = transcript_data.get("duration", 0)
language = transcript_data.get("language", "en")
if not full_text.strip():
raise ValueError("Transcript text is empty")
chain = build_summarization_chain(model=model)
result = chain.invoke({
"transcript": full_text,
"duration_seconds": duration,
"language": language
})
with open(output_path, "w") as f:
json.dump(result, f, indent=2)
return result
if __name__ == "__main__":
import sys
if len(sys.argv) < 2:
print("Usage: python summarize.py <transcript.json> [model]")
sys.exit(1)
transcript_path = sys.argv[1]
model = sys.argv[2] if len(sys.argv) > 2 else "gpt-4o-mini"
output_path = transcript_path.replace(".transcript.json", ".summary.json")
print(f"Summarizing with {model}...")
summary = summarize_transcript(transcript_path, output_path, model)
print(f"Summary saved to {output_path}")
print(json.dumps(summary, indent=2))Run the summarization:
python summarize.py sample_call.transcript.json gpt-4o-miniVerify the output:
cat sample_call.summary.json | jq '.key_topics, .action_items, .sentiment'You should see valid JSON with all CallSummary fields populated. If the transcript exceeds the model’s context window, the chain truncates at ~120k characters (roughly 30k tokens). For longer calls, see Step 6.
Step 5: Add speaker diarization for multi-party calls
Whisper alone doesn’t identify speakers. For calls with 2+ participants, add pyannote.audio diarization to label segments before summarization. This dramatically improves summary quality.
# diarize.py
from pyannote.audio import Pipeline
from pyannote.core import Segment
import torch
import json
from pathlib import Path
def diarize_audio(audio_path: str, hf_token: str) -> list:
"""
Run speaker diarization on audio file.
Returns list of {start, end, speaker} segments.
"""
pipeline = Pipeline.from_pretrained(
"pyannote/speaker-diarization-3.1",
use_auth_token=hf_token
)
# Move to GPU if available
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
pipeline.to(device)
diarization = pipeline(audio_path)
segments = []
for turn, _, speaker in diarization.itertracks(yield_label=True):
segments.append({
"start": turn.start,
"end": turn.end,
"speaker": speaker
})
return segments
def merge_transcript_with_diarization(
transcript_path: str,
diarization_segments: list
) -> str:
"""
Merge Whisper word-level timestamps with diarization segments
to produce a speaker-labeled transcript.
"""
with open(transcript_path) as f:
data = json.load(f)
words = []
for segment in data.get("segments", []):
for word in segment.get("words", []):
words.append({
"start": word["start"],
"end": word["end"],
"text": word["word"]
})
# Assign speaker to each word based on overlap
for word in words:
word_center = (word["start"] + word["end"]) / 2
for seg in diarization_segments:
if seg["start"] <= word_center <= seg["end"]:
word["speaker"] = seg["speaker"]
break
# Group consecutive words by speaker
labeled_lines = []
current_speaker = None
current_text = []
current_start = None
for word in words:
speaker = word.get("speaker", "UNKNOWN")
if speaker != current_speaker:
if current_text:
labeled_lines.append(f"[{current_start:.1f}s] {current_speaker}: {''.join(current_text).strip()}")
current_speaker = speaker
current_text = [word["text"]]
current_start = word["start"]
else:
current_text.append(word["text"])
if current_text:
labeled_lines.append(f"[{current_start:.1f}s] {current_speaker}: {''.join(current_text).strip()}")
return "\n".join(labeled_lines)
if __name__ == "__main__":
import sys
import os
if len(sys.argv) < 3:
print("Usage: python diarize.py <audio.wav> <transcript.json> [hf_token]")
sys.exit(1)
audio_path = sys.argv[1]
transcript_path = sys.argv[2]
hf_token = sys.argv[3] if len(sys.argv) > 3 else os.getenv("HF_TOKEN")
if not hf_token:
print("Error: Hugging Face token required (pass as arg or set HF_TOKEN env var)")
sys.exit(1)
print("Running diarization...")
segments = diarize_audio(audio_path, hf_token)
print("Merging with transcript...")
labeled_transcript = merge_transcript_with_diarization(transcript_path, segments)
output_path = Path(transcript_path).with_suffix(".diarized.txt")
with open(output_path, "w") as f:
f.write(labeled_transcript)
print(f"Diarized transcript saved to {output_path}")
print(labeled_transcript[:500] + "..." if len(labeled_transcript) > 500 else labeled_transcript)You’ll need a Hugging Face token with access to pyannote/speaker-diarization-3.1 (accept the user conditions on the model page). Install the extra dependency:
pip install pyannote.audio torchRun it:
export HF_TOKEN=your_token_here
python diarize.py sample_call.wav sample_call.transcript.jsonVerify the output — you should see speaker-labeled lines like [12.3s] SPEAKER_00: Hello, thanks for calling....
Step 6: Handle long calls with map-reduce summarization
Calls over ~30 minutes exceed the context window even for 128k models. Use LangChain’s map-reduce pattern: chunk the transcript, summarize each chunk, then combine.
# summarize_long.py
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from pydantic import BaseModel, Field
from typing import List, Dict, Any
import json
class ChunkSummary(BaseModel):
key_points: List[str] = Field(description="Key points from this chunk")
action_items: List[Dict[str, str]] = Field(description="Action items with owner/deadline")
decisions: List[str] = Field(description="Decisions made")
speakers: List[str] = Field(description="Speakers in this chunk")
MAP_PROMPT = ChatPromptTemplate.from_messages([
("system", """Summarize this transcript chunk. Extract key points, action items, decisions, and speakers.
Output JSON only."""),
("human", "Chunk:\n{chunk}"),
])
REDUCE_PROMPT = ChatPromptTemplate.from_messages([
("system", """Combine multiple chunk summaries into a final call summary.
Deduplicate action items and decisions. Identify overall topics and sentiment.
Output the complete CallSummary JSON schema."""),
("human", "Chunk summaries:\n{chunk_summaries}"),
])
def build_map_reduce_chain(model: str = "gpt-4o-mini"):
llm = ChatOpenAI(model=model, temperature=0.1)
map_parser = JsonOutputParser(pydantic_object=ChunkSummary)
reduce_parser = JsonOutputParser(pydantic_object=CallSummary)
map_chain = MAP_PROMPT | llm | map_parser
reduce_chain = REDUCE_PROMPT | llm | reduce_parser
def map_chunks(inputs: Dict[str, Any]) -> List[Dict]:
chunks = inputs["chunks"]
return map_chain.batch([{"chunk": c} for c in chunks])
def format_for_reduce(inputs: Dict[str, Any]) -> Dict[str, str]:
summaries = inputs["chunk_summaries"]
formatted = "\n\n---\n\n".join(json.dumps(s, indent=2) for s in summaries)
return {"chunk_summaries": formatted}
full_chain = (
RunnablePassthrough.assign(chunk_summaries=RunnableLambda(map_chunks))
| RunnableLambda(format_for_reduce)
| reduce_chain
)
return full_chain
def summarize_long_transcript(
transcript_path: str,
output_path: str,
chunk_size: int = 8000,
chunk_overlap: int = 500,
model: str = "gpt-4o-mini"
) -> CallSummary:
with open(transcript_path) as f:
data = json.load(f)
full_text = data.get("text", "")
splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
separators=["\n\n", "\n", ". ", " ", ""]
)
chunks = splitter.split_text(full_text)
print(f"Split into {len(chunks)} chunks")
chain = build_map_reduce_chain(model)
result = chain.invoke({"chunks": chunks})
with open(output_path, "w") as f:
json.dump(result, f, indent=2)
return resultUse this instead of summarize.py for calls longer than 25-30 minutes. The chunk size of 8000 characters leaves room for the prompt and response within a 128k context window.
Step 7: Wire it into a single CLI pipeline
Combine all steps into one command for production use. This script handles the full flow: normalize → transcribe → (optional diarize) → summarize.
# pipeline.py
#!/usr/bin/env python3
"""
End-to-end call transcription and summarization pipeline.
Usage: python pipeline.py <input_audio> [--model small] [--summarize-model gpt-4o-mini] [--diarize]
"""
import argparse
import subprocess
import sys
from pathlib import Path
import os
from transcribe import transcribe_audio, save_transcript
from summarize import summarize_transcript, summarize_long_transcript
def run_pipeline(
input_path: str,
whisper_model: str = "small",
summarize_model: str = "gpt-4o-mini",
diarize: bool = False,
long_call: bool = False
) -> dict:
base = Path(input_path).stem
work_dir = Path("output") / base
work_dir.mkdir(parents=True, exist_ok=True)
# Step 1: Normalize audio
wav_path = work_dir / "normalized.wav"
print(f"[1/4] Normalizing audio to {wav_path}...")
subprocess.run([
"ffmpeg", "-y", "-i", input_path,
"-ar", "16000", "-ac", "1", "-c:a", "pcm_s16le",
str(wav_path)
], check=True, capture_output=True)
# Step 2: Transcribe
transcript_path = work_dir / "transcript.json"
print(f"[2/4] Transcribing with {whisper_model}...")
result = transcribe_audio(str(wav_path), model_size=whisper_model)
save_transcript(result, str(transcript_path))
print(f" Duration: {result.get('duration', 0):.1f}s, Segments: {len(result.get('segments', []))}")
# Step 3: Optional diarization
final_transcript = transcript_path
if diarize:
diarized_path = work_dir / "transcript.diarized.txt"
print(f"[3/4] Running diarization...")
hf_token = os.getenv("HF_TOKEN")
if not hf_token:
print(" WARNING: HF_TOKEN not set, skipping diarization")
else:
from diarize import diarize_audio, merge_transcript_with_diarization
segments = diarize_audio(str(wav_path), hf_token)
labeled = merge_transcript_with_diarization(str(transcript_path), segments)
with open(diarized_path, "w") as f:
f.write(labeled)
# For summarization, use the labeled text by creating a modified transcript
result["text"] = labeled
final_transcript = work_dir / "transcript.for_summary.json"
save_transcript(result, str(final_transcript))
print(f" Diarized transcript saved to {diarized_path}")
# Step 4: Summarize
summary_path = work_dir / "summary.json"
print(f"[4/4] Summarizing with {summarize_model}...")
if long_call or result.get("duration", 0) > 1800:
summary = summarize_long_transcript(str(final_transcript), str(summary_path), model=summarize_model)
else:
summary = summarize_transcript(str(final_transcript), str(summary_path), model=summarize_model)
print(f"\n✓ Pipeline complete. Outputs in {work_dir}:")
print(f" - Normalized audio: {wav_path}")
print(f" - Transcript: {transcript_path}")
if diarize:
print(f" - Diarized transcript: {diarized_path}")
print(f" - Summary: {summary_path}")
return summary
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Call transcription + summarization pipeline")
parser.add_argument("input", help="Input audio file (any format ffmpeg supports)")
parser.add_argument("--whisper-model", default="small", choices=["tiny", "base", "small", "medium", "large-v3"])
parser.add_argument("--summarize-model", default="gpt-4o-mini")
parser.add_argument("--diarize", action="store_true", help="Enable speaker diarization (requires HF_TOKEN)")
parser.add_argument("--long-call", action="store_true", help="Force map-reduce summarization")
args = parser.parse_args()
try:
run_pipeline(
args.input,
whisper_model=args.whisper_model,
summarize_model=args.summarize_model,
diarize=args.diarize,
long_call=args.long_call
)
except subprocess.CalledProcessError as e:
print(f"ffmpeg error: {e.stderr.decode() if e.stderr else e}", file=sys.stderr)
sys.exit(1)
except Exception as e:
print(f"Pipeline failed: {e}", file=sys.stderr)
sys.exit(1)Make it executable and run:
chmod +x pipeline.py
python pipeline.py recorded_call.mp3 --whisper-model small --summarize-model gpt-4o-mini --diarizeFor a 45-minute board meeting:
python pipeline.py board_meeting.wav --whisper-model medium --summarize-model gpt-4o --long-call --diarizeStep 8: Verify and monitor in production
Add structured logging and basic quality checks so you can detect regressions.
# verify.py
import json
from pathlib import Path
from typing import Dict, Any
def verify_summary(summary_path: str) -> Dict[str, Any]:
"""Run quality checks on a generated summary."""
with open(summary_path) as f:
summary = json.load(f)
checks = {
"has_participants": len(summary.get("participants", [])) > 0,
"has_key_topics": len(summary.get("key_topics", [])) > 0,
"has_action_items": len(summary.get("action_items", [])) > 0,
"has_decisions": len(summary.get("decisions", [])) >= 0, # zero is valid
"has_sentiment": summary.get("sentiment") in ["positive", "neutral", "negative"],
"has_narrative": len(summary.get("summary", "").strip()) > 50,
"action_items_have_owners": all(
"owner" in item and item["owner"]
for item in summary.get("action_items", [])
),
}
all_passed = all(checks.values())
return {
"passed": all_passed,
"checks": checks,
"summary_stats": {
"participants": len(summary.get("participants", [])),
"key_topics": len(summary.get("key_topics", [])),
"action_items": len(summary.get("action_items", [])),
"decisions": len(summary.get("decisions", [])),
"summary_length": len(summary.get("summary", ""))
}
}
if __name__ == "__main__":
import sys
if len(sys.argv) < 2:
print("Usage: python verify.py <summary.json>")
sys.exit(1)
result = verify_summary(sys.argv[1])
print(json.dumps(result, indent=2))
sys.exit(0 if result["passed"] else 1)Run verification as a post-step:
python verify.py output/recorded_call/summary.jsonExit code 0 means all checks passed. Wire this into your CI/CD or cron job to catch model drift or prompt regressions.
Operational notes
Model selection: whisper-small transcribes ~1 hour of audio in 2-3 minutes on a modern CPU. medium improves accuracy on accented speech and noisy lines but doubles latency. large-v3 rarely justifies its cost for telephony audio.
Cost control: gpt-4o-mini summarizes a 30-minute call for ~$0.02. gpt-4o is ~10x more expensive; reserve it for high-stakes calls (legal, medical, executive).
Diarization latency: pyannote adds 30-60 seconds per hour of audio. Run it asynchronously if your pipeline is user-facing.
Provider routing: If you’re running this at scale and want automatic fallback when your primary LLM provider is rate-limited or degraded, n4n.ai forwards provider cache-control hints and honors client routing directives so you can swap models without code changes.
Storage: Keep the normalized WAV and raw transcript JSON. Re-summarization with improved prompts is cheaper than re-transcription.
Next steps
- Add PII redaction before sending transcripts to the summarization model (use
presidioorspacyNER) - Build a simple UI with Streamlit or FastAPI + React for non-technical reviewers
- Store summaries in a vector database (pgvector, Pinecone) for semantic search across calls
- Implement few-shot prompting with your best human-written summaries to improve consistency
The pipeline above runs end-to-end on a laptop and scales to hundreds of calls/day on a modest GPU instance. Start with small + gpt-4o-mini, measure your WER and summary quality, then upgrade components only where the data justifies it.