Voice-to-text pipelines are deceptively simple in demos but painful in production. You need reliable transcription, speaker diarization, timestamp alignment, and a way to make the resulting text queryable. This tutorial builds a complete pipeline: Deepgram handles the audio-to-text heavy lifting with its Nova-2 model, and LlamaIndex structures the output for semantic search and RAG. By the end you’ll have a runnable system that ingests audio files, transcribes with speaker labels and word-level timestamps, indexes the segments, and answers questions grounded in the source audio.
Prerequisites
- Python 3.10+
- A Deepgram API key (sign up at console.deepgram.com)
- An OpenAI API key for embeddings and LLM (or substitute your preferred provider)
- ffmpeg installed and on PATH for audio preprocessing
Install dependencies:
pip install deepgram-sdk llama-index llama-index-llms-openai llama-index-embeddings-openai python-dotenv ffmpeg-pythonCreate a .env file:
DEEPGRAM_API_KEY=your_deepgram_key
OPENAI_API_KEY=your_openai_keyProject structure
voice_pipeline/
├── config.py
├── transcribe.py
├── index.py
├── query.py
├── pipeline.py
└── main.pyConfiguration
Centralize settings so you can swap models without hunting through code.
# config.py
import os
from dataclasses import dataclass
from dotenv import load_dotenv
load_dotenv()
@dataclass
class Settings:
deepgram_api_key: str = os.getenv("DEEPGRAM_API_KEY", "")
openai_api_key: str = os.getenv("OPENAI_API_KEY", "")
# Deepgram Nova-2 is the current best general-purpose model
dg_model: str = "nova-2"
dg_language: str = "en"
# LlamaIndex defaults
embed_model: str = "text-embedding-3-small"
llm_model: str = "gpt-4o-mini"
chunk_size: int = 512
chunk_overlap: int = 50
# Pipeline
sample_rate: int = 16000
max_file_size_mb: int = 100
settings = Settings()Audio preprocessing
Deepgram accepts many formats but normalizing to 16 kHz mono WAV avoids edge cases and reduces payload size.
# transcribe.py
import ffmpeg
import os
from pathlib import Path
from config import settings
def normalize_audio(input_path: str, output_path: str | None = None) -> str:
"""Convert any audio to 16kHz mono WAV."""
input_path = Path(input_path)
if output_path is None:
output_path = input_path.with_suffix(".wav")
(
ffmpeg
.input(str(input_path))
.output(str(output_path), ac=1, ar=settings.sample_rate, format="wav")
.overwrite_output()
.run(quiet=True, capture_stdout=True, capture_stderr=True)
)
return str(output_path)
def validate_audio(file_path: str) -> bool:
"""Basic sanity check before sending to Deepgram."""
path = Path(file_path)
if not path.exists():
raise FileNotFoundError(f"Audio file not found: {file_path}")
size_mb = path.stat().st_size / (1024 * 1024)
if size_mb > settings.max_file_size_mb:
raise ValueError(f"File {size_mb:.1f}MB exceeds {settings.max_file_size_mb}MB limit")
return TrueDeepgram transcription with diarization and timestamps
The Nova-2 model supports speaker diarization, word-level timestamps, and punctuation out of the box. We’ll request all three and structure the response into segments suitable for indexing.
# transcribe.py (continued)
from deepgram import DeepgramClient, PrerecordedOptions, FileSource
from typing import List, Dict, Any
import json
def transcribe_audio(file_path: str) -> List[Dict[str, Any]]:
"""
Transcribe audio with speaker diarization and word timestamps.
Returns a list of segments, each with speaker, text, start, end, and words.
"""
validate_audio(file_path)
wav_path = normalize_audio(file_path)
deepgram = DeepgramClient(settings.deepgram_api_key)
with open(wav_path, "rb") as audio:
source = FileSource({"buffer": audio.read()})
options = PrerecordedOptions(
model=settings.dg_model,
language=settings.dg_language,
diarize=True,
punctuate=True,
utterances=True, # groups words by speaker turn
smart_format=True, # numbers, currencies, etc.
filler_words=True, # keep "um", "uh" if you need them
profanity_filter=False,
)
response = deepgram.listen.rest.v("1").transcribe_file(source, options)
# Clean up temp WAV if we created it
if wav_path != file_path:
os.unlink(wav_path)
return parse_utterances(response)
def parse_utterances(response) -> List[Dict[str, Any]]:
"""Extract structured utterances from Deepgram response."""
results = response.results
utterances = results.utterances if hasattr(results, "utterances") else []
segments = []
for u in utterances:
segment = {
"speaker": u.speaker,
"text": u.transcript.strip(),
"start": u.start,
"end": u.end,
"confidence": u.confidence,
"words": [
{"word": w.word, "start": w.start, "end": w.end, "confidence": w.confidence}
for w in u.words
] if hasattr(u, "words") and u.words else [],
}
segments.append(segment)
return segments
def save_transcript(segments: List[Dict[str, Any]], output_path: str) -> None:
"""Persist transcript for debugging and audit."""
with open(output_path, "w") as f:
json.dump(segments, f, indent=2)
def format_transcript(segments: List[Dict[str, Any]]) -> str:
"""Human-readable transcript with speaker labels and timestamps."""
lines = []
for s in segments:
start = f"{s['start']:.2f}"
end = f"{s['end']:.2f}"
lines.append(f"[{start}s - {end}s] Speaker {s['speaker']}: {s['text']}")
return "\n".join(lines)Checkpoint — run a test transcription:
# test_transcribe.py
from transcribe import transcribe_audio, format_transcript, save_transcript
segments = transcribe_audio("sample_audio.mp3")
print(format_transcript(segments))
save_transcript(segments, "transcript.json")Expected output (truncated):
[0.00s - 4.23s] Speaker 0: Welcome to the quarterly review. Let's start with the numbers.
[4.23s - 9.81s] Speaker 1: Revenue came in at 12.4 million, up 18% year over year.
[9.81s - 15.67s] Speaker 0: Good. What about churn?
[15.67s - 22.10s] Speaker 1: Churn dropped to 3.2%, the lowest in three years.The transcript.json contains full word-level timestamps and confidence scores for downstream alignment tasks.
LlamaIndex document construction
Each utterance becomes a Document with metadata preserving speaker, time range, and confidence. This enables filtered retrieval (e.g., “only Speaker 1”) and citation with timestamps.
# index.py
from llama_index.core import Document, VectorStoreIndex, Settings as LlamaSettings
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.llms.openai import OpenAI
from llama_index.core.node_parser import SentenceSplitter
from typing import List, Dict, Any
from config import settings
def configure_llama_index():
LlamaSettings.embed_model = OpenAIEmbedding(
model=settings.embed_model,
api_key=settings.openai_api_key,
)
LlamaSettings.llm = OpenAI(
model=settings.llm_model,
api_key=settings.openai_api_key,
temperature=0.1,
)
LlamaSettings.node_parser = SentenceSplitter(
chunk_size=settings.chunk_size,
chunk_overlap=settings.chunk_overlap,
)
def segments_to_documents(segments: List[Dict[str, Any]], source_name: str) -> List[Document]:
"""Convert Deepgram utterances to LlamaIndex Documents with rich metadata."""
docs = []
for i, seg in enumerate(segments):
metadata = {
"source": source_name,
"speaker": seg["speaker"],
"start_time": seg["start"],
"end_time": seg["end"],
"confidence": seg["confidence"],
"segment_index": i,
"word_count": len(seg["text"].split()),
}
# Include word-level timestamps in metadata for precise citation
if seg["words"]:
metadata["words"] = seg["words"]
doc = Document(
text=seg["text"],
metadata=metadata,
doc_id=f"{source_name}_seg_{i}",
)
docs.append(doc)
return docs
def build_index(documents: List[Document]) -> VectorStoreIndex:
"""Build a vector index from documents."""
configure_llama_index()
index = VectorStoreIndex.from_documents(documents, show_progress=True)
return index
def persist_index(index: VectorStoreIndex, persist_dir: str) -> None:
index.storage_context.persist(persist_dir=persist_dir)
def load_index(persist_dir: str) -> VectorStoreIndex:
configure_llama_index()
from llama_index.core import StorageContext, load_index_from_storage
storage_context = StorageContext.from_defaults(persist_dir=persist_dir)
return load_index_from_storage(storage_context)Checkpoint — build and persist the index:
# test_index.py
from transcribe import transcribe_audio
from index import segments_to_documents, build_index, persist_index
segments = transcribe_audio("sample_audio.mp3")
docs = segments_to_documents(segments, "q3_review")
index = build_index(docs)
persist_index(index, "./storage")
print(f"Indexed {len(docs)} segments")Expected output:
Indexed 47 segmentsThe ./storage directory now contains docstore.json, vector_store.json, and index_store.json — portable across environments.
Query engine with citation and filtering
A raw vector query returns text chunks. We need a query engine that:
- Filters by speaker or time range when requested
- Returns citations with timestamps
- Grounds answers in the transcript
# query.py
from llama_index.core import VectorStoreIndex
from llama_index.core.retrievers import VectorIndexRetriever
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.postprocessor import MetadataReplacementPostProcessor
from llama_index.core.response_synthesizers import get_response_synthesizer
from llama_index.core.prompts import PromptTemplate
from typing import Optional, List, Dict, Any
from index import load_index, configure_llama_index
CITATION_PROMPT = PromptTemplate(
"You are an analyst reviewing meeting transcripts.\n"
"Answer the question using ONLY the provided context.\n"
"Each context chunk has metadata: speaker, start_time, end_time.\n"
"Cite sources inline like [Speaker 0, 12.34s-15.67s].\n"
"If the answer isn't in the context, say you don't know.\n\n"
"Context:\n{context_str}\n\n"
"Question: {query_str}\n\n"
"Answer:"
)
def create_query_engine(
index: VectorStoreIndex,
speaker_filter: Optional[int] = None,
time_range: Optional[tuple[float, float]] = None,
similarity_top_k: int = 5,
) -> RetrieverQueryEngine:
"""Build a query engine with optional metadata filters."""
configure_llama_index()
# Build metadata filters
filters = []
if speaker_filter is not None:
from llama_index.core.vector_stores import MetadataFilter, FilterOperator
filters.append(MetadataFilter(key="speaker", value=speaker_filter, operator=FilterOperator.EQ))
if time_range:
from llama_index.core.vector_stores import MetadataFilter, FilterOperator
start, end = time_range
filters.append(MetadataFilter(key="start_time", value=start, operator=FilterOperator.GTE))
filters.append(MetadataFilter(key="end_time", value=end, operator=FilterOperator.LTE))
retriever = VectorIndexRetriever(
index=index,
similarity_top_k=similarity_top_k,
filters=filters if filters else None,
)
response_synthesizer = get_response_synthesizer(
response_mode="compact",
text_qa_template=CITATION_PROMPT,
)
return RetrieverQueryEngine(
retriever=retriever,
response_synthesizer=response_synthesizer,
)
def query_with_citations(
engine: RetrieverQueryEngine,
question: str,
) -> Dict[str, Any]:
"""Run query and return structured response with source nodes."""
response = engine.query(question)
sources = []
for node in response.source_nodes:
meta = node.metadata
sources.append({
"speaker": meta.get("speaker"),
"start_time": meta.get("start_time"),
"end_time": meta.get("end_time"),
"text": node.text[:200] + "..." if len(node.text) > 200 else node.text,
"score": node.score,
})
return {
"answer": str(response),
"sources": sources,
}Checkpoint — query the index:
# test_query.py
from index import load_index
from query import create_query_engine, query_with_citations
import json
index = load_index("./storage")
# General question
engine = create_query_engine(index)
result = query_with_citations(engine, "What was the revenue figure mentioned?")
print(json.dumps(result, indent=2))
# Speaker-filtered question
engine_s1 = create_query_engine(index, speaker_filter=1)
result_s1 = query_with_citations(engine_s1, "What metrics did Speaker 1 report?")
print(json.dumps(result_s1, indent=2))
# Time-range question
engine_time = create_query_engine(index, time_range=(0, 10))
result_time = query_with_citations(engine_time, "What was discussed in the first 10 seconds?")
print(json.dumps(result_time, indent=2))Expected output:
{
"answer": "Revenue came in at 12.4 million, up 18% year over year [Speaker 1, 4.23s-9.81s].",
"sources": [
{
"speaker": 1,
"start_time": 4.23,
"end_time": 9.81,
"text": "Revenue came in at 12.4 million, up 18% year over year.",
"score": 0.92
}
]
}End-to-end pipeline
Wire everything into a single class that handles ingestion, indexing, and querying. This is what you’d import in your application.
# pipeline.py
from pathlib import Path
from typing import Optional, List, Dict, Any
from transcribe import transcribe_audio, format_transcript, save_transcript
from index import segments_to_documents, build_index, persist_index, load_index
from query import create_query_engine, query_with_citations
from config import settings
class VoicePipeline:
def __init__(self, storage_dir: str = "./storage"):
self.storage_dir = storage_dir
self.index: Optional[VectorStoreIndex] = None
self._source_name: Optional[str] = None
def ingest(self, audio_path: str, source_name: Optional[str] = None) -> List[Dict[str, Any]]:
"""Transcribe audio, build index, persist to disk."""
if source_name is None:
source_name = Path(audio_path).stem
print(f"Transcribing {audio_path}...")
segments = transcribe_audio(audio_path)
print(f"Saving transcript...")
save_transcript(segments, f"{source_name}_transcript.json")
print(f"Building index...")
docs = segments_to_documents(segments, source_name)
self.index = build_index(docs)
print(f"Persisting index to {self.storage_dir}...")
persist_index(self.index, self.storage_dir)
self._source_name = source_name
print(f"Done. Indexed {len(segments)} segments.")
return segments
def load_existing(self) -> bool:
"""Load a previously persisted index."""
try:
self.index = load_index(self.storage_dir)
return True
except Exception:
return False
def query(
self,
question: str,
speaker: Optional[int] = None,
time_range: Optional[tuple[float, float]] = None,
top_k: int = 5,
) -> Dict[str, Any]:
"""Query the indexed audio with optional filters."""
if self.index is None:
if not self.load_existing():
raise RuntimeError("No index loaded. Call ingest() first.")
engine = create_query_engine(
self.index,
speaker_filter=speaker,
time_range=time_range,
similarity_top_k=top_k,
)
return query_with_citations(engine, question)
def get_transcript(self, formatted: bool = True) -> str | List[Dict[str, Any]]:
"""Retrieve the full transcript for the current source."""
if self._source_name is None:
raise RuntimeError("No source loaded.")
transcript_path = f"{self._source_name}_transcript.json"
import json
with open(transcript_path) as f:
segments = json.load(f)
return format_transcript(segments) if formatted else segmentsCLI entry point
A thin CLI makes the pipeline scriptable and testable without writing new code each time.
# main.py
import argparse
import json
import sys
from pipeline import VoicePipeline
def main():
parser = argparse.ArgumentParser(description="Voice-to-text pipeline with Deepgram + LlamaIndex")
subparsers = parser.add_subparsers(dest="command", required=True)
# Ingest command
ingest_parser = subparsers.add_parser("ingest", help="Transcribe and index an audio file")
ingest_parser.add_argument("audio", help="Path to audio file")
ingest_parser.add_argument("--name", help="Source name (default: filename)")
ingest_parser.add_argument("--storage", default="./storage", help="Index storage directory")
# Query command
query_parser = subparsers.add_parser("query", help="Query an existing index")
query_parser.add_argument("question", help="Question to ask")
query_parser.add_argument("--storage", default="./storage", help="Index storage directory")
query_parser.add_argument("--speaker", type=int, help="Filter by speaker ID")
query_parser.add_argument("--start", type=float, help="Start time filter (seconds)")
query_parser.add_argument("--end", type=float, help="End time filter (seconds)")
query_parser.add_argument("--top-k", type=int, default=5, help="Number of chunks to retrieve")
# Transcript command
transcript_parser = subparsers.add_parser("transcript", help="Print full transcript")
transcript_parser.add_argument("--storage", default="./storage", help="Index storage directory")
transcript_parser.add_argument("--raw", action="store_true", help="Output raw JSON")
args = parser.parse_args()
pipeline = VoicePipeline(storage_dir=args.storage)
if args.command == "ingest":
segments = pipeline.ingest(args.audio, args.name)
print(json.dumps(segments, indent=2))
elif args.command == "query":
time_range = None
if args.start is not None or args.end is not None:
time_range = (args.start or 0, args.end or float("inf"))
result = pipeline.query(
args.question,
speaker=args.speaker,
time_range=time_range,
top_k=args.top_k,
)
print(json.dumps(result, indent=2))
elif args.command == "transcript":
if not pipeline.load_existing():
print("No index found. Run ingest first.", file=sys.stderr)
sys.exit(1)
transcript = pipeline.get_transcript(formatted=not args.raw)
print(transcript)
if __name__ == "__main__":
main()Usage examples:
# Ingest a meeting recording
python -m main ingest board_meeting.mp3 --name q3_board
# Ask a general question
python -m main query "What was the revenue forecast for Q4?"
# Ask about a specific speaker
python -m main query "What concerns were raised?" --speaker 2
# Ask about a time window
python -m main query "What was decided in the first five minutes?" --start 0 --end 300
# Print full transcript with timestamps
python -m main transcriptHandling long files and streaming
Deepgram’s prerecorded endpoint handles files up to 2 hours, but large files benefit from chunking. For real-time use cases, swap the REST client for Deepgram’s live streaming WebSocket API — the rest of the pipeline (LlamaIndex ingestion, querying) remains identical.
# transcribe.py (addition for large files)
async def transcribe_streaming(file_path: str, chunk_seconds: int = 600) -> List[Dict[str, Any]]:
"""Chunk large files and stitch utterances together."""
# Implementation uses ffmpeg to split, then processes each chunk
# Deepgram's live API would replace this for true streaming
passProduction considerations
Error handling: Wrap Deepgram calls in retries with exponential backoff. The SDK raises DeepgramApiError with status codes — treat 429 and 5xx as retryable.
Cost control: Nova-2 pricing is per minute. For high-volume pipelines, batch transcription jobs and cache transcripts keyed by file hash.
Speaker consistency: Deepgram’s diarization assigns speaker IDs per file (Speaker 0, Speaker 1…). Across files, Speaker 0 isn’t the same person. For multi-file corpora, add a speaker enrollment step or post-process with a speaker recognition model.
Evaluation: Build a small eval set (audio + ground truth Q/A pairs) and measure citation accuracy and answer faithfulness before deploying.
What’s next
- Add a web UI with
streamlitorgradiofor non-technical stakeholders - Swap OpenAI embeddings for local models (BGE, E5) via
llama-index-embeddings-huggingface - Implement incremental indexing: append new audio to an existing index without rebuilding
- Add structured extraction: use LlamaIndex’s
StructuredOutputParserto pull entities (action items, decisions, owners) from each segment
The pipeline you’ve built here is the same architecture that powers production meeting intelligence tools. Deepgram handles the acoustic complexity; LlamaIndex handles the retrieval complexity. Your job is the glue — and now you have it.