Tháng 3/2026, một đêm trước sự kiện ra mắt hệ thống RAG cho doanh nghiệp thương mại điện tử quy mô 2 triệu sản phẩm, tôi nhận ra rằng kiến trúc retrieval cũ không còn đủ. Chunk 512 tokens truyền thống không thể nắm bắt ngữ cảnh sản phẩm, so sánh giá, hay chuỗi hành vi người dùng. Cần một cách tiếp cận hoàn toàn mới — và đó là lúc tôi bắt đầu khám phá sức mạnh của context window 1 triệu token.
Tại Sao 1M Token Thay Đổi Cuộc Chơi
Với context 1 triệu token, bạn có thể đưa vào 750.000 từ văn bản thuần — tương đương 3 cuốn sách dày hay toàn bộ codebase 50.000 dòng. Điều này mở ra ba tầng ứng dụng hoàn toàn mới:
- RAG Đầy Đủ (Full-Context RAG): Thay vì trích xuất chunk nhỏ, hệ thống có thể đưa toàn bộ tài liệu vào context và để model tự suy luận.
- Phân Tích Codebase Toàn Diện: Một model có thể đọc toàn bộ repo, hiểu architecture, và đề xuất refactoring chính xác.
- Hồ Sơ Người Dùng Đa Phiên: Tổng hợp 30 ngày tương tác, lịch sử mua hàng, và preference vào một lần gọi duy nhất.
So Sánh Chi Phí: HolySheep AI vs OpenAI
| Model | HolySheep ($/MTok) | OpenAI ($/MTok) | Tiết kiệm |
|---|---|---|---|
| GPT-4.1 | $8.00 | $60.00 | 86.7% |
| Claude Sonnet 4.5 | $15.00 | $75.00 | 80% |
| Gemini 2.5 Flash | $2.50 | $15.00 | 83.3% |
| DeepSeek V3.2 | $0.42 | $2.00 | 79% |
Với 1 triệu token context, một yêu cầu đầy đủ có thể tiêu tốn tới $0.06 (với DeepSeek V3.2 trên HolyShehe) — vẫn rẻ hơn một tách cà phê. Nhưng để tận dụng tối đa, bạn cần thiết kế API thông minh.
Kiến Trúc API Tối Ưu Cho Context 1M Token
1. Streaming Response Với Chunked Context
Thay vì gửi toàn bộ 1M token cùng lúc, hãy implement progressive loading để giảm perceived latency. Dưới đây là implementation hoàn chỉnh:
#!/usr/bin/env python3
"""
HolySheep AI - Full-Context RAG System
Hỗ trợ context 1M token với streaming response
"""
import httpx
import json
import asyncio
from typing import AsyncGenerator, List, Dict, Optional
class HolySheepContextRAG:
"""Hệ thống RAG tối ưu cho context 1M token"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self.client = httpx.AsyncClient(
timeout=httpx.Timeout(300.0, connect=30.0) # 5 phút cho 1M context
)
async def stream_full_context_query(
self,
query: str,
documents: List[str],
system_prompt: Optional[str] = None
) -> AsyncGenerator[str, None]:
"""
Gửi query với toàn bộ documents vào context và streaming response
Args:
query: Câu hỏi của người dùng
documents: Danh sách documents (tổng có thể lên tới 800K tokens)
system_prompt: Custom system prompt
Yields:
Response chunks từ API
"""
# Tính toán approximate tokens (1 token ~ 4 ký tự tiếng Việt)
context_size = sum(len(doc) // 4 for doc in documents)
if context_size > 950_000:
print(f"Cảnh báo: Context {context_size} tokens gần đạt giới hạn")
# Build messages với full context
default_system = """Bạn là trợ lý AI chuyên phân tích tài liệu.
Đọc kỹ toàn bộ context được cung cấp và trả lời chính xác.
Nếu thông tin không có trong context, hãy nói rõ rằng bạn không biết."""
messages = [
{"role": "system", "content": system_prompt or default_system},
{"role": "user", "content": self._build_context_prompt(query, documents)}
]
async with self.client.stream(
"POST",
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": "deepseek-v3.2", # Model rẻ nhất cho context dài
"messages": messages,
"stream": True,
"temperature": 0.3,
"max_tokens": 4096
}
) as response:
async for line in response.aiter_lines():
if line.startswith("data: "):
if line.strip() == "data: [DONE]":
break
data = json.loads(line[6:])
if delta := data.get("choices", [{}])[0].get("delta", {}).get("content"):
yield delta
def _build_context_prompt(self, query: str, documents: List[str]) -> str:
"""Build prompt với full context"""
context_header = f"=== CONTEXT ({len(documents)} documents) ===\n"
context_docs = "\n\n---\n\n".join(
f"[Doc {i+1}]\n{doc}" for i, doc in enumerate(documents)
)
context_footer = "\n=== END CONTEXT ===\n\n"
question = f"Câu hỏi: {query}\n\nTrả lời dựa trên context trên:"
# Tổng prompt phải fit trong 1M token
return f"{context_header}{context_docs}{context_footer}{question}"
async def batch_process_queries(
self,
queries: List[Dict[str, any]],
max_concurrent: int = 5
) -> List[Dict]:
"""
Xử lý batch queries với concurrency control
Args:
queries: List of {"query": str, "documents": List[str]}
max_concurrent: Số request đồng thời tối đa
Returns:
List of {"query": str, "response": str, "latency_ms": float}
"""
semaphore = asyncio.Semaphore(max_concurrent)
async def process_single(item: Dict) -> Dict:
async with semaphore:
start = asyncio.get_event_loop().time()
response_chunks = []
async for chunk in self.stream_full_context_query(
item["query"],
item["documents"]
):
response_chunks.append(chunk)
latency_ms = (asyncio.get_event_loop().time() - start) * 1000
return {
"query": item["query"],
"response": "".join(response_chunks),
"latency_ms": round(latency_ms, 2),
"tokens_processed": sum(len(d)//4 for d in item["documents"])
}
return await asyncio.gather(*[process_single(q) for q in queries])
async def close(self):
await self.client.aclose()
=== DEMO USAGE ===
async def main():
rag = HolySheepContextRAG(api_key="YOUR_HOLYSHEEP_API_KEY")
# Demo: Phân tích 100 sản phẩm cùng lúc
products = [
f"""Sản phẩm {i+1}: Điện thoại XYZ Model {i}
- Giá: {199 + i * 10} USD
- RAM: {8 + (i % 4) * 2}GB
- Bộ nhớ: {128 + (i % 8) * 64}GB
- Camera: {48 + (i % 3) * 12}MP
- Đánh giá: {4 + (i % 10) / 10} sao
- Số đánh giá: {100 + i * 50}
"""
for i in range(100)
]
query = "So sánh top 5 điện thoại có camera tốt nhất dưới 300 USD?"
print("Đang xử lý query với context 1M token...")
print(f"Số documents: {len(products)}")
print(f"Context size: ~{sum(len(p)//4 for p in products)} tokens\n")
async for chunk in rag.stream_full_context_query(query, products):
print(chunk, end="", flush=True)
await rag.close()
if __name__ == "__main__":
asyncio.run(main())
2. Smart Context Trimming Với Priority Scoring
Khi context vượt ngưỡng, bạn cần intelligent trimming. Implementation này sử dụng semantic similarity để giữ lại chunks quan trọng nhất:
#!/usr/bin/env python3
"""
Smart Context Manager - Tối ưu context cho 1M token window
Implement priority-based trimming và cached embeddings
"""
import httpx
import hashlib
import json
from typing import List, Tuple, Optional
from dataclasses import dataclass
from collections import OrderedDict
@dataclass
class DocumentChunk:
"""Represent a document chunk với metadata"""
id: str
content: str
token_count: int
relevance_score: float = 0.0
metadata: dict = None
class SmartContextManager:
"""Quản lý context thông minh với automatic trimming"""
def __init__(self, api_key: str, max_tokens: int = 950_000):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.max_tokens = max_tokens
self.embedding_cache = OrderedDict()
self.cache_max_size = 1000
def _estimate_tokens(self, text: str) -> int:
"""Estimate token count - 1 token ~ 4 chars for Vietnamese"""
return len(text) // 4
def _generate_chunk_id(self, content: str) -> str:
"""Generate deterministic ID for caching"""
return hashlib.md5(content.encode()).hexdigest()[:16]
async def get_embedding(self, text: str) -> List[float]:
"""
Lấy embedding từ HolySheep API với caching
Latency thực tế: <50ms với cache hit
"""
chunk_id = self._generate_chunk_id(text)
# Check cache
if chunk_id in self.embedding_cache:
return self.embedding_cache[chunk_id]
# Fetch from API
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{self.base_url}/embeddings",
headers={"Authorization": f"Bearer {self.api_key}"},
json={
"model": "embedding-v2",
"input": text[:8192] # Max input for embeddings
}
)
data = response.json()
embedding = data["data"][0]["embedding"]
# Update cache
self.embedding_cache[chunk_id] = embedding
if len(self.embedding_cache) > self.cache_max_size:
self.embedding_cache.popitem(last=False)
return embedding
def cosine_similarity(self, a: List[float], b: List[float]) -> float:
"""Tính cosine similarity giữa 2 vectors"""
dot = 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 / (norm_a * norm_b + 1e-8)
async def score_and_rank_chunks(
self,
chunks: List[DocumentChunk],
query: str,
top_k: Optional[int] = None
) -> List[DocumentChunk]:
"""
Score chunks theo relevance với query
Returns:
Sorted list of chunks by relevance (descending)
"""
# Get query embedding once
query_embedding = await self.get_embedding(query)
# Score each chunk
for chunk in chunks:
if chunk.embedding_cache:
chunk_embedding = chunk.embedding_cache
else:
chunk_embedding = await self.get_embedding(chunk.content)
chunk.embedding_cache = chunk_embedding
chunk.relevance_score = self.cosine_similarity(
query_embedding,
chunk_embedding
)
# Sort by score
ranked = sorted(chunks, key=lambda x: x.relevance_score, reverse=True)
if top_k:
ranked = ranked[:top_k]
return ranked
async def build_optimal_context(
self,
chunks: List[DocumentChunk],
query: str,
include_summary: bool = True
) -> Tuple[str, int, List[str]]:
"""
Build optimal context từ chunks
Strategy:
1. Score all chunks by relevance
2. Greedily add highest-scoring chunks until limit
3. Optionally prepend summary of excluded chunks
Returns:
(context_string, total_tokens, included_chunk_ids)
"""
# Score and rank
ranked_chunks = await self.score_and_rank_chunks(chunks, query)
selected_chunks = []
total_tokens = 0
included_ids = []
# Greedy selection
for chunk in ranked_chunks:
if total_tokens + chunk.token_count <= self.max_tokens - 5000: # Reserve for prompt
selected_chunks.append(chunk)
total_tokens += chunk.token_count
included_ids.append(chunk.id)
# Add summary if many chunks excluded
excluded_count = len(chunks) - len(selected_chunks)
if include_summary and excluded_count > 5:
summary_prompt = f"\n[Tổng cộng {len(chunks)} chunks, hiển thị {len(selected_chunks)} chunks liên quan nhất. {excluded_count} chunks ít liên quan hơn đã bị cắt bỏ.]\n"
total_tokens += self._estimate_tokens(summary_prompt)
else:
summary_prompt = ""
# Build context string
context_parts = [f"=== CONTEXT ({len(selected_chunks)}/{len(chunks)} chunks) ===\n"]
for i, chunk in enumerate(selected_chunks):
context_parts.append(f"[{i+1}] {chunk.content}\n")
context_parts.append(summary_prompt)
context_parts.append("===\n")
return "".join(context_parts), total_tokens, included_ids
async def batch_build_contexts(
self,
chunk_sets: List[List[DocumentChunk]],
queries: List[str]
) -> List[Tuple[str, int]]:
"""
Batch process multiple context builds
Performance:
- Batch embeddings để giảm API calls
- Cache sharing giữa các batches
- Target: <500ms cho 10 contexts với 100 chunks each
"""
results = []
for chunks, query in zip(chunk_sets, queries):
context, tokens, _ = await self.build_optimal_context(chunks, query)
results.append((context, tokens))
return results
=== PERFORMANCE TEST ===
async def benchmark():
"""Test performance với realistic data"""
import time
manager = SmartContextManager(api_key="YOUR_HOLYSHEEP_API_KEY")
# Generate 500 realistic chunks (e-commerce scenario)
chunks = [
DocumentChunk(
id=f"chunk_{i}",
content=f"Sản phẩm {i}: Mô tả chi tiết với thông số kỹ thuật, đánh giá, "
f"so sánh với sản phẩm tương tự. " * 20,
token_count=200,
metadata={"category": "electronics", "price_range": i % 5}
)
for i in range(500)
]
queries = [
"Điện thoại tốt nhất cho chụp ảnh dưới 500$?",
"So sánh laptop gaming và laptop văn phòng",
"Tai nghe không dây chống ồn tốt nhất 2026"
]
print("=== Smart Context Manager Benchmark ===")
print(f"Chunks: {len(chunks)}")
print(f"Total tokens (all): {sum(c.token_count for c in chunks):,}")
print(f"Max allowed: {manager.max_tokens:,}\n")
for query in queries:
start = time.perf_counter()
context, tokens, ids = await manager.build_optimal_context(chunks, query)
elapsed_ms = (time.perf_counter() - start) * 1000
print(f"Query: {query[:50]}...")
print(f" Selected: {len(ids)} chunks")
print(f" Tokens used: {tokens:,}")
print(f" Latency: {elapsed_ms:.2f}ms")
print()
if __name__ == "__main__":
import asyncio
asyncio.run(benchmark())
3. Monitoring Dashboard Data
Đoạn code này gửi metrics lên dashboard để track chi phí và hiệu suất theo thời gian thực:
#!/usr/bin/env python3
"""
Context Usage Monitor - Theo dõi chi phí và hiệu suất
Integration với HolySheep Analytics
"""
import time
import asyncio
import json
from datetime import datetime, timedelta
from dataclasses import dataclass, asdict
from typing import Dict, List, Optional
import httpx
@dataclass
class ContextMetrics:
"""Metrics cho một context request"""
timestamp: str
request_id: str
context_tokens: int
response_tokens: int
total_cost: float
latency_ms: float
model: str
hit_cache: bool = False
class ContextMonitor:
"""Monitor và optimize chi phí context 1M"""
# HolySheep 2026 Pricing (USD per million tokens)
PRICING = {
"deepseek-v3.2": {"input": 0.42, "output": 1.68},
"gpt-4.1": {"input": 8.00, "output": 24.00},
"claude-sonnet-4.5": {"input": 15.00, "output": 75.00},
"gemini-2.5-flash": {"input": 2.50, "output": 10.00}
}
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.metrics: List[ContextMetrics] = []
self.cost_alert_threshold = 100.0 # Alert khi > $100/ngày
def calculate_cost(
self,
model: str,
input_tokens: int,
output_tokens: int
) -> float:
"""Tính chi phí theo HolySheep pricing"""
pricing = self.PRICING.get(model, {"input": 1.0, "output": 4.0})
input_cost = (input_tokens / 1_000_000) * pricing["input"]
output_cost = (output_tokens / 1_000_000) * pricing["output"]
return round(input_cost + output_cost, 6) # 6 decimal places
async def track_request(
self,
model: str,
context_tokens: int,
output_tokens: int,
latency_ms: float,
hit_cache: bool = False
) -> ContextMetrics:
"""Track một request và trả về metrics"""
cost = self.calculate_cost(model, context_tokens, output_tokens)
metrics = ContextMetrics(
timestamp=datetime.now().isoformat(),
request_id=f"req_{int(time.time() * 1000)}",
context_tokens=context_tokens,
response_tokens=output_tokens,
total_cost=cost,
latency_ms=round(latency_ms, 2),
model=model,
hit_cache=hit_cache
)
self.metrics.append(metrics)
return metrics
def get_daily_summary(self, date: Optional[datetime] = None) -> Dict:
"""Tổng hợp chi phí theo ngày"""
if date is None:
date = datetime.now()
day_metrics = [
m for m in self.metrics
if datetime.fromisoformat(m.timestamp).date() == date.date()
]
if not day_metrics:
return {"date": date.date(), "total_cost": 0, "requests": 0}
model_costs = {}
total_cost = 0
total_tokens = 0
for m in day_metrics:
total_cost += m.total_cost
total_tokens += m.context_tokens + m.response_tokens
model_costs[m.model] = model_costs.get(m.model, 0) + m.total_cost
return {
"date": date.date().isoformat(),
"total_cost": round(total_cost, 4),
"total_requests": len(day_metrics),
"total_tokens": total_tokens,
"avg_latency_ms": round(
sum(m.latency_ms for m in day_metrics) / len(day_metrics), 2
),
"cache_hit_rate": round(
sum(1 for m in day_metrics if m.hit_cache) / len(day_metrics) * 100, 1
),
"cost_by_model": {k: round(v, 4) for k, v in model_costs.items()},
"alert": total_cost > self.cost_alert_threshold
}
async def send_to_dashboard(self, metrics: ContextMetrics):
"""Gửi metrics lên dashboard (simulation)"""
# Trong production, gửi lên monitoring service
print(f"[MONITOR] {metrics.request_id} | "
f"Tokens: {metrics.context_tokens:,} | "
f"Cost: ${metrics.total_cost:.6f} | "
f"Latency: {metrics.latency_ms}ms | "
f"Model: {metrics.model}")
async def run_cost_optimizer(self):
"""
Suggest optimizations để giảm chi phí
Chạy periodic để đề xuất model switching
"""
summary = self.get_daily_summary()
suggestions = []
# Check nếu đang dùng model đắt cho context dài
expensive_requests = [
m for m in self.metrics[-100:]
if m.model == "gpt-4.1" and m.context_tokens > 100_000
]
if len(expensive_requests) > 10:
potential_savings = sum(
self.calculate_cost("deepseek-v3.2", m.context_tokens, m.response_tokens)
- m.total_cost
for m in expensive_requests
)
suggestions.append({
"type": "model_switch",
"message": f"Switch {len(expensive_requests)} requests sang DeepSeek V3.2 "
f"để tiết kiệm ${potential_savings:.2f}/ngày",
"impact": "high",
"savings_percent": 95
})
# Check cache hit rate thấp
if summary["cache_hit_rate"] < 30:
suggestions.append({
"type": "cache_improvement",
"message": "Cache hit rate chỉ ở " + str(summary["cache_hit_rate"]) + "%. "
"Xem xét tăng cache size hoặc batch similar requests.",
"impact": "medium",
"potential_savings": "10-20%"
})
return suggestions
=== INTEGRATION EXAMPLE ===
async def example_with_monitoring():
"""Example: Sử dụng monitor với RAG system"""
import random
monitor = ContextMonitor(api_key="YOUR_HOLYSHEEP_API_KEY")
# Simulate 50 requests trong ngày
models = ["deepseek-v3.2", "gpt-4.1", "gemini-2.5-flash"]
print("=== Simulating Daily Requests ===\n")
for i in range(50):
model = random.choice(models)
context_tokens = random.randint(50_000, 500_000)
output_tokens = random.randint(500, 4000)
latency_ms = random.uniform(30, 200)
hit_cache = random.random() > 0.7
metrics = await monitor.track_request(
model=model,
context_tokens=context_tokens,
output_tokens=output_tokens,
latency_ms=latency_ms,
hit_cache=hit_cache
)
await monitor.send_to_dashboard(metrics)
await asyncio.sleep(0.01) # Simulate real-time
print("\n=== Daily Summary ===")
summary = monitor.get_daily_summary()
print(json.dumps(summary, indent=2))
print("\n=== Optimization Suggestions ===")
suggestions = await monitor.run_cost_optimizer()
for s in suggestions:
print(f"• [{s['impact'].upper()}] {s['message']}")
if __name__ == "__main__":
asyncio.run(example_with_monitoring())
Lỗi Thường Gặp Và Cách Khắc Phục
1. Lỗi "Context Length Exceeded" - Quá Tải Context
Mã lỗi: context_length_exceeded hoặc HTTP 400
Nguyên nhân: Tổng tokens (input + output) vượt quá limit của model. Với 1M context, bạn cần trừ đi phần system prompt, chat history, và reserved output tokens.
# ❌ SAI: Không tính toán buffer
def build_prompt_unsafe(chunks: List[str], query: str) -> str:
return f"Context: {' '.join(chunks)}\n\nQuestion: {query}"
✅ ĐÚNG: Luôn reserve tokens cho response
def build_prompt_safe(
chunks: List[str],
query: str,
max_context: int = 950_000,
reserved_output: int = 4000
) -> str:
"""
Build prompt với safety margin
"""
# System + query base tokens
base_overhead = 500
available_for_context = max_context - reserved_output - base_overhead
# Progressive truncation nếu cần
current_context = ""
current_tokens = 0
for chunk in chunks:
chunk_tokens = len(chunk) // 4
if current_tokens + chunk_tokens <= available_for_context:
current_context += chunk + "\n\n"
current_tokens += chunk_tokens
else:
break # Dừng lại, không raise error
return f"Context: {current_context}\n\nQuestion: {query}"
2. Lỗi Timeout Khi Xử Lý Context Lớn
Triệu chứng: Request chờ > 60 giây rồi failed với timeout_error
Giải pháp: Tăng timeout và implement retry với exponential backoff:
import httpx
import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=30)
)
async def safe_context_call(
messages: List[Dict],
api_key: str,
max_context_tokens: int = 950_000
) -> Dict:
"""
Gọi API với retry logic cho context lớn
"""
# Validate trước
total_tokens = sum(
len(msg["content"]) // 4
for msg in messages
if msg.get("content")
)
if total_tokens > max_context_tokens:
raise ValueError(
f"Context {total_tokens} tokens vượt limit {max_context_tokens}"
)
async with httpx.AsyncClient(
timeout=httpx.Timeout(180.0, connect=30.0) # 3 phút
) as client:
try:
response = await client.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
},
json={
"model": "deepseek-v3.2",
"messages": messages,
"temperature": 0.3,
"max_tokens": 4000
}
)
response.raise_for_status()
return response.json()
except httpx.TimeoutException as e:
print(f"Timeout sau 180s, retry...")
raise
except httpx.HTTPStatusError as e:
if e.response.status_code == 429: # Rate limit
await asyncio.sleep(5)
raise
raise
3. Lỗi Chi Phí Không Kiểm Soát - Billing Spike
Triệu chứng: Chi phí API tăng đột ngột, không có warning
Giải pháp: Implement pre-check và budget cap:
class BudgetController:
"""
Kiểm soát chi phí cho context 1M operations
HolySheep ưu đãi: DeepSeek V3.2 chỉ $0.42/MTok input
"""
def __init__(self, daily_budget: float = 50.0):
self.daily_budget = daily_budget
self.daily_spent = 0.0
self.last_reset = datetime.now().date()
def _check_budget(self, estimated_cost: float) -> bool:
"""Kiểm tra budget trước khi gọi API"""
today = datetime.now().date()
if today > self.last_reset:
self.daily_spent = 0.0
self.last_reset = today
if self.daily_spent + estimated_cost > self.daily_budget:
raise BudgetExceededError(
f"Daily budget ${self.daily_budget} sẽ bị vượt. "
f"Hiện tại: ${self.daily_spent:.2f}, "
f"Dự kiến: ${estimated_cost:.4f}"
)
return True
async def execute_with_budget_check(
self,
context_tokens: int,
output_tokens: int,
model: str
) -> float:
"""
Execute request chỉ khi trong budget
Returns:
Actual cost incurred
"""
# Tính chi phí dự kiến
pricing = {
"deepseek-v3.2": {"input": 0.42, "output": 1.68},
"gemini-2.5-flash": {"input": 2.50, "output": 10.00},
"gpt-4.1": {"input": 8.00, "output": 24.00}
}
p = pricing.get(model, {"input": 1.0, "output": 4.0})
estimated = (context_tokens / 1e6) * p["input"] + \
(output_tokens / 1e6) * p["output"]
self._check_budget(estimated)
# Execute request...
# Sau khi thành công, cập nhật spent
self.daily_spent += estimated
return estimated
Best Practices Tổng Hợp
- Luôn reserve tokens: Để 5-10% buffer cho response và system prompt.
Tài nguyên liên quan
Bài viết liên quan