How to Build a RAG Pipeline with YouTube Transcripts

ダッシュボードからキーを取得し（100無料クレジット、カード不要）、各動画の文字起こしをIDで取得します。クレジットは動画の長さではなくリクエストごとに課金されるため、3時間の講義動画も2分のクリップも同じ1クレジットです - チャンネルのバックカタログ全体を取り込む際に重要になります:

async function fetchTranscript(videoId: string) {
  const res = await fetch(
    `https://getyoutubetranscript.com/api/v1/transcript?v=${videoId}`,
    { headers: { Authorization: `Bearer ${process.env.GETYOUTUBETRANSCRIPT_API_KEY}` } }
  );
  const json = await res.json();
  if (!json.success) throw new Error(json.message);
  return json.data as { video_id: string; title: string; transcript: string; word_count: number };
}

const videoIds = ['dQw4w9WgXcQ', 'jNQXAC9IVRw' /* ... */];
const videos = await Promise.all(videoIds.map(fetchTranscript));

video_idとtitleが文字起こしテキストと一緒に返ってきます - これがステップ5で回答を引用する対象です。

エンベディングする前に、各文字起こしを重複のあるウィンドウに分割します - 動画全体を1つのエンベディングにすると、特定の回答を取得するには粗すぎます。これにライブラリは不要です:

function chunkText(text: string, size = 1200, overlap = 200) {
  const chunks: string[] = [];
  let start = 0;
  while (start < text.length) {
    const end = Math.min(start + size, text.length);
    chunks.push(text.slice(start, end));
    start += size - overlap;
  }
  return chunks;
}

const records = videos.flatMap((video) =>
  chunkText(video.transcript).map((text, chunkIndex) => ({
    videoId: video.video_id,
    title: video.title,
    chunkIndex,
    text,
  }))
);

1,200文字のウィンドウに200文字のオーバーラップは、話し言葉の文字起こしに対する妥当なデフォルトです - 短く密度の高いクリップでは小さくし、より多くの周辺コンテキストが役立つ長い講義形式のコンテンツでは大きくしてください。

各チャンクをエンベディングし、動画IDとタイトルをベクトルと一緒に保持します。これが後で取得したチャンクを引用に変えるものだからです:

import OpenAI from 'openai';
const openai = new OpenAI();

async function embed(text: string) {
  const res = await openai.embeddings.create({ model: 'text-embedding-3-small', input: text });
  return res.data[0].embedding;
}

const index = await Promise.all(
  records.map(async (r) => ({ ...r, embedding: await embed(r.text) }))
);

ここではわかりやすさのためベクトルをメモリ内に保持しています。数百チャンクを超えたら、この配列をpgvector、Pinecone、または他のベクトルストアに置き換えてください - 以下の取得ステップは、この正確な形ではなく、上位k件の最も近いベクトルを返す関数だけを必要とします。

クエリ時には、同じモデルで質問をエンベディングし、コサイン類似度で保存済みのチャンクをランク付けします:

function cosineSimilarity(a: number[], b: number[]) {
  let dot = 0, normA = 0, normB = 0;
  for (let i = 0; i < a.length; i++) {
    dot += a[i] * b[i];
    normA += a[i] ** 2;
    normB += b[i] ** 2;
  }
  return dot / (Math.sqrt(normA) * Math.sqrt(normB));
}

async function retrieve(query: string, k = 5) {
  const queryEmbedding = await embed(query);
  return index
    .map((r) => ({ ...r, score: cosineSimilarity(r.embedding, queryEmbedding) }))
    .sort((a, b) => b.score - a.score)
    .slice(0, k);
}

k=5は出発点です - チャンクが少なすぎるとモデルが十分な文脈を得られず完全に答えられませんし、多すぎると無関係なチャンクが回答を薄めてしまいます。

取得したチャンクを番号付きコンテキストとしてモデルに渡し、そこにある内容だけから回答し、出典を引用するよう指示します - これにより、回答がモデル自体の一般知識ではなく実在する動画にまで遡って追跡できるようになります:

async function answer(query: string) {
  const matches = await retrieve(query);
  const context = matches
    .map((m, i) => `[${i + 1}] From "${m.title}":\n${m.text}`)
    .join('\n\n');

  const completion = await openai.chat.completions.create({
    model: 'gpt-4.1-mini',
    messages: [
      {
        role: 'system',
        content:
          "Answer only using the numbered transcript excerpts below. Cite sources as [1], [2], etc. " +
          "If the excerpts don't contain the answer, say so.",
      },
      { role: 'user', content: `${context}\n\nQuestion: ${query}` },
    ],
  });

  return completion.choices[0].message.content;
}

タイムスタンプをでっち上げるのではなく動画タイトルで引用することは、文字起こしAPIが実際に返す内容と一致します - モデルに与えられていないタイムスタンプを捏造させないでください。

更新が続くチャンネルについては、/channel/latestエンドポイント（1回の確認につき1クレジット）でスケジュールに沿って新しいアップロードを確認し、すでにインデックス済みの動画IDをスキップして、新しいものだけを取得・分割・エンベディングします。これにより、取り込みコストは毎回のフル再インデックスではなく、新しいコンテンツに比例したものになります。

動画は検索対象として構築するにはやや扱いにくい形式の一つです - 動画をgrepすることはできません。文字起こしは、それを既存のエンベディングと検索スタックがすでに扱い方を知っているテキストに戻してくれます。しかも自前で音声認識ジョブを実行する必要もありません。さらに、リクエストごとに課金されるAPIは動画の長さで従量課金しないため、長尺コンテンツ（ポッドキャスト、講義、数時間に及ぶ配信）のバックログを取り込んでも、短いクリップを取り込むより体系的に高くつくことはありません - これはまさにRAGパイプラインが通常検索対象として構築されるアーカイブの種類です。