Build a RAG Pipeline on YouTube Transcripts

Retrieval-Augmented Generation works best when your knowledge source is dense, well-structured, and citable. YouTube transcripts check every box. There are over 800 million videos on YouTube, with more than 500 hours of new content uploaded every minute—a corpus that dwarfs most enterprise document stores. The catch is getting that content out reliably and into a form your vector database can index.

This guide walks through the complete pipeline: fetching transcripts with timestamps, chunking them intelligently, embedding and storing in a vector DB, retrieving with deep-link citations, and keeping the index fresh as channels publish new videos.

Why Video Transcripts Make Great RAG Sources

Most RAG tutorials start with PDFs or web pages. Transcripts are better for several reasons:

• Dense signal, minimal noise. Transcripts contain almost no navigation chrome, cookie banners, or unrelated boilerplate. Nearly every token is content.
• Built-in segmentation. Each segment comes with start and duration fields, so you always know where in the video a passage lives.
• Citable by design. Because segments carry timestamps, you can generate a deep link (e.g., https://youtu.be/VIDEO_ID?t=142) that takes users directly to the moment a retrieved chunk was spoken. That's a citation UX most document RAG systems can't match.
• Multilingual. A single video often has caption tracks in dozens of languages. You can build a multilingual index from the same source video by requesting different language codes.

The main friction isn't the data quality—it's getting the data out. YouTube aggressively rate-limits and blocks automated requests, and open-source libraries like youtube-transcript-api go down regularly when YouTube rotates its anti-bot measures. A managed API that handles proxy rotation and retries for you is worth it once you're indexing more than a handful of videos.

Fetching Transcripts with Timestamps

The YouTube Transcriber transcript endpoint (GET /api/v2/transcript) returns segments with text, start (seconds), and duration (seconds) out of the box. Here's a minimal fetch in Python:

import httpx, os

API_KEY = os.environ["YT_TRANSCRIBER_KEY"]
BASE    = "https://getyoutubetranscriber.com"

def fetch_transcript(video_id: str, lang: str = "en") -> dict:
    resp = httpx.get(
        f"{BASE}/api/v2/transcript",
        params={
            "video_url": video_id,
            "lang": lang,
            "include_timestamp": "true",
            "send_metadata": "true",
        },
        headers={"Authorization": f"Bearer {API_KEY}"},
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json()

data = fetch_transcript("dQw4w9WgXcQ")
# data["transcript"] → [{"text": "...", "start": 0, "duration": 2.5}, ...]
# data["metadata"]   → {"title": "...", "author_name": "...", ...}

The response looks like this:

{
  "video_id": "dQw4w9WgXcQ",
  "language": "en",
  "transcript": [
    { "text": "Never gonna give you up", "start": 0,   "duration": 2.5 },
    { "text": "Never gonna let you down", "start": 2.5, "duration": 2.3 }
  ],
  "metadata": {
    "title": "Rick Astley - Never Gonna Give You Up",
    "author_name": "Rick Astley",
    "author_url": "https://youtube.com/@RickAstleyYT"
  }
}

If you're building a dataset from a large channel, use the bulk endpoint (POST /api/v2/transcripts/bulk), which accepts up to 50 video IDs per request and is significantly more efficient than one-at-a-time calls.

Chunking for Retrieval

Raw transcript segments are too granular to embed well—a 2-second segment is rarely a semantically complete thought. The standard approach is to group segments into token-sized chunks while keeping the earliest start timestamp for each chunk as metadata.

import tiktoken

def chunk_transcript(
    segments: list[dict],
    video_id: str,
    max_tokens: int = 300,
    overlap_tokens: int = 50,
) -> list[dict]:
    enc = tiktoken.get_encoding("cl100k_base")
    chunks, current_text, current_start, current_tokens = [], [], None, 0

    for seg in segments:
        tokens = enc.encode(seg["text"])
        if current_start is None:
            current_start = seg["start"]

        if current_tokens + len(tokens) > max_tokens and current_text:
            chunk_text = " ".join(current_text)
            chunks.append({
                "text":     chunk_text,
                "video_id": video_id,
                "start":    current_start,
                "cite_url": f"https://youtu.be/{video_id}?t={int(current_start)}",
            })
            # overlap: keep last N tokens worth of segments
            overlap_text = chunk_text.split()[-overlap_tokens:]
            current_text   = overlap_text
            current_tokens = len(enc.encode(" ".join(overlap_text)))
            current_start  = seg["start"]

        current_text.append(seg["text"])
        current_tokens += len(tokens)

    if current_text:
        chunks.append({
            "text":     " ".join(current_text),
            "video_id": video_id,
            "start":    current_start,
            "cite_url": f"https://youtu.be/{video_id}?t={int(current_start)}",
        })

    return chunks

Gotcha: Don't strip start when you do the overlap—carry the first segment's timestamp forward into the new chunk, not the overlap segment's timestamp. Your citation link should point to where the new thought starts.

A 200–400 token window with 50-token overlap is a good default. Dense lecture content can push to 512; short-form video with fast cuts may need 150 or less. Benchmark against your actual query set.

Embedding and Storing in a Vector Database

With chunks in hand, embed them and upsert into your vector store. This example uses OpenAI embeddings and Pinecone, but the pattern is the same for Weaviate, Qdrant, or pgvector:

from openai import OpenAI
from pinecone import Pinecone, ServerlessSpec

openai_client = OpenAI()
pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])

