RAG (Retrieval-Augmented Generation)

Giới thiệu

RAG (Retrieval-Augmented Generation) là một kỹ thuật trong AI kết hợp khả năng tìm kiếm thông tin với khả năng sinh văn bản của các mô hình ngôn ngữ lớn.

Kiến trúc cơ bản

RAG gồm hai thành phần chính:

• Retriever: Tìm kiếm và lấy thông tin liên quan từ cơ sở tri thức
• Generator: Mô hình ngôn ngữ sinh ra câu trả lời dựa trên thông tin đã tìm được

Quy trình hoạt động

Bước 1: Chuẩn bị dữ liệu

• Chia nhỏ tài liệu thành các đoạn (chunks)
• Chuyển đổi các đoạn thành vector embedding
• Lưu trữ trong vector database

Bước 2: Xử lý câu hỏi

• Nhận câu hỏi từ người dùng
• Chuyển câu hỏi thành vector embedding

Bước 3: Tìm kiếm (Retrieval)

• So sánh vector của câu hỏi với các vector trong database
• Tìm ra những đoạn văn bản có độ tương đồng cao nhất (thường 3-10 đoạn)

Bước 4: Sinh câu trả lời (Generation)

• Kết hợp câu hỏi gốc với các đoạn thông tin đã tìm được
• Đưa vào mô hình ngôn ngữ để sinh ra câu trả lời coherent và accurate

Ưu điểm

• Truy cập được thông tin cập nhật và domain-specific
• Giảm hallucination của mô hình ngôn ngữ
• Có thể truy vết nguồn gốc thông tin
• Không cần fine-tune mô hình chính

Ứng dụng phổ biến

• Chatbot doanh nghiệp
• Hệ thống Q&A trên tài liệu
• Tìm kiếm thông tin thông minh
• Hỗ trợ khách hàng tự động

Các Model Embedding Text

• text-embedding-3-large/small (OpenAI): Hỗ trợ tốt đa ngôn ngữ, thay thế ada-002
• multilingual-e5-large/base (Microsoft): Được thiết kế đặc biệt cho multilingual
• BGE-M3 (BAAI): Xuất sắc cho multilingual retrieval
• Cohere embed-multilingual-v3.0: Hiệu suất mạnh mẽ trên nhiều ngôn ngữ

Ví dụ thực tế

1. Indexing Phase (Giai đoạn lập chỉ mục)

Document Chunking

Chia tài liệu lớn thành các chunks nhỏ (thường 200-1000 tokens)

Các strategies phổ biến: - Fixed-size chunking (cố định kích thước) - Sentence-based chunking (theo câu) - Paragraph-based chunking (theo đoạn văn) - Semantic chunking (dựa trên ngữ nghĩa)

Embedding Conversion

• Text-embedding-ada-002 tạo ra vector 1536 dimensions
• Mỗi chunk → một vector đại diện cho semantic meaning
• Process: text → tokenization → embedding model → dense vector

Vector Storage trong ChromaDB

# Ví dụ với ChromaDB
collection.add(
    documents=chunks,
    embeddings=embeddings,
    metadatas=metadata,
    ids=unique_ids
)

2. Retrieval Phase (Giai đoạn truy xuất)

Query Processing

• User query cũng được convert thành embedding vector cùng model
• Đảm bảo consistency trong embedding space

Similarity Search

• Sử dụng cosine similarity hoặc euclidean distance
• ChromaDB hỗ trợ các similarity metrics khác nhau
• Top-k retrieval (thường k=3-10) dựa trên similarity scores

Advanced Retrieval Techniques

• Hybrid search (kết hợp semantic + keyword search)
• Re-ranking với cross-encoder models
• Query expansion và reformulation

3. Generation Phase (Giai đoạn sinh câu trả lời)

Context Construction

System: You are a helpful assistant. Use the following context to answer the question.

Context: [Retrieved chunks joined together]

Question: [Original user query]

Answer:

LLM Processing

• GPT-4/Gemini nhận được full context
• Model sử dụng retrieved information để generate accurate answer
• Có thể include citations đến source chunks

Quality Control

• Answer relevance checking
• Hallucination detection
• Source attribution

Đánh giá và Metrics (Evaluation & Metrics)

Metrics quan trọng cần track

