جمع‌آوری داده‌های یوتیوب برای RAG: راهنمایی با پایتون و پروکسی

```html

سیستم های RAG (Retrieval-Augmented Generation) به داده های با کیفیت برای آموزش نیاز دارند. YouTube یک منبع بزرگ از محتوای ساختار یافته است: ویدیوها با زیرنویس، متاداده و نظرات. در این مقاله بررسی می کنیم که چگونه به طور مؤثر داده های YouTube را برای RAG جمع آوری کنیم، از مسدود شدن جلوگیری کنیم و محدودیت های API را رعایت کنیم.

RAG چیست و داده های YouTube چه کاربردی دارند

RAG (Retrieval-Augmented Generation) یک رویکرد برای ساخت سیستم های هوش مصنوعی است که در آن مدل زبانی با پایگاه دانش تکمیل می شود. به جای تکیه بر داده هایی که مدل بر روی آن آموزش دیده است، RAG اطلاعات مرتبط را از منبع خارجی استخراج کرده و از آن برای تولید پاسخ ها استفاده می کند.

YouTube شامل میلیون ها ساعت محتوا با زیرنویس به زبان های مختلف است. این پلتفرم را به منبع ارزشمندی برای داده های سیستم های RAG در زمینه های مختلف تبدیل می کند:

سیستم های آموزشی — سخنرانی ها، آموزش ها، دوره ها با زمان بندی
مستندات فنی — ویدیوهای راهنما در برنامه نویسی، DevOps، تنظیم نرم افزار
پایگاه های دانش پزشکی — سخنرانی های پزشکان، تحلیل موارد بالینی
تحلیل کسب و کار — مصاحبه با کارشناسان، موارد، بررسی بازارها
پشتیبانی محصول — بررسی محصولات، سوالات متداول به صورت ویدیو

مزیت داده های YouTube وجود ساختار است: زیرنویس ها با زمان بندی، متاداده (دسته بندی ها، برچسب ها)، زمینه اجتماعی (نظرات، لایک ها). همه اینها به سیستم RAG کمک می کند تا نه تنها محتوا بلکه زمینه اطلاعات را نیز درک کند.

کدام داده های YouTube برای سیستم های RAG مفید هستند

برای عملکرد مؤثر سیستم RAG باید چند نوع داده جمع آوری شود. هر نوع وظایف خاص خود را در فرآیند استخراج و تولید اطلاعات دارد.

زیرنویس ها (Transcripts)

منبع اصلی داده های متنی. YouTube دو نوع زیرنویس ارائه می دهد:

خودکار — توسط الگوریتم های تشخیص گفتار Google تولید می شوند. برای اکثر ویدیوها به زبان انگلیسی و دیگر زبان های محبوب در دسترس هستند. دقت 85-95% بسته به کیفیت صدا.
دستی — توسط نویسندگان یا جامعه بارگذاری می شوند. دقیق تر هستند و اغلب شامل فرمت بندی و زمینه اضافی می باشند.

زیرنویس ها شامل زمان بندی (timestamps) هستند که امکان ارتباط متن با لحظات خاص ویدیو را فراهم می کند. این برای ایجاد لینک های دقیق به منابع در پاسخ های RAG حیاتی است.

متاداده ویدیو

متاداده به سیستم RAG کمک می کند تا زمینه و ارتباط اطلاعات را درک کند:

نوع داده	کاربرد در RAG
عنوان و توضیحات	جستجوی معنایی، تعیین موضوع
برچسب ها و دسته بندی ها	طبقه بندی محتوا، فیلتر کردن
تاریخ انتشار	مرتبط بودن اطلاعات (مهم برای موضوعات فنی)
مدت زمان	ارزیابی عمق پردازش موضوع
آمار (بازدیدها، لایک ها)	ارزیابی کیفیت و محبوبیت منبع
اطلاعات کانال	تعیین اعتبار منبع

نظرات

نظرات شامل زمینه اضافی هستند: سوالات بینندگان، توضیحات نویسندگان، بحث ها. برای سیستم های RAG این ارزشمند است، زیرا:

نظرات اغلب شامل سوالات متداول در مورد موضوع ویدیو هستند
نویسندگان می توانند اصلاحات و اضافات را منتشر کنند
بحث ها دیدگاه های مختلفی را در مورد مشکل روشن می کنند

کار با YouTube Data API v3: تنظیمات و محدودیت ها

YouTube Data API v3 — روش رسمی برای دریافت داده ها است. این API به متاداده، آمار، نظرات دسترسی می دهد. زیرنویس ها از طریق روش های جداگانه به دست می آیند.

دریافت کلید API

برای کار با API به کلیدی از Google Cloud Console نیاز دارید:

به console.cloud.google.com بروید
یک پروژه جدید ایجاد کنید یا پروژه موجود را انتخاب کنید
YouTube Data API v3 را در بخش "APIs & Services" فعال کنید
گواهی نامه ها (Credentials) را ایجاد کنید → کلید API
کلید را کپی کنید — این کلید برای تمام درخواست ها لازم است

محدودیت ها و سهمیه ها

YouTube API از سیستم سهمیه استفاده می کند. هر درخواست "هزینه" معینی از واحدها دارد:

عملیات	هزینه در سهمیه ها
جستجوی ویدیو (search.list)	100 واحد
دریافت داده های ویدیو (videos.list)	1 واحد
دریافت نظرات (commentThreads.list)	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

ساده ترین روش — کتابخانه 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]:  # سه چانک اول
    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 'بسیار تازه'
    elif age_days < 180:
        return 'تازه'
    elif age_days < 365:
        return 'متوسط'
    else:
        return 'قدیمی'

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 == 'بسیار تازه':
        score += 2
    elif freshness == 'تازه':
        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 برای تسریع فرآیند

انتخاب نوع پروکسی

نوع پروکسی	مزایا	چه زمانی استفاده کنیم
پروکسی دیتاسنتر	سرعت بالا، قیمت پایین	کار با 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 برای 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 کلمه بهینه است
از متاداده برای رتبه بندی استفاده کنید — محتوای تازه تر و محبوب تر می تواند اولویت داشته باشد
نظرات را اضافه کنید — آنها اغلب شامل توضیحات مهم و سوالات متداول هستند
دسترس پذیری ویدیوها را بررسی کنید — برخی ویدیوها ممکن است حذف شوند یا خصوصی شوند

نکته: برای جمع آوری داده های مقیاس بزرگ YouTube توصیه می شود از ترکیب API رسمی (برای متاداده) و تجزیه از طریق پروکسی (برای زیرنویس ها) استفاده کنید. این به دور زدن محدودیت ها و دریافت حداکثر اطلاعات کمک می کند.

نتیجه گیری

جمع آوری داده های YouTube برای سیستم های RAG یک فرآیند چند مرحله ای است که شامل کار با API، تجزیه زیرنویس ها، پردازش متاداده و ایجاد جاسازی های برداری است. نکات کلیدی:

YouTube Data API v3 متاداده، آمار و نظرات را با محدودیت 10,000 سهمیه در روز ارائه می دهد
زیرنویس ها از طریق کتابخانه youtube-transcript-api یا درخواست های مستقیم تجزیه می شوند
زمان بندی ها برای ایجاد لینک های دقیق به منابع حیاتی هستند
متاداده (بازدیدها، لایک ها، تاریخ) به ارزیابی کیفیت و مرتبط بودن محتوا کمک می کند
نظرات زمینه را اضافه کرده و اغلب شامل سوالات متداول هستند
پروکسی ها برای مقیاس بندی و دور زدن محدودیت ها ضروری هستند
جاسازی های برداری و پایگاه داده های تخصصی جستجوی سریع معنایی را فراهم می کنند

با تنظیم صحیح فرآیند، می توان ده ها هزار بخش با کیفیت از محتوا را در روز جمع آوری کرد و یک پایگاه دانش قدرتمند برای سیستم های RAG در هر زمینه ای ایجاد کرد.

اگر قصد دارید جمع آوری داده های مقیاس بزرگ YouTube را با دور زدن محدودیت ها و مسدود شدن ها انجام دهید، توصیه می کنیم از پروکسی های مسکونی استفاده کنید — آنها ثبات را در تجزیه هزاران ویدیو فراهم کرده و خطر مسدود شدن توسط YouTube را به حداقل می رسانند.

```

چگونه داده‌های یوتیوب را برای سیستم‌های RAG جمع‌آوری کنیم: استخراج زیرنویس‌ها و متاداده‌ها با پایتون