Mở đầu: Khi hệ thống RAG của tôi "chết lâm sàng" vào đêm launch
Tôi vẫn nhớ rõ cái đêm tháng 3 năm 2025 — ngày ra mắt hệ thống RAG cho nền tảng thương mại điện tử của một startup Việt Nam. 23:47, ngay trước giờ peak traffic, toàn bộ API trả về 500 error. Đó là khoảnh khắc tôi hiểu rằng: debug AI không chỉ là đọc log — mà là nghệ thuật đọc "triệu chứng" của một hệ thống đang chết dần.
Bài viết này chia sẻ 2 năm kinh nghiệm debug hệ thống AI tích hợp, từ lỗi timeout nhỏ nhất đến crash toàn bộ pipeline. Tất cả code mẫu sử dụng
HolySheep AI — nền tảng tôi tin dùng với độ trễ dưới 50ms và chi phí chỉ từ $0.42/MTok với DeepSeek V3.2.
Tại sao Windsurf AI Debugging lại khó?
Windsurf AI là công cụ AI coding assistant mạnh mẽ, nhưng khi tích hợp vào production, bạn sẽ gặp những thách thức đặc thù:
- Context window overflow khi prompt quá dài
- Token limit exceeded không mong đợi
- Rate limiting từ phía API provider
- Streaming response bị interrupt
- Semantic search trả về kết quả không liên quan
Kiến trúc Debug Pipeline hoàn chỉnh
Dưới đây là architecture tôi xây dựng sau hơn 50 dự án tích hợp AI:
┌─────────────────────────────────────────────────────────────┐
│ DEBUG PIPELINE ARCHITECTURE │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌───────┐ │
│ │ Request │───▶│ Validate │───▶│ Route │───▶│Model │ │
│ │ Layer │ │ & Count │ │ Check │ │Inference│
│ └──────────┘ └──────────┘ └──────────┘ └───────┘ │
│ │ │ │ │ │
│ ▼ ▼ ▼ ▼ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ CENTRALIZED LOGGING │ │
│ │ • Request ID • Token Count • Latency │ │
│ │ • Error Type • Retry Count • Cost │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ │
│ │ Alert System │ │
│ │ Slack/Email │ │
│ └──────────────┘ │
└─────────────────────────────────────────────────────────────┘
Code Implementation: Production-Ready Debug System
1. Core Debug Client với Error Tracking
import requests
import time
import json
from datetime import datetime
from typing import Optional, Dict, Any
from dataclasses import dataclass, field
from enum import Enum
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ErrorType(Enum):
TIMEOUT = "timeout_error"
RATE_LIMIT = "rate_limit_error"
TOKEN_EXCEEDED = "token_exceeded_error"
CONTEXT_OVERFLOW = "context_overflow_error"
AUTH_FAILED = "authentication_error"
NETWORK_ERROR = "network_error"
UNKNOWN = "unknown_error"
@dataclass
class RequestMetrics:
request_id: str
timestamp: datetime
model: str
prompt_tokens: int = 0
completion_tokens: int = 0
total_tokens: int = 0
latency_ms: float = 0.0
cost_usd: float = 0.0
error_type: Optional[ErrorType] = None
error_message: Optional[str] = None
retry_count: int = 0
success: bool = False
class HolySheepDebugClient:
"""
Production-grade client với comprehensive debugging
Đăng ký: https://www.holysheep.ai/register
"""
PRICING = {
"gpt-4.1": {"input": 8.0, "output": 8.0}, # $/MTok
"claude-sonnet-4.5": {"input": 15.0, "output": 15.0},
"gemini-2.5-flash": {"input": 2.50, "output": 2.50},
"deepseek-v3.2": {"input": 0.42, "output": 0.42} # Tiết kiệm 85%+
}
def __init__(
self,
api_key: str,
base_url: str = "https://api.holysheep.ai/v1",
max_retries: int = 3,
timeout: int = 30
):
self.api_key = api_key
self.base_url = base_url.rstrip('/')
self.max_retries = max_retries
self.timeout = timeout
self.metrics: list[RequestMetrics] = []
self._session = requests.Session()
self._session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
})
def _calculate_cost(
self,
model: str,
prompt_tokens: int,
completion_tokens: int
) -> float:
"""Tính chi phí theo thời gian thực"""
pricing = self.PRICING.get(model, self.PRICING["deepseek-v3.2"])
input_cost = (prompt_tokens / 1_000_000) * pricing["input"]
output_cost = (completion_tokens / 1_000_000) * pricing["output"]
return round(input_cost + output_cost, 6)
def _classify_error(self, status_code: int, error_body: dict) -> ErrorType:
"""Tự động phân loại lỗi"""
if status_code == 429:
return ErrorType.RATE_LIMIT
elif status_code == 401 or status_code == 403:
return ErrorType.AUTH_FAILED
elif "token" in str(error_body).lower() and "limit" in str(error_body).lower():
return ErrorType.TOKEN_EXCEEDED
elif "context" in str(error_body).lower():
return ErrorType.CONTEXT_OVERFLOW
elif status_code >= 500:
return ErrorType.UNKNOWN
return ErrorType.UNKNOWN
def _make_request(
self,
model: str,
messages: list,
temperature: float = 0.7,
max_tokens: int = 2048,
request_id: Optional[str] = None
) -> tuple[dict, RequestMetrics]:
"""Internal request handler với retry logic"""
import uuid
request_id = request_id or str(uuid.uuid4())[:8]
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
metrics = RequestMetrics(
request_id=request_id,
timestamp=datetime.now(),
model=model
)
url = f"{self.base_url}/chat/completions"
for retry in range(self.max_retries):
start_time = time.perf_counter()
try:
response = self._session.post(
url,
json=payload,
timeout=self.timeout
)
metrics.latency_ms = (time.perf_counter() - start_time) * 1000
if response.status_code == 200:
data = response.json()
usage = data.get("usage", {})
metrics.prompt_tokens = usage.get("prompt_tokens", 0)
metrics.completion_tokens = usage.get("completion_tokens", 0)
metrics.total_tokens = usage.get("total_tokens", 0)
metrics.cost_usd = self._calculate_cost(
model,
metrics.prompt_tokens,
metrics.completion_tokens
)
metrics.success = True
logger.info(
f"[{request_id}] ✓ Success | "
f"Tokens: {metrics.total_tokens} | "
f"Latency: {metrics.latency_ms:.2f}ms | "
f"Cost: ${metrics.cost_usd:.6f}"
)
self.metrics.append(metrics)
return data, metrics
else:
error_body = response.json() if response.content else {}
metrics.error_type = self._classify_error(
response.status_code,
error_body
)
metrics.error_message = str(error_body)
metrics.retry_count = retry + 1
logger.warning(
f"[{request_id}] ⚠ Error {response.status_code}: {error_body} | "
f"Retry {retry + 1}/{self.max_retries}"
)
if response.status_code == 429:
import random
wait_time = (2 ** retry) + random.uniform(0, 1)
logger.info(f"[{request_id}] Waiting {wait_time:.2f}s before retry...")
time.sleep(wait_time)
continue
if response.status_code in [401, 403]:
metrics.success = False
self.metrics.append(metrics)
return {}, metrics
except requests.exceptions.Timeout:
metrics.error_type = ErrorType.TIMEOUT
metrics.error_message = f"Request timeout after {self.timeout}s"
metrics.retry_count = retry + 1
logger.error(f"[{request_id}] ⏱ Timeout | Retry {retry + 1}/{self.max_retries}")
except requests.exceptions.ConnectionError as e:
metrics.error_type = ErrorType.NETWORK_ERROR
metrics.error_message = str(e)
metrics.retry_count = retry + 1
logger.error(f"[{request_id}] 🌐 Network Error: {e}")
except Exception as e:
metrics.error_type = ErrorType.UNKNOWN
metrics.error_message = str(e)
logger.exception(f"[{request_id}] ❌ Unexpected error")
metrics.success = False
self.metrics.append(metrics)
return {}, metrics
def chat(
self,
prompt: str,
model: str = "deepseek-v3.2",
system_prompt: Optional[str] = None,
context: Optional[list] = None
) -> tuple[str, RequestMetrics]:
"""Main chat method với automatic error recovery"""
messages = []
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
if context:
messages.extend(context)
messages.append({"role": "user", "content": prompt})
data, metrics = self._make_request(model, messages)
if metrics.success and data:
return data["choices"][0]["message"]["content"], metrics
# Fallback to smaller model if primary fails
if not metrics.success and model != "deepseek-v3.2":
logger.info(f"[{metrics.request_id}] Falling back to deepseek-v3.2...")
fallback_data, fallback_metrics = self._make_request(
"deepseek-v3.2",
messages,
request_id=f"{metrics.request_id}-fallback"
)
if fallback_metrics.success:
return fallback_data["choices"][0]["message"]["content"], fallback_metrics
raise RuntimeError(
f"Request failed after {metrics.retry_count} retries: {metrics.error_message}"
)
def get_statistics(self) -> Dict[str, Any]:
"""Tổng hợp statistics cho debugging"""
if not self.metrics:
return {"message": "No requests recorded"}
total_requests = len(self.metrics)
successful = sum(1 for m in self.metrics if m.success)
failed = total_requests - successful
error_counts = {}
for metric in self.metrics:
if metric.error_type:
error_counts[metric.error_type.value] = \
error_counts.get(metric.error_type.value, 0) + 1
return {
"total_requests": total_requests,
"successful": successful,
"failed": failed,
"success_rate": f"{(successful/total_requests)*100:.1f}%",
"total_cost_usd": sum(m.cost_usd for m in self.metrics),
"avg_latency_ms": sum(m.latency_ms for m in self.metrics) / total_requests,
"total_tokens": sum(m.total_tokens for m in self.metrics),
"error_breakdown": error_counts
}
============ USAGE EXAMPLE ============
if __name__ == "__main__":
client = HolySheepDebugClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
timeout=30,
max_retries=3
)
# Test với production prompt
response, metrics = client.chat(
prompt="Debug: API returns 500 on /api/search endpoint",
model="deepseek-v3.2",
system_prompt="You are an expert debugging assistant."
)
print(f"Response: {response[:200]}...")
print(f"Latency: {metrics.latency_ms:.2f}ms")
print(f"Cost: ${metrics.cost_usd:.6f}")
print(f"Stats: {client.get_statistics()}")
2. Windsurf AI RAG Debugging System
import numpy as np
from typing import List, Dict, Tuple, Optional
from dataclasses import dataclass
import hashlib
import logging
logger = logging.getLogger(__name__)
@dataclass
class RetrievalResult:
chunk_id: str
content: str
score: float
metadata: dict
latency_ms: float
@dataclass
class DebugReport:
query: str
total_chunks: int
retrieved_chunks: int
avg_score: float
min_score: float
score_threshold: float
low_relevance_chunks: List[RetrievalResult]
context_window_usage: float # percentage
suggestions: List[str]
class WindsurfRAGDebugger:
"""
Specialized debugger cho RAG systems
Phát hiện và fix common RAG issues
"""
def __init__(
self,
embedding_client,
llm_client: HolySheepDebugClient,
default_threshold: float = 0.7,
max_context_chunks: int = 10
):
self.embedding = embedding_client
self.llm = llm_client
self.default_threshold = default_threshold
self.max_context_chunks = max_context_chunks
self._debug_logs: List[DebugReport] = []
def diagnose_retrieval(
self,
query: str,
chunks: List[Dict],
embeddings: np.ndarray,
top_k: int = 5
) -> DebugReport:
"""
Phân tích retrieval quality và đưa ra recommendations
"""
# Tính query embedding
query_embedding = self.embedding.encode([query])
# Tính cosine similarity
similarities = np.dot(embeddings, query_embedding.T).flatten()
# Sort và lấy top_k
top_indices = np.argsort(similarities)[::-1][:top_k]
retrieval_results = []
for idx in top_indices:
result = RetrievalResult(
chunk_id=chunks[idx].get("id", hashlib.md5(
chunks[idx]["content"].encode()).hexdigest()[:8]
),
content=chunks[idx]["content"],
score=float(similarities[idx]),
metadata=chunks[idx].get("metadata", {}),
latency_ms=0.0
)
retrieval_results.append(result)
# Phân tích
avg_score = np.mean([r.score for r in retrieval_results])
min_score = min([r.score for r in retrieval_results])
# Phát hiện low-relevance chunks
low_relevance = [r for r in retrieval_results if r.score < self.default_threshold]
# Estimate context usage
total_tokens = sum(
len(chunk["content"].split()) * 1.3 # rough token estimate
for chunk in chunks[:top_k]
)
context_usage = (total_tokens / 128000) * 100 # Assuming 128k context
# Generate suggestions
suggestions = self._generate_fix_suggestions(
avg_score, min_score, low_relevance, context_usage
)
report = DebugReport(
query=query,
total_chunks=len(chunks),
retrieved_chunks=len(retrieval_results),
avg_score=avg_score,
min_score=min_score,
score_threshold=self.default_threshold,
low_relevance_chunks=low_relevance,
context_window_usage=context_usage,
suggestions=suggestions
)
self._debug_logs.append(report)
self._log_diagnosis(report)
return report
def _generate_fix_suggestions(
self,
avg_score: float,
min_score: float,
low_relevance: List[RetrievalResult],
context_usage: float
) -> List[str]:
"""Sinh suggestions dựa trên diagnosis"""
suggestions = []
if avg_score < 0.5:
suggestions.append(
"⚠️ LOW AVG SCORE: Cân nhắc sử dụng semantic search engine khác "
"hoặc fine-tune embedding model cho domain-specific vocabulary"
)
if min_score < 0.3:
suggestions.append(
f"🔍 CRITICAL: {len(low_relevance)} chunks có relevance < 0.3. "
"Kiểm tra lại chunking strategy và data quality"
)
if context_usage > 80:
suggestions.append(
"📊 HIGH CONTEXT: Context window sử dụng >80%. "
"Cân nhắc giảm top_k hoặc summarize chunks trước"
)
if len(low_relevance) > 2:
suggestions.append(
"🔄 MULTI-STEP: Thử hybrid search (dense + sparse) "
"hoặc reranking với cross-encoder"
)
if avg_score >= 0.7 and min_score >= 0.5:
suggestions.append(
"✅ RETRIEVAL QUALITY: Tốt, proceed với generation"
)
return suggestions if suggestions else ["No specific issues detected"]
def _log_diagnosis(self, report: DebugReport):
"""Log diagnosis results"""
logger.info("=" * 60)
logger.info(f"RAG DIAGNOSIS: {report.query[:50]}...")
logger.info(f"Avg Score: {report.avg_score:.3f} | Min Score: {report.min_score:.3f}")
logger.info(f"Context Usage: {report.context_window_usage:.1f}%")
logger.info(f"Low Relevance Chunks: {len(report.low_relevance_chunks)}")
if report.suggestions:
logger.info("Suggestions:")
for s in report.suggestions:
logger.info(f" • {s}")
logger.info("=" * 60)
def auto_fix_and_retry(
self,
query: str,
chunks: List[Dict],
embeddings: np.ndarray,
max_retries: int = 3
) -> Tuple[Optional[str], DebugReport]:
"""
Auto-fix pipeline: diagnose -> apply fixes -> retry
"""
best_result = None
best_report = None
strategies = [
{"top_k": 5, "threshold": 0.7},
{"top_k": 3, "threshold": 0.6},
{"top_k": 7, "threshold": 0.5},
]
for i, strategy in enumerate(strategies[:max_retries]):
logger.info(f"Attempt {i+1}: Trying strategy {strategy}")
# Get retrieval results
results = []
for idx, chunk in enumerate(chunks[:strategy["top_k"]]):
score = float(np.dot(
embeddings[idx:idx+1],
self.embedding.encode([query]).T
).flatten()[0])
if score >= strategy["threshold"]:
results.append((chunk, score))
if not results:
logger.warning(f"Attempt {i+1}: No results above threshold")
continue
# Build context
context = "\n\n".join([
f"[Source {i+1}]: {chunk['content']}"
for chunk, score in sorted(results, key=lambda x: -x[1])
])
# Try generation
try:
prompt = f"""Based on the following context, answer the query.
Context:
{context}
Query: {query}
Answer:"""
response, metrics = self.llm.chat(
prompt=prompt,
model="deepseek-v3.2",
system_prompt="You are a helpful assistant. Answer based ONLY on the provided context."
)
# Validate response quality
if metrics.success and len(response) > 50:
best_result = response
best_report = DebugReport(
query=query,
total_chunks=len(chunks),
retrieved_chunks=len(results),
avg_score=sum(s for _, s in results) / len(results),
min_score=min(s for _, s in results),
score_threshold=strategy["threshold"],
low_relevance_chunks=[],
context_window_usage=0,
suggestions=[f"✓ Success với strategy: {strategy}"]
)
logger.info(f"✓ Attempt {i+1} successful!")
break
except Exception as e:
logger.error(f"Attempt {i+1} failed: {e}")
continue
if best_report:
self._debug_logs.append(best_report)
return best_result, best_report
============ INTEGRATION WITH HOLYSHEEP ============
Khởi tạo clients
holy_sheep = HolySheepDebugClient(
api_key="YOUR_HOLYSHEEP_API_KEY"
)
Mock embedding (thay bằng actual embedding model)
class MockEmbedding:
def encode(self, texts):
return np.random.randn(len(texts), 384)
rag_debugger = WindsurfRAGDebugger(
embedding_client=MockEmbedding(),
llm_client=holy_sheep,
default_threshold=0.7
)
Sample chunks
sample_chunks = [
{"id": "c1", "content": "Python list comprehension syntax: [x for x in items]", "metadata": {"file": "basics.py"}},
{"id": "c2", "content": "Debugging with pdb: set_trace(), step, next, continue", "metadata": {"file": "debug.py"}},
{"id": "c3", "content": "Error handling: try-except-finally blocks", "metadata": {"file": "errors.py"}},
]
Run diagnosis
report = rag_debugger.diagnose_retrieval(
query="How to debug Python code?",
chunks=sample_chunks,
embeddings=np.random.rand(len(sample_chunks), 384),
top_k=3
)
print(f"Report: avg_score={report.avg_score:.3f}, suggestions={report.suggestions}")
3. Real-time Monitoring Dashboard
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import List, Optional
import uvicorn
from datetime import datetime, timedelta
import asyncio
app = FastAPI(title="Windsurf AI Debug Monitor")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
Global state
class MonitorState:
def __init__(self):
self.active_requests = {}
self.request_history = []
self.alert_thresholds = {
"latency_ms": 2000,
"error_rate_percent": 10,
"cost_per_hour_usd": 100
}
self.alerts = []
def add_request(self, request_id: str, metadata: dict):
self.active_requests[request_id] = {
**metadata,
"started_at": datetime.now(),
"status": "running"
}
def complete_request(self, request_id: str, success: bool, metrics: dict):
if request_id in self.active_requests:
self.active_requests[request_id]["status"] = "completed" if success else "failed"
self.active_requests[request_id]["completed_at"] = datetime.now()
self.active_requests[request_id].update(metrics)
# Move to history (keep last 1000)
self.request_history.append(self.active_requests[request_id])
if len(self.request_history) > 1000:
self.request_history = self.request_history[-1000:]
def get_health_status(self) -> dict:
if not self.request_history:
return {"status": "healthy", "message": "No requests yet"}
recent = [r for r in self.request_history
if r.get("started_at", datetime.min) > datetime.now() - timedelta(minutes=5)]
if not recent:
return {"status": "idle", "message": "No recent activity"}
error_count = sum(1 for r in recent if r.get("status") == "failed")
error_rate = (error_count / len(recent)) * 100
avg_latency = sum(r.get("latency_ms", 0) for r in recent) / len(recent)
status = "healthy"
if error_rate > self.alert_thresholds["error_rate_percent"]:
status = "degraded"
if error_rate > 50:
status = "critical"
return {
"status": status,
"requests_5min": len(recent),
"error_rate_percent": round(error_rate, 2),
"avg_latency_ms": round(avg_latency, 2),
"active_requests": len([r for r in self.active_requests.values()
if r["status"] == "running"])
}
state = MonitorState()
class ChatRequest(BaseModel):
prompt: str
model: str = "deepseek-v3.2"
temperature: float = 0.7
max_tokens: int = 2048
class ChatResponse(BaseModel):
request_id: str
response: str
metrics: dict
@app.post("/api/chat", response_model=ChatResponse)
async def chat_endpoint(request: ChatRequest):
import uuid
request_id = str(uuid.uuid4())[:8]
# Track request
state.add_request(request_id, {
"model": request.model,
"prompt_length": len(request.prompt)
})
try:
# Call HolySheep AI
client = HolySheepDebugClient(api_key="YOUR_HOLYSHEEP_API_KEY")
response, metrics = client.chat(
prompt=request.prompt,
model=request.model,
temperature=request.temperature,
max_tokens=request.max_tokens
)
state.complete_request(request_id, True, {
"latency_ms": metrics.latency_ms,
"total_tokens": metrics.total_tokens,
"cost_usd": metrics.cost_usd
})
return ChatResponse(
request_id=request_id,
response=response,
metrics={
"latency_ms": metrics.latency_ms,
"tokens": metrics.total_tokens,
"cost_usd": metrics.cost_usd
}
)
except Exception as e:
state.complete_request(request_id, False, {"error": str(e)})
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/health")
async def health_check():
return state.get_health_status()
@app.get("/api/stats")
async def get_stats():
"""Real-time statistics"""
if not state.request_history:
return {"message": "No data available"}
last_hour = [r for r in state.request_history
if r.get("started_at", datetime.min) > datetime.now() - timedelta(hours=1)]
total_cost = sum(r.get("cost_usd", 0) for r in last_hour)
total_tokens = sum(r.get("total_tokens", 0) for r in last_hour)
return {
"requests_last_hour": len(last_hour),
"total_cost_usd": round(total_cost, 4),
"total_tokens": total_tokens,
"avg_latency_ms": round(
sum(r.get("latency_ms", 0) for r in last_hour) / max(len(last_hour), 1),
2
),
"error_count": sum(1 for r in last_hour if r.get("status") == "failed"),
"model_breakdown": _get_model_breakdown(last_hour)
}
def _get_model_breakdown(requests: List[dict]) -> dict:
breakdown = {}
for r in requests:
model = r.get("model", "unknown")
if model not in breakdown:
breakdown[model] = {"count": 0, "cost": 0, "tokens": 0}
breakdown[model]["count"] += 1
breakdown[model]["cost"] += r.get("cost_usd", 0)
breakdown[model]["tokens"] += r.get("total_tokens", 0)
return breakdown
@app.get("/api/alerts")
async def get_alerts():
"""Get active alerts"""
health = state.get_health_status()
alerts = []
if health.get("error_rate_percent", 0) > state.alert_thresholds["error_rate_percent"]:
alerts.append({
"level": "warning",
"message": f"High error rate: {health['error_rate_percent']}%"
})
if health.get("avg_latency_ms", 0) > state.alert_thresholds["latency_ms"]:
alerts.append({
"level": "warning",
"message": f"High latency: {health['avg_latency_ms']}ms"
})
return {"alerts": alerts, "total": len(alerts)}
@app.get("/api/active-requests")
async def get_active_requests():
"""Get all running requests"""
return {
"count": len([r for r in state.active_requests.values() if r["status"] == "running"]),
"requests": [
{
"request_id": rid,
"started_at": r["started_at"].isoformat(),
"model": r.get("model"),
"status": r["status"]
}
for rid, r in state.active_requests.items()
if r["status"] == "running"
]
}
if __name__ == "__main__":
print("🚀 Starting Windsurf AI Debug Monitor")
print(f"📊 API: http://localhost:8000")
print(f"📚 Docs: http://localhost:8000/docs")
uvicorn.run(app, host="0.0.0.0", port=8000)
Lỗi thường gặp và cách khắc phục
1. Lỗi 429 - Rate Limit Exceeded
Mô tả: API trả về lỗi "Too Many Requests" sau khi gọi liên tục.
Nguyên nhân gốc: Vượt quota hoặc request/minute limit của tài khoản.
Giải pháp:
Implement exponential backoff với jitter
import time
import random
def robust_api_call_with_rate_limit_handling():
"""
Retry logic với exponential backoff cho rate limit errors
"""
max_retries = 5
base_delay = 1 # second
for attempt in range(max_retries):
try:
response = holy_sheep._session.post(
f"{holy_sheep.base_url}/chat/completions",
json={"model": "deepseek-v3.2", "messages": [{"role": "user", "content": "test"}]},
timeout=30
)
if response.status_code == 429:
# Parse Retry-After header nếu có
retry_after = response.headers.get("Retry-After")
if retry_after:
wait_time = int(retry_after)
else:
# Exponential backoff với jitter
wait_time = (base_delay * (2 ** attempt)) + random.uniform(0, 1)
print(f"Rate limited. Waiting {wait_time:.2f}s before retry...")
time.sleep(wait_time)
continue
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
if attempt == max_retries - 1:
raise
time.sleep(base_delay * (attempt + 1))
raise Exception("Max retries exceeded")
2. Lỗi Token Limit Exceeded
Mô tả: Response trả về context window overflow, không thể process prompt dài.
Nguyên nhân gốc: Prompt + context vượt quá model context window (thường 128k tokens với deepseek-v3.2).
Giải pháp:
def smart_context_truncation(messages: list, max_tokens: int = 100000) -> list:
"""
Intelligent truncation giữ lại system prompt và recent context
"""
total_tokens = 0
truncated_messages = []
# Luôn giữ system prompt
system_prompt = None
for msg in messages:
if msg["role"] == "system":
system_prompt = msg
truncated_messages.append(msg)
total_tokens += estimate_tokens(msg["content"])
Tài nguyên liên quan
Bài viết liên quan