Để triển khai RAG Pipeline thành production thực sự, bạn cần hơn là chỉ kết nối vector database với LLM. Bài viết này sẽ hướng dẫn bạn xây dựng hệ thống production-grade với giám sát toàn diện, caching thông minh và chiến lược fallback khi dịch vụ gặp sự cố.
So sánh các giải pháp API
| Tiêu chí | HolySheep AI | API chính thức | Dịch vụ Relay khác |
|---|---|---|---|
| Tỷ giá | ¥1 = $1 (85%+ tiết kiệm) | $1 = $1 | Biến đổi, thường cao hơn |
| Độ trễ P50 | <50ms | 200-500ms | 100-300ms |
| Thanh toán | WeChat/Alipay/Tech/visa | Thẻ quốc tế | Hạn chế |
| Tín dụng miễn phí | Có khi đăng ký | $5 demo | Ít khi có |
| GPT-4.1 | $8/MTok | $60/MTok | $45-55/MTok |
| Claude Sonnet 4.5 | $15/MTok | $30/MTok | $20-25/MTok |
| Gemini 2.5 Flash | $2.50/MTok | $3.50/MTok | $2.80/MTok |
| DeepSeek V3.2 | $0.42/MTok | Không có | $0.60/MTok |
Như bạn thấy, HolySheep AI không chỉ tiết kiệm chi phí mà còn cung cấp độ trễ thấp hơn đáng kể — yếu tố quan trọng cho RAG pipeline production.
Tại sao RAG Pipeline Production cần Giám sát?
Trong môi trường production, RAG pipeline của bạn sẽ đối mặt với nhiều thách thức:
- Độ trễ không nhất quán — Query phức tạp có thể mất 5-10 giây
- Chi phí leo thang — Mỗi retrieval + generation đều tốn token
- Chất lượng phản hồi thất thường — Context window overflow, retrieval noise
- System failure — API downtime, rate limit exceeded
Kiến trúc tổng thể RAG Pipeline Production
Đây là kiến trúc mà tôi đã triển khai cho nhiều dự án, bao gồm cả hệ thống xử lý 10,000+ queries/ngày:
┌─────────────────────────────────────────────────────────────────┐
│ RAG PIPELINE PRODUCTION │
├─────────────────────────────────────────────────────────────────┤
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Query │───▶│ Retrieve │───▶│ Rerank │───▶│ Generate │ │
│ │ Input │ │ (Vector) │ │ (Filter) │ │ (LLM) │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
│ │ │ │ │ │
│ ▼ ▼ ▼ ▼ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ MONITORING LAYER │ │
│ │ Latency | Cost | Quality | Error Rate | Rate Limit │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │ │ │ │ │
│ ▼ ▼ ▼ ▼ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ CACHE LAYER │ │
│ │ Semantic Cache | Exact Match | Result Deduplication │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │ │ │ │ │
│ ▼ ▼ ▼ ▼ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ FALLBACK LAYER │ │
│ │ Circuit Breaker | Model Fallback | Cache Fallback │ │
│ └─────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘Triển khai Code Production
1. Cấu hình HolySheep API Client
import httpx
import asyncio
from typing import List, Dict, Optional, Any
from dataclasses import dataclass, field
from datetime import datetime, timedelta
import hashlib
import json
import logging
from collections import OrderedDict
from enum import Enum
Cấu hình HolySheep - KHÔNG dùng api.openai.com
HOLYSHEEP_CONFIG = {
"base_url": "https://api.holysheep.ai/v1",
"api_key": "YOUR_HOLYSHEEP_API_KEY", # Thay bằng key thực tế
"timeout": 120,
"max_retries": 3,
"default_model": "gpt-4.1"
}
class CircuitState(Enum):
CLOSED = "closed" # Hoạt động bình thường
OPEN = "open" # Đang bị ngắt
HALF_OPEN = "half_open" # Thử nghiệm phục hồi
@dataclass
class MonitoringMetrics:
"""Metrics cho việc giám sát pipeline"""
total_requests: int = 0
successful_requests: int = 0
failed_requests: int = 0
cache_hits: int = 0
cache_misses: int = 0
total_latency_ms: float = 0.0
total_cost: float = 0.0
latency_history: List[float] = field(default_factory=list)
error_counts: Dict[str, int] = field(default_factory=dict)
def record_request(self, latency_ms: float, success: bool,
cached: bool, cost: float, error: Optional[str] = None):
self.total_requests += 1
self.total_latency_ms += latency_ms
self.latency_history.append(latency_ms)
# Chỉ giữ 1000 measurement gần nhất
if len(self.latency_history) > 1000:
self.latency_history = self.latency_history[-1000:]
if success:
self.successful_requests += 1
else:
self.failed_requests += 1
if error:
self.error_counts[error] = self.error_counts.get(error, 0) + 1
if cached:
self.cache_hits += 1
else:
self.cache_misses += 1
self.total_cost += cost
def get_stats(self) -> Dict[str, Any]:
avg_latency = self.total_latency_ms / max(self.total_requests, 1)
sorted_latencies = sorted(self.latency_history)
p50 = sorted_latencies[len(sorted_latencies) // 2] if sorted_latencies else 0
p95 = sorted_latencies[int(len(sorted_latencies) * 0.95)] if sorted_latencies else 0
p99 = sorted_latencies[int(len(sorted_latencies) * 0.99)] if sorted_latencies else 0
cache_total = self.cache_hits + self.cache_misses
cache_hit_rate = self.cache_hits / max(cache_total, 1)
return {
"total_requests": self.total_requests,
"success_rate": self.successful_requests / max(self.total_requests, 1),
"cache_hit_rate": cache_hit_rate,
"avg_latency_ms": avg_latency,
"p50_latency_ms": p50,
"p95_latency_ms": p95,
"p99_latency_ms": p99,
"total_cost_usd": self.total_cost,
"error_counts": self.error_counts
}2. Semantic Cache Implementation
class SemanticCache:
"""
Semantic cache sử dụng embedding similarity
Cache hit nếu cosine similarity > threshold
"""
def __init__(self, max_size: int = 10000, similarity_threshold: float = 0.92):
self.max_size = max_size
self.similarity_threshold = similarity_threshold
self.cache: OrderedDict[str, Dict] = OrderedDict()
self.embeddings: Dict[str, List[float]] = {}
self._stats = {"hits": 0, "misses": 0, "evictions": 0}
def _get_cache_key(self, query: str, top_k: int, filters: Optional[Dict] = None) -> str:
"""Tạo cache key dựa trên query và parameters"""
content = json.dumps({
"query": query.lower().strip(),
"top_k": top_k,
"filters": filters or {}
}, sort_keys=True)
return hashlib.sha256(content.encode()).hexdigest()
def _compute_embedding(self, text: str) -> List[float]:
"""Gọi HolySheep API để lấy embedding"""
client = HolySheepRAGPipeline.http_client
response = client.post(
f"{HOLYSHEEP_CONFIG['base_url']}/embeddings",
json={
"model": "text-embedding-3-small",
"input": text
},
headers={
"Authorization": f"Bearer {HOLYSHEEP_CONFIG['api_key']}",
"Content-Type": "application/json"
}
)
response.raise_for_status()
return response.json()["data"][0]["embedding"]
def _cosine_similarity(self, a: List[float], b: List[float]) -> float:
"""Tính cosine similarity giữa 2 vectors"""
dot_product = sum(x * y for x, y in zip(a, b))
norm_a = sum(x * x for x in a) ** 0.5
norm_b = sum(x * x for x in b) ** 0.5
return dot_product / (norm_a * norm_b + 1e-8)
async def get(self, query: str, top_k: int,
filters: Optional[Dict] = None) -> Optional[Dict]:
"""Kiểm tra cache, trả về cached result nếu tìm thấy"""
cache_key = self._get_cache_key(query, top_k, filters)
# Exact match check
if cache_key in self.cache:
self.cache.move_to_end(cache_key)
self._stats["hits"] += 1
cached = self.cache[cache_key]
cached["hits"] += 1
return cached["result"]
# Semantic similarity check
if self.embeddings:
try:
query_embedding = self._compute_embedding(query)
best_match = None
best_similarity = 0
for cached_key, cached_embedding in self.embeddings.items():
similarity = self._cosine_similarity(query_embedding, cached_embedding)
if similarity > best_similarity:
best_similarity = similarity
best_match = cached_key
if best_match and best_similarity >= self.similarity_threshold:
self.cache.move_to_end(best_match)
self._stats["hits"] += 1
cached = self.cache[best_match]
cached["hits"] += 1
cached["similarity"] = best_similarity
return cached["result"]
except Exception as e:
logging.warning(f"Semantic cache lookup failed: {e}")
self._stats["misses"] += 1
return None
def set(self, query: str, top_k: int, result: Dict,
filters: Optional[Dict] = None):
"""Lưu result vào cache"""
cache_key = self._get_cache_key(query, top_k, filters)
# Evict oldest entries if cache is full
while len(self.cache) >= self.max_size:
evicted_key, _ = self.cache.popitem(last=False)
self.embeddings.pop(evicted_key, None)
self._stats["evictions"] += 1
try:
query_embedding = self._compute_embedding(query)
self.embeddings[cache_key] = query_embedding
except Exception:
pass
self.cache[cache_key] = {
"result": result,
"created_at": datetime.now(),
"hits": 0,
"similarity": 1.0
}
def get_stats(self) -> Dict:
total = self._stats["hits"] + self._stats["misses"]
return {
**self._stats,
"size": len(self.cache),
"hit_rate": self._stats["hits"] / max(total, 1)
}
def clear_expired(self, max_age_hours: int = 24):
"""Xóa các cache entries cũ"""
now = datetime.now()
expired_keys = [
key for key, value in self.cache.items()
if (now - value["created_at"]).total_seconds() / 3600 > max_age_hours
]
for key in expired_keys:
del self.cache[key]
self.embeddings.pop(key, None)3. Circuit Breaker và Fallback Strategy
class CircuitBreaker:
"""
Circuit Breaker pattern để ngăn chặn cascading failures
"""
def __init__(
self,
failure_threshold: int = 5,
recovery_timeout: int = 60,
half_open_max_calls: int = 3
):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.half_open_max_calls = half_open_max_calls
self.state = CircuitState.CLOSED
self.failure_count = 0
self.last_failure_time: Optional[datetime] = None
self.half_open_calls = 0
def record_success(self):
"""Ghi nhận request thành công"""
if self.state == CircuitState.HALF_OPEN:
self.half_open_calls += 1
if self.half_open_calls >= self.half_open_max_calls:
self._transition_to(CircuitState.CLOSED)
elif self.state == CircuitState.CLOSED:
self.failure_count = 0
def record_failure(self):
"""Ghi nhận request thất bại"""
self.failure_count += 1
self.last_failure_time = datetime.now()
if self.state == CircuitState.HALF_OPEN:
self._transition_to(CircuitState.OPEN)
elif (self.failure_count >= self.failure_threshold and
self.state == CircuitState.CLOSED):
self._transition_to(CircuitState.OPEN)
def can_attempt(self) -> bool:
"""Kiểm tra xem có nên thử request không"""
if self.state == CircuitState.CLOSED:
return True
if self.state == CircuitState.OPEN:
if self.last_failure_time:
elapsed = (datetime.now() - self.last_failure_time).total_seconds()
if elapsed >= self.recovery_timeout:
self._transition_to(CircuitState.HALF_OPEN)
return True
return False
if self.state == CircuitState.HALF_OPEN:
return self.half_open_calls < self.half_open_max_calls
return False
def _transition_to(self, new_state: CircuitState):
logging.info(f"Circuit breaker: {self.state.value} -> {new_state.value}")
self.state = new_state
self.failure_count = 0
self.half_open_calls = 0
class ModelFallbackChain:
"""
Chain of models với fallback priority
Thử lần lượt từ model chính đến các model backup
"""
def __init__(self):
# Ưu tiên: Model rẻ nhất -> đắt nhất
self.fallback_chain = [
{"model": "deepseek-v3.2", "max_cost": 0.42, "latency_priority": "low"},
{"model": "gemini-2.5-flash", "max_cost": 2.50, "latency_priority": "high"},
{"model": "gpt-4.1", "max_cost": 8.0, "latency_priority": "medium"},
{"model": "claude-sonnet-4.5", "max_cost": 15.0, "latency_priority": "medium"},
]
self.circuit_breakers: Dict[str, CircuitBreaker] = {
model["model"]: CircuitBreaker()
for model in self.fallback_chain
}
async def call_with_fallback(
self,
messages: List[Dict],
system_prompt: Optional[str] = None,
force_model: Optional[str] = None
) -> Dict[str, Any]:
"""
Gọi model với fallback chain
Trả về response cùng với metadata về model nào được sử dụng
"""
models_to_try = []
if force_model:
models_to_try = [m for m in self.fallback_chain if m["model"] == force_model]
else:
models_to_try = self.fallback_chain
last_error = None
for model_config in models_to_try:
model_name = model_config["model"]
circuit = self.circuit_breakers[model_name]
if not circuit.can_attempt():
logging.info(f"Skipping {model_name} - circuit breaker is {circuit.state.value}")
continue
try:
result = await self._call_model(model_name, messages, system_prompt)
circuit.record_success()
return {
"success": True,
"model": model_name,
"response": result["content"],
"usage": result.get("usage", {}),
"latency_ms": result.get("latency_ms", 0),
"fallback_attempts": len(models_to_try) - len([m for m in models_to_try if m == model_config])
}
except Exception as e:
circuit.record_failure()
last_error = e
logging.warning(f"Model {model_name} failed: {str(e)}")
continue
# Tất cả models đều thất bại
raise RuntimeError(
f"All fallback models exhausted. Last error: {last_error}"
)
async def _call_model(
self,
model: str,
messages: List[Dict],
system_prompt: Optional[str] = None
) -> Dict[str, Any]:
"""Gọi HolySheep API với model cụ thể"""
import time
start_time = time.time()
all_messages = []
if system_prompt:
all_messages.append({"role": "system", "content": system_prompt})
all_messages.extend(messages)
async with httpx.AsyncClient(timeout=HOLYSHEEP_CONFIG["timeout"]) as client:
response = await client.post(
f"{HOLYSHEEP_CONFIG['base_url']}/chat/completions",
json={
"model": model,
"messages": all_messages,
"temperature": 0.7,
"max_tokens": 2048
},
headers={
"Authorization": f"Bearer {HOLYSHEEP_CONFIG['api_key']}",
"Content-Type": "application/json"
}
)
response.raise_for_status()
data = response.json()
latency_ms = (time.time() - start_time) * 1000
return {
"content": data["choices"][0]["message"]["content"],
"usage": data.get("usage", {}),
"latency_ms": latency_ms
}4. Complete RAG Pipeline Class
class HolySheepRAGPipeline:
"""
Production-ready RAG Pipeline với monitoring, caching và fallback
"""
http_client: httpx.AsyncClient = None
def __init__(
self,
vector_store, # ChromaDB, Pinecone, v.v.
embedding_model: str = "text-embedding-3-small",
default_top_k: int = 5,
cache_enabled: bool = True,
fallback_enabled: bool = True
):
self.vector_store = vector_store
self.embedding_model = embedding_model
self.default_top_k = default_top_k
# Monitoring
self.metrics = MonitoringMetrics()
# Caching
self.cache = SemanticCache(max_size=5000) if cache_enabled else None
# Fallback
self.fallback_chain = ModelFallbackChain() if fallback_enabled else None
self.circuit_breaker = CircuitBreaker()
# Logging
self.logger = logging.getLogger(__name__)
async def initialize(self):
"""Khởi tạo HTTP client"""
if HolySheepRAGPipeline.http_client is None:
HolySheepRAGPipeline.http_client = httpx.AsyncClient(
timeout=HOLYSHEEP_CONFIG["timeout"]
)
async def close(self):
"""Đóng HTTP client"""
if HolySheepRAGPipeline.http_client:
await HolySheepRAGPipeline.http_client.aclose()
async def get_embedding(self, text: str) -> List[float]:
"""Lấy embedding từ HolySheep API"""
await self.initialize()
response = await HolySheepRAGPipeline.http_client.post(
f"{HOLYSHEEP_CONFIG['base_url']}/embeddings",
json={
"model": self.embedding_model,
"input": text
},
headers={
"Authorization": f"Bearer {HOLYSHEEP_CONFIG['api_key']}"
}
)
response.raise_for_status()
return response.json()["data"][0]["embedding"]
async def retrieve(
self,
query: str,
top_k: Optional[int] = None,
filters: Optional[Dict] = None,
use_cache: bool = True
) -> List[Dict]:
"""
Retrieve documents từ vector store
Có hỗ trợ cache
"""
top_k = top_k or self.default_top_k
# Check cache first
if use_cache and self.cache:
cached_result = await self.cache.get(query, top_k, filters)
if cached_result:
self.logger.info("Cache HIT for retrieval")
return cached_result["documents"]
# Get query embedding
query_embedding = await self.get_embedding(query)
# Query vector store
results = self.vector_store.query(
query_embeddings=[query_embedding],
n_results=top_k,
where=filters
)
# Format results
documents = []
if results and results.get("documents"):
for i, doc in enumerate(results["documents"][0]):
documents.append({
"content": doc,
"metadata": results.get("metadatas", [[{}]])[0][i],
"distance": results.get("distances", [[0]])[0][i]
})
# Save to cache
if use_cache and self.cache:
await self.cache.set(query, top_k, {"documents": documents}, filters)
return documents
async def generate(
self,
query: str,
context_documents: List[Dict],
system_prompt: Optional[str] = None,
model: Optional[str] = None
) -> Dict[str, Any]:
"""
Generate response từ context
Sử dụng fallback chain nếu enabled
"""
# Build context from documents
context = "\n\n".join([
f"[Document {i+1}]\n{doc['content']}"
for i, doc in enumerate(context_documents)
])
# Build messages
user_message = f"Query: {query}\n\nContext:\n{context}\n\nBased on the context above, please answer the query."
messages = [{"role": "user", "content": user_message}]
# Estimate cost
estimated_tokens = len(context.split()) + len(query.split())
estimated_cost = estimated_tokens / 1_000_000 * 8.0 # GPT-4.1 rate
# Call with fallback
if self.fallback_chain:
result = await self.fallback_chain.call_with_fallback(
messages=messages,
system_prompt=system_prompt,
force_model=model
)
# Record metrics
actual_cost = (result["usage"].get("total_tokens", 0) / 1_000_000 *
next((m["max_cost"] for m in self.fallback_chain.fallback_chain
if m["model"] == result["model"]), 8.0))
self.metrics.record_request(
latency_ms=result["latency_ms"],
success=True,
cached=False,
cost=actual_cost
)
return {
"response": result["response"],
"model_used": result["model"],
"latency_ms": result["latency_ms"],
"usage": result["usage"],
"cost_usd": actual_cost
}
else:
# Direct call without fallback
return await self._direct_generate(messages, system_prompt, model)
async def query(
self,
query: str,
top_k: Optional[int] = None,
filters: Optional[Dict] = None,
system_prompt: Optional[str] = None,
model: Optional[str] = None,
use_cache: bool = True
) -> Dict[str, Any]:
"""
Full RAG pipeline: retrieve + generate
"""
import time
start_time = time.time()
try:
# Retrieve
documents = await self.retrieve(
query=query,
top_k=top_k,
filters=filters,
use_cache=use_cache
)
if not documents:
return {
"response": "Không tìm thấy thông tin liên quan trong cơ sở dữ liệu.",
"sources": [],
"latency_ms": (time.time() - start_time) * 1000,
"cache_hit": False
}
# Generate
result = await self.generate(
query=query,
context_documents=documents,
system_prompt=system_prompt,
model=model
)
return {
"response": result["response"],
"sources": documents,
"model_used": result.get("model_used"),
"latency_ms": (time.time() - start_time) * 1000,
"total_cost_usd": result.get("cost_usd", 0),
"usage": result.get("usage", {})
}
except Exception as e:
total_latency = (time.time() - start_time) * 1000
self.metrics.record_request(
latency_ms=total_latency,
success=False,
cached=False,
cost=0,
error=str(e)
)
raise
def get_metrics(self) -> Dict[str, Any]:
"""Lấy metrics hiện tại"""
metrics = self.metrics.get_stats()
if self.cache:
metrics["cache"] = self.cache.get_stats()
if self.fallback_chain:
metrics["circuit_breakers"] = {
model: cb.state.value
for model, cb in self.fallback_chain.circuit_breakers.items()
}
return metrics
async def health_check(self) -> Dict[str, Any]:
"""Kiểm tra health của pipeline"""
try:
# Test embedding endpoint
await self.get_embedding("health check")
# Check circuit breakers
circuit_states = {}
if self.fallback_chain:
for model, cb in self.fallback_chain.circuit_breakers.items():
circuit_states[model] = {
"state": cb.state.value,
"failure_count": cb.failure_count
}
return {
"status": "healthy",
"circuit_breakers": circuit_states,
"metrics": self.get_metrics()
}
except Exception as e:
return {
"status": "unhealthy",
"error": str(e)
}Sử dụng Pipeline trong Production
# ==================== USAGE EXAMPLE ====================
import asyncio
from chromadb import Client as ChromaClient
async def main():
# Khởi tạo vector store (ví dụ ChromaDB)
chroma_client = ChromaClient()
collection = chroma_client.get_collection("my_documents")
# Khởi tạo RAG Pipeline
pipeline = HolySheepRAGPipeline(
vector_store=collection,
embedding_model="text-embedding-3-small",
default_top_k=5,
cache_enabled=True,
fallback_enabled=True
)
try:
# Query đầu tiên - sẽ cache kết quả
result1 = await pipeline.query(
query="Cách đăng ký tài khoản HolySheep AI?",
top_k=3,
system_prompt="Bạn là trợ lý AI hỗ trợ người dùng."
)
print(f"Response: {result1['response']}")
print(f"Latency: {result1['latency_ms']:.2f}ms")
print(f"Model: {result1['model_used']}")
print(f"Sources: {len(result1['sources'])} documents")
# Query thứ 2 - cùng query (cache hit)
result2 = await pipeline.query(
query="Cách đăng ký tài khoản HolySheep AI?",
top_k=3
)
print(f"\nCache hit! Latency: {result2['latency_ms']:.2f}ms")
# Lấy metrics
metrics = pipeline.get_metrics()
print(f"\n=== Pipeline Metrics ===")
print(f"Total requests: {metrics['total_requests']}")
print(f"Cache hit rate: {metrics['cache_hit_rate']:.2%}")
print(f"P95 latency: {metrics['p95_latency_ms']:.2f}ms")
print(f"Total cost: ${metrics['total_cost_usd']:.4f}")
# Health check
health = await pipeline.health_check()
print(f"\n=== Health Status ===")
print(f"Status: {health['status']}")
finally:
await pipeline.close()
Chạy
asyncio.run(main())Lỗi thường gặp và cách khắc phục
1. Lỗi 429 - Rate Limit Exceeded
# Triệu chứng: API trả về HTTP 429 Too Many Requests
Nguyên nhân: Vượt quá rate limit của HolySheep
Cách khắc phục - Implement retry with exponential backoff
import asyncio
import random
class RateLimitHandler:
def __init__(self, max_retries: int = 5, base_delay: float = 1.0):
self.max_retries = max_retries
self.base_delay = base_delay
async def execute_with_retry(self, func, *args, **kwargs):
for attempt in range(self.max_retries):
try:
return await func(*args, **kwargs)
except httpx.HTTPStatusError as e:
if e.response.status_code == 429:
# Exponential backoff với jitter
delay = self.base_delay * (2 ** attempt) + random.uniform(0, 1)
wait_time = min(delay, 60) # Max 60 giây
logging.warning(
f"Rate limit hit. Attempt {attempt + 1}/{self.max_retries}. "
f"Waiting {wait_time:.2f}s"
)
await asyncio.sleep(wait_time)
else:
raise
except Exception as e:
raise
raise RuntimeError(f"Failed after {self.max_retries} retries")2. Lỗi Context Overflow - Quá dài
# Triệu chứng: LLM trả về response bị cắt hoặc lỗi context_length
Nguyên nhân: Context vượt quá max tokens của model
Cách khắc phục - Intelligent context truncation
async def smart_truncate_context(
documents: List[Dict],
query: str,
max_tokens: int = 6000,
model: str = "gpt-4.1"
) -> List[Dict]:
"""
Thông minh truncate context:
1. Ưu tiên documents có relevance cao hơn
2. Giữ metadata quan trọng
3. Cắt từ documents ít relevant nhất
"""
# Token limits theo model
model_limits = {
"gpt-4.1": 128000,
"gpt-4o": 128000,
"gpt-3.5-turbo": 16385,
"deepseek-v3.2": 64000,
"gemini-2.5-flash": 1000000,
}
max_context = min(max_tokens, model_limits.get(model, 6000))
# Sort by relevance (distance)
sorted_docs = sorted(documents, key=lambda x: x.get("distance