# Core RAG Metrics
rag_metrics = {
    "retrieval_precision": 0.85,      # Độ chính xác của retrieval
    "retrieval_recall": 0.78,         # Độ phủ của retrieval  
    "answer_relevance": 0.90,         # Câu trả lời có liên quan không
    "answer_faithfulness": 0.88,      # Câu trả lời có trung thực không
    "context_utilization": 0.75,      # Sử dụng context hiệu quả không
    "hallucination_rate": 0.05        # Tỷ lệ hallucination (< 5% là tốt)
}

Evaluation Framework

def evaluate_rag_system(test_dataset):
    results = []
    for test_case in test_dataset:
        query = test_case["question"]
        expected_answer = test_case["ground_truth"]

        # Get RAG response
        retrieved_chunks = retrieve(query)
        generated_answer = generate(query, retrieved_chunks)

        # Calculate metrics
        metrics = {
            "retrieval_score": calculate_retrieval_quality(retrieved_chunks, test_case["relevant_docs"]),
            "answer_score": calculate_answer_quality(generated_answer, expected_answer),
            "latency": measure_response_time()
        }
        results.append(metrics)

    return aggregate_results(results)

So sánh Models Embedding

Model	Ngôn ngữ	Performance	Cost	Kích thước	Use Case
text-embedding-3-large	Excellent	95%	$$	3072 dim	Production enterprise
text-embedding-3-small	Very Good	90%	$	1536 dim	Cost-effective production
multilingual-e5-large	Excellent	92%	Free	1024 dim	Multilingual, open source
BGE-M3	Very Good	91%	Free	1024 dim	Chinese + English
Cohere embed-v3	Good	89%	$	1024 dim	Good balance cost/performance

Recommendation:

• Startup/MVP: multilingual-e5-large (free, good quality)
• Production: text-embedding-3-small (best cost/performance)
• Enterprise: text-embedding-3-large (highest quality)

Error Handling & Edge Cases

Common Issues và Solutions

class RAGErrorHandler:
    def handle_retrieval_errors(self, query):
        try:
            chunks = self.retrieve(query)

            if len(chunks) == 0:
                # Case 1: Không tìm thấy relevant chunks
                return self.fallback_search(query)

            if self.is_low_confidence(chunks):
                # Case 2: Confidence score thấp
                return self.expand_query_and_retry(query)

            return chunks

        except EmbeddingError:
            # Case 3: Embedding model failure
            return self.use_backup_embedding_model(query)

        except DatabaseError:
            # Case 4: Vector DB connection issue
            return self.use_cached_results(query)

    def handle_generation_errors(self, query, context):
        try:
            response = self.llm.generate(query, context)

            if self.detect_hallucination(response, context):
                # Case 5: Detect hallucination
                return self.regenerate_with_stricter_prompt(query, context)

            return response

        except TokenLimitError:
            # Case 6: Context quá dài
            return self.truncate_context_and_retry(query, context)

Edge Cases cần xử lý

1. Empty retrieval: Query không match với bất kỳ chunk nào
2. Low confidence scores: Tất cả chunks đều có similarity thấp
3. Context overflow: Tổng chunks vượt quá token limit
4. Contradictory information: Chunks có thông tin mâu thuẫn
5. Outdated information: Chunks chứa thông tin cũ

Production Considerations

Performance Optimization

# Caching Strategy
class RAGCache:
    def __init__(self):
        self.query_cache = {}  # Cache cho frequent queries
        self.embedding_cache = {}  # Cache cho embeddings

    def get_cached_response(self, query):
        query_hash = hash(query)
        if query_hash in self.query_cache:
            return self.query_cache[query_hash]
        return None

    def cache_response(self, query, response):
        self.query_cache[hash(query)] = response

# Load Balancing
class RAGLoadBalancer:
    def __init__(self, embedding_models, llm_models):
        self.embedding_models = embedding_models
        self.llm_models = llm_models

    def get_best_model(self, current_load):
        # Route requests based on load
        return min(self.embedding_models, key=lambda m: m.current_load)

Cost Management

# Cost Tracking
daily_costs = {
    "embedding_api_calls": 1250,      # $0.05
    "llm_api_calls": 340,             # $2.40  
    "vector_db_operations": 5000,     # $0.10
    "total_daily_cost": "$2.55"
}