index = pc.Index("youtube-rag")

def embed_and_upsert(chunks: list[dict]) -> None:
    texts = [c["text"] for c in chunks]
    resp  = openai_client.embeddings.create(
        input=texts,
        model="text-embedding-3-small",
    )
    vectors = []
    for i, chunk in enumerate(chunks):
        vectors.append({
            "id":     f"{chunk['video_id']}_{int(chunk['start'])}",
            "values": resp.data[i].embedding,
            "metadata": {
                "text":     chunk["text"],
                "video_id": chunk["video_id"],
                "start":    chunk["start"],
                "cite_url": chunk["cite_url"],
            },
        })
    index.upsert(vectors=vectors)

Store cite_url in metadata—it's tiny and it means you never have to reconstruct the deep link at query time.

According to Pinecone's RAG guide, keeping source metadata co-located with vectors is critical for production pipelines because it eliminates a secondary database lookup on every retrieval. Transcript chunks make this easy because all the metadata you need (video_id, start) fits in a small, flat record.

Retrieving with Citations Back to Video Moments

At query time, embed the user's question, retrieve the top-K chunks, and pass them to your LLM with an instruction to include the citation URL in its answer:

def retrieve_and_answer(question: str, top_k: int = 5) -> str:
    q_embedding = openai_client.embeddings.create(
        input=[question],
        model="text-embedding-3-small",
    ).data[0].embedding

    results = index.query(
        vector=q_embedding,
        top_k=top_k,
        include_metadata=True,
    )

    context_blocks = []
    for match in results.matches:
        md = match.metadata
        context_blocks.append(
            f"[Source: {md['cite_url']}]\n{md['text']}"
        )
    context = "\n\n---\n\n".join(context_blocks)

    messages = [
        {"role": "system", "content": (
            "Answer using only the provided context. "
            "Include the source URL in your answer so users can verify."
        )},
        {"role": "user", "content": f"Context:\n{context}\n\nQuestion: {question}"},
    ]
    resp = openai_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages,
    )
    return resp.choices[0].message.content

