How to build a RAG pipeline with YouTube transcripts

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: TranscriptFetch for the source data, an embedding model, a vector store, and a chat model for the final answer.

The pipeline at a glance

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

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.

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.

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.

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.

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.

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

Step 4: retrieve

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

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]

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.

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?"))

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.

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.

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. Swap the in memory store for a real vector database when you outgrow the prototype, and schedule the ingest so your index stays current.