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:
- GPT-4.1 Output: $8.00/MTok
- Claude Sonnet 4.5 Output: $15.00/MTok
- Gemini 2.5 Flash Output: $2.50/MTok
- DeepSeek V3.2 Output: $0.42/MTok
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
- Text Injection: Chèn keyword để thao túng BM25/reranker scoring
- Vector Perturbation: Thêm embedding gần giống để đẩy kết quả chính xác ra khỏi top-k
- Semantic Drift: Thay đổi semantic meaning của toàn bộ cluster
- Backdoor Trigger: Nhúng pattern đặc biệt kích hoạt hành vi sai trong LLM
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