Module 5: Audio Processing
8. Project: Audio Pipeline
Description
This project integrates everything learned in Module 5 into a complete and configurable Audio Pipeline. The system receives an audio file, validates and preprocesses it, transcribes it with Whisper, processes the transcript with an LLM (summary, action items, Q&A), and optionally generates audio of the result with TTS. It's a production system with flexible configuration, cost tracking, error handling, and extensions for batch processing and diarization.
Why it matters: This pipeline is the audio component of the Document Analyzer from Module 8. When the final system receives an audio file, it will delegate to exactly this pipeline. Besides, it's a useful system in its own right: any team that records meetings can use it to generate automatic summaries.
Specifications
Input
| Parameter | Type | Required | Description |
|---|---|---|---|
audio_path | str | Yes | Path to the audio file |
task | str | No | "summarize", "action_items", "qa", "full" |
questions | list[str] | Only if task="qa" | Questions about the audio |
generate_audio | bool | No | If True, generates TTS of the result |
language | str | No | ISO code of the language (default: "es") |
voice | str | No | Voice for TTS (default: "nova") |
vocabulary | list[str] | No | Technical terms to improve transcription |
Output
{
"status": "success",
"transcript": "Full transcribed text...",
"result": {
"summary": "Summary in key points...",
"action_items": ["Task 1", "Task 2"],
},
"audio_output_path": "summary.mp3",
"metadata": {
"audio_duration_min": 15.3,
"transcript_words": 2340,
"language_detected": "es",
"chunks_processed": 1
},
"costs": {
"whisper": 0.0918,
"llm": 0.0012,
"tts": 0.0045,
"total": 0.0975
},
"timing": {
"preprocessing_s": 0.5,
"transcription_s": 8.2,
"llm_s": 1.3,
"tts_s": 2.1,
"total_s": 12.1
}
}Step 1: Configuration
from openai import OpenAI
from pydub import AudioSegment
from pathlib import Path
from dataclasses import dataclass, field
import tempfile
import time
import json
client = OpenAI()
WHISPER_FORMATS = {".mp3", ".mp4", ".mpeg", ".mpga", ".m4a", ".wav", ".webm", ".flac", ".ogg", ".oga"}
WHISPER_MAX_SIZE = 25 * 1024 * 1024
@dataclass
class PipelineConfig:
task: str = "summarize"
language: str = "es"
voice: str = "nova"
tts_model: str = "tts-1"
llm_model: str = "gpt-4o-mini"
generate_audio: bool = False
questions: list[str] = field(default_factory=list)
vocabulary: list[str] = field(default_factory=list)
max_transcript_chars: int = 12000
max_tts_chars: int = 4000
output_dir: str = "pipeline_output"
auto_preprocess: bool = True
@dataclass
class PipelineMetrics:
costs: dict = field(default_factory=dict)
timing: dict = field(default_factory=dict)
def add_cost(self, service: str, amount: float):
self.costs[service] = self.costs.get(service, 0) + amount
self.costs["total"] = sum(v for k, v in self.costs.items() if k != "total")
def start_timer(self, name: str):
self.timing[f"_start_{name}"] = time.time()
def stop_timer(self, name: str):
start_key = f"_start_{name}"
if start_key in self.timing:
self.timing[name] = round(time.time() - self.timing.pop(start_key), 2)Step 2: Validation and Preprocessing
def validate_audio(audio_path: str) -> dict:
path = Path(audio_path)
if not path.exists():
return {"valid": False, "error": "File not found", "path": audio_path}
suffix = path.suffix.lower()
size_bytes = path.stat().st_size
size_mb = size_bytes / (1024 * 1024)
issues = []
if suffix not in WHISPER_FORMATS:
issues.append({
"type": "unsupported_format",
"detail": f"Format {suffix} not supported",
"auto_fixable": True
})
if size_bytes > WHISPER_MAX_SIZE:
issues.append({
"type": "file_too_large",
"detail": f"File of {size_mb:.1f}MB exceeds 25MB",
"auto_fixable": True
})
try:
audio = AudioSegment.from_file(audio_path)
duration_s = len(audio) / 1000
if audio.dBFS < -35:
issues.append({
"type": "low_volume",
"detail": f"Low volume ({audio.dBFS:.1f} dBFS)",
"auto_fixable": True
})
if audio.frame_rate < 16000:
issues.append({
"type": "low_sample_rate",
"detail": f"Low sample rate ({audio.frame_rate} Hz)",
"auto_fixable": True
})
except Exception as e:
return {"valid": False, "error": f"Could not read the audio: {str(e)}"}
return {
"valid": len([i for i in issues if not i["auto_fixable"]]) == 0,
"size_mb": round(size_mb, 2),
"duration_s": round(duration_s, 2),
"format": suffix,
"issues": issues,
"needs_preprocessing": len(issues) > 0
}
def preprocess_audio(audio_path: str, output_dir: str) -> dict:
Path(output_dir).mkdir(parents=True, exist_ok=True)
path = Path(audio_path)
steps = []
current = audio_path
if path.suffix.lower() not in WHISPER_FORMATS:
converted = str(Path(output_dir) / f"{path.stem}.mp3")
AudioSegment.from_file(current).export(converted, format="mp3", bitrate="128k")
current = converted
steps.append("format_conversion")
audio = AudioSegment.from_file(current)
needs_normalize = (
audio.dBFS < -30
or audio.channels > 1
or audio.frame_rate < 16000
)
if needs_normalize:
normalized = str(Path(output_dir) / f"{path.stem}_normalized.mp3")
audio = audio.set_frame_rate(16000).set_channels(1).normalize()
audio.export(normalized, format="mp3", bitrate="64k")
current = normalized
steps.append("normalization")
size_mb = Path(current).stat().st_size / (1024 * 1024)
if size_mb > 24:
compressed = str(Path(output_dir) / f"{path.stem}_compressed.mp3")
audio = AudioSegment.from_file(current)
audio = audio.set_channels(1).set_frame_rate(16000)
audio.export(compressed, format="mp3", bitrate="48k")
current = compressed
steps.append("compression")
return {
"output_path": current,
"steps": steps,
"final_size_mb": round(Path(current).stat().st_size / (1024 * 1024), 2)
}Step 3: Transcription with Whisper
def transcribe_audio(
audio_path: str,
config: PipelineConfig,
metrics: PipelineMetrics
) -> dict:
metrics.start_timer("transcription")
path = Path(audio_path)
size_mb = path.stat().st_size / (1024 * 1024)
audio = AudioSegment.from_file(audio_path)
duration_min = len(audio) / 1000 / 60
if size_mb > 24:
result = _transcribe_chunked(audio, config, metrics)
else:
result = _transcribe_single(audio_path, config)
metrics.stop_timer("transcription")
metrics.add_cost("whisper", duration_min * 0.006)
return {
"text": result["text"],
"language": result.get("language", config.language),
"duration_min": round(duration_min, 2),
"word_count": len(result["text"].split()),
"chunks": result.get("chunks", 1)
}
def _transcribe_single(audio_path: str, config: PipelineConfig) -> dict:
kwargs = {
"model": "whisper-1",
"response_format": "verbose_json",
}
if config.language:
kwargs["language"] = config.language
if config.vocabulary:
kwargs["prompt"] = ", ".join(config.vocabulary)
with open(audio_path, "rb") as f:
kwargs["file"] = f
response = client.audio.transcriptions.create(**kwargs)
return {
"text": response.text,
"language": response.language,
"duration": response.duration,
"chunks": 1
}
def _transcribe_chunked(
audio: AudioSegment,
config: PipelineConfig,
metrics: PipelineMetrics,
chunk_duration_ms: int = 10 * 60 * 1000
) -> dict:
chunks = [
audio[i:i + chunk_duration_ms]
for i in range(0, len(audio), chunk_duration_ms)
]
transcripts = []
detected_language = None
with tempfile.TemporaryDirectory() as tmp_dir:
for i, chunk in enumerate(chunks):
chunk_path = str(Path(tmp_dir) / f"chunk_{i:03d}.mp3")
chunk.export(chunk_path, format="mp3", bitrate="128k")
result = _transcribe_single(chunk_path, config)
transcripts.append(result["text"])
if detected_language is None:
detected_language = result.get("language")
return {
"text": " ".join(transcripts),
"language": detected_language,
"chunks": len(chunks)
}Step 4: Processing with the LLM
def process_with_llm(
transcript: str,
config: PipelineConfig,
metrics: PipelineMetrics
) -> dict:
metrics.start_timer("llm")
truncated = transcript[:config.max_transcript_chars]
if config.task == "summarize":
result = _summarize(truncated, config)
elif config.task == "action_items":
result = _extract_action_items(truncated, config)
elif config.task == "qa":
result = _answer_questions(truncated, config)
elif config.task == "full":
result = _full_analysis(truncated, config)
else:
result = _summarize(truncated, config)
input_tokens = len(truncated.split())
metrics.add_cost("llm", input_tokens * 0.00000015 + 1000 * 0.0000006)
metrics.stop_timer("llm")
return result
def _summarize(transcript: str, config: PipelineConfig) -> dict:
response = client.chat.completions.create(
model=config.llm_model,
messages=[{
"role": "user",
"content": (
"Summarize the following text in 5-7 key points. "
"Be specific: include names, dates and data mentioned.\n\n"
+ transcript
)
}],
max_tokens=800
)
return {"summary": response.choices[0].message.content}
def _extract_action_items(transcript: str, config: PipelineConfig) -> dict:
response = client.chat.completions.create(
model=config.llm_model,
messages=[{
"role": "user",
"content": (
"Extract ALL the action items from the following text.\n"
"Format for each one:\n"
"- [ ] [Owner if mentioned]: [Concrete task] "
"[Date if mentioned]\n\n"
"If there are no clear action items, state 'No explicit "
"action items identified' and suggest possible implicit tasks.\n\n"
+ transcript
)
}],
max_tokens=600
)
return {"action_items": response.choices[0].message.content}
def _answer_questions(transcript: str, config: PipelineConfig) -> dict:
answers = []
for question in config.questions:
response = client.chat.completions.create(
model=config.llm_model,
messages=[
{
"role": "system",
"content": (
"Answer based ONLY on the provided context. "
"If the information isn't available, say "
"'Not mentioned in the audio'."
)
},
{
"role": "user",
"content": f"Context:\n{transcript}\n\nQuestion: {question}"
}
],
max_tokens=300
)
answers.append({
"question": question,
"answer": response.choices[0].message.content
})
return {"qa": answers}
def _full_analysis(transcript: str, config: PipelineConfig) -> dict:
response = client.chat.completions.create(
model=config.llm_model,
messages=[{
"role": "user",
"content": (
"Analyze the following text and generate a complete report with "
"EXACTLY these sections:\n\n"
"## Executive Summary\n"
"(3-5 sentences with the most important points)\n\n"
"## Main Topics\n"
"(Numbered list)\n\n"
"## Decisions Made\n"
"(List. If there are none, state 'No explicit decisions')\n\n"
"## Action Items\n"
"(Format: - [ ] [Owner]: [Task])\n\n"
"## Data and Figures Mentioned\n"
"(Dates, amounts, percentages, etc.)\n\n"
+ transcript
)
}],
max_tokens=1200
)
return {"full_analysis": response.choices[0].message.content}Step 5: Audio Generation (TTS)
def generate_audio_output(
text: str,
config: PipelineConfig,
metrics: PipelineMetrics,
output_path: str = None
) -> dict:
metrics.start_timer("tts")
if output_path is None:
output_path = str(Path(config.output_dir) / "audio_output.mp3")
Path(output_path).parent.mkdir(parents=True, exist_ok=True)
tts_text = text[:config.max_tts_chars]
tts_text = _prepare_text_for_tts(tts_text)
if len(tts_text) > 4000:
_generate_long_audio(tts_text, config, output_path)
else:
response = client.audio.speech.create(
model=config.tts_model,
voice=config.voice,
input=tts_text
)
response.stream_to_file(output_path)
cost_per_char = 0.000015 if config.tts_model == "tts-1" else 0.00003
metrics.add_cost("tts", len(tts_text) * cost_per_char)
metrics.stop_timer("tts")
return {
"audio_path": output_path,
"characters": len(tts_text),
"voice": config.voice,
"model": config.tts_model
}
def _prepare_text_for_tts(text: str) -> str:
text = text.strip()
text = text.replace("- [ ]", ".")
text = text.replace("##", ".")
text = text.replace("**", "")
text = text.replace("*", "")
text = text.replace("\n\n", ". ")
text = text.replace("\n", " ")
import re
text = re.sub(r'\s+', ' ', text)
if not text.endswith((".", "!", "?")):
text += "."
return text
def _generate_long_audio(text: str, config: PipelineConfig, output_path: str):
sentences = text.replace(". ", ".\n").split("\n")
sentences = [s.strip() for s in sentences if s.strip()]
chunks = []
current = ""
for sentence in sentences:
if len(current) + len(sentence) + 1 <= 4000:
current = f"{current} {sentence}".strip()
else:
if current:
chunks.append(current)
current = sentence
if current:
chunks.append(current)
segments = []
with tempfile.TemporaryDirectory() as tmp_dir:
for i, chunk in enumerate(chunks):
chunk_path = str(Path(tmp_dir) / f"tts_chunk_{i:03d}.mp3")
response = client.audio.speech.create(
model=config.tts_model,
voice=config.voice,
input=chunk
)
response.stream_to_file(chunk_path)
segments.append(AudioSegment.from_mp3(chunk_path))
combined = segments[0]
for seg in segments[1:]:
combined += seg
combined.export(output_path, format="mp3")Step 6: Integrated Pipeline
class AudioPipeline:
def __init__(self, config: PipelineConfig = None):
self.config = config or PipelineConfig()
Path(self.config.output_dir).mkdir(parents=True, exist_ok=True)
def run(self, audio_path: str) -> dict:
metrics = PipelineMetrics()
metrics.start_timer("total")
try:
validation = validate_audio(audio_path)
if not validation["valid"] and not self.config.auto_preprocess:
return self._error_result(
f"Invalid audio: {validation.get('error', validation.get('issues'))}",
metrics
)
if validation.get("needs_preprocessing") and self.config.auto_preprocess:
metrics.start_timer("preprocessing")
prep = preprocess_audio(audio_path, self.config.output_dir)
working_path = prep["output_path"]
metrics.stop_timer("preprocessing")
else:
working_path = audio_path
transcript_data = transcribe_audio(working_path, self.config, metrics)
llm_result = process_with_llm(
transcript_data["text"], self.config, metrics
)
audio_output = None
if self.config.generate_audio:
result_text = self._get_text_for_tts(llm_result)
if result_text:
audio_output = generate_audio_output(
result_text, self.config, metrics
)
metrics.stop_timer("total")
return {
"status": "success",
"transcript": transcript_data["text"],
"result": llm_result,
"audio_output_path": audio_output["audio_path"] if audio_output else None,
"metadata": {
"audio_duration_min": transcript_data["duration_min"],
"transcript_words": transcript_data["word_count"],
"language_detected": transcript_data["language"],
"chunks_processed": transcript_data["chunks"],
"task": self.config.task
},
"costs": metrics.costs,
"timing": {k: v for k, v in metrics.timing.items() if not k.startswith("_")}
}
except Exception as e:
metrics.stop_timer("total")
return self._error_result(str(e), metrics)
def _get_text_for_tts(self, llm_result: dict) -> str:
if "summary" in llm_result:
return llm_result["summary"]
if "full_analysis" in llm_result:
return llm_result["full_analysis"]
if "action_items" in llm_result:
return llm_result["action_items"]
if "qa" in llm_result:
return "\n".join(
f"Question: {qa['question']}. Answer: {qa['answer']}"
for qa in llm_result["qa"]
)
return None
def _error_result(self, error: str, metrics: PipelineMetrics) -> dict:
return {
"status": "error",
"error": error,
"costs": metrics.costs,
"timing": {k: v for k, v in metrics.timing.items() if not k.startswith("_")}
}Step 7: Demo
Meeting summary
config = PipelineConfig(
task="summarize",
language="es",
generate_audio=True,
voice="nova",
vocabulary=["sprint", "deployment", "QA", "staging"]
)
pipeline = AudioPipeline(config)
result = pipeline.run("team_meeting.mp3")
if result["status"] == "success":
print("=== SUMMARY ===")
print(result["result"]["summary"])
print(f"\nAudio duration: {result['metadata']['audio_duration_min']} min")
print(f"Transcribed words: {result['metadata']['transcript_words']}")
print(f"Total cost: ${result['costs']['total']:.4f}")
print(f"Total time: {result['timing']['total']}s")
if result["audio_output_path"]:
print(f"Summary audio: {result['audio_output_path']}")Action-item extraction
config = PipelineConfig(task="action_items", language="es")
pipeline = AudioPipeline(config)
result = pipeline.run("standup.mp3")
if result["status"] == "success":
print(result["result"]["action_items"])Q&A over the audio
config = PipelineConfig(
task="qa",
language="es",
questions=[
"What was the decision about the budget?",
"Who is responsible for the deploy?",
"When is the delivery date?"
]
)
pipeline = AudioPipeline(config)
result = pipeline.run("planning.mp3")
if result["status"] == "success":
for qa in result["result"]["qa"]:
print(f"Q: {qa['question']}")
print(f"A: {qa['answer']}\n")Full analysis
config = PipelineConfig(
task="full",
language="es",
generate_audio=True,
tts_model="tts-1-hd",
voice="echo"
)
pipeline = AudioPipeline(config)
result = pipeline.run("board_meeting.mp3")
if result["status"] == "success":
print(result["result"]["full_analysis"])Extension 1: Batch Processing
Process multiple audio files in batch:
from concurrent.futures import ThreadPoolExecutor
class BatchAudioPipeline:
def __init__(self, config: PipelineConfig = None, max_workers: int = 3):
self.config = config or PipelineConfig()
self.max_workers = max_workers
def run_batch(self, audio_paths: list[str]) -> dict:
pipeline = AudioPipeline(self.config)
results = []
def process_one(path: str) -> dict:
result = pipeline.run(path)
result["source_file"] = Path(path).name
return result
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
results = list(executor.map(process_one, audio_paths))
total_cost = sum(
r.get("costs", {}).get("total", 0)
for r in results
)
total_time = sum(
r.get("timing", {}).get("total", 0)
for r in results
)
successful = sum(1 for r in results if r["status"] == "success")
return {
"results": results,
"summary": {
"total_files": len(audio_paths),
"successful": successful,
"failed": len(audio_paths) - successful,
"total_cost_usd": round(total_cost, 4),
"total_time_s": round(total_time, 2)
}
}Usage
batch = BatchAudioPipeline(
config=PipelineConfig(task="summarize", language="es"),
max_workers=3
)
audio_files = list(Path("recordings").glob("*.mp3"))
results = batch.run_batch([str(f) for f in audio_files])
print(f"Processed: {results['summary']['successful']}/{results['summary']['total_files']}")
print(f"Total cost: ${results['summary']['total_cost_usd']}")Extension 2: Diarization with AssemblyAI
Extend the pipeline to identify speakers:
import assemblyai as aai
class DiarizedAudioPipeline(AudioPipeline):
def __init__(self, config: PipelineConfig = None):
super().__init__(config)
aai.settings.api_key = os.getenv("ASSEMBLYAI_API_KEY")
def run_with_diarization(self, audio_path: str) -> dict:
base_result = self.run(audio_path)
if base_result["status"] != "success":
return base_result
try:
diarization = self._diarize(audio_path)
base_result["diarization"] = diarization
if diarization["utterances"]:
dialogue = self._format_dialogue(diarization["utterances"])
base_result["result"]["dialogue"] = dialogue
except Exception as e:
base_result["diarization_error"] = str(e)
return base_result
def _diarize(self, audio_path: str) -> dict:
config = aai.TranscriptionConfig(
speaker_labels=True,
language_code=self.config.language
)
transcript = aai.Transcriber().transcribe(audio_path, config=config)
if transcript.status == aai.TranscriptStatus.error:
raise RuntimeError(f"Diarization error: {transcript.error}")
utterances = [
{
"speaker": u.speaker,
"text": u.text,
"start_ms": u.start,
"end_ms": u.end
}
for u in (transcript.utterances or [])
]
speakers = list(set(u["speaker"] for u in utterances))
return {
"utterances": utterances,
"speakers": speakers,
"speaker_count": len(speakers)
}
def _format_dialogue(self, utterances: list[dict]) -> str:
lines = []
for u in utterances:
start_min = u["start_ms"] // 60000
start_sec = (u["start_ms"] % 60000) // 1000
timestamp = f"[{start_min:02d}:{start_sec:02d}]"
lines.append(f"{timestamp} Speaker {u['speaker']}: {u['text']}")
return "\n\n".join(lines)Extension 3: Export Results
def export_results(result: dict, output_dir: str, formats: list[str] = None):
if formats is None:
formats = ["json", "md", "txt"]
Path(output_dir).mkdir(parents=True, exist_ok=True)
if "json" in formats:
json_path = Path(output_dir) / "pipeline_result.json"
with open(json_path, "w", encoding="utf-8") as f:
json.dump(result, f, ensure_ascii=False, indent=2)
if "md" in formats:
md_path = Path(output_dir) / "pipeline_result.md"
md_content = _build_markdown_report(result)
with open(md_path, "w", encoding="utf-8") as f:
f.write(md_content)
if "txt" in formats:
txt_path = Path(output_dir) / "transcript.txt"
with open(txt_path, "w", encoding="utf-8") as f:
f.write(result.get("transcript", ""))
def _build_markdown_report(result: dict) -> str:
lines = ["# Audio Pipeline Report\n"]
meta = result.get("metadata", {})
lines.append(f"**Duration:** {meta.get('audio_duration_min', 'N/A')} min")
lines.append(f"**Words:** {meta.get('transcript_words', 'N/A')}")
lines.append(f"**Language:** {meta.get('language_detected', 'N/A')}")
lines.append(f"**Task:** {meta.get('task', 'N/A')}\n")
lines.append("## Result\n")
llm_result = result.get("result", {})
for key, value in llm_result.items():
lines.append(f"### {key.replace('_', ' ').title()}\n")
if isinstance(value, list):
for item in value:
if isinstance(item, dict):
lines.append(f"**{item.get('question', '')}**")
lines.append(f"{item.get('answer', '')}\n")
else:
lines.append(f"- {item}")
else:
lines.append(str(value))
lines.append("")
costs = result.get("costs", {})
if costs:
lines.append("## Costs\n")
lines.append("| Service | Cost USD |")
lines.append("|----------|-----------|")
for service, cost in costs.items():
lines.append(f"| {service} | ${cost:.6f} |")
lines.append("")
timing = result.get("timing", {})
if timing:
lines.append("## Timing\n")
lines.append("| Step | Seconds |")
lines.append("|------|----------|")
for step, secs in timing.items():
lines.append(f"| {step} | {secs}s |")
return "\n".join(lines)Project Troubleshooting
Problem: Pipeline fails silently
Symptom: status: "error" without useful information.
Solution: Add detailed logging:
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("audio_pipeline")
class AudioPipelineDebug(AudioPipeline):
def run(self, audio_path: str) -> dict:
logger.info(f"Starting pipeline for: {audio_path}")
logger.info(f"Config: task={self.config.task}, lang={self.config.language}")
result = super().run(audio_path)
if result["status"] == "error":
logger.error(f"Pipeline failed: {result['error']}")
else:
logger.info(f"Pipeline succeeded. Cost: ${result['costs'].get('total', 0):.4f}")
return resultProblem: Partial transcription on long audio
Symptom: Only part of the audio gets transcribed.
Solution: Verify that the chunking works correctly. The pipeline automatically handles files >25MB, but if the file is already under the limit and is long (e.g., 50 min of audio compressed into 20MB), the transcription may lose context during long pauses.
config = PipelineConfig(
language="es",
vocabulary=["sprint", "deployment"],
auto_preprocess=True
)Problem: Unexpectedly high cost in batch
Symptom: A batch of 50 files costs much more than expected.
Solution: Estimate costs before running:
def estimate_batch_cost(audio_paths: list[str]) -> dict:
total_duration = 0
for path in audio_paths:
audio = AudioSegment.from_file(path)
total_duration += len(audio) / 1000 / 60
return {
"files": len(audio_paths),
"total_duration_min": round(total_duration, 1),
"estimated_whisper_cost": round(total_duration * 0.006, 2),
"estimated_llm_cost": round(len(audio_paths) * 0.002, 2),
"estimated_total": round(total_duration * 0.006 + len(audio_paths) * 0.002, 2)
}Completeness Checklist
Core Functionality
- Audio validation (format, size, quality)
- Automatic preprocessing (conversion, normalization)
- Transcription with Whisper (single and chunked)
- Processing with the LLM (summarize, action_items, qa, full)
- Audio generation with TTS (optional)
- Error handling at each step
Quality
- Cost tracking per service
- Timing tracking per step
- Support for technical vocabulary
- Text preprocessed for TTS (no markdown, no bullets)
Extensions
- Batch processing with ThreadPoolExecutor
- Diarization with AssemblyAI
- Export results (JSON, Markdown, TXT)
- Pre-execution cost estimation
Testing
- Works with short audio (<1 min)
- Works with long audio (>25MB, chunking)
- Works with a non-standard format (conversion)
- Handles errors gracefully (file not found, API error)
- Generates output audio correctly
Exercises
Exercise 1: Pipeline with cache
Extend AudioPipeline to cache transcriptions on disk using the audio file's MD5 hash as the key. If the file was already transcribed, don't call Whisper again.
See solution
import hashlib
class CachedAudioPipeline(AudioPipeline):
def __init__(self, config: PipelineConfig = None, cache_dir: str = "transcript_cache"):
super().__init__(config)
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(exist_ok=True)
def _get_cache_key(self, audio_path: str) -> str:
h = hashlib.md5()
with open(audio_path, "rb") as f:
for chunk in iter(lambda: f.read(8192), b""):
h.update(chunk)
return h.hexdigest()
def run(self, audio_path: str) -> dict:
metrics = PipelineMetrics()
metrics.start_timer("total")
validation = validate_audio(audio_path)
working_path = audio_path
if validation.get("needs_preprocessing") and self.config.auto_preprocess:
prep = preprocess_audio(audio_path, self.config.output_dir)
working_path = prep["output_path"]
cache_key = self._get_cache_key(working_path)
cache_file = self.cache_dir / f"{cache_key}.json"
if cache_file.exists():
cached = json.loads(cache_file.read_text(encoding="utf-8"))
transcript_data = cached
transcript_data["from_cache"] = True
else:
transcript_data = transcribe_audio(working_path, self.config, metrics)
cache_file.write_text(
json.dumps(transcript_data, ensure_ascii=False),
encoding="utf-8"
)
transcript_data["from_cache"] = False
llm_result = process_with_llm(transcript_data["text"], self.config, metrics)
audio_output = None
if self.config.generate_audio:
result_text = self._get_text_for_tts(llm_result)
if result_text:
audio_output = generate_audio_output(result_text, self.config, metrics)
metrics.stop_timer("total")
return {
"status": "success",
"transcript": transcript_data["text"],
"result": llm_result,
"audio_output_path": audio_output["audio_path"] if audio_output else None,
"from_cache": transcript_data.get("from_cache", False),
"metadata": {
"audio_duration_min": transcript_data.get("duration_min"),
"transcript_words": transcript_data.get("word_count"),
"language_detected": transcript_data.get("language"),
"chunks_processed": transcript_data.get("chunks", 1),
"task": self.config.task
},
"costs": metrics.costs,
"timing": {k: v for k, v in metrics.timing.items() if not k.startswith("_")}
}Exercise 2: Pipeline with comparative report
Create a function that runs the same audio with different configurations (summarize, action_items, full) and generates a comparative report of costs and results.
See solution
def compare_pipeline_tasks(audio_path: str) -> dict:
tasks = ["summarize", "action_items", "full"]
results = {}
for task in tasks:
config = PipelineConfig(task=task, language="es")
pipeline = AudioPipeline(config)
result = pipeline.run(audio_path)
results[task] = {
"status": result["status"],
"result": result.get("result"),
"cost": result.get("costs", {}).get("total", 0),
"time": result.get("timing", {}).get("total", 0),
"transcript_words": result.get("metadata", {}).get("transcript_words", 0)
}
comparison = {
"results": results,
"cheapest_task": min(results.items(), key=lambda x: x[1]["cost"])[0],
"fastest_task": min(results.items(), key=lambda x: x[1]["time"])[0],
"total_cost": round(sum(r["cost"] for r in results.values()), 4),
"total_time": round(sum(r["time"] for r in results.values()), 2)
}
return comparisonSummary
- The Audio Pipeline integrates validation → preprocessing → transcription → LLM → TTS into a configurable system.
- It supports 4 tasks: summarize, action_items, qa, full analysis.
- It includes cost tracking and timing for each step of the pipeline.
- Automatic preprocessing: format conversion, audio normalization, handling of large files.
- Extensions: batch processing, diarization with AssemblyAI, multi-format export.
- The pipeline is the audio component of the Document Analyzer from Module 8.
- Typical cost: ~$0.10-0.20 for 15 minutes of audio with a summary and TTS.
Next module: Module 6 — Multimodal RAG. You already process text, images and audio; now you'll build an intelligent search system that combines all the modalities to answer questions about documents with figures.