Trong bối cảnh các mô hình ngôn ngữ lớn (LLM) ngày càng được ứng dụng rộng rãi, hệ thống RAG (Retrieval-Augmented Generation) đã trở thành kiến trúc tiêu chuẩn cho việc xây dựng ứng dụng AI tư vấn tri thức. Tuy nhiên, một mối đe dọa nghiêm trọng đang âm thầm phá hoại độ chính xác của RAG: retrieval pollution attack. Bài viết này sẽ phân tích sâu cơ chế tấn công, chiến lược phòng thủ, và cung cấp mã nguồn thực chiến để bảo vệ hệ thống của bạn.

Bối Cảnh Thị Trường và Chi Phí Vận Hành 2026

Trước khi đi sâu vào kỹ thuật, chúng ta cần hiểu rõ bối cảnh chi phí khi triển khai RAG system. Dưới đây là bảng so sánh giá của các provider hàng đầu:

So sánh chi phí cho hệ thống xử lý 10 triệu token/tháng:

+-------------------+---------------+----------------+
| Provider          | Chi phí/MTok  | Tổng/tháng     |
+-------------------+---------------+----------------+
| GPT-4.1           | $8.00         | $80,000         |
| Claude Sonnet 4.5 | $15.00        | $150,000        |
| Gemini 2.5 Flash  | $2.50         | $25,000         |
| DeepSeek V3.2     | $0.42         | $4,200          |
+-------------------+---------------+----------------+
Tiết kiệm vs GPT-4.1: DeepSeek V3.2 giảm 94.75% chi phí

Đây là lý do tại sao việc bảo vệ RAG system khỏi pollution attack không chỉ là vấn đề bảo mật mà còn ảnh hưởng trực tiếp đến chi phí vận hành. Khi hệ thống bị tấn công, model có thể generate ra nội dung sai lệch, buộc phải gọi lại nhiều lần hoặc tệ hơn là trả lời sai hoàn toàn khiến người dùng mất niềm tin.

Retrieval Pollution Attack Là Gì?

Retrieval pollution attack là kỹ thuật tấn công nhằm mục đích thao túng kết quả trả về từ vector database bằng cách chèn các tài liệu độc hại vào index. Khi retrieval layer trả về kết quả bị ô nhiễm, LLM sẽ dựa vào đó để generate ra câu trả lời sai lệch hoặc có hại.

Các Loại Pollution Attack Phổ Biến

Kiến Trúc Phòng Thủ Nhiều Lớp

Để bảo vệ RAG system hiệu quả, chúng ta cần triển khai kiến trúc phòng thủ theo chiến lược defense-in-depth với 4 lớp chính.

Lớp 1: Input Validation và Sanitization

import re
from typing import List, Dict, Tuple
import hashlib

class InputSanitizer:
    """
    Lớp sanitization đầu vào - loại bỏ các pattern tấn công phổ biến
    """
    
    DANGEROUS_PATTERNS = [
        r'\b(SQL|HTML|JavaScript|Injection)\s*:',
        r'<script|javascript:',
        r'\beval\s*\(',
        r'\bunion\s+select\b',
        r'[\x00-\x1f\x7f-\x9f]',  # Control characters
        r'(\b){5,}',  # Excessive word repetition (spam indicator)
    ]
    
    def __init__(self):
        self.compiled_patterns = [
            re.compile(p, re.IGNORECASE) for p in self.DANGEROUS_PATTERNS
        ]
        self.known_poisoned_hashes: set = set()
    
    def sanitize(self, text: str) -> Tuple[str, List[str]]:
        """
        Sanitize input text và trả về danh sách warnings
        
        Returns:
            Tuple[str, List[str]]: (sanitized_text, list_of_warnings)
        """
        warnings = []
        sanitized = text
        
        # Check against dangerous patterns
        for i, pattern in enumerate(self.compiled_patterns):
            matches = pattern.findall(sanitized)
            if matches:
                warnings.append(f"Pattern {i} detected: {matches[:3]}")
                sanitized = pattern.sub('[REDACTED]', sanitized)
        
        # Check text length
        if len(sanitized) > 10000:
            warnings.append("Text truncated due to length limit")
            sanitized = sanitized[:10000]
        
        # Compute content hash for poisoning tracking
        content_hash = hashlib.sha256(sanitized.encode()).hexdigest()[:16]
        
        if content_hash in self.known_poisoned_hashes:
            warnings.append(f"Content hash {content_hash} flagged as poisoned")
        
        return sanitized, warnings

