YouTubeチャンネル全体でRAGを構築: RSSから出典付き回答まで

YouTubeチャンネル全体でRAGを構築するというのは、「このクリエイターはXについて何を語っていて、それはどの動画でか?」という問いに、公開された全動画を対象として答えられるようにすることです。パイプラインは4ステップ: RSSで動画を発見し、文字起こしを取得し、チャンク分割して埋め込み、出典付きで回答します。この記事では各ステップを実行可能なPythonで示します。多くの解説記事が省略する部分 — クォータの扱いと、タイムスタンプなしでの引用が正直なところどうなるか — も含めて解説します。

ステップ1: チャンネルのRSSフィードから動画IDを取得

YouTubeは今でもチャンネルごとに本物のRSSフィードを提供しており、APIキーは不要です。

import requests
import xml.etree.ElementTree as ET

CHANNEL_ID = "UCxxxxxxxxxxxxxxxxxxxxxx"
NS = "{http://www.youtube.com/xml/schemas/2015}"

feed = requests.get(
    f"https://www.youtube.com/feeds/videos.xml?channel_id={CHANNEL_ID}",
    timeout=30,
).text
video_ids = [el.text for el in ET.fromstring(feed).iter(f"{NS}videoId")]

注意点: フィードには直近およそ15本のアップロードしか含まれません。活発なチャンネルなら、これを定期実行してIDをファイルやテーブルに蓄積してください。それより古いバックカタログには、YouTube Data APIか一度きりの手動リストが必要です。ただし、インデックスを最新に保つ用途にはRSSは最適で、チャンネルからニュースレターへのパイプラインの屋台骨でもあります。

ステップ2: 文字起こしを取得し、クォータを尊重する

import time

API_KEY = "yt_your_key_here"

def fetch(video_id: str) -> dict | None:
    resp = requests.get(
        "https://youtube2text.org/api/transcribe",
        params={"url": f"https://www.youtube.com/watch?v={video_id}"},
        headers={"x-api-key": API_KEY},
        timeout=90,
    )
    if resp.status_code == 404:
        return None  # TRANSCRIPT_UNAVAILABLE — no captions, skip
    if resp.status_code == 429:
        time.sleep(resp.json()["error"].get("retryAfterSeconds", 30))
        return fetch(video_id)
    resp.raise_for_status()
    return resp.json()["result"]

transcripts = [t for vid in video_ids if (t := fetch(vid))]

チャンネル全体をループする前に、自分のプランで計算しておきましょう。無料プランは月5本で、概念実証には足りますがそれ以上は無理です。ベーシック($5.99)で50本、プロ($9.99)で500本 — チャンネルのインデックス構築と継続的な新着処理を両立させるなら、現実的な下限はプロです。

ステップ3: チャンク分割と埋め込み

ベクトルストアは何でも構いませんが、例を短くするためにChromaを使います。自動字幕には句読点が欠けていることが多いので、文単位ではなく単語数でチャンク分割しています。

import chromadb

def chunk(text: str, size: int = 300, overlap: int = 50):
    words = text.split()
    for i in range(0, len(words), size - overlap):
        yield " ".join(words[i:i + size])

client = chromadb.PersistentClient(path="./channel_index")
col = client.get_or_create_collection("channel")

for t in transcripts:
    for i, piece in enumerate(chunk(t["content"])):
        col.add(
            ids=[f'{t["videoId"]}-{i}'],
            documents=[piece],
            metadatas=[{"videoId": t["videoId"], "title": t["title"]}],
        )

ステップ4: 出典付きで回答する — 正直バージョン

APIはタイムスタンプなしのプレーンテキストを返すので、率直に言います。このデータから「23分41秒の時点で」といった引用はできません。できること — そして通常はそれで十分なこと — は、チャンクレベルの動画帰属です。取得された各チャンクは自分の videoId と title を知っているので、回答は「どの動画がそう言ったか」を、有効なリンク付きで引用できます。

hits = col.query(query_texts=["what's their take on electric trucks?"], n_results=5)

context = "\n\n".join(
    f'[{m["title"]}] (https://www.youtube.com/watch?v={m["videoId"]})\n{doc}'
    for doc, m in zip(hits["documents"][0], hits["metadatas"][0])
)
prompt = (
    "Answer using only the excerpts below. After each claim, "
    f"cite the video title in brackets.\n\n{context}\n\nQuestion: ..."
)

prompt はお好みのLLMに渡してください。ローカルモデルでも問題なく動きます。セットアップはOllamaでの要約パイプラインと同じです。

組み立てのメモ

フレームワーク派の方へ: チャンク分割を手書きしたくなければ、この構成はLlamaIndexやLangChainにそのまま一対一で対応します。いずれにせよ文字起こしのレイヤーは同一で、そこだけが従量制です。youtube2text.org/app/keys でキーを取得し、300本の動画に向ける前に、まず無料プランでパイプラインを検証しましょう。