# Cost Optimization Strategies
def optimize_costs():
    # 1. Batch embedding generation
    # 2. Cache frequent queries  
    # 3. Use smaller models for simple queries
    # 4. Implement query routing based on complexity

Monitoring & Alerting

# Key metrics để monitor
monitoring_dashboard = {
    "response_time_p95": "< 2s",
    "error_rate": "< 1%", 
    "daily_cost": "< $5",
    "user_satisfaction": "> 4.0/5",
    "hallucination_rate": "< 3%"
}

# Alerts
alerts = [
    {"metric": "response_time", "threshold": "5s", "action": "scale_up"},
    {"metric": "error_rate", "threshold": "5%", "action": "investigate"},
    {"metric": "cost", "threshold": "$10/day", "action": "review_usage"}
]

Security & Privacy

Data Protection

class RAGSecurity:
    def sanitize_input(self, query):
        # Remove PII, malicious code, etc.
        cleaned_query = self.remove_pii(query)
        cleaned_query = self.sanitize_sql_injection(cleaned_query)
        return cleaned_query

    def filter_sensitive_chunks(self, chunks):
        # Remove chunks containing sensitive information
        safe_chunks = []
        for chunk in chunks:
            if not self.contains_pii(chunk.text):
                safe_chunks.append(chunk)
        return safe_chunks

    def anonymize_response(self, response):
        # Replace any remaining PII in response
        return self.mask_sensitive_data(response)

Access Control

# Role-based access to documents
access_control = {
    "public_docs": ["company_blog", "product_docs"],
    "internal_docs": ["hr_policies", "financial_reports"], 
    "confidential_docs": ["strategic_plans", "legal_docs"]
}

def check_document_access(user_role, document_id):
    if user_role == "employee" and document_id in access_control["internal_docs"]:
        return True
    elif user_role == "manager" and document_id in access_control["confidential_docs"]:
        return True
    return False

Limitations & Challenges

Known Limitations

1. Context Window: Limited context size (4k-32k tokens)
2. Multi-hop Reasoning: Khó xử lý câu hỏi cần nhiều bước suy luận
3. Real-time Data: Không có access to real-time information
4. Complex Calculations: Yếu trong math và complex reasoning
5. Language Mixing: Performance giảm khi mix nhiều ngôn ngữ

Mitigation Strategies

# Solutions cho limitations
solutions = {
    "context_overflow": "Implement hierarchical chunking",
    "multi_hop": "Use iterative retrieval approach", 
    "real_time": "Integrate with live data APIs",
    "calculations": "Route math queries to specialized tools",
    "language_mixing": "Use language-specific models"
}

Complete Example Workflow

def complete_rag_pipeline():
    # Real-world example
    document = """
    Annual Report Q4 2023
    Revenue: $2.5M (up 25% from Q3)
    Profit: $450K (up 30% from Q3)  
    New customers: 1,200
    Employee count: 45
    """

    # Step 1: Process document
    chunks = chunk_document(document)
    embeddings = generate_embeddings(chunks)
    store_in_vectordb(chunks, embeddings)

    # Step 2: User query
    query = "What was the revenue growth in Q4?"

    # Step 3: Retrieve relevant chunks
    query_embedding = generate_embedding(query)
    relevant_chunks = search_similar(query_embedding, top_k=3)

    # Step 4: Generate response
    context = "\n".join([chunk.text for chunk in relevant_chunks])
    prompt = f"""
    Based on this context: {context}
    Answer: {query}
    """

    response = llm.generate(prompt)
    # Expected: "Revenue grew 25% in Q4 2023, reaching $2.5M compared to Q3"

    return response

Các tối ưu hóa quan trọng

Chunking Strategy

• Overlap giữa các chunks (50-200 tokens)
• Preserve context boundaries
• Include metadata (title, section, page number)

ChromaDB Optimizations

• Proper indexing parameters
• Batch processing cho large datasets
• Persistent storage configuration

Retrieval Improvements

• Query preprocessing (typo correction, expansion)
• Multi-query retrieval
• Parent-child chunking strategy

Generation Enhancements

• Prompt engineering for better context usage
• Temperature và top-p tuning
• Response formatting instructions