Как собирать данные YouTube для RAG-систем: парсинг субтитров и метаданных с Python

RAG (Retrieval-Augmented Generation) системы требуют качественных данных для обучения. YouTube — огромный источник структурированного контента: видео с субтитрами, метаданными, комментариями. В этой статье разберём, как эффективно собирать YouTube-данные для RAG, избегая блокировок и соблюдая лимиты API.

Что такое RAG и зачем нужны данные YouTube

RAG (Retrieval-Augmented Generation) — это подход к построению AI-систем, где языковая модель дополняется базой знаний. Вместо того чтобы полагаться только на данные, на которых модель обучена, RAG извлекает релевантную информацию из внешнего источника и использует её для генерации ответов.

YouTube содержит миллионы часов контента с субтитрами на разных языках. Это делает платформу ценным источником данных для RAG-систем в различных областях:

• Образовательные системы — лекции, туториалы, курсы с временными метками
• Техническая документация — видео-гайды по программированию, DevOps, настройке ПО
• Медицинские базы знаний — лекции врачей, разборы клинических случаев
• Бизнес-аналитика — интервью с экспертами, кейсы, обзоры рынков
• Продуктовая поддержка — обзоры продуктов, FAQ в видеоформате

Преимущество YouTube-данных — наличие структуры: субтитры с временными метками, метаданные (категории, теги), социальный контекст (комментарии, лайки). Всё это помогает RAG-системе понимать не только содержание, но и контекст информации.

Какие данные YouTube полезны для RAG-систем

Для эффективной работы RAG-системы нужно собирать несколько типов данных. Каждый тип решает свои задачи в процессе извлечения и генерации информации.

Субтитры (Transcripts)

Основной источник текстовых данных. YouTube предоставляет два типа субтитров:

• Автоматические — генерируются алгоритмами распознавания речи Google. Доступны для большинства видео на английском и других популярных языках. Точность 85-95% в зависимости от качества звука.
• Ручные — загружаются авторами или сообществом. Более точные, часто содержат форматирование и дополнительный контекст.

Субтитры включают временные метки (timestamps), что позволяет связывать текст с конкретными моментами видео. Это критично для создания точных ссылок на источники в RAG-ответах.

Метаданные видео

Метаданные помогают RAG-системе понимать контекст и релевантность информации:

Комментарии

Комментарии содержат дополнительный контекст: вопросы зрителей, уточнения авторов, дискуссии. Для RAG-систем это ценно, так как:

• Комментарии часто содержат FAQ по теме видео
• Авторы могут публиковать исправления и дополнения
• Обсуждения раскрывают разные точки зрения на проблему

Работа с YouTube Data API v3: настройка и лимиты

YouTube Data API v3 — официальный способ получения данных. Он предоставляет доступ к метаданным, статистике, комментариям. Субтитры получаются через отдельные методы.

Получение API ключа

Для работы с API нужен ключ из Google Cloud Console:

1. Перейдите на console.cloud.google.com
2. Создайте новый проект или выберите существующий
3. Включите YouTube Data API v3 в разделе "APIs & Services"
4. Создайте учетные данные (Credentials) → API key
5. Скопируйте ключ — он понадобится для всех запросов

Лимиты и квоты

YouTube API использует систему квот. Каждый запрос "стоит" определённое количество единиц:

Дневной лимит по умолчанию — 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('machine learning tutorial', max_results=5)
for item in results.get('items', []):
    video_id = item['id']['videoId']
    title = item['snippet']['title']
    print(f'ID: {video_id}, Title: {title}')
    
    # Получаем детальную информацию
    details = get_video_details(video_id)
    stats = details['items'][0]['statistics']
    print(f"Views: {stats.get('viewCount')}, Likes: {stats.get('likeCount')}")

Парсинг субтитров видео: автоматические и ручные

YouTube Data API v3 не предоставляет прямого доступа к субтитрам. Для их получения используются альтернативные методы.

Использование библиотеки youtube-transcript-api

Самый простой способ — библиотека youtube-transcript-api для Python. Она извлекает субтитры напрямую, без 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
    
    # Engagement (лайки относительно просмотров)
    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])

Использование прокси для масштабирования сбора данных

При масштабном сборе данных возникают две проблемы: лимиты YouTube API (10,000 квот в день) и блокировки при парсинге субтитров. Прокси помогают решить обе задачи.

Когда нужны прокси для YouTube-парсинга

• Превышение квот API — используя несколько API-ключей через разные IP, можно увеличить дневной лимит
• Парсинг субтитров в обход API — библиотека youtube-transcript-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"Подготовлено {len(rag_data)} фрагментов для RAG")

Сохранение в векторную базу данных

Для эффективного поиска эмбеддинги сохраняются в специализированные базы данных. Популярные варианты: Pinecone, Weaviate, Qdrant, ChromaDB.

import chromadb
from chromadb.config import Settings

# Инициализация ChromaDB (локальная векторная БД)
client = chromadb.Client(Settings(
    chroma_db_impl="duckdb+parquet",
    persist_directory="./youtube_rag_db"
))

# Создание коллекции
collection = client.create_collection(
    name="youtube_transcripts",
    metadata={"description": "YouTube video transcripts for RAG"}
)

def store_in_vector_db(embeddings_data, collection):
    """Сохранение эмбеддингов в векторную БД"""
    
    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)} эмбеддингов в векторную БД")

# Использование
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()
    
    # Поиск в векторной БД
    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

Несколько советов для улучшения качества RAG-системы на YouTube-данных:

• Фильтруйте низкокачественный контент — используйте метрики просмотров, лайков, давности публикации
• Сохраняйте контекст — добавляйте название видео и канала к каждому чанку текста
• Оптимизируйте размер чанков — для технического контента 300-500 слов оптимально
• Используйте метаданные для ранжирования — более свежий и популярный контент может получать приоритет
• Добавляйте комментарии — они часто содержат важные уточнения и FAQ
• Проверяйте доступность видео — некоторые видео могут быть удалены или стать приватными

Заключение

Сбор данных YouTube для RAG-систем — многоступенчатый процесс, включающий работу с API, парсинг субтитров, обработку метаданных и создание векторных эмбеддингов. Ключевые моменты:

• YouTube Data API v3 предоставляет метаданные, статистику и комментарии с лимитом 10,000 квот в день
• Субтитры парсятся через библиотеку youtube-transcript-api или прямые запросы
• Временные метки критически важны для создания точных ссылок на источники
• Метаданные (просмотры, лайки, дата) помогают оценивать качество и актуальность контента
• Комментарии добавляют контекст и часто содержат FAQ
• Прокси необходимы для масштабирования и обхода лимитов
• Векторные эмбеддинги и специализированные БД обеспечивают быстрый семантический поиск

При правильной настройке процесса можно собирать десятки тысяч качественных фрагментов контента в день, создавая мощную базу знаний для RAG-систем в любой предметной области.

Если вы планируете масштабный сбор данных YouTube с обходом лимитов и блокировок, рекомендуем использовать резидентные прокси — они обеспечивают стабильность при парсинге тысяч видео и минимизируют риск блокировок со стороны YouTube.