```html
```
RAG（Retrieval-Augmented Generation）システムは、学習のために質の高いデータを必要とします。YouTubeは、字幕、メタデータ、コメントを含む構造化されたコンテンツの巨大なソースです。本記事では、ブロックを回避し、APIの制限を守りながら、YouTubeデータをRAGのために効率的に収集する方法を解説します。
RAGとは何か、YouTubeデータが必要な理由
RAG（Retrieval-Augmented Generation）は、AIシステムを構築するためのアプローチであり、言語モデルが知識ベースで補完されます。モデルが訓練されたデータにのみ依存するのではなく、RAGは外部ソースから関連情報を抽出し、それを使用して回答を生成します。
YouTubeには、さまざまな言語の字幕付きで数百万時間のコンテンツが含まれています。これにより、さまざまな分野のRAGシステムにとって貴重なデータソースとなります：
- 教育システム — 講義、チュートリアル、タイムスタンプ付きのコース
- 技術文書 — プログラミング、DevOps、ソフトウェア設定に関するビデオガイド
- 医療知識ベース — 医師の講義、臨床ケースの分析
- ビジネス分析 — 専門家とのインタビュー、ケーススタディ、市場レビュー
- 製品サポート — 製品レビュー、動画形式のFAQ
YouTubeデータの利点は、構造があることです：タイムスタンプ付きの字幕、メタデータ（カテゴリ、タグ）、社会的文脈（コメント、いいね）。これにより、RAGシステムは情報の内容だけでなく、文脈も理解するのに役立ちます。
RAGシステムに有用なYouTubeデータの種類
RAGシステムを効果的に機能させるためには、いくつかのデータタイプを収集する必要があります。各タイプは、情報の抽出と生成のプロセスでそれぞれのタスクを解決します。
字幕（Transcripts）
テキストデータの主要なソースです。YouTubeは2種類の字幕を提供しています：
- 自動字幕 — Googleの音声認識アルゴリズムによって生成されます。英語や他の人気のある言語のほとんどの動画で利用可能です。音質に応じて85-95％の精度です。
- 手動字幕 — 作者やコミュニティによってアップロードされます。より正確で、しばしばフォーマットや追加の文脈を含みます。
字幕にはタイムスタンプが含まれており、テキストを動画の特定の瞬間に関連付けることができます。これは、RAGの回答での情報源への正確なリンクを作成するために重要です。
動画のメタデータ
メタデータは、RAGシステムが情報の文脈と関連性を理解するのに役立ちます：
| データタイプ | RAGでの用途 |
|---|---|
| タイトルと説明 | セマンティック検索、テーマの特定 |
| タグとカテゴリ | コンテンツの分類、フィルタリング |
| 公開日 | 情報の関連性（技術的なテーマに重要） |
| 長さ | テーマの深さの評価 |
| 統計（視聴回数、いいね） | 情報源の質と人気の評価 |
| チャンネル情報 | 情報源の権威性の特定 |
コメント
コメントには追加の文脈が含まれています：視聴者の質問、作者の明確化、ディスカッション。RAGシステムにとってこれは貴重です：
- コメントには動画のテーマに関するFAQが含まれていることが多い
- 作者が修正や追加を投稿することができる
- ディスカッションは問題に対する異なる視点を明らかにする
YouTube Data API v3の操作：設定と制限
YouTube Data API v3は、データを取得するための公式な方法です。メタデータ、統計、コメントにアクセスできます。字幕は別のメソッドを通じて取得されます。
APIキーの取得
APIを使用するには、Google Cloud Consoleからキーが必要です：
- console.cloud.google.comに移動します
- 新しいプロジェクトを作成するか、既存のプロジェクトを選択します
- 「APIs & Services」セクションでYouTube Data API v3を有効にします
- 資格情報（Credentials）を作成 → APIキー
- キーをコピーします — すべてのリクエストに必要です
制限とクォータ
YouTube APIはクォータシステムを使用しています。各リクエストには特定の単位数が「かかります」：
| 操作 | クォータのコスト |
|---|---|
| 動画検索（search.list） | 100単位 |
| 動画データの取得（videos.list） | 1単位 |
| コメントの取得（commentThreads.list） | 1単位 |
デフォルトの1日の制限は10,000単位です。これは約100の検索リクエストまたは10,000のメタデータリクエストに相当します。クォータを増やすには、Googleに申請する必要があります。
APIの基本的な使用例
import requests
API_KEY = 'あなたの_api_キー'
BASE_URL = 'https://www.googleapis.com/youtube/v3'
# クエリによる動画検索
def search_videos(query, max_results=10):
url = f'{BASE_URL}/search'
params = {
'part': 'snippet',
'q': query,
'type': 'video',
'maxResults': max_results,
'key': API_KEY
}
response = requests.get(url, params=params)
return response.json()
# 動画のメタデータを取得
def get_video_details(video_id):
url = f'{BASE_URL}/videos'
params = {
'part': 'snippet,contentDetails,statistics',
'id': video_id,
'key': API_KEY
}
response = requests.get(url, params=params)
return response.json()
# 使用例
results = search_videos('機械学習 チュートリアル', max_results=5)
for item in results.get('items', []):
video_id = item['id']['videoId']
title = item['snippet']['title']
print(f'ID: {video_id}, タイトル: {title}')
# 詳細情報を取得
details = get_video_details(video_id)
stats = details['items'][0]['statistics']
print(f"視聴回数: {stats.get('viewCount')}, いいね: {stats.get('likeCount')}")
動画の字幕のパース：自動と手動
YouTube Data API v3は字幕への直接アクセスを提供していません。字幕を取得するためには代替手段を使用します。
youtube-transcript-apiライブラリの使用
最も簡単な方法は、Python用のライブラリyoutube-transcript-apiです。これは、APIキーなしで字幕を直接抽出します：
from youtube_transcript_api import YouTubeTranscriptApi
# 字幕の取得
video_id = 'dQw4w9WgXcQ'
try:
# ロシア語の字幕を取得しようとし、なければ英語の字幕を取得
transcript = YouTubeTranscriptApi.get_transcript(video_id, languages=['ru', 'en'])
# タイムスタンプ付きの字幕を出力
for entry in transcript:
start_time = entry['start']
duration = entry['duration']
text = entry['text']
print(f"[{start_time:.2f}s] {text}")
except Exception as e:
print(f"字幕取得エラー: {e}")
# 利用可能な言語のリストを取得
transcript_list = YouTubeTranscriptApi.list_transcripts(video_id)
for transcript in transcript_list:
print(f"言語: {transcript.language}, 自動生成: {transcript.is_generated}")
このライブラリは自動的に利用可能な字幕を特定し、YouTubeがそのような機能を提供している場合、他の言語に翻訳することもできます。
RAGのためのタイムスタンプの処理
RAGシステムにとって、テキストとタイムスタンプの関連を保持することが重要です。これにより、情報源への正確なリンクを作成できます：
def format_timestamp(seconds):
"""秒をMM:SS形式に変換"""
minutes = int(seconds // 60)
secs = int(seconds % 60)
return f"{minutes:02d}:{secs:02d}"
def create_chunks_with_timestamps(transcript, chunk_size=500):
"""タイムスタンプを保持しながら字幕をチャンクに分割"""
chunks = []
current_chunk = ""
chunk_start_time = 0
for i, entry in enumerate(transcript):
if len(current_chunk) == 0:
chunk_start_time = entry['start']
current_chunk += entry['text'] + " "
# 必要なサイズに達したか、字幕の終わりに達した場合
if len(current_chunk) >= chunk_size or i == len(transcript) - 1:
chunks.append({
'text': current_chunk.strip(),
'start_time': chunk_start_time,
'timestamp': format_timestamp(chunk_start_time),
'video_id': video_id
})
current_chunk = ""
return chunks
# 使用例
transcript = YouTubeTranscriptApi.get_transcript(video_id)
chunks = create_chunks_with_timestamps(transcript)
for chunk in chunks[:3]: # 最初の3つのチャンク
print(f"[{chunk['timestamp']}] {chunk['text'][:100]}...")
メタデータの収集：タイトル、説明、タグ
メタデータはRAGシステムに文脈を豊かにします。以下は、必要なすべてのデータを収集する完全な例です：
import requests
from datetime import datetime
def collect_video_metadata(video_id, api_key):
"""動画の完全なメタデータを収集"""
url = f'https://www.googleapis.com/youtube/v3/videos'
params = {
'part': 'snippet,contentDetails,statistics,topicDetails',
'id': video_id,
'key': api_key
}
response = requests.get(url, params=params)
data = response.json()
if 'items' not in data or len(data['items']) == 0:
return None
item = data['items'][0]
snippet = item['snippet']
stats = item.get('statistics', {})
content = item.get('contentDetails', {})
metadata = {
'video_id': video_id,
'title': snippet['title'],
'description': snippet['description'],
'channel_title': snippet['channelTitle'],
'channel_id': snippet['channelId'],
'published_at': snippet['publishedAt'],
'tags': snippet.get('tags', []),
'category_id': snippet.get('categoryId'),
'duration': content.get('duration'),
'view_count': int(stats.get('viewCount', 0)),
'like_count': int(stats.get('likeCount', 0)),
'comment_count': int(stats.get('commentCount', 0)),
'topics': item.get('topicDetails', {}).get('topicCategories', [])
}
return metadata
# 使用例
metadata = collect_video_metadata('dQw4w9WgXcQ', API_KEY)
print(f"タイトル: {metadata['title']}")
print(f"チャンネル: {metadata['channel_title']}")
print(f"視聴回数: {metadata['view_count']:,}")
print(f"タグ: {', '.join(metadata['tags'][:5])}")
コンテンツの関連性の特定
技術的なテーマにとって、情報の新鮮さは重要です。関連性を評価する関数を追加します：
from datetime import datetime, timedelta
def calculate_content_freshness(published_date_str):
"""コンテンツの関連性を評価"""
published_date = datetime.fromisoformat(published_date_str.replace('Z', '+00:00'))
age_days = (datetime.now(published_date.tzinfo) - published_date).days
if age_days < 30:
return 'very_fresh'
elif age_days < 180:
return 'fresh'
elif age_days < 365:
return 'moderate'
else:
return 'old'
def calculate_quality_score(metadata):
"""情報源の質の評価を計算"""
score = 0
# 人気
views = metadata['view_count']
if views > 100000:
score += 3
elif views > 10000:
score += 2
elif views > 1000:
score += 1
# エンゲージメント（視聴回数に対するいいね）
if views > 0:
like_ratio = metadata['like_count'] / views
if like_ratio > 0.05:
score += 2
elif like_ratio > 0.02:
score += 1
# 新鮮さ
freshness = calculate_content_freshness(metadata['published_at'])
if freshness == 'very_fresh':
score += 2
elif freshness == 'fresh':
score += 1
return score
# 使用例
metadata = collect_video_metadata('dQw4w9WgXcQ', API_KEY)
quality = calculate_quality_score(metadata)
freshness = calculate_content_freshness(metadata['published_at'])
print(f"質の評価: {quality}/7")
print(f"関連性: {freshness}")
文脈分析のためのコメントのパース
コメントには貴重な情報が含まれていることがあります：動画の誤りの修正、追加リソース、よくある質問。RAGシステムにとってこれは追加の文脈です。
def get_video_comments(video_id, api_key, max_results=100):
"""動画のコメントを取得"""
url = 'https://www.googleapis.com/youtube/v3/commentThreads'
comments = []
next_page_token = None
while len(comments) < max_results:
params = {
'part': 'snippet',
'videoId': video_id,
'maxResults': min(100, max_results - len(comments)),
'order': 'relevance', # 関連性でソート
'key': api_key
}
if next_page_token:
params['pageToken'] = next_page_token
response = requests.get(url, params=params)
data = response.json()
if 'items' not in data:
break
for item in data['items']:
top_comment = item['snippet']['topLevelComment']['snippet']
comments.append({
'author': top_comment['authorDisplayName'],
'text': top_comment['textDisplay'],
'like_count': top_comment['likeCount'],
'published_at': top_comment['publishedAt'],
'reply_count': item['snippet']['totalReplyCount']
})
next_page_token = data.get('nextPageToken')
if not next_page_token:
break
return comments
def filter_valuable_comments(comments, min_likes=5):
"""価値のあるコメントをフィルタリング"""
valuable = []
for comment in comments:
# 価値の基準：
# 1. たくさんのいいね（人気）
# 2. 返信がある（議論を引き起こした）
# 3. 長いテキスト（詳細なコメント）
if (comment['like_count'] >= min_likes or
comment['reply_count'] > 0 or
len(comment['text']) > 200):
valuable.append(comment)
return valuable
# 使用例
comments = get_video_comments('dQw4w9WgXcQ', API_KEY, max_results=50)
valuable_comments = filter_valuable_comments(comments)
print(f"コメントの総数: {len(comments)}")
print(f"価値のあるコメントの数: {len(valuable_comments)}")
for comment in valuable_comments[:3]:
print(f"\n[{comment['like_count']} いいね] {comment['author']}:")
print(comment['text'][:200])
データ収集のスケーリングのためのプロキシの使用
大規模なデータ収集では、2つの問題が発生します：YouTube APIの制限（1日あたり10,000クォータ）と、字幕のパース時のブロック。プロキシは両方の問題を解決するのに役立ちます。
YouTubeパースのためにプロキシが必要なとき
- APIクォータの超過 — 異なるIPを介して複数のAPIキーを使用することで、1日の制限を増やすことができます
- APIを回避した字幕のパース — youtube-transcript-apiライブラリは、高頻度でブロックされる可能性のある直接リクエストを行います
- 異なる地域からのデータ収集 — 一部の動画は特定の国でのみ利用可能です
- 並行収集 — 複数のIPに負荷を分散させてプロセスを加速します
プロキシの種類の選択
| プロキシの種類 | 利点 | 使用するタイミング |
|---|---|---|
| データセンター | 高速、低価格 | APIとの作業、小規模なボリューム |
| レジデンシャル | ブロックのリスクが低い、実際のIP | 大量の字幕パース、制限の回避 |
| モバイル | 最大の信頼性、ブロックが少ない | モバイルアプリからのデータ収集、重要なタスク |
RAGシステムのほとんどのタスクには、レジデンシャルプロキシが適しています — これは、大規模なパース時のコストと信頼性のバランスを提供します。
コード内のプロキシの設定
import requests
from youtube_transcript_api import YouTubeTranscriptApi
from youtube_transcript_api._api import TranscriptListFetcher
# プロキシの設定
PROXY = {
'http': 'http://username:password@proxy-server:port',
'https': 'http://username:password@proxy-server:port'
}
# プロキシを介してAPIを使用するための関数
def get_video_details_with_proxy(video_id, api_key, proxy):
url = f'https://www.googleapis.com/youtube/v3/videos'
params = {
'part': 'snippet,statistics',
'id': video_id,
'key': api_key
}
response = requests.get(url, params=params, proxies=proxy, timeout=10)
return response.json()
# プロキシを介して字幕をパースするためのクラス
class ProxiedTranscriptApi:
def __init__(self, proxy):
self.proxy = proxy
def get_transcript(self, video_id, languages=['en']):
# プロキシを使用してカスタムセッションを作成
session = requests.Session()
session.proxies = self.proxy
# リクエストにセッションを使用
fetcher = TranscriptListFetcher(session)
transcript_list = fetcher.fetch(video_id)
# 必要な言語を取得
for lang in languages:
try:
transcript = transcript_list.find_transcript([lang])
return transcript.fetch()
except:
continue
raise Exception(f"言語に対する字幕が見つかりませんでした: {languages}")
# 使用例
api = ProxiedTranscriptApi(PROXY)
transcript = api.get_transcript('dQw4w9WgXcQ', languages=['ru', 'en'])
print(f"取得した字幕セグメントの数: {len(transcript)}")
スケーリングのためのプロキシのローテーション
数千の動画からデータを収集する際には、複数のプロキシ間で負荷を分散することが重要です：
import random
import time
class ProxyRotator:
def __init__(self, proxy_list):
self.proxies = proxy_list
self.current_index = 0
def get_next_proxy(self):
"""順次ローテーション"""
proxy = self.proxies[self.current_index]
self.current_index = (self.current_index + 1) % len(self.proxies)
return proxy
def get_random_proxy(self):
"""ランダムローテーション"""
return random.choice(self.proxies)
# プロキシのリスト
PROXY_LIST = [
{'http': 'http://user:pass@proxy1:port', 'https': 'http://user:pass@proxy1:port'},
{'http': 'http://user:pass@proxy2:port', 'https': 'http://user:pass@proxy2:port'},
{'http': 'http://user:pass@proxy3:port', 'https': 'http://user:pass@proxy3:port'},
]
rotator = ProxyRotator(PROXY_LIST)
def collect_data_with_rotation(video_ids):
results = []
for video_id in video_ids:
proxy = rotator.get_next_proxy()
try:
# メタデータを取得
metadata = get_video_details_with_proxy(video_id, API_KEY, proxy)
# 字幕を取得
api = ProxiedTranscriptApi(proxy)
transcript = api.get_transcript(video_id)
results.append({
'video_id': video_id,
'metadata': metadata,
'transcript': transcript
})
# リクエスト間の遅延
time.sleep(1)
except Exception as e:
print(f"{video_id}のエラー: {e}")
continue
return results
# 使用例
video_ids = ['video1', 'video2', 'video3', 'video4', 'video5']
data = collect_data_with_rotation(video_ids)
print(f"{len(data)}の動画のデータを収集しました")
RAGのためのデータの処理と準備
データを収集した後、それらを処理し、RAGシステムが効果的に機能するために構造化する必要があります。
ベクトル埋め込みの作成
RAGシステムは、関連するフラグメントを見つけるためにベクトル検索を使用します。テキストを埋め込みに変換する必要があります：
from sentence_transformers import SentenceTransformer
import numpy as np
# 埋め込みを作成するためのモデルをロード
model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')
def create_embeddings_from_transcript(transcript_chunks):
"""字幕のチャンクのための埋め込みを作成"""
embeddings = []
for chunk in transcript_chunks:
# 文脈のためにテキストとメタデータを組み合わせる
text_with_context = f"{chunk['title']} | {chunk['text']}"
# 埋め込みを作成
embedding = model.encode(text_with_context)
embeddings.append({
'video_id': chunk['video_id'],
'timestamp': chunk['timestamp'],
'text': chunk['text'],
'embedding': embedding.tolist(),
'metadata': {
'title': chunk['title'],
'channel': chunk['channel'],
'views': chunk['views']
}
})
return embeddings
# データの準備
def prepare_rag_data(video_data):
"""RAGのためのすべてのデータを準備"""
all_chunks = []
for video in video_data:
metadata = video['metadata']
transcript = video['transcript']
# 字幕をチャンクに分割
chunks = create_chunks_with_timestamps(transcript)
# 各チャンクにメタデータを追加
for chunk in chunks:
chunk['title'] = metadata['title']
chunk['channel'] = metadata['channel_title']
chunk['views'] = metadata['view_count']
all_chunks.append(chunk)
# 埋め込みを作成
embeddings = create_embeddings_from_transcript(all_chunks)
return embeddings
# 使用例
rag_data = prepare_rag_data(collected_videos)
print(f"RAGのために{len(rag_data)}のフラグメントを準備しました")
ベクトルデータベースへの保存
効率的な検索のために、埋め込みは専門のデータベースに保存されます。人気のあるオプション：Pinecone、Weaviate、Qdrant、ChromaDB。
import chromadb
from chromadb.config import Settings
# ChromaDBの初期化（ローカルベクトルDB）
client = chromadb.Client(Settings(
chroma_db_impl="duckdb+parquet",
persist_directory="./youtube_rag_db"
))
# コレクションの作成
collection = client.create_collection(
name="youtube_transcripts",
metadata={"description": "RAGのためのYouTube動画の字幕"}
)
def store_in_vector_db(embeddings_data, collection):
"""埋め込みをベクトルDBに保存"""
ids = []
embeddings = []
documents = []
metadatas = []
for i, item in enumerate(embeddings_data):
ids.append(f"{item['video_id']}_{i}")
embeddings.append(item['embedding'])
documents.append(item['text'])
metadatas.append({
'video_id': item['video_id'],
'timestamp': item['timestamp'],
'title': item['metadata']['title'],
'channel': item['metadata']['channel'],
'views': str(item['metadata']['views']),
'youtube_url': f"https://youtube.com/watch?v={item['video_id']}&t={int(float(item['timestamp'].split(':')[0])*60 + float(item['timestamp'].split(':')[1]))}s"
})
# コレクションに追加
collection.add(
ids=ids,
embeddings=embeddings,
documents=documents,
metadatas=metadatas
)
print(f"{len(ids)}の埋め込みをベクトルDBに保存しました")
# 使用例
store_in_vector_db(rag_data, collection)
検索と回答の生成
最終ステップは、RAG検索と回答生成の実装です：
def search_youtube_knowledge(query, collection, model, top_k=3):
"""YouTubeから関連するフラグメントを検索"""
# クエリの埋め込みを作成
query_embedding = model.encode(query).tolist()
# ベクトルDBで検索
results = collection.query(
query_embeddings=[query_embedding],
n_results=top_k
)
# 結果のフォーマット
sources = []
for i in range(len(results['ids'][0])):
sources.append({
'text': results['documents'][0][i],
'metadata': results['metadatas'][0][i],
'distance': results['distances'][0][i] if 'distances' in results else None
})
return sources
def generate_rag_answer(query, sources, llm_api_key):
"""見つかった情報源に基づいて回答を生成"""
# 見つかった情報源から文脈を形成
context = "\n\n".join([
f"情報源: {s['metadata']['title']} ({s['metadata']['timestamp']})\n{s['text']}"
for s in sources
])
# LLMのためのプロンプト
prompt = f"""次のYouTube動画のフラグメントに基づいて、ユーザーの質問に答えてください。
必ず、タイムスタンプ付きの情報源を示してください。
文脈:
{context}
質問: {query}
回答:"""
# ここであなたのLLMを呼び出します（OpenAI、Claude、ローカルモデル）
# OpenAIの例：
# response = openai.ChatCompletion.create(
# model="gpt-4",
# messages=[{"role": "user", "content": prompt}]
# )
# answer = response.choices[0].message.content
# 例としてプロンプトを返します
return {
'answer': 'ここにLLMの回答が表示されます',
'sources': sources
}
# 使用例
query = "Pythonでプロキシを設定するには？"
sources = search_youtube_knowledge(query, collection, model, top_k=3)
print("見つかった情報源:")
for source in sources:
print(f"\n{source['metadata']['title']}")
print(f"時間: {source['metadata']['timestamp']}")
print(f"リンク: {source['metadata']['youtube_url']}")
print(f"テキスト: {source['text'][:200]}...")
RAGの質の最適化
YouTubeデータに基づくRAGシステムの質を向上させるためのいくつかのヒント：
- 低品質のコンテンツをフィルタリング — 視聴回数、いいね、公開日などのメトリクスを使用してください
- 文脈を保持 — 各テキストチャンクに動画とチャンネルのタイトルを追加してください
- チャンクのサイズを最適化 — 技術的なコンテンツには300-500語が最適です
- メタデータを使用してランク付け — より新しく人気のあるコンテンツが優先される場合があります
- コメントを追加 — 重要な明確化やFAQが含まれていることが多いです
- 動画の可用性を確認 — 一部の動画は削除されるか、プライベートになる可能性があります
アドバイス： 大規模なYouTubeデータ収集には、公式API（メタデータ用）とプロキシを介したパース（字幕用）の組み合わせを使用することをお勧めします。これにより、制限を回避し、最大限の情報を得ることができます。
結論
RAGシステムのためのYouTubeデータ収集は、APIの操作、字幕のパース、メタデータの処理、ベクトル埋め込みの作成を含む多段階のプロセスです。重要なポイントは：
- YouTube Data API v3は、1日あたり10,000の制限でメタデータ、統計、コメントを提供します
- 字幕はyoutube-transcript-apiライブラリまたは直接リクエストを介してパースされます
- タイムスタンプは情報源への正確なリンクを作成するために重要です
- メタデータ（視聴回数、いいね、日付）はコンテンツの質と関連性を評価するのに役立ちます
- コメントは文脈を追加し、FAQを含むことが多いです
- プロキシはスケーリングと制限の回避に必要です
- ベクトル埋め込みと専門のDBが迅速なセマンティック検索を提供します
プロセスを適切に設定すれば、1日に数万の質の高いコンテンツフラグメントを収集し、あらゆる分野のRAGシステムのための強力な知識ベースを構築できます。
制限やブロックを回避しながら大規模なYouTubeデータ収集を計画している場合は、レジデンシャルプロキシの使用をお勧めします — これは数千の動画をパースする際の安定性を提供し、YouTubeからのブロックのリスクを最小限に抑えます。