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 डेटा API v3 के साथ काम करना: सेटअप और सीमाएं

YouTube डेटा API v3 डेटा प्राप्त करने का आधिकारिक तरीका है। यह मेटाडेटा, आंकड़े, टिप्पणियों तक पहुँच प्रदान करता है। सबटाइटल अलग-अलग तरीकों से प्राप्त होते हैं।

API कुंजी प्राप्त करना

API के साथ काम करने के लिए Google Cloud Console से एक कुंजी की आवश्यकता होती है:

1. console.cloud.google.com पर जाएं
2. एक नया प्रोजेक्ट बनाएं या मौजूदा प्रोजेक्ट चुनें
3. "APIs & Services" अनुभाग में YouTube डेटा 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"Views: {stats.get('viewCount')}, Likes: {stats.get('likeCount')}")

वीडियो सबटाइटल का पार्सिंग: स्वचालित और मैनुअल

YouTube डेटा 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
    
    # 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 कोटा का उल्लंघन — विभिन्न IP के माध्यम से कई API कुंजियों का उपयोग करके दैनिक सीमा बढ़ाई जा सकती है
• 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": "YouTube वीडियो ट्रांसक्रिप्ट RAG के लिए"}
)

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"वेक्टर DB में {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()
    
    # वेक्टर 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 डेटा API v3 मेटाडेटा, आंकड़े और टिप्पणियाँ 10,000 कोटा प्रति दिन के साथ प्रदान करता है
• सबटाइटल को youtube-transcript-api पुस्तकालय या सीधे अनुरोधों के माध्यम से पार्स किया जाता है
• समय चिह्न स्रोतों के लिए सटीक लिंक बनाने के लिए महत्वपूर्ण हैं
• मेटाडेटा (दृश्य, लाइक, तारीख) सामग्री की गुणवत्ता और प्रासंगिकता का मूल्यांकन करने में मदद करते हैं
• टिप्पणियाँ संदर्भ जोड़ती हैं और अक्सर FAQ शामिल करती हैं
• स्केलिंग और सीमाओं को बायपास करने के लिए प्रॉक्सी आवश्यक हैं
• वेक्टर एम्बेडिंग और विशेष डेटाबेस त्वरित सेमांटिक खोज सुनिश्चित करते हैं

सही प्रक्रिया सेटअप के साथ, आप प्रति दिन हजारों गुणवत्ता वाले सामग्री खंड एकत्र कर सकते हैं, जो किसी भी विषय क्षेत्र के लिए RAG सिस्टम के लिए एक शक्तिशाली ज्ञान आधार बनाते हैं।

यदि आप सीमाओं और ब्लॉकिंग को बायपास करते हुए बड़े पैमाने पर YouTube डेटा संग्रहण की योजना बना रहे हैं, तो रिहायशी प्रॉक्सी का उपयोग करने की सिफारिश की जाती है — ये हजारों वीडियो के पार्सिंग के दौरान स्थिरता प्रदान करते हैं और YouTube द्वारा ब्लॉकिंग के जोखिम को न्यूनतम करते हैं।