YouTube 데이터 수집 방법: Python을 이용한 자막 및 메타데이터 파싱

RAG (Retrieval-Augmented Generation) 시스템은 학습을 위한 품질 높은 데이터가 필요합니다. YouTube는 자막, 메타데이터, 댓글이 포함된 구조화된 콘텐츠의 방대한 출처입니다. 이 기사에서는 차단을 피하고 API 한계를 준수하면서 RAG를 위한 YouTube 데이터를 효과적으로 수집하는 방법을 살펴보겠습니다.

RAG란 무엇이며 YouTube 데이터가 필요한 이유

RAG (Retrieval-Augmented Generation)는 AI 시스템 구축을 위한 접근 방식으로, 언어 모델이 지식 기반으로 보완됩니다. 모델이 학습한 데이터에만 의존하는 대신, RAG는 외부 출처에서 관련 정보를 추출하고 이를 사용하여 응답을 생성합니다.

YouTube에는 다양한 언어로 된 수백만 시간의 콘텐츠가 포함되어 있습니다. 이는 다양한 분야의 RAG 시스템에 유용한 데이터 출처로서 플랫폼의 가치를 높입니다:

• 교육 시스템 — 강의, 튜토리얼, 타임스탬프가 포함된 과정
• 기술 문서 — 프로그래밍, DevOps, 소프트웨어 설정에 대한 비디오 가이드
• 의료 지식 기반 — 의사의 강의, 임상 사례 분석
• 비즈니스 분석 — 전문가 인터뷰, 사례 연구, 시장 리뷰
• 제품 지원 — 제품 리뷰, 비디오 형식의 FAQ

YouTube 데이터의 장점은 구조가 있다는 것입니다: 타임스탬프가 있는 자막, 메타데이터(카테고리, 태그), 사회적 맥락(댓글, 좋아요). 이는 RAG 시스템이 내용뿐만 아니라 정보의 맥락을 이해하는 데 도움을 줍니다.

RAG 시스템에 유용한 YouTube 데이터 종류

RAG 시스템의 효과적인 작업을 위해서는 여러 유형의 데이터를 수집해야 합니다. 각 유형은 정보 추출 및 생성 과정에서 고유한 역할을 합니다.

자막 (Transcripts)

텍스트 데이터의 주요 출처입니다. YouTube는 두 가지 유형의 자막을 제공합니다:

• 자동 자막 — Google 음성 인식 알고리즘에 의해 생성됩니다. 대부분의 영어 및 기타 인기 언어 비디오에서 사용할 수 있습니다. 정확도는 음질에 따라 85-95%입니다.
• 수동 자막 — 저자 또는 커뮤니티에 의해 업로드됩니다. 더 정확하며, 종종 포맷팅 및 추가 맥락을 포함합니다.

자막에는 타임스탬프가 포함되어 있어 텍스트를 비디오의 특정 순간과 연결할 수 있습니다. 이는 RAG 응답에서 출처에 대한 정확한 링크를 생성하는 데 중요합니다.

비디오 메타데이터

메타데이터는 RAG 시스템이 정보의 맥락과 관련성을 이해하는 데 도움을 줍니다:

댓글

댓글은 추가적인 맥락을 제공합니다: 시청자의 질문, 저자의 설명, 토론. RAG 시스템에 이는 귀중한 정보로, 다음과 같은 이유로 중요합니다:

• 댓글은 종종 비디오 주제에 대한 FAQ를 포함합니다
• 저자는 수정 및 추가 정보를 게시할 수 있습니다
• 토론은 문제에 대한 다양한 관점을 드러냅니다

YouTube Data API v3 작업: 설정 및 한계

YouTube Data API v3는 데이터를 얻기 위한 공식적인 방법입니다. 메타데이터, 통계, 댓글에 접근할 수 있습니다. 자막은 별도의 메소드를 통해 얻습니다.

API 키 받기

API를 사용하려면 Google Cloud Console에서 키가 필요합니다:

1. console.cloud.google.com로 이동합니다
2. 새 프로젝트를 만들거나 기존 프로젝트를 선택합니다
3. "APIs & Services" 섹션에서 YouTube Data API v3를 활성화합니다
4. 자격 증명(Credentials) → API 키를 생성합니다
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"조회수: {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])

데이터 수집 확장을 위한 프록시 사용

대규모 데이터 수집 시 두 가지 문제가 발생합니다: 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"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를 포함
• 비디오 접근성 확인 — 일부 비디오는 삭제되거나 비공개로 전환될 수 있음

결론

RAG 시스템을 위한 YouTube 데이터 수집은 API 작업, 자막 파싱, 메타데이터 처리 및 벡터 임베딩 생성이 포함된 다단계 프로세스입니다. 주요 사항은 다음과 같습니다:

• YouTube Data API v3는 메타데이터, 통계 및 댓글을 하루 10,000 쿼트 한도로 제공합니다
• 자막은 youtube-transcript-api 라이브러리 또는 직접 요청을 통해 파싱됩니다
• 타임스탬프는 출처에 대한 정확한 링크 생성을 위해 매우 중요합니다
• 메타데이터(조회수, 좋아요, 날짜)는 콘텐츠의 품질과 적시성을 평가하는 데 도움을 줍니다
• 댓글은 맥락을 추가하고 종종 FAQ를 포함합니다
• 프록시는 확장 및 한계 우회를 위해 필요합니다
• 벡터 임베딩 및 전문 데이터베이스는 빠른 의미 검색을 제공합니다

프로세스를 올바르게 설정하면 하루에 수만 개의 품질 높은 콘텐츠 프래그먼트를 수집하여 모든 주제 영역의 RAG 시스템을 위한 강력한 지식 기반을 구축할 수 있습니다.

YouTube 데이터의 대규모 수집을 계획하고 한계 및 차단을 우회하려는 경우, 주거용 프록시를 사용하는 것이 좋습니다 — 이는 수천 개의 비디오를 파싱할 때 안정성을 제공하고 YouTube의 차단 위험을 최소화합니다.