对整个 YouTube 频道做 RAG：从 RSS 到带引用的答案

对一个 YouTube 频道做 RAG，意味着回答"这位创作者对 X 说过什么、在哪个视频里说的？"——覆盖他们发布过的所有内容。流水线分四步：通过 RSS 发现视频、获取转录文本、分块并嵌入、带出处地回答。下面每一步都是可直接运行的 Python，还包括大多数文章跳过的部分：配额处理，以及在没有时间戳的情况下引用到底能做到什么程度。

第一步：从频道 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 是完美的——它也是频道转 Newsletter 流水线的骨架。

第二步：获取转录文本，尊重配额

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 个，pro 档（$9.99）是 500 个——要索引一个频道及其持续更新，pro 是现实的起点。

第三步：分块并嵌入

任何向量库都可以；用 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"]}],
        )

第四步：带引用地回答——诚实版

这个 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 个视频。