RAG na kanale YouTube: od RSS do odpowiedzi z cytowaniami

Zbudowanie RAG na kanale YouTube oznacza odpowiadanie na pytanie „co ten twórca powiedział o X i w którym filmie?" — w obrębie wszystkiego, co opublikował. Pipeline to cztery kroki: odkrywanie filmów przez RSS, pobieranie transkrypcji, dzielenie na fragmenty i embeddingi, odpowiadanie z atrybucją. Oto każdy krok w działającym Pythonie, łącznie z częściami, które większość opracowań pomija: obsługą limitów i tym, jak uczciwie wygląda cytowanie bez znaczników czasu.

Krok 1: identyfikatory filmów z kanału RSS

YouTube wciąż udostępnia prawdziwy feed RSS dla każdego kanału, bez klucza 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")]

Zastrzeżenie: feed zwraca tylko ~15 najnowszych uploadów. Dla aktywnego kanału uruchamiaj to według harmonogramu i gromadź identyfikatory w pliku albo tabeli; starsze archiwum wymaga YouTube Data API albo jednorazowej ręcznej listy. Do utrzymywania świeżości indeksu RSS jest jednak idealny — to także kręgosłup pipeline'u od kanału do newslettera.

Krok 2: pobierz transkrypcje, szanuj limit

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))]

Policz swój plan, zanim puścisz pętlę po kanale: darmowy poziom to 5 filmów miesięcznie, co wystarcza na proof of concept i nic więcej. Basic (5,99 USD) kupuje 50, pro (9,99 USD) kupuje 500 — pro to realistyczne minimum do zaindeksowania kanału wraz z jego bieżącymi uploadami.

Krok 3: fragmenty i embeddingi

Zadziała dowolny magazyn wektorów; Chroma trzyma przykład krótko. Automatyczne napisy często nie mają interpunkcji, więc dzielę po liczbie słów zamiast po zdaniach:

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"]}],
        )

Krok 4: odpowiedzi z cytowaniami — wersja uczciwa

API zwraca czysty tekst bez znaczników czasu, więc powiem wprost: z tych danych nie zacytujesz „w 23:41". Co możesz zrobić — i co zwykle wystarcza — to atrybucja do filmu na poziomie fragmentu: każdy pobrany fragment zna swoje videoId i title, więc odpowiedzi wskazują, który film to powiedział, z działającym linkiem.

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: ..."
)

Przekaż prompt do dowolnego LLM — model lokalny sprawdzi się tu świetnie, ta sama konfiguracja co w potoku streszczania z Ollama.

Notatki montażowe

Zwolennicy frameworków: to mapuje się jeden do jednego na LlamaIndex albo LangChain, jeśli wolisz nie składać chunkingu ręcznie. Tak czy inaczej warstwa transkrypcji jest identyczna i to jedyna część z licznikiem — odbierz klucz na youtube2text.org/app/keys i zacznij od darmowego poziomu, żeby zweryfikować pipeline, zanim wycelujesz go w 300 filmów.