Retrieval augmented generation is only as good as the text you feed it. Generic web pages are noisy and shallow. YouTube transcripts, by contrast, are dense, spoken explanations from people who often know a subject deeply: conference talks, tutorials, lectures, and expert interviews. This tutorial walks through building a working RAG pipeline on top of YouTube transcripts, from fetching the source text to answering questions with citations.

We will keep the stack small and swappable: the YouTube transcript API for the source data, an embedding model, a vector store, and a chat model for the final answer.

TL;DR: Use TranscriptFetch’s YouTube Transcript API as the retrieval layer for a YouTube RAG pipeline, then split transcripts into passages, embed them, and retrieve relevant chunks before generating an answer. Keep video IDs and timestamps with each chunk so answers can cite the source moment. Save processed IDs and ingest new uploads without rebuilding the archive.

At a glance

The pipeline at a glance

Collect transcripts for the videos you care about.
Chunk each transcript into passages.
Embed every chunk and store the vectors.
Retrieve the most relevant chunks for a question.
Generate an answer grounded in those chunks.

Guide

Step 1: collect transcripts

Start by pulling transcripts for a topic. You can fetch a specific list of videos, or search for a keyword and ingest the top results. If you want every upload from one creator instead of a keyword search, pulling a whole channel is the same two steps.

python

import os, requests

KEY = os.environ["TRANSCRIPTFETCH_KEY"]
HEAD = {"Authorization": f"Bearer {KEY}", "Content-Type": "application/json"}

def search_videos(query, limit=10):
    r = requests.post("https://transcriptfetch.com/api/transcripts", headers=HEAD,
                      json={"endpoint": "search", "value": query, "limit": limit})
    return [v["videoId"] for v in r.json().get("videos", [])]

def fetch_transcripts(ids):
    r = requests.post("https://transcriptfetch.com/api/transcripts/batch", headers=HEAD,
                      json={"videoIds": ids})
    return [x for x in r.json()["results"] if x["outcome"] == "ok" and x["text"]]

ids = search_videos("retrieval augmented generation explained")
docs = fetch_transcripts(ids)

You now have a list of documents, each with a videoId, title, and full text.

Guide

Step 2: chunk the transcripts

Transcripts are long, and embedding a whole video as one vector loses detail. Split each transcript into overlapping passages of a few hundred words. The overlap keeps ideas that straddle a boundary intact.

python

def chunk_text(text, size=900, overlap=150):
    words = text.split()
    step = size - overlap
    for i in range(0, len(words), step):
        piece = " ".join(words[i:i + size])
        if piece:
            yield piece

passages = []
for d in docs:
    for piece in chunk_text(d["text"]):
        passages.append({
            "video_id": d["videoId"],
            "title": d["title"],
            "text": piece,
        })

If you store the timestamped segments alongside each chunk, you can later link an answer back to the exact moment in the video. That is one of the nicest properties of video as a source.

Guide

Step 3: embed and store

Turn each passage into a vector and keep it in a store. For a prototype, an in memory list with cosine similarity is enough. For production, use a real vector database.

python

from openai import OpenAI
import numpy as np

client = OpenAI()

def embed(texts):
    resp = client.embeddings.create(model="text-embedding-3-small", input=texts)
    return [np.array(d.embedding) for d in resp.data]

vectors = embed([p["text"] for p in passages])
for p, v in zip(passages, vectors):
    p["vector"] = v

Guide

Step 4: retrieve

Given a question, embed it and find the closest passages.

python

def cosine(a, b):
    return float(a @ b / (np.linalg.norm(a) * np.linalg.norm(b)))

def retrieve(question, k=5):
    qv = embed([question])[0]
    ranked = sorted(passages, key=lambda p: cosine(qv, p["vector"]), reverse=True)
    return ranked[:k]

Guide

Step 5: generate a grounded answer

Feed the retrieved passages to the chat model as context, and ask it to cite the videos it used.

python

def answer(question):
    hits = retrieve(question)
    context = "\n\n".join(
        f"[{h['title']}] {h['text']}" for h in hits
    )
    prompt = (
        "Answer the question using only the context below. "
        "Cite the video titles you used.\n\n"
        f"Context:\n{context}\n\nQuestion: {question}"
    )
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
    )
    return resp.choices[0].message.content

print(answer("What problem does RAG solve that fine-tuning does not?"))

Guide

Keeping the index fresh

A channel keeps publishing, so a one time ingest goes stale. Schedule the collection step to run on a cadence, list new uploads with the channel endpoint, and only embed transcripts you have not seen before. Because repeat fetches are served from cache, re-running the job is cheap.

Guide

Why transcripts work well for RAG

Density. A 20 minute talk often contains more focused explanation than a dozen blog posts.
Recency. Conference and tutorial videos cover new tools long before the books catch up.
Citations people trust. Linking an answer to a timestamped moment in a known video is more convincing than a generic URL.

Guide

Wrapping up

RAG on YouTube transcripts is the same five steps as any RAG system, with one advantage: the source material is unusually rich. Fetch with a batch call, chunk with overlap, embed, retrieve, and ground the answer. If you want the fetch step on its own first, fetching transcripts in Python covers it without the vector store. Swap the in memory store for a real vector database when you outgrow the prototype, and schedule the ingest so your index stays current.