The LLM will naturally weave the [Source: https://youtu.be/VIDEO_ID?t=142] URL into its answer because it appears right next to the relevant text in the context. Users can click the link and land on the exact moment the information was spoken—that's a trust signal no web-page RAG system easily replicates.

Keeping the Index Fresh as Channels Post

A static index goes stale fast on active channels. The /api/v2/channel/latest endpoint is backed by YouTube's public RSS feed and returns the most recent uploads for any channel:

def poll_and_index_new_videos(channel_handle: str, seen_ids: set[str]) -> set[str]:
    resp = httpx.get(
        f"{BASE}/api/v2/channel/latest",
        params={"channel": channel_handle},
        headers={"Authorization": f"Bearer {API_KEY}"},
        timeout=15,
    )
    resp.raise_for_status()
    data = resp.json()

    new_ids = set()
    for video in data["results"]:
        vid_id = video["video_id"]
        if vid_id not in seen_ids:
            transcript_data = fetch_transcript(vid_id)
            if "transcript" in transcript_data:
                chunks = chunk_transcript(transcript_data["transcript"], vid_id)
                embed_and_upsert(chunks)
                new_ids.add(vid_id)

    return seen_ids | new_ids

Run this on a cron or in a background worker. For channels that post daily, a 15–30 minute polling interval is sufficient. The endpoint is free, so the cost is purely your ingestion throughput and embedding calls for new videos.

For a full treatment of monitoring patterns—including webhook-style architectures and channel resolution from @handles—see Monitor a YouTube Channel for New Videos via API.

End-to-End Pipeline Summary

Here's the complete flow stitched together:

import schedule, time

CHANNELS    = ["@lexfridman", "@3blue1brown"]
seen_videos: set[str] = set()

def ingest_all():
    global seen_videos
    for channel in CHANNELS:
        seen_videos = poll_and_index_new_videos(channel, seen_videos)

# Initial full ingest (optional: pull all channel videos, not just latest)
ingest_all()

# Keep fresh
schedule.every(30).minutes.do(ingest_all)
while True:
    schedule.run_pending()
    time.sleep(60)

The key design points:

1. Fetch via /api/v2/transcript with include_timestamp=true
2. Chunk into 200–400 token windows, preserving start as metadata
3. Embed with a consistent model (don't mix embedding models within an index)
4. Upsert with cite_url in vector metadata so retrieval is self-contained
5. Poll /api/v2/channel/latest on a schedule for freshness

According to LangChain's RAG documentation, metadata filtering at retrieval time—for example, restricting search to a specific video_id or channel—is one of the most effective ways to improve precision. The flat metadata structure above integrates cleanly with LangChain's SelfQueryRetriever or Pinecone's filter syntax.

Practical Considerations

Token costs add up. A 1-hour lecture transcript runs roughly 10,000–15,000 tokens. At text-embedding-3-small pricing, embedding a hundred such videos costs under a dollar. Watch your chunking overlap ratio—50 tokens of overlap on a 300-token chunk means ~17% redundancy, which is reasonable.

Deduplication matters for bulk channel ingestion. Store embedded video_id values in a persistent set (Redis, a simple SQLite table, or your vector DB's ID namespace) so reruns don't double-embed videos.

Caption quality varies. Auto-generated captions on technical content can have transcription errors for domain-specific terms. If accuracy is critical, consider requesting a manual caption track by passing lang explicitly, or post-processing with a spell-correction step tuned to your domain vocabulary.

Language selection. Pass lang=fr, lang=de, or any BCP-47 code to the transcript endpoint to request a specific caption track. For multilingual indexes, use a multilingual embedding model (e.g., multilingual-e5-large) and store chunks from all languages in the same index.

Next Steps

The pipeline above is production-ready at small to medium scale. To push further:

• Bulk ingest an entire channel's history. The /api/v2/channel/videos endpoint returns paginated video lists; combine with the bulk transcript endpoint to process a channel's full archive in parallel batches.
• Add a summarization layer. Before embedding, run each transcript through a summarizer to create a second, higher-level chunk alongside the raw segments—useful for broad queries that need an overview rather than a specific passage.
• Hybrid search. Combine dense vector search with BM25 keyword search (Pinecone hybrid, Weaviate BM25, or Elasticsearch) to improve recall on proper nouns and technical terms that embeddings sometimes miss.

Start with the 100 free credits at getyoutubetranscriber.com—no credit card required—and the transcript and channel endpoints cover everything this pipeline needs.