Thu thập dữ liệu YouTube cho RAG: hướng dẫn với Python và proxy

```html

Hệ thống RAG (Retrieval-Augmented Generation) cần dữ liệu chất lượng để đào tạo. YouTube là một nguồn tài nguyên khổng lồ với nội dung có cấu trúc: video có phụ đề, siêu dữ liệu, bình luận. Trong bài viết này, chúng ta sẽ tìm hiểu cách thu thập dữ liệu YouTube hiệu quả cho RAG, tránh bị chặn và tuân thủ giới hạn API.

RAG là gì và tại sao cần dữ liệu YouTube

RAG (Retrieval-Augmented Generation) là một phương pháp xây dựng hệ thống AI, trong đó mô hình ngôn ngữ được bổ sung bằng một cơ sở tri thức. Thay vì chỉ dựa vào dữ liệu mà mô hình đã được đào tạo, RAG trích xuất thông tin liên quan từ nguồn bên ngoài và sử dụng nó để tạo ra câu trả lời.

YouTube chứa hàng triệu giờ nội dung với phụ đề bằng nhiều ngôn ngữ khác nhau. Điều này làm cho nền tảng trở thành một nguồn dữ liệu quý giá cho các hệ thống RAG trong nhiều lĩnh vực:

Các hệ thống giáo dục — bài giảng, hướng dẫn, khóa học với dấu thời gian
Tài liệu kỹ thuật — video hướng dẫn lập trình, DevOps, cấu hình phần mềm
Các cơ sở tri thức y tế — bài giảng của bác sĩ, phân tích các trường hợp lâm sàng
Phân tích kinh doanh — phỏng vấn với các chuyên gia, trường hợp, đánh giá thị trường
Hỗ trợ sản phẩm — đánh giá sản phẩm, FAQ dưới dạng video

Ưu điểm của dữ liệu YouTube là có cấu trúc: phụ đề với dấu thời gian, siêu dữ liệu (thể loại, thẻ), ngữ cảnh xã hội (bình luận, lượt thích). Tất cả những điều này giúp hệ thống RAG hiểu không chỉ nội dung mà còn cả ngữ cảnh của thông tin.

Các loại dữ liệu YouTube hữu ích cho hệ thống RAG

Để hệ thống RAG hoạt động hiệu quả, cần thu thập nhiều loại dữ liệu khác nhau. Mỗi loại giải quyết các nhiệm vụ riêng trong quá trình trích xuất và tạo thông tin.

Phụ đề (Transcripts)

Nguồn dữ liệu văn bản chính. YouTube cung cấp hai loại phụ đề:

Tự động — được tạo ra bởi các thuật toán nhận diện giọng nói của Google. Có sẵn cho hầu hết các video bằng tiếng Anh và các ngôn ngữ phổ biến khác. Độ chính xác từ 85-95% tùy thuộc vào chất lượng âm thanh.
Thủ công — được tải lên bởi các tác giả hoặc cộng đồng. Chính xác hơn, thường chứa định dạng và ngữ cảnh bổ sung.

Phụ đề bao gồm dấu thời gian (timestamps), cho phép liên kết văn bản với các khoảnh khắc cụ thể trong video. Điều này rất quan trọng để tạo ra các liên kết chính xác đến các nguồn trong các câu trả lời RAG.

Siêu dữ liệu video

Siêu dữ liệu giúp hệ thống RAG hiểu ngữ cảnh và tính liên quan của thông tin:

Loại dữ liệu	Ứng dụng trong RAG
Tiêu đề và mô tả	Tìm kiếm ngữ nghĩa, xác định chủ đề
Thẻ và thể loại	Phân loại nội dung, lọc
Ngày xuất bản	Tính cập nhật của thông tin (quan trọng cho các chủ đề kỹ thuật)
Thời gian	Đánh giá độ sâu của chủ đề được đề cập
Thống kê (lượt xem, lượt thích)	Đánh giá chất lượng và độ phổ biến của nguồn
Thông tin về kênh	Xác định độ tin cậy của nguồn

Bình luận

Bình luận chứa ngữ cảnh bổ sung: câu hỏi của người xem, làm rõ từ tác giả, thảo luận. Đối với hệ thống RAG, điều này rất quý giá, vì:

Bình luận thường chứa FAQ về chủ đề video
Tác giả có thể công bố sửa chữa và bổ sung
Thảo luận tiết lộ các quan điểm khác nhau về vấn đề

Làm việc với YouTube Data API v3: cấu hình và giới hạn

YouTube Data API v3 là cách chính thức để lấy dữ liệu. Nó cung cấp quyền truy cập vào siêu dữ liệu, thống kê, bình luận. Phụ đề được lấy qua các phương thức riêng biệt.

Lấy khóa API

Để làm việc với API, cần có khóa từ Google Cloud Console:

Truy cập console.cloud.google.com
Tạo một dự án mới hoặc chọn dự án hiện có
Bật YouTube Data API v3 trong phần "APIs & Services"
Tạo thông tin xác thực (Credentials) → API key
Sao chép khóa — nó sẽ cần cho tất cả các yêu cầu

Giới hạn và hạn ngạch

YouTube API sử dụng hệ thống hạn ngạch. Mỗi yêu cầu "có giá" một số đơn vị nhất định:

Hoạt động	Chi phí trong hạn ngạch
Tìm kiếm video (search.list)	100 đơn vị
Lấy dữ liệu video (videos.list)	1 đơn vị
Lấy bình luận (commentThreads.list)	1 đơn vị

Giới hạn hàng ngày mặc định là 10,000 đơn vị. Điều này tương đương khoảng 100 yêu cầu tìm kiếm hoặc 10,000 yêu cầu siêu dữ liệu. Để tăng hạn ngạch, cần gửi yêu cầu đến Google.

Ví dụ cơ bản về cách sử dụng API

import requests

API_KEY = 'khóa_api_của_bạn'
BASE_URL = 'https://www.googleapis.com/youtube/v3'

# Tìm kiếm video theo truy vấn
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()

# Lấy siêu dữ liệu video
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()

# Ví dụ sử dụng
results = search_videos('hướng dẫn học máy', max_results=5)
for item in results.get('items', []):
    video_id = item['id']['videoId']
    title = item['snippet']['title']
    print(f'ID: {video_id}, Tiêu đề: {title}')
    
    # Lấy thông tin chi tiết
    details = get_video_details(video_id)
    stats = details['items'][0]['statistics']
    print(f"Lượt xem: {stats.get('viewCount')}, Lượt thích: {stats.get('likeCount')}")

Phân tích phụ đề video: tự động và thủ công

YouTube Data API v3 không cung cấp quyền truy cập trực tiếp vào phụ đề. Để lấy chúng, cần sử dụng các phương pháp thay thế.

Sử dụng thư viện youtube-transcript-api

Cách đơn giản nhất là sử dụng thư viện youtube-transcript-api cho Python. Nó trích xuất phụ đề trực tiếp, không cần khóa API:

from youtube_transcript_api import YouTubeTranscriptApi

# Lấy phụ đề
video_id = 'dQw4w9WgXcQ'

try:
    # Cố gắng lấy phụ đề tiếng Nga, nếu không có thì lấy tiếng Anh
    transcript = YouTubeTranscriptApi.get_transcript(video_id, languages=['ru', 'en'])
    
    # Xuất phụ đề với dấu thời gian
    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"Lỗi khi lấy phụ đề: {e}")

# Lấy danh sách các ngôn ngữ có sẵn
transcript_list = YouTubeTranscriptApi.list_transcripts(video_id)
for transcript in transcript_list:
    print(f"Ngôn ngữ: {transcript.language}, Tự động: {transcript.is_generated}")

Thư viện tự động xác định các phụ đề có sẵn và có thể dịch chúng sang các ngôn ngữ khác (nếu YouTube cung cấp khả năng này).

Xử lý dấu thời gian cho RAG

Đối với hệ thống RAG, điều quan trọng là duy trì mối liên hệ giữa văn bản và dấu thời gian. Điều này cho phép tạo ra các liên kết chính xác đến các nguồn:

def format_timestamp(seconds):
    """Chuyển đổi giây thành định dạng 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):
    """Chia phụ đề thành các khối với dấu thời gian"""
    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'] + " "
        
        # Nếu đạt kích thước cần thiết hoặc hết phụ đề
        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

# Sử dụng
transcript = YouTubeTranscriptApi.get_transcript(video_id)
chunks = create_chunks_with_timestamps(transcript)

for chunk in chunks[:3]:  # Ba khối đầu tiên
    print(f"[{chunk['timestamp']}] {chunk['text'][:100]}...")

Thu thập siêu dữ liệu: tiêu đề, mô tả, thẻ

Siêu dữ liệu làm phong phú thêm ngữ cảnh cho hệ thống RAG. Dưới đây là ví dụ đầy đủ về việc thu thập tất cả các dữ liệu cần thiết:

import requests
from datetime import datetime

def collect_video_metadata(video_id, api_key):
    """Thu thập siêu dữ liệu đầy đủ của video"""
    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

# Ví dụ sử dụng
metadata = collect_video_metadata('dQw4w9WgXcQ', API_KEY)
print(f"Tiêu đề: {metadata['title']}")
print(f"Kênh: {metadata['channel_title']}")
print(f"Lượt xem: {metadata['view_count']:,}")
print(f"Thẻ: {', '.join(metadata['tags'][:5])}")

Xác định tính cập nhật của nội dung

Đối với các chủ đề kỹ thuật, tính mới của thông tin là rất quan trọng. Hãy thêm một hàm để đánh giá tính cập nhật:

from datetime import datetime, timedelta

def calculate_content_freshness(published_date_str):
    """Đánh giá tính cập nhật của nội dung"""
    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 'rất mới'
    elif age_days < 180:
        return 'mới'
    elif age_days < 365:
        return 'vừa phải'
    else:
        return 'cũ'

def calculate_quality_score(metadata):
    """Tính toán điểm chất lượng của nguồn"""
    score = 0
    
    # Độ phổ biến
    views = metadata['view_count']
    if views > 100000:
        score += 3
    elif views > 10000:
        score += 2
    elif views > 1000:
        score += 1
    
    # Engagement (lượt thích so với lượt xem)
    if views > 0:
        like_ratio = metadata['like_count'] / views
        if like_ratio > 0.05:
            score += 2
        elif like_ratio > 0.02:
            score += 1
    
    # Tính cập nhật
    freshness = calculate_content_freshness(metadata['published_at'])
    if freshness == 'rất mới':
        score += 2
    elif freshness == 'mới':
        score += 1
    
    return score

# Sử dụng
metadata = collect_video_metadata('dQw4w9WgXcQ', API_KEY)
quality = calculate_quality_score(metadata)
freshness = calculate_content_freshness(metadata['published_at'])
print(f"Điểm chất lượng: {quality}/7")
print(f"Tính cập nhật: {freshness}")

Phân tích bình luận để phân tích ngữ cảnh

Bình luận có thể chứa thông tin quý giá: sửa lỗi trong video, tài nguyên bổ sung, câu hỏi thường gặp. Đối với hệ thống RAG, đây là ngữ cảnh bổ sung.

def get_video_comments(video_id, api_key, max_results=100):
    """Lấy bình luận cho video"""
    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',  # Sắp xếp theo độ liên quan
            '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):
    """Lọc bình luận có giá trị"""
    valuable = []
    
    for comment in comments:
        # Tiêu chí giá trị:
        # 1. Nhiều lượt thích (độ phổ biến)
        # 2. Có phản hồi (gây ra thảo luận)
        # 3. Văn bản dài (bình luận chi tiết)
        
        if (comment['like_count'] >= min_likes or 
            comment['reply_count'] > 0 or 
            len(comment['text']) > 200):
            valuable.append(comment)
    
    return valuable

# Sử dụng
comments = get_video_comments('dQw4w9WgXcQ', API_KEY, max_results=50)
valuable_comments = filter_valuable_comments(comments)

print(f"Tổng số bình luận: {len(comments)}")
print(f"Bình luận có giá trị: {len(valuable_comments)}")

for comment in valuable_comments[:3]:
    print(f"\n[{comment['like_count']} lượt thích] {comment['author']}:")
    print(comment['text'][:200])

Sử dụng proxy để mở rộng thu thập dữ liệu

Khi thu thập dữ liệu quy mô lớn, có hai vấn đề phát sinh: giới hạn API YouTube (10,000 hạn ngạch mỗi ngày) và việc bị chặn khi phân tích phụ đề. Proxy giúp giải quyết cả hai vấn đề này.

Khi nào cần proxy cho phân tích YouTube

Vượt quá hạn ngạch API — sử dụng nhiều khóa API qua các IP khác nhau có thể tăng giới hạn hàng ngày
Phân tích phụ đề mà không qua API — thư viện youtube-transcript-api thực hiện các yêu cầu trực tiếp, có thể bị chặn khi tần suất cao
Thu thập dữ liệu từ các khu vực khác nhau — một số video chỉ có sẵn ở một số quốc gia nhất định
Thu thập song song — phân phối tải trên nhiều IP để tăng tốc độ

Lựa chọn loại proxy

Loại proxy	Ưu điểm	Khi nào sử dụng
Data-center	Tốc độ cao, giá thấp	Làm việc với API, khối lượng nhỏ
Residential	Rủi ro bị chặn thấp, IP thực	Phân tích hàng loạt phụ đề, vượt qua hạn chế
Mobile	Tin cậy tối đa, ít bị chặn	Thu thập dữ liệu từ ứng dụng di động, nhiệm vụ quan trọng

Đối với hầu hết các nhiệm vụ của hệ thống RAG, proxy residential là lựa chọn phù hợp — chúng cung cấp sự cân bằng giữa chi phí và độ tin cậy khi phân tích quy mô lớn.

Cấu hình proxy trong mã

import requests
from youtube_transcript_api import YouTubeTranscriptApi
from youtube_transcript_api._api import TranscriptListFetcher

# Cấu hình proxy
PROXY = {
    'http': 'http://username:password@proxy-server:port',
    'https': 'http://username:password@proxy-server:port'
}

# Để làm việc với API qua proxy
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()

# Để phân tích phụ đề qua proxy
class ProxiedTranscriptApi:
    def __init__(self, proxy):
        self.proxy = proxy
    
    def get_transcript(self, video_id, languages=['en']):
        # Tạo phiên tùy chỉnh với proxy
        session = requests.Session()
        session.proxies = self.proxy
        
        # Sử dụng phiên cho các yêu cầu
        fetcher = TranscriptListFetcher(session)
        transcript_list = fetcher.fetch(video_id)
        
        # Lấy ngôn ngữ cần thiết
        for lang in languages:
            try:
                transcript = transcript_list.find_transcript([lang])
                return transcript.fetch()
            except:
                continue
        
        raise Exception(f"Không tìm thấy phụ đề cho các ngôn ngữ: {languages}")

# Sử dụng
api = ProxiedTranscriptApi(PROXY)
transcript = api.get_transcript('dQw4w9WgXcQ', languages=['ru', 'en'])
print(f"Đã nhận {len(transcript)} đoạn phụ đề")

Luân phiên proxy để mở rộng

Khi thu thập dữ liệu từ hàng ngàn video, điều quan trọng là phân phối tải giữa nhiều proxy:

import random
import time

class ProxyRotator:
    def __init__(self, proxy_list):
        self.proxies = proxy_list
        self.current_index = 0
    
    def get_next_proxy(self):
        """Luân phiên theo thứ tự"""
        proxy = self.proxies[self.current_index]
        self.current_index = (self.current_index + 1) % len(self.proxies)
        return proxy
    
    def get_random_proxy(self):
        """Luân phiên ngẫu nhiên"""
        return random.choice(self.proxies)

# Danh sách proxy
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:
            # Lấy siêu dữ liệu
            metadata = get_video_details_with_proxy(video_id, API_KEY, proxy)
            
            # Lấy phụ đề
            api = ProxiedTranscriptApi(proxy)
            transcript = api.get_transcript(video_id)
            
            results.append({
                'video_id': video_id,
                'metadata': metadata,
                'transcript': transcript
            })
            
            # Đợi giữa các yêu cầu
            time.sleep(1)
            
        except Exception as e:
            print(f"Lỗi cho {video_id}: {e}")
            continue
    
    return results

# Sử dụng
video_ids = ['video1', 'video2', 'video3', 'video4', 'video5']
data = collect_data_with_rotation(video_ids)
print(f"Đã thu thập dữ liệu cho {len(data)} video")

Xử lý và chuẩn bị dữ liệu cho RAG

Sau khi thu thập dữ liệu, cần xử lý và cấu trúc chúng để hệ thống RAG hoạt động hiệu quả.

Tạo nhúng vector

Hệ thống RAG sử dụng tìm kiếm vector để tìm các đoạn liên quan. Cần chuyển đổi văn bản thành nhúng:

from sentence_transformers import SentenceTransformer
import numpy as np

# Tải mô hình để tạo nhúng
model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')

def create_embeddings_from_transcript(transcript_chunks):
    """Tạo nhúng cho các khối phụ đề"""
    embeddings = []
    
    for chunk in transcript_chunks:
        # Kết hợp văn bản với siêu dữ liệu để có ngữ cảnh tốt hơn
        text_with_context = f"{chunk['title']} | {chunk['text']}"
        
        # Tạo nhúng
        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

# Chuẩn bị dữ liệu
def prepare_rag_data(video_data):
    """Chuẩn bị tất cả dữ liệu cho RAG"""
    all_chunks = []
    
    for video in video_data:
        metadata = video['metadata']
        transcript = video['transcript']
        
        # Chia phụ đề thành các khối
        chunks = create_chunks_with_timestamps(transcript)
        
        # Thêm siêu dữ liệu vào mỗi khối
        for chunk in chunks:
            chunk['title'] = metadata['title']
            chunk['channel'] = metadata['channel_title']
            chunk['views'] = metadata['view_count']
            all_chunks.append(chunk)
    
    # Tạo nhúng
    embeddings = create_embeddings_from_transcript(all_chunks)
    
    return embeddings

# Sử dụng
rag_data = prepare_rag_data(collected_videos)
print(f"Đã chuẩn bị {len(rag_data)} đoạn cho RAG")

Lưu vào cơ sở dữ liệu vector

Để tìm kiếm hiệu quả, các nhúng được lưu vào các cơ sở dữ liệu chuyên dụng. Các tùy chọn phổ biến: Pinecone, Weaviate, Qdrant, ChromaDB.

import chromadb
from chromadb.config import Settings

# Khởi tạo ChromaDB (cơ sở dữ liệu vector cục bộ)
client = chromadb.Client(Settings(
    chroma_db_impl="duckdb+parquet",
    persist_directory="./youtube_rag_db"
))

# Tạo bộ sưu tập
collection = client.create_collection(
    name="youtube_transcripts",
    metadata={"description": "Phụ đề video YouTube cho RAG"}
)

def store_in_vector_db(embeddings_data, collection):
    """Lưu nhúng vào cơ sở dữ liệu vector"""
    
    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"
        })
    
    # Thêm vào bộ sưu tập
    collection.add(
        ids=ids,
        embeddings=embeddings,
        documents=documents,
        metadatas=metadatas
    )
    
    print(f"Đã lưu {len(ids)} nhúng vào cơ sở dữ liệu vector")

# Sử dụng
store_in_vector_db(rag_data, collection)

Tìm kiếm và tạo câu trả lời

Bước cuối cùng là thực hiện tìm kiếm RAG và tạo câu trả lời:

def search_youtube_knowledge(query, collection, model, top_k=3):
    """Tìm kiếm các đoạn liên quan từ YouTube"""
    
    # Tạo nhúng cho truy vấn
    query_embedding = model.encode(query).tolist()
    
    # Tìm kiếm trong cơ sở dữ liệu vector
    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=top_k
    )
    
    # Định dạng kết quả
    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):
    """Tạo câu trả lời dựa trên các nguồn đã tìm thấy"""
    
    # Tạo ngữ cảnh từ các nguồn đã tìm thấy
    context = "\n\n".join([
        f"Nguồn: {s['metadata']['title']} ({s['metadata']['timestamp']})\n{s['text']}"
        for s in sources
    ])
    
    # Prompt cho LLM
    prompt = f"""Dựa trên các đoạn sau từ video YouTube, hãy trả lời câu hỏi của người dùng.
Nhất định phải chỉ ra các nguồn với dấu thời gian.

Ngữ cảnh:
{context}

Câu hỏi: {query}

Câu trả lời:"""
    
    # Đây là nơi gọi LLM của bạn (OpenAI, Claude, mô hình cục bộ)
    # Ví dụ với OpenAI:
    # response = openai.ChatCompletion.create(
    #     model="gpt-4",
    #     messages=[{"role": "user", "content": prompt}]
    # )
    # answer = response.choices[0].message.content
    
    # Để ví dụ, trả về prompt
    return {
        'answer': 'Đây sẽ là câu trả lời của LLM',
        'sources': sources
    }

# Sử dụng
query = "Cách cấu hình proxy trong Python?"
sources = search_youtube_knowledge(query, collection, model, top_k=3)

print("Các nguồn đã tìm thấy:")
for source in sources:
    print(f"\n{source['metadata']['title']}")
    print(f"Thời gian: {source['metadata']['timestamp']}")
    print(f"Liên kết: {source['metadata']['youtube_url']}")
    print(f"Nội dung: {source['text'][:200]}...")

Tối ưu hóa chất lượng RAG

Một vài mẹo để cải thiện chất lượng hệ thống RAG dựa trên dữ liệu YouTube:

Lọc nội dung chất lượng thấp — sử dụng các chỉ số lượt xem, lượt thích, độ mới của bài đăng
Giữ ngữ cảnh — thêm tiêu đề video và kênh vào mỗi khối văn bản
Tối ưu hóa kích thước khối — đối với nội dung kỹ thuật, 300-500 từ là tối ưu
Sử dụng siêu dữ liệu để xếp hạng — nội dung mới và phổ biến hơn có thể được ưu tiên
Thêm bình luận — chúng thường chứa các làm rõ quan trọng và FAQ
Kiểm tra tính khả dụng của video — một số video có thể bị xóa hoặc trở thành riêng tư

Mẹo: Đối với việc thu thập dữ liệu YouTube quy mô lớn, nên sử dụng sự kết hợp giữa API chính thức (cho siêu dữ liệu) và phân tích qua proxy (cho phụ đề). Điều này cho phép vượt qua các giới hạn và thu thập tối đa thông tin.

Kết luận

Thu thập dữ liệu YouTube cho hệ thống RAG là một quá trình nhiều bước, bao gồm làm việc với API, phân tích phụ đề, xử lý siêu dữ liệu và tạo nhúng vector. Những điểm chính:

YouTube Data API v3 cung cấp siêu dữ liệu, thống kê và bình luận với giới hạn 10,000 hạn ngạch mỗi ngày
Phụ đề được phân tích qua thư viện youtube-transcript-api hoặc các yêu cầu trực tiếp
Dấu thời gian rất quan trọng để tạo ra các liên kết chính xác đến các nguồn
Siêu dữ liệu (lượt xem, lượt thích, ngày đăng) giúp đánh giá chất lượng và tính cập nhật của nội dung
Bình luận thêm ngữ cảnh và thường chứa FAQ
Proxy là cần thiết để mở rộng và vượt qua các giới hạn
Nhúng vector và cơ sở dữ liệu chuyên dụng cung cấp tìm kiếm ngữ nghĩa nhanh chóng

Với việc thiết lập quy trình đúng cách, có thể thu thập hàng chục nghìn đoạn nội dung chất lượng mỗi ngày, tạo ra một cơ sở tri thức mạnh mẽ cho các hệ thống RAG trong bất kỳ lĩnh vực nào.

Nếu bạn dự định thu thập dữ liệu YouTube quy mô lớn với việc vượt qua các giới hạn và chặn, chúng tôi khuyên bạn nên sử dụng proxy residential — chúng cung cấp sự ổn định khi phân tích hàng ngàn video và giảm thiểu rủi ro bị chặn từ YouTube.

```

Cách thu thập dữ liệu YouTube cho hệ thống RAG: phân tích phụ đề và siêu dữ liệu bằng Python