Bối Cảnh Thực Chiến: Khi Context Window Trở Thành Con Dao Hai Lưỡi
Tuần trước, tôi nhận được một yêu cầu từ khách hàng enterprise về việc xây dựng hệ thống RAG (Retrieval Augmented Generation) cho kho tài liệu pháp lý với hơn 50,000 trang PDF. Đội ngũ của họ đã thử nghiệm với Gemini 3.1 Pro và gặp phải một lỗi nghiêm trọng:
ConnectionError: HTTPSConnectionPool(host='generativelanguage.googleapis.com', port=443):
Max retries exceeded with url: /v1beta/models/gemini-1.5-pro-002:generateContent
(Caused by NewConnectionError: <requests.packages.urllib3.connection.VerifiedHTTPSConnection
object at 0x7f8a3c291050>: Failed to establish a new connection: [Errno 110] Connection timed out))
GeminiAPIError: 504 Gateway Timeout - Request timeout after 120 seconds
Total tokens: 1,847,293 | Processing time: 127.4s | Estimated cost: $23.47
Kết quả? Một yêu cầu đơn lẻ đã tiêu tốn **$23.47** chỉ vì timeout khi xử lý 1.8 triệu token. Đây là bài học đắt giá về việc tại sao chi phí context window cần được tính toán kỹ lưỡng trước khi triển khai production.
Gemini 3.1 Pro 2M: Tổng Quan Kỹ Thuật
Gemini 3.1 Pro hỗ trợ context window lên đến **2 triệu token**, cho phép xử lý toàn bộ tài liệu dài mà không cần chunking phức tạp. Tuy nhiên, đây là con dao hai lưỡi:
- **Ưu điểm**: Không cần RAG retrieval phức tạp, giữ nguyên ngữ cảnh toàn bộ
- **Nhược điểm**: Chi phí tăng theo cấp số nhân với độ dài document
- **Rủi ro**: Timeout, rate limit, và chi phí phát sinh không lường trước
So Sánh Chi Phí Thực Tế: Gemini 3.1 Pro vs Các Đối Thủ
Dựa trên bảng giá chính thức 2026 (tính theo $1 = ¥7.2):
Bảng So Sánh Chi Phí Cho 1 Triệu Token Đầu Vào:
┌─────────────────────┬──────────────┬────────────────┬───────────────┐
│ Model │ Input $/MTok │ Output $/MTok │ Savings vs GCP│
├─────────────────────┼──────────────┼────────────────┼───────────────┤
│ Gemini 3.1 Pro 2M │ $1.25 │ $5.00 │ Baseline │
│ GPT-4.1 │ $8.00 │ $32.00 │ -85% cheaper │
│ Claude Sonnet 4.5 │ $15.00 │ $75.00 │ -91% cheaper │
│ Gemini 2.5 Flash │ $2.50 │ $10.00 │ -50% cheaper │
└─────────────────────┴──────────────┴────────────────┴───────────────┘
Chi phí cho document 500K tokens (1 phút audio):
• Gemini 3.1 Pro: $0.625 input + $2.50 output = $3.125/request
• GPT-4.1: $4.00 + $16.00 = $20.00/request (538% đắt hơn)
• Claude Sonnet: $7.50 + $37.50 = $45.00/request (1340% đắt hơn)
Với tỷ giá **¥1 = $1** trên HolySheep AI, bạn có thể tiết kiệm **85%+** so với API gốc của Google.
Triển Khai RAG Với HolySheep AI: Code Mẫu Toàn Diện
Setup Environment và Configuration
import os
import time
import httpx
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from datetime import datetime
@dataclass
class RAGConfig:
"""Cấu hình cho RAG system với Gemini 3.1 Pro 2M"""
base_url: str = "https://api.holysheep.ai/v1"
api_key: str = "YOUR_HOLYSHEEP_API_KEY" # Thay thế bằng key thực tế
model: str = "gemini-3.1-pro-2m" # Model context 2M
max_retries: int = 3
timeout: float = 180.0 # 3 phút cho document dài
chunk_size: int = 50000 # Chunk 50K tokens để tối ưu chi phí
# Chi phí tracking
input_cost_per_mtok: float = 1.25 # $1.25/MTok input
output_cost_per_mtok: float = 5.00 # $5.00/MTok output
class CostTracker:
"""Theo dõi chi phí API thời gian thực"""
def __init__(self):
self.total_input_tokens = 0
self.total_output_tokens = 0
self.total_requests = 0
self.failed_requests = 0
self.request_history = []
def log_request(self, input_tokens: int, output_tokens: int,
latency_ms: float, success: bool = True,
error: Optional[str] = None):
"""Ghi log mỗi request để phân tích chi phí"""
self.total_input_tokens += input_tokens
self.total_output_tokens += output_tokens
self.total_requests += 1
if not success:
self.failed_requests += 1
request_cost = (input_tokens / 1_000_000 * 1.25 +
output_tokens / 1_000_000 * 5.00)
self.request_history.append({
'timestamp': datetime.now().isoformat(),
'input_tokens': input_tokens,
'output_tokens': output_tokens,
'latency_ms': latency_ms,
'cost': request_cost,
'success': success,
'error': error
})
def get_summary(self) -> Dict[str, Any]:
"""Tổng hợp báo cáo chi phí"""
total_input_cost = self.total_input_tokens / 1_000_000 * 1.25
total_output_cost = self.total_output_tokens / 1_000_000 * 5.00
total_cost = total_input_cost + total_output_cost
success_rate = (self.total_requests - self.failed_requests) / self.total_requests * 100
return {
'total_requests': self.total_requests,
'failed_requests': self.failed_requests,
'success_rate': f"{success_rate:.2f}%",
'total_input_tokens': self.total_input_tokens,
'total_output_tokens': self.total_output_tokens,
'total_input_cost': f"${total_input_cost:.4f}",
'total_output_cost': f"${total_output_cost:.4f}",
'total_cost': f"${total_cost:.4f}",
'avg_cost_per_request': f"${total_cost / max(self.total_requests, 1):.4f}"
}
Khởi tạo tracker toàn cục
cost_tracker = CostTracker()
print("✅ RAG Configuration initialized")
print(f"📊 Model: Gemini 3.1 Pro 2M Context")
print(f"💰 Input: $1.25/MToken | Output: $5.00/MToken")
RAG Engine Với Chunking Strategy Tối Ưu
import hashlib
import json
from collections import defaultdict
class GeminiRAGEngine:
"""
RAG Engine tối ưu cho Gemini 3.1 Pro 2M Context
Sử dụng chiến lược chunking thông minh để giảm chi phí
"""
def __init__(self, config: RAGConfig):
self.config = config
self.client = httpx.Client(
base_url=config.base_url,
headers={
"Authorization": f"Bearer {config.api_key}",
"Content-Type": "application/json"
},
timeout=config.timeout
)
self.document_cache = {} # Cache documents đã xử lý
def estimate_cost(self, text: str, include_context: bool = True) -> Dict[str, Any]:
"""Ước tính chi phí trước khi gọi API"""
# Đếm tokens (xấp xỉ 4 ký tự = 1 token cho tiếng Anh)
estimated_input_tokens = len(text) // 4
if include_context:
# Thêm overhead cho system prompt và examples
estimated_input_tokens += 2000
estimated_output_tokens = 4000 # Ước tính output trung bình
input_cost = estimated_input_tokens / 1_000_000 * self.config.input_cost_per_mtok
output_cost = estimated_output_tokens / 1_000_000 * self.config.output_cost_per_mtok
total_cost = input_cost + output_cost
return {
'estimated_input_tokens': estimated_input_tokens,
'estimated_output_tokens': estimated_output_tokens,
'estimated_input_cost': f"${input_cost:.4f}",
'estimated_output_cost': f"${output_cost:.4f}",
'total_cost': f"${total_cost:.4f}",
'within_2m_limit': estimated_input_tokens < 2_000_000
}
def smart_chunk(self, document: str, overlap: int = 500) -> List[Dict[str, Any]]:
"""
Chia document thành chunks với overlap để giữ ngữ cảnh
Sử dụng chiến lược semantic-aware chunking
"""
chunks = []
chunk_size = self.config.chunk_size
start = 0
chunk_id = 0
while start < len(document):
end = min(start + chunk_size, len(document))
# Tìm boundary gần nhất (dấu chấm, dòng mới, paragraph)
if end < len(document):
# Tìm paragraph boundary gần nhất
for sep in ['\n\n', '\n', '. ']:
last_sep = document.rfind(sep, start + chunk_size - 200, end)
if last_sep > start + chunk_size // 2:
end = last_sep + len(sep)
break
chunk_text = document[start:end].strip()
chunk_hash = hashlib.md5(chunk_text.encode()).hexdigest()[:8]
chunks.append({
'id': f"chunk_{chunk_id}_{chunk_hash}",
'text': chunk_text,
'start': start,
'end': end,
'token_count': len(chunk_text) // 4
})
start = end - overlap # Overlap để giữ ngữ cảnh
chunk_id += 1
return chunks
def retrieve_relevant_chunks(self, chunks: List[Dict], query: str,
top_k: int = 3) -> List[Dict]:
"""
Retrieval đơn giản dựa trên keyword matching
Trong production nên dùng embeddings vector search
"""
query_words = set(query.lower().split())
scored_chunks = []
for chunk in chunks:
chunk_words = set(chunk['text'].lower().split())
# Tính Jaccard similarity đơn giản
intersection = query_words & chunk_words
union = query_words | chunk_words
score = len(intersection) / len(union) if union else 0
scored_chunks.append((score, chunk))
# Sắp xếp theo điểm và lấy top_k
scored_chunks.sort(key=lambda x: x[0], reverse=True)
return [chunk for _, chunk in scored_chunks[:top_k]]
def generate_with_rag(self, query: str, document: str,
use_cache: bool = True) -> Dict[str, Any]:
"""
Generation với RAG - xử lý document dài bằng chunking
"""
doc_hash = hashlib.md5(document.encode()).hexdigest()
# Kiểm tra cache
if use_cache and doc_hash in self.document_cache:
cached_chunks = self.document_cache[doc_hash]
else:
cached_chunks = self.smart_chunk(document)
if use_cache:
self.document_cache[doc_hash] = cached_chunks
# Retrieve relevant chunks
relevant_chunks = self.retrieve_relevant_chunks(cached_chunks, query, top_k=3)
# Build context từ chunks
context = "\n\n---\n\n".join([c['text'] for c in relevant_chunks])
# Build prompt
system_prompt = """Bạn là trợ lý phân tích tài liệu chuyên nghiệp.
Dựa vào ngữ cảnh được cung cấp, hãy trả lời câu hỏi một cách chính xác và chi tiết.
Nếu thông tin không có trong ngữ cảnh, hãy nói rõ điều này."""
user_prompt = f"""NGỮ CẢNH:
{context}
CÂU HỎI: {query}
TRẢ LỜI:"""
# Ước tính chi phí trước
cost_estimate = self.estimate_cost(context + user_prompt)
if not cost_estimate['within_2m_limit']:
return {
'success': False,
'error': 'Document quá lớn cho context window',
'estimated_cost': cost_estimate
}
# Gọi API với retry logic
return self._call_api_with_retry(system_prompt, user_prompt, cost_estimate)
def _call_api_with_retry(self, system_prompt: str, user_prompt: str,
cost_estimate: Dict) -> Dict[str, Any]:
"""Gọi API với retry logic và error handling"""
payload = {
"model": self.config.model,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
"max_tokens": 4096,
"temperature": 0.3
}
for attempt in range(self.config.max_retries):
start_time = time.time()
try:
response = self.client.post("/chat/completions", json=payload)
latency_ms = (time.time() - start_time) * 1000
if response.status_code == 200:
data = response.json()
result = data['choices'][0]['message']['content']
usage = data.get('usage', {})
input_tokens = usage.get('prompt_tokens', 0)
output_tokens = usage.get('completion_tokens', 0)
# Log thành công
cost_tracker.log_request(
input_tokens, output_tokens, latency_ms, success=True
)
return {
'success': True,
'result': result,
'latency_ms': round(latency_ms, 2),
'input_tokens': input_tokens,
'output_tokens': output_tokens,
'cost': cost_estimate['total_cost'],
'chunks_used': len(user_prompt) // self.config.chunk_size + 1
}
elif response.status_code == 429:
# Rate limit - exponential backoff
wait_time = 2 ** attempt * 5
print(f"⚠️ Rate limited. Waiting {wait_time}s before retry...")
time.sleep(wait_time)
elif response.status_code == 401:
return {
'success': False,
'error': 'Invalid API key. Kiểm tra YOUR_HOLYSHEEP_API_KEY'
}
else:
error_msg = f"API Error {response.status_code}: {response.text}"
print(f"❌ {error_msg}")
cost_tracker.log_request(0, 0, 0, success=False, error=error_msg)
except httpx.TimeoutException as e:
error_msg = f"Timeout after {self.config.timeout}s"
print(f"⏱️ {error_msg}")
cost_tracker.log_request(0, 0, self.config.timeout * 1000,
success=False, error=error_msg)
if attempt < self.config.max_retries - 1:
time.sleep(5)
except httpx.ConnectError as e:
error_msg = f"Connection error: {str(e)}"
print(f"🔌 {error_msg}")
cost_tracker.log_request(0, 0, 0, success=False, error=error_msg)
return {'success': False, 'error': error_msg}
return {
'success': False,
'error': f'Failed after {self.config.max_retries} attempts'
}
Ví dụ sử dụng
if __name__ == "__main__":
config = RAGConfig()
rag = GeminiRAGEngine(config)
# Test với sample document
sample_doc = """
Gemini 3.1 Pro là mô hình ngôn ngữ lớn của Google DeepMind với khả năng xử lý
context window lên đến 2 triệu token. Điều này cho phép mô hình hiểu và phân tích
các tài liệu dài một cách toàn diện mà không cần chia nhỏ.
Ứng dụng trong RAG:
1. Xử lý document dài: Báo cáo tài chính, hợp đồng pháp lý
2. Phân tích code base: Hiểu toàn bộ repository
3. Tổng hợp kiến thức: Từ nhiều nguồn cùng lúc
Chi phí: Input $1.25/MToken, Output $5.00/MToken
So với GPT-4.1 tiết kiệm 85% chi phí
"""
query = "Giải thích khả năng context window của Gemini 3.1 Pro"
result = rag.generate_with_rag(query, sample_doc)
if result['success']:
print(f"✅ Thành công!")
print(f"⏱️ Latency: {result['latency_ms']}ms")
print(f"💰 Chi phí: {result['cost']}")
print(f"📊 Tokens: {result['input_tokens']} in / {result['output_tokens']} out")
print(f"\n📝 Kết quả:\n{result['result']}")
else:
print(f"❌ Lỗi: {result['error']}")
Batch Processing Với Cost Optimization
import asyncio
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List, Tuple
class BatchRAGProcessor:
"""
Xử lý batch nhiều documents với cost optimization
Tự động chọn chiến lược tối ưu chi phí
"""
def __init__(self, rag_engine: GeminiRAGEngine):
self.rag = rag_engine
self.batch_results = []
def analyze_documents(self, documents: List[Tuple[str, str]],
max_concurrent: int = 5) -> Dict[str, Any]:
"""
Xử lý batch documents với concurrency limit
Args:
documents: List of (doc_id, content) tuples
max_concurrent: Số request đồng thời tối đa
Returns:
Báo cáo chi phí và kết quả chi tiết
"""
start_time = time.time()
results = []
errors = []
# Pre-check: Ước tính tổng chi phí
total_cost_estimate = 0
oversized_docs = []
for doc_id, content in documents:
cost = self.rag.estimate_cost(content)
total_cost_estimate += float(cost['total_cost'].replace('$', ''))
if not cost['within_2m_limit']:
oversized_docs.append((doc_id, cost))
print(f"📊 Batch Analysis Report")
print(f" Documents: {len(documents)}")
print(f" Estimated total cost: ${total_cost_estimate:.4f}")
print(f" Oversized documents: {len(oversized_docs)}")
if oversized_docs:
print(f"\n⚠️ Documents vượt quá 2M limit:")
for doc_id, cost in oversized_docs[:3]:
print(f" - {doc_id}: {cost['estimated_input_tokens']:,} tokens")
# Process với concurrency control
with ThreadPoolExecutor(max_workers=max_concurrent) as executor:
futures = {
executor.submit(self._process_single_doc, doc_id, content): doc_id
for doc_id, content in documents
}
for future in as_completed(futures):
doc_id = futures[future]
try:
result = future.result()
results.append(result)
if result['success']:
print(f"✅ {doc_id}: ${result['actual_cost']:.4f}")
else:
print(f"❌ {doc_id}: {result['error']}")
errors.append(result)
except Exception as e:
error_result = {
'doc_id': doc_id,
'success': False,
'error': str(e)
}
errors.append(error_result)
print(f"💥 {doc_id}: {str(e)}")
total_time = time.time() - start_time
total_actual_cost = sum(r.get('actual_cost', 0) for r in results if r['success'])
return {
'summary': {
'total_documents': len(documents),
'successful': len([r for r in results if r['success']]),
'failed': len(errors),
'total_time_seconds': round(total_time, 2),
'estimated_cost': f"${total_cost_estimate:.4f}",
'actual_cost': f"${total_actual_cost:.4f}",
'cost_difference': f"${total_cost_estimate - total_cost_actual:.4f}",
'avg_cost_per_doc': f"${total_actual_cost / len(documents):.4f}",
'avg_time_per_doc': f"{total_time / len(documents):.2f}s"
},
'results': results,
'errors': errors,
'cost_tracker_summary': cost_tracker.get_summary()
}
def _process_single_doc(self, doc_id: str, content: str) -> Dict[str, Any]:
"""Xử lý một document đơn lẻ"""
query = "Tóm tắt nội dung chính và các điểm quan trọng"
result = self.rag.generate_with_rag(query, content)
return {
'doc_id': doc_id,
'success': result['success'],
'result': result.get('result', ''),
'latency_ms': result.get('latency_ms', 0),
'input_tokens': result.get('input_tokens', 0),
'output_tokens': result.get('output_tokens', 0),
'actual_cost': float(result.get('cost', '$0').replace('$', '')),
'error': result.get('error')
}
Ví dụ batch processing
if __name__ == "__main__":
config = RAGConfig()
rag = GeminiRAGEngine(config)
batch_processor = BatchRAGProcessor(rag)
# Sample batch documents (trong thực tế sẽ load từ database)
sample_documents = [
("doc_001", "Nội dung tài liệu 1..." * 100),
("doc_002", "Nội dung tài liệu 2..." * 150),
("doc_003", "Nội dung tài liệu 3..." * 200),
("doc_004", "Nội dung tài liệu 4..." * 80),
("doc_005", "Nội dung tài liệu 5..." * 120),
]
print("🚀 Bắt đầu batch processing...")
batch_report = batch_processor.analyze_documents(sample_documents, max_concurrent=3)
print(f"\n{'='*60}")
print("📋 BÁO CÁO TỔNG KẾT")
print(f"{'='*60}")
for key, value in batch_report['summary'].items():
print(f" {key}: {value}")
# Xuất báo cáo chi phí chi tiết
print(f"\n💰 Chi phí chi tiết từ tracker:")
tracker_summary = cost_tracker.get_summary()
for key, value in tracker_summary.items():
print(f" {key}: {value}")
Chiến Lược Tối Ưu Chi Phí RAG
1. Chunking Strategy Tối Ưu
Với Gemini 3.1 Pro 2M, chiến lược chunking ảnh hưởng lớn đến chi phí:
"""
So sánh chi phí giữa các chiến lược chunking khác nhau
"""
CHUNKING_STRATEGIES = {
'naive_10k': {
'description': 'Chia đều 10K tokens/chunk',
'chunk_size': 10000,
'overlap': 500,
'avg_chunks_per_doc_100k': 10
},
'semantic_50k': {
'description': 'Chunk semantic 50K với overlap',
'chunk_size': 50000,
'overlap': 2000,
'avg_chunks_per_doc_100k': 2
},
'full_context': {
'description': 'Đưa toàn bộ vào context (nếu <2M)',
'chunk_size': 2000000,
'overlap': 0,
'avg_chunks_per_doc_100k': 1
}
}
def calculate_strategy_cost(strategy: Dict, doc_size_tokens: int,
requests_per_day: int) -> Dict:
"""Tính chi phí theo chiến lược"""
chunks_needed = max(1, doc_size_tokens // strategy['chunk_size'])
# Chi phí cho 1 document
cost_per_doc = chunks_needed * 0.5 # Ước tính $0.50/chunk average
# Chi phí hàng ngày
daily_cost = cost_per_doc * requests_per_day
monthly_cost = daily_cost * 30
yearly_cost = daily_cost * 365
return {
'chunks_needed': chunks_needed,
'cost_per_doc': f"${cost_per_doc:.4f}",
'daily_cost': f"${daily_cost:.2f}",
'monthly_cost': f"${monthly_cost:.2f}",
'yearly_cost': f"${yearly_cost:.2f}"
}
So sánh chi phí cho document 500K tokens, 100 requests/ngày
doc_size = 500000
daily_requests = 100
print(f"📊 So sánh chi phí cho document {doc_size:,} tokens")
print(f" Tần suất: {daily_requests} requests/ngày\n")
for name, strategy in CHUNKING_STRATEGIES.items():
cost = calculate_strategy_cost(strategy, doc_size, daily_requests)
print(f"🔹 {strategy['description']}")
print(f" Chunks cần thiết: {cost['chunks_needed']}")
print(f" Chi phí/document: {cost['cost_per_doc']}")
print(f" Chi phí tháng: {cost['monthly_cost']}")
print(f" Chi phí năm: {cost['yearly_cost']}\n")
"""
Kết quả mong đợi:
- Naive 10K: ~$150/tháng (10 chunks × $0.50 × 100 × 30)
- Semantic 50K: ~$30/tháng (2 chunks × $0.50 × 100 × 30)
- Full Context: ~$15/tháng (1 chunk × $0.50 × 100 × 30)
→ Semantic 50K là sweet spot giữa chi phí và chất lượng
""
2. Caching Strategy Để Giảm Chi Phí
class SemanticCache:
"""
Cache thông minh dựa trên semantic similarity
Giảm chi phí bằng cách trả lời query trùng lặp từ cache
"""
def __init__(self, similarity_threshold: float = 0.95):
self.cache = {} # query_hash -> response
self.access_count = {}
self.hit_count = 0
self.miss_count = 0
self.similarity_threshold = similarity_threshold
def get_cache_key(self, query: str, context: str) -> str:
"""Tạo cache key từ query và context"""
# Hash kết hợp để đảm bảo uniqueness
combined = f"{query}|{hash(context)}"
return hashlib.sha256(combined.encode()).hexdigest()
def get(self, query: str, context: str) -> Optional[Dict]:
"""Lấy response từ cache nếu có"""
cache_key = self.get_cache_key(query, context)
if cache_key in self.cache:
self.hit_count += 1
self.access_count[cache_key] = self.access_count.get(cache_key, 0) + 1
cached = self.cache[cache_key]
cached['cache_hit'] = True
return cached
self.miss_count += 1
return None
def set(self, query: str, context: str, response: Dict):
"""Lưu response vào cache"""
cache_key = self.get_cache_key(query, context)
self.cache[cache_key] = {
'response': response,
'timestamp': datetime.now().isoformat(),
'access_count': 0
}
def get_stats(self) -> Dict:
"""Thống kê cache performance"""
total = self.hit_count + self.miss_count
hit_rate = self.hit_count / total * 100 if total > 0 else 0
return {
'total_requests': total,
'cache_hits': self.hit_count,
'cache_misses': self.miss_count,
'hit_rate': f"{hit_rate:.2f}%",
'cached_responses': len(self.cache),
'estimated_savings': f"${self.hit_count * 0.5:.2f}" # Ước tính $0.50/response
}
Integration với RAG engine
class CachedRAGEngine(GeminiRAGEngine):
"""RAG Engine với semantic caching"""
def __init__(self, config: RAGConfig, cache: SemanticCache):
super().__init__(config)
self.cache = cache
def generate_with_caching(self, query: str, document: str) -> Dict[str, Any]:
"""Generation với cache lookup trước"""
# 1. Check cache
cached_response = self.cache.get(query, document)
if cached_response:
print(f"🎯 Cache HIT! Bỏ qua API call")
return cached_response
# 2. Cache miss - gọi API
result = self.generate_with_rag(query, document)
# 3. Save to cache
if result['success']:
self.cache.set(query, document, result)
print(f"💾 Saved to cache")
return result
Ví dụ sử dụng caching
if __name__ == "__main__":
cache = SemanticCache(similarity_threshold=0.95)
config = RAGConfig()
cached_rag = CachedRAGEngine(config, cache)
doc = "Nội dung tài liệu mẫu..." * 100
query = "Tóm tắt nội dung chính"
# Request 1 - cache miss
print("Request 1:")
result1 = cached_rag.generate_with_caching(query, doc)
# Request 2 - cache hit
print("\nRequest 2 (cùng query và context):")
result2 = cached_rag.generate_with_caching(query, doc)
# Stats
print(f"\n📊 Cache Statistics:")
for key, value in cache.get_stats().items():
print(f" {key}: {value}")
Lỗi Thường Gặp và Cách Khắc Phục
1. Lỗi Timeout Khi Xử Lý Document Dài
"""
LỖI: ConnectionError: HTTPSConnectionPool Timeout
NGUYÊN NHÂN: Document vượt quá 2M tokens hoặc network timeout quá ngắn
GIẢI PHÁP:
"""
❌ SAI: Timeout quá ngắn cho document lớn
config_bad = {
"timeout": 30.0, # Chỉ 30s - không đủ cho document 500K tokens
"max_retries": 1
}
✅ ĐÚNG: Timeout phù hợp với document size
config_good = {
Tài nguyên liên quan
Bài viết liên quan