YouTube Transcripts to Vector Database — A Complete Python Pipeline
A vector database turns a collection of YouTube transcripts into something you can ask questions. Not search by keyword — ask by meaning. "What did the speaker say about managing technical debt?" finds the passage even if the words "managing" and "technical debt" don't appear in the same sentence.
This guide builds a complete pipeline: extract transcripts from a YouTube playlist, generate embeddings, store them in a vector database, and query with natural language. All code is working Python. The pipeline handles single videos or entire playlists and scales from a laptop to production infrastructure.
What We're Building
YouTube playlist
↓
INDXR.AI RAG JSON export (chunked, metadata-rich)
↓
OpenAI text-embedding-3-small
↓
Pinecone (or ChromaDB for local development)
↓
Semantic search + LLM-generated answers with timestamp citationsThe key design choice: INDXR.AI handles the hard parts of transcript processing before the data enters this pipeline. Raw YouTube transcripts require merging 2–5 second fragments, cleaning filler words, detecting sentence boundaries, computing token counts, and attaching metadata per chunk. The RAG JSON export delivers all of that pre-processed. We write embedding and storage code, not transcript processing code.
Prerequisites
pip install openai pinecone chromadb tiktoken requestsYou'll need:
- An INDXR.AI account with credits for AI Transcription and RAG JSON export
- An OpenAI API key for embeddings
- For production: a Pinecone account (free tier available)
- For local development: ChromaDB runs in-process, no account needed
Step 1: Extract Transcripts as RAG JSON
Extract your target playlist in INDXR.AI's Playlist tab with the RAG JSON toggle enabled. Download the bulk ZIP file. The ZIP contains one JSON file per video. Each file has this structure:
{
"metadata": {
"video_id": "okHkUIW46ks",
"title": "How to Remember Everything You Read",
"duration_seconds": 1574,
"extracted_at": "2026-08-27T12:43:09.577Z",
"chunking_config": {
"chunk_size_seconds": 60,
"overlap_seconds": 9,
"overlap_strategy": "segment_boundary",
"total_chunks": 30
},
"channel": "Justin Sung",
"language": "en",
"extraction_method": "youtube_captions"
},
"chunks": [
{
"chunk_index": 0,
"chunk_id": "okHkUIW46ks_chunk_000",
"text": "in this video I'll teach you a system for remembering everything you read study or learn I've been using the system for the last 7 years …",
"start_time": 0.04,
"end_time": 61.519,
"deep_link": "https://youtu.be/okHkUIW46ks?t=0",
"token_count_estimate": 209,
"metadata": {
"video_id": "okHkUIW46ks",
"title": "How to Remember Everything You Read",
"channel": "Justin Sung",
"chunk_index": 0,
"start_time": 0.04,
"end_time": 61.519,
"language": "en",
"total_chunks": 30
}
}
]
}Each chunk is 60 seconds of speech by default (adjustable to 30, 90 or 120), which lands in the dense-retrieval range established by NVIDIA's chunking benchmark and the Vectara NAACL 2025 study. See How to Chunk YouTube Transcripts for RAG for the research behind these defaults.
Step 2: Load and Parse JSON Files
import json
import glob
import os
def load_rag_json_files(directory: str) -> list[dict]:
"""Load all RAG JSON files from a directory."""
all_chunks = []
for filepath in glob.glob(os.path.join(directory, "*.json")):
with open(filepath, encoding="utf-8") as f:
data = json.load(f)
video = data["metadata"]
chunks = data["chunks"]
for chunk in chunks:
enriched_metadata = {
**chunk["metadata"],
"deep_link": chunk["deep_link"],
"token_count": chunk["token_count_estimate"],
"extraction_method": video.get("extraction_method"),
}
all_chunks.append({
"id": chunk["chunk_id"],
"text": chunk["text"],
"metadata": enriched_metadata
})
print(f"Loaded {len(all_chunks)} chunks from {len(glob.glob(os.path.join(directory, '*.json')))} videos")
return all_chunks
chunks = load_rag_json_files("./transcripts/")Step 3a: Local Development with ChromaDB
ChromaDB runs in-process — no server, no account. Good for development and small corpora (under ~100,000 chunks).
import chromadb
from openai import OpenAI
openai_client = OpenAI()
chroma_client = chromadb.PersistentClient(path="./chroma_db")
collection = chroma_client.get_or_create_collection(
name="youtube_transcripts",
metadata={"hnsw:space": "cosine"}
)
def embed_batch(texts: list[str]) -> list[list[float]]:
response = openai_client.embeddings.create(
input=texts,
model="text-embedding-3-small"
)
return [item.embedding for item in response.data]
# Index all chunks
BATCH_SIZE = 100
for i in range(0, len(chunks), BATCH_SIZE):
batch = chunks[i:i + BATCH_SIZE]
texts = [c["text"] for c in batch]
ids = [c["id"] for c in batch]
metadatas = [c["metadata"] for c in batch]
embeddings = embed_batch(texts)
collection.add(
embeddings=embeddings,
documents=texts,
metadatas=metadatas,
ids=ids
)
print(f"Indexed {min(i + BATCH_SIZE, len(chunks))}/{len(chunks)} chunks")
print(f"Total indexed: {collection.count()}")Step 3b: Production with Pinecone
Pinecone scales to millions of vectors and supports filtering by metadata fields.
from pinecone import Pinecone, ServerlessSpec
pc = Pinecone(api_key="YOUR_PINECONE_API_KEY")
index_name = "youtube-transcripts"
if index_name not in [i.name for i in pc.list_indexes()]:
pc.create_index(
name=index_name,
dimension=1536, # text-embedding-3-small dimension
metric="cosine",
spec=ServerlessSpec(cloud="aws", region="us-east-1")
)
index = pc.Index(index_name)
BATCH_SIZE = 100
for i in range(0, len(chunks), BATCH_SIZE):
batch = chunks[i:i + BATCH_SIZE]
texts = [c["text"] for c in batch]
embeddings = embed_batch(texts)
vectors = [
{
"id": chunk["id"],
"values": embedding,
"metadata": {
**chunk["metadata"],
"text": chunk["text"]
}
}
for chunk, embedding in zip(batch, embeddings)
]
index.upsert(vectors=vectors)
print(f"Upserted {min(i + BATCH_SIZE, len(chunks))}/{len(chunks)} chunks")
stats = index.describe_index_stats()
print(f"Index stats: {stats.total_vector_count} vectors")Step 4: Query with Natural Language
def search(query: str, n_results: int = 5, filter_channel: str = None):
"""Search the knowledge base with optional channel filter."""
query_embedding = openai_client.embeddings.create(
input=[query],
model="text-embedding-3-small"
).data[0].embedding
where_filter = {}
if filter_channel:
where_filter = {"channel": {"$eq": filter_channel}}
results = collection.query(
query_embeddings=[query_embedding],
n_results=n_results,
where=where_filter if where_filter else None
)
output = []
for doc, metadata in zip(results["documents"][0], results["metadatas"][0]):
output.append({
"text": doc,
"title": metadata.get("title"),
"channel": metadata.get("channel"),
"deep_link": metadata.get("deep_link"),
"start_time": metadata.get("start_time"),
})
return output
results = search("how do attention mechanisms work in transformers")
for r in results:
print(f"\n{r['title']} — {r['deep_link']}")
print(f"{r['text'][:200]}...")Step 5: Add an LLM Answer Layer
def answer(question: str, n_chunks: int = 4) -> str:
"""Generate an answer with source citations."""
relevant_chunks = search(question, n_results=n_chunks)
context = "\n\n".join([
f"[Source: {c['title']} at {int(c['start_time'] // 60)}:{int(c['start_time'] % 60):02d}]\n{c['text']}"
for c in relevant_chunks
])
response = openai_client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "system",
"content": (
"Answer questions based on the provided YouTube transcript excerpts. "
"Cite the video title and timestamp for each claim. "
"If the context doesn't contain enough information, say so."
)
},
{
"role": "user",
"content": f"Context:\n{context}\n\nQuestion: {question}"
}
],
temperature=0.1
)
answer_text = response.choices[0].message.content
source_links = "\n".join([
f"- [{c['title']}]({c['deep_link']})"
for c in relevant_chunks
])
return f"{answer_text}\n\n**Sources:**\n{source_links}"
print(answer("Explain the role of positional encoding in transformers"))Embedding Cost Estimate
text-embedding-3-small costs $0.02 per million tokens. For a 20-video playlist with an average of 30 minutes per video:
- ~20 videos × 30 min × 2.5 chunks/min × 350 tokens/chunk = ~525,000 tokens
- Embedding cost: ~$0.01 for the full corpus
The embedding cost is negligible. The primary cost is INDXR.AI credits for AI Transcription (€0.025/credit at Plus, 1 credit/minute = ~€0.75 for 30 minutes). See pricing for package details.
Keeping the Index Current
INDXR.AI's duplicate detection means re-running a playlist extraction only processes new videos. Re-embed and upsert only the new chunks:
# Track already-indexed chunk IDs to avoid re-embedding
existing_ids = set() # Load from your tracking store
new_chunks = [c for c in load_rag_json_files("./new_transcripts/")
if c["id"] not in existing_ids]
if new_chunks:
print(f"Indexing {len(new_chunks)} new chunks")
# Index using same batching logic aboveTo start: extract any YouTube video as RAG JSON. For the chunking research behind the 60-second default, see How to Chunk YouTube Transcripts for RAG.