Usage example

sanitizer = InputSanitizer() clean_text, warnings = sanitizer.sanitize( "Tìm kiếm về UNION SELECT vulnerability trong database" ) print(f"Sanitized: {clean_text}") print(f"Warnings: {warnings}")

Lớp 2: Retrieval Verification System

Lớp này đảm bảo rằng kết quả retrieval không bị ô nhiễm bằng cách cross-verify với multiple sources và kiểm tra consistency.

import numpy as np
from typing import List, Dict, Tuple
from dataclasses import dataclass

@dataclass
class RetrievalResult:
    chunk_id: str
    content: str
    score: float
    source: str
    embedding_hash: str

class RetrievalVerifier:
    """
    Verify retrieval results against poisoning attempts
    """
    
    def __init__(self, min_consistency_score: float = 0.7):
        self.min_consistency_score = min_consistency_score
        self.metadata_store: Dict[str, Dict] = {}
    
    def verify_batch(
        self, 
        results: List[RetrievalResult], 
        query: str,
        context_window: int = 3
    ) -> Tuple[List[RetrievalResult], List[str]]:
        """
        Verify batch of retrieval results
        
        Args:
            results: List of retrieved chunks
            query: Original query for consistency check
            context_window: Number of neighboring chunks to check
            
        Returns:
            Tuple of (verified_results, verification_warnings)
        """
        verified = []
        warnings = []
        
        for i, result in enumerate(results):
            checks_passed = 0
            total_checks = 0
            
            # Check 1: Metadata consistency
            if result.chunk_id in self.metadata_store:
                stored = self.metadata_store[result.chunk_id]
                if stored.get('embedding_hash') == result.embedding_hash:
                    checks_passed += 1
                total_checks += 1
            
            # Check 2: Content quality (length, readability)
            if 50 < len(result.content) < 5000:
                checks_passed += 1
            total_checks += 1
            
            # Check 3: Source credibility
            credible_sources = {'official_docs', 'verified_kb', 'trusted_cms'}
            if result.source in credible_sources:
                checks_passed += 1
            total_checks += 1
            
            # Check 4: No obvious injection markers
            injection_markers = ['REDACTED', '[MALICIOUS]', '...truncated']
            if not any(marker in result.content for marker in injection_markers):
                checks_passed += 1
            total_checks += 1
            
            consistency_score = checks_passed / total_checks
            
            if consistency_score >= self.min_consistency_score:
                verified.append(result)
            else:
                warnings.append(
                    f"Chunk {result.chunk_id} failed verification "
                    f"(score: {consistency_score:.2f})"
                )
        
        return verified, warnings
    
    def register_chunk(self, chunk_id: str, metadata: Dict):
        """Register chunk metadata for future verification"""
        self.metadata_store[chunk_id] = metadata

Cross-verification with semantic consistency

class SemanticConsistencyChecker: """Ensure retrieved chunks are semantically consistent""" def __init__(self, threshold: float = 0.6): self.threshold = threshold def check_consistency( self, chunks: List[str], query: str ) -> Dict: """ Check if retrieved chunks are consistent with each other """ # Simplified consistency check # In production, use embedding similarity between chunks word_overlap_scores = [] for i, chunk in enumerate(chunks): for j, other_chunk in enumerate(chunks[i+1:], i+1): words_a = set(chunk.lower().split()) words_b = set(other_chunk.lower().split()) if len(words_a) > 0 and len(words_b) > 0: overlap = len(words_a & words_b) score = overlap / min(len(words_a), len(words_b)) word_overlap_scores.append(score) avg_consistency = np.mean(word_overlap_scores) if word_overlap_scores else 0 return { 'consistent': avg_consistency >= self.threshold, 'score': float(avg_consistency), 'num_pairs_checked': len(word_overlap_scores) }

Lớp 3: Hybrid Retrieval Với Re-ranking Defense

Triển khai hybrid retrieval kết hợp sparse (BM25) và dense (embedding) retrieval, sau đó áp dụng re-ranker có khả năng phát hiện pollution.

import asyncio
from enum import Enum

class TrustLevel(Enum):
    """Mức độ tin cậy của nguồn dữ liệu"""
    TRUSTED = 1.0
    VERIFIED = 0.8
    STANDARD = 0.6
    UNVERIFIED = 0.3
    FLAGGED = 0.0

class HybridRetrievalWithDefense:
    """
    Hybrid retrieval system với built-in poisoning defense
    """
    
    def __init__(self, api_key: str, base_url: str):
        self.base_url = base_url
        self.api_key = api_key
        self.trust_weights = {}
    
    async def retrieve_with_defense(
        self,
        query: str,
        top_k: int = 10,
        min_trust_score: float = 0.5
    ) -> Dict:
        """
        Retrieve documents với multi-layer defense
        """
        # Step 1: Parallel sparse và dense retrieval
        sparse_results = await self._bm25_search(query, top_k * 2)
        dense_results = await self._embedding_search(query, top_k * 2)
        
        # Step 2: Merge results với trust scoring
        merged = self._merge_with_trust(
            sparse_results, 
            dense_results,
            min_trust_score
        )
        
        # Step 3: Apply defensive re-ranking
        reranked = await self._defensive_rerank(query, merged)
        
        # Step 4: Final filtering
        final_results = [
            r for r in reranked 
            if r['trust_score'] >= min_trust_score
        ][:top_k]
        
        return {
            'results': final_results,
            'total_candidates': len(merged),
            'defense_flags': self._get_defense_flags(merged)
        }
    
    async def _embedding_search(self, query: str, limit: int):
        """Dense retrieval sử dụng embedding model"""
        # Call HolySheep API cho embedding
        import aiohttp
        
        headers = {
            'Authorization': f'Bearer {self.api_key}',
            'Content-Type': 'application/json'
        }
        
        async with aiohttp.ClientSession() as session:
            # Get query embedding
            embed_payload = {
                'model': 'text-embedding-3-large',
                'input': query
            }
            
            async with session.post(
                f'{self.base_url}/embeddings',
                json=embed_payload,
                headers=headers
            ) as resp:
                embed_result = await resp.json()
                query_embedding = embed_result['data'][0]['embedding']
            
            # Search vector store (simplified)
            return await self._vector_search(query_embedding, limit)
    
    async def _defensive_rerank(
        self, 
        query: str, 
        candidates: List[Dict]
    ) -> List[Dict]:
        """Defensive re-ranking với poisoning detection"""
        # Gọi HolySheep API cho re-ranking
        import aiohttp
        
        headers = {
            'Authorization': f'Bearer {self.api_key}',
            'Content-Type': 'application/json'
        }
        
        # Prepare pairs cho cross-encoder
        pairs = [[query, cand['content']] for cand in candidates]
        
        rerank_payload = {
            'model': 'cross-encoder/ms-marco-MiniLML-6-v2',
            'query': query,
            'documents': [p[1] for p in pairs]
        }
        
        async with aiohttp.ClientSession() as session:
            async with session.post(
                f'{self.base_url}/rerank',
                json=rerank_payload,
                headers=headers
            ) as resp:
                rerank_result = await resp.json()
        
        # Apply trust score adjustment
        for i, result in enumerate(rerank_result['results']):
            candidates[i]['rerank_score'] = result['relevance_score']
            # Reduce score nếu trust thấp
            candidates[i]['final_score'] = (
                result['relevance_score'] * 
                candidates[i].get('trust_score', 0.5)
            )
        
        # Sort by final score
        return sorted(candidates, key=lambda x: x['final_score'], reverse=True)
    
    def _merge_with_trust(
        self,
        sparse_results: List[Dict],
        dense_results: List[Dict],
        min_trust: float
    ) -> List[Dict]:
        """Merge sparse và dense results với trust scoring"""
        seen_ids = set()
        merged = []
        
        for source_name, results in [('sparse', sparse_results), ('dense', dense_results)]:
            for r in results:
                chunk_id = r['chunk_id']
                if chunk_id in seen_ids:
                    continue
                
                trust_score = self.trust_weights.get(
                    r.get('source', 'unknown'), 
                    TrustLevel.STANDARD.value
                )
                
                r['trust_score'] = trust_score
                r['source_type'] = source_name
                
                if trust_score >= min_trust:
                    merged.append(r)
                    seen_ids.add(chunk_id)
        
        return merged
    
    def set_source_trust(self, source: str, level: TrustLevel):
        """Cập nhật trust level cho một nguồn"""
        self.trust_weights[source] = level.value
    
    def _get_defense_flags(self, candidates: List[Dict]) -> List[str]:
        """Extract defense flags từ candidates"""
        flags = []
        
        low_trust_count = sum(
            1 for c in candidates 
            if c.get('trust_score', 0) < 0.3
        )
        if low_trust_count > len(candidates) * 0.3:
            flags.append('HIGH_POLLUTION_SUSPECTED')
        
        return flags

Khởi tạo với HolySheep API

config = { 'api_key': 'YOUR_HOLYSHEEP_API_KEY', 'base_url': 'https://api.holysheep.ai/v1' } retrieval_system = HybridRetrievalWithDefense( api_key=config['api_key'], base_url=config['base_url'] )

Cấu hình trust levels

retrieval_system.set_source_trust('official_docs', TrustLevel.TRUSTED) retrieval_system.set_source_trust('verified_kb', TrustLevel.VERIFIED) retrieval_system.set_source_trust('user_uploads', TrustLevel.UNVERIFIED)

Lớp 4: Output Guardrails

from typing import Optional, List
from dataclasses import dataclass
import re

@dataclass
class GuardrailConfig:
    enable_toxicity_check: bool = True
    enable_factuality_check: bool = True
    enable_similarity_check: bool = True
    max_response_length: int = 4000

class RAGOutputGuardrails:
    """
    Final layer defense - kiểm tra output trước khi trả về user
    """
    
    def __init__(self, config: GuardrailConfig):
        self.config = config
        self.retrieval_context: List[str] = []
    
    def set_context(self, contexts: List[str]):
        """Set retrieval context để so sánh với output"""
        self.retrieval_context = contexts
    
    async def validate_output(self, output: str) -> Dict:
        """
        Validate LLM output against various attack vectors
        """
        issues = []
        
        # Check 1: Toxicity
        if self.config.enable_toxicity_check:
            toxicity_issues = self._check_toxicity(output)
            issues.extend(toxicity_issues)
        
        # Check 2: Factuality consistency
        if self.config.enable_factuality_check:
            factuality_issues = await self._check_factuality(output)
            issues.extend(factuality_issues)
        
        # Check 3: Prompt injection patterns
        injection_patterns = self._detect_prompt_injection(output)
        if injection_patterns:
            issues.append({
                'type': 'PROMPT_INJECTION',
                'details': injection_patterns
            })
        
        # Check 4: Length limit
        if len(output) > self.config.max_response_length:
            issues.append({
                'type': 'LENGTH_EXCEEDED',
                'details': f'{len(output)} > {self.config.max_response_length}'
            })
        
        return {
            'passed': len(issues) == 0,
            'issues': issues,
            'sanitized_output': self._sanitize_output(output, issues)
        }
    
    def _check_toxicity(self, text: str) -> List[Dict]:
        """Simple toxicity check"""
        # Production nên dùng dedicated toxicity API
        toxic_keywords = ['hate', 'violent', 'explicit']
        found = [kw for kw in toxic_keywords if kw.lower() in text.lower()]
        
        return [{
            'type': 'TOXICITY',
            'details': found
        }] if found else []
    
    async def _check_factuality(self, output: str) -> List[Dict]:
        """Check factuality against retrieval context"""
        issues = []
        
        for context in self.retrieval_context:
            # Simple overlap check
            context_words = set(context.lower().split())
            output_words = set(output.lower().split())
            
            overlap_ratio = len(context_words & output_words) / max(len(context_words), 1)
            
            # Output quá khác bi