YouTube 채널로 RAG 구축하기: RSS에서 출처 있는 답변까지

YouTube 채널로 RAG를 구축한다는 것은 "이 크리에이터가 X에 대해 뭐라고 했고, 어느 영상에서 했나?"라는 질문에 그가 게시한 모든 콘텐츠에 걸쳐 답한다는 뜻입니다. 파이프라인은 네 단계입니다. 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개 영상에 들이대기 전에 무료 요금제로 파이프라인부터 검증하세요.