Đêm 26/4, ngay trước đợt flash sale lớn nhất quý, hệ thống chatbot chăm sóc khách hàng của một thương mại điện tử Việt Nam đã xử lý 47,000 tư vấn đơn hàng trong 4 giờ đồng hồ. Tỷ lệ thành công: 99.2%. Đó là lúc tôi nhận ra rằng kiến trúc multi-agent với AutoGen 0.4 đã trưởng thành đủ để chạy production thực sự — không còn là proof-of-concept.
Tại Sao AutoGen 0.4 + MCP Server Là Lựa Chọn Đúng Cho Production
AutoGen phiên bản 0.4 đánh dấu bước tiến lớn với kiến trúc agent orchestration được tối ưu hóa cho môi trường phân tán. MCP (Model Context Protocol) server đóng vai trò như lớp trung gian chuẩn hóa, cho phép các agent giao tiếp với nhiều LLM provider một cách nhất quán.
Ưu Điểm Kiến Trúc
- Model routing thông minh: Agent tự chọn model phù hợp dựa trên độ phức tạp task
- Streaming response: Xử lý real-time với latency dưới 200ms cho prompts ngắn
- Conversation state management: Duy trì context qua nhiều agent turn
- Built-in retry và fallback: Tự động chuyển model khi một provider gặp sự cố
Cài Đặt Môi Trường Và Cấu Hình
# Cài đặt AutoGen 0.4 và dependencies
pip install autogen-agentchat==0.4.0
pip install "autogen-agentchat[ollama]>=0.4.0"
pip install mcp-server==1.0.0
pip install fastapi uvicorn aiohttp
Kiểm tra phiên bản
python -c "import autogen; print(autogen.__version__)"# Cấu hình .env cho multi-model setup
cat > .env << 'EOF'
HolySheep AI - Multi-model Configuration
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1
Model Routing
PRIMARY_MODEL=gpt-4.1
FALLBACK_MODEL=claude-sonnet-4.5
CHEAP_MODEL=deepseek-v3.2
REASONING_MODEL=gemini-2.5-flash
MCP Server
MCP_SERVER_PORT=8080
MCP_TRANSPORT=stdio
Production Settings
MAX_CONCURRENT_AGENTS=50
REQUEST_TIMEOUT=30
ENABLE_STREAMING=true
EOFTriển Khai Kiến Trúc Multi-Agent Với Model Routing
import os
import asyncio
from autogen_agentchat import AssistantAgent, UserProxyAgent, Team
from autogen_agentchat.conditions import MaxMessageTermination
from autogen_agentchat.llm import LLMConfig
from typing import Optional
import httpx
Load configuration
API_KEY = os.getenv("HOLYSHEEP_API_KEY")
BASE_URL = os.getenv("HOLYSHEEP_BASE_URL", "https://api.holysheep.ai/v1")
class ModelRouter:
"""Router thông minh chọn model dựa trên độ phức tạp task"""
# Định nghĩa ngưỡng routing
COMPLEXITY_THRESHOLDS = {
"simple": 50, # tokens dự đoán
"medium": 200,
"complex": 800
}
# Pricing reference (USD/MTok)
MODEL_COSTS = {
"gpt-4.1": 8.0,
"claude-sonnet-4.5": 15.0,
"deepseek-v3.2": 0.42,
"gemini-2.5-flash": 2.50
}
def __init__(self):
self.client = httpx.AsyncClient(
base_url=BASE_URL,
headers={"Authorization": f"Bearer {API_KEY}"},
timeout=30.0
)
async def estimate_complexity(self, prompt: str) -> str:
"""Ước tính độ phức tạp dựa trên độ dài và từ khóa"""
token_count = len(prompt.split()) * 1.3 # Approximate
# Keywords chỉ ra task phức tạp
complex_keywords = [
"analyze", "compare", "evaluate", "design", "architect",
"explain", "debug", "refactor", "optimize", "strategy"
]
prompt_lower = prompt.lower()
complexity_bonus = sum(2 for kw in complex_keywords if kw in prompt_lower)
if token_count + complexity_bonus < self.COMPLEXITY_THRESHOLDS["simple"]:
return "simple"
elif token_count + complexity_bonus < self.COMPLEXITY_THRESHOLDS["medium"]:
return "medium"
else:
return "complex"
async def get_model_for_task(self, prompt: str, task_type: Optional[str] = None):
"""Chọn model tối ưu về chi phí và hiệu suất"""
if task_type == "reasoning":
return "gemini-2.5-flash" # Tốt cho chain-of-thought
elif task_type == "coding":
return "gpt-4.1" # Mạnh nhất cho code
elif task_type == "fast_response":
return "deepseek-v3.2" # Rẻ và nhanh
complexity = await self.estimate_complexity(prompt)
if complexity == "simple":
return "deepseek-v3.2" # Tiết kiệm 95% chi phí
elif complexity == "medium":
return "gemini-2.5-flash" # Cân bằng
else:
return "gpt-4.1" # Chất lượng cao nhất
async def chat_completion(self, model: str, messages: list, stream: bool = True):
"""Gọi API với model được chọn"""
payload = {
"model": model,
"messages": messages,
"stream": stream,
"temperature": 0.7,
"max_tokens": 2000
}
response = await self.client.post("/chat/completions", json=payload)
response.raise_for_status()
return response.json()
Khởi tạo router
router = ModelRouter()Xây Dựng MCP Server Cho Multi-Provider Access
# mcp_server.py - Model Context Protocol Server
import asyncio
import json
from typing import Any, Dict, List, Optional
from dataclasses import dataclass, asdict
from enum import Enum
class TaskType(Enum):
COMPLETION = "completion"
EMBEDDING = "embedding"
REASONING = "reasoning"
MULTIMODAL = "multimodal"
@dataclass
class MCPRequest:
method: str
params: Dict[str, Any]
task_type: Optional[TaskType] = None
@dataclass
class MCPResponse:
success: bool
data: Any
error: Optional[str] = None
latency_ms: float = 0
class MCPServer:
"""
MCP Server quản lý kết nối đến nhiều LLM provider
qua HolySheep AI unified API
"""
def __init__(self, api_key: str, base_url: str):
self.api_key = api_key
self.base_url = base_url
self.request_count = 0
self.total_cost = 0.0
self._metrics = {"requests": 0, "errors": 0, "total_tokens": 0}
async def handle_request(self, request: MCPRequest) -> MCPResponse:
"""Xử lý request theo MCP protocol"""
import time
start = time.time()
try:
if request.method == "chat.completion":
result = await self._handle_chat(request.params)
elif request.method == "embedding.create":
result = await self._handle_embedding(request.params)
elif request.method == "model.list":
result = await self._list_models()
else:
raise ValueError(f"Unknown method: {request.method}")
latency = (time.time() - start) * 1000
# Update metrics
self._metrics["requests"] += 1
self._update_cost_estimate(request.params)
return MCPResponse(
success=True,
data=result,
latency_ms=round(latency, 2)
)
except Exception as e:
self._metrics["errors"] += 1
return MCPResponse(
success=False,
data=None,
error=str(e),
latency_ms=round((time.time() - start) * 1000, 2)
)
async def _handle_chat(self, params: Dict) -> Dict:
"""Xử lý chat completion với automatic model selection"""
model = params.get("model", "gpt-4.1")
messages = params["messages"]
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": model,
"messages": messages,
"stream": params.get("stream", False),
"temperature": params.get("temperature", 0.7),
"max_tokens": params.get("max_tokens", 2048)
},
timeout=params.get("timeout", 30.0)
)
if response.status_code != 200:
raise Exception(f"API Error: {response.status_code} - {response.text}")
return response.json()
async def _handle_embedding(self, params: Dict) -> Dict:
"""Tạo embeddings cho RAG systems"""
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.base_url}/embeddings",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": "embedding-3",
"input": params["input"]
}
)
return response.json()
async def _list_models(self) -> Dict:
"""Liệt kê models khả dụng"""
return {
"models": [
{"id": "gpt-4.1", "context_length": 128000, "pricing": 8.0},
{"id": "claude-sonnet-4.5", "context_length": 200000, "pricing": 15.0},
{"id": "deepseek-v3.2", "context_length": 64000, "pricing": 0.42},
{"id": "gemini-2.5-flash", "context_length": 1000000, "pricing": 2.50}
]
}
def _update_cost_estimate(self, params: Dict):
"""Cập nhật ước tính chi phí"""
model = params.get("model", "gpt-4.1")
messages = params.get("messages", [])
# Rough token estimation
input_tokens = sum(len(m.get("content", "").split()) * 1.3 for m in messages)
output_tokens = params.get("max_tokens", 2048)
total_tokens = input_tokens + output_tokens
pricing = {
"gpt-4.1": 0.008, # $8/MTok
"claude-sonnet-4.5": 0.015,
"deepseek-v3.2": 0.00042,
"gemini-2.5-flash": 0.0025
}.get(model, 0.008)
self.total_cost += (total_tokens / 1_000_000) * pricing * 2 # Input + Output
def get_metrics(self) -> Dict:
"""Lấy metrics hiệu tại"""
return {
**self._metrics,
"total_cost_usd": round(self.total_cost, 4),
"avg_cost_per_request": round(
self.total_cost / max(self._metrics["requests"], 1), 6
)
}
Khởi chạy server
if __name__ == "__main__":
mcp = MCPServer(
api_key=os.getenv("HOLYSHEEP_API_KEY"),
base_url=os.getenv("HOLYSHEEP_BASE_URL", "https://api.holysheep.ai/v1")
)
print("MCP Server khởi động thành công!")
print(f"Models khả dụng: {asyncio.run(mcp.handle_request(MCPRequest('model.list', {})))}")AutoGen Team Với Smart Model Selection
# main.py - AutoGen 0.4 Production Setup
import asyncio
import os
from autogen_agentchat import AssistantAgent, UserProxyAgent, Team
from autogen_agentchat.conditions import TextMentionTermination, MaxMessageTermination
from autogen_agentchat.llm import LLMConfig
from mcp_server import MCPServer, ModelRouter
Cấu hình LLM cho từng agent
def get_llm_config(model: str) -> LLMConfig:
return LLMConfig(
model=model,
api_key=os.getenv("HOLYSHEEP_API_KEY"),
base_url=os.getenv("HOLYSHEEP_BASE_URL", "https://api.holysheep.ai/v1"),
temperature=0.7,
max_tokens=2048
)
Khởi tạo MCP Server
mcp_server = MCPServer(
api_key=os.getenv("HOLYSHEEP_API_KEY"),
base_url=os.getenv("HOLYSHEEP_BASE_URL", "https://api.holysheep.ai/v1")
)
router = ModelRouter()
=== Agent Definitions ===
1. Triage Agent - Phân loại request ban đầu
triage_agent = AssistantAgent(
name="TriageAgent",
description="Phân loại và routing request khách hàng",
system_message="""Bạn là agent phân loại request.
Phân tích yêu cầu và xác định loại task:
- ORDER: Thông tin đơn hàng, tracking, cancel
- PRODUCT: Hỏi về sản phẩm, so sánh, review
- TECHNICAL: Hỗ trợ kỹ thuật, troubleshoot
- BILLING: Thanh toán, hoàn tiền, voucher
- COMPLEX: Cần xử lý phức tạp, escalate
Trả lời ngắn gọn với format: [TASK_TYPE]: brief_summary""",
llm_config=get_llm_config("deepseek-v3.2") # Fast, cheap cho classification
)
2. Product Expert - Tư vấn sản phẩm
product_expert = AssistantAgent(
name="ProductExpert",
description="Chuyên gia tư vấn sản phẩm",
system_message="""Bạn là chuyên gia sản phẩm thương mại điện tử.
Cung cấp thông tin chi tiết về sản phẩm, so sánh features,
đưa ra recommendations phù hợp với nhu cầu khách hàng.
Luôn bao gồm: price, availability, alternatives""",
llm_config=get_llm_config("gpt-4.1") # Best quality cho product咨询
)
3. Order Manager - Xử lý đơn hàng
order_manager = AssistantAgent(
name="OrderManager",
description="Quản lý và xử lý đơn hàng",
system_message="""Bạn là order management agent.
Kiểm tra trạng thái đơn hàng, xử lý cancel/modify requests.
Tuân thủ strict business rules:
- Cancel only within 24h of placement
- Max 3 address changes per order
- Provide accurate ETAs
Kết thúc mỗi response với: [ACTION_REQUIRED: yes/no]""",
llm_config=get_llm_config("gemini-2.5-flash") # Fast + good reasoning
)
4. Billing Specialist - Xử lý thanh toán
billing_specialist = AssistantAgent(
name="BillingSpecialist",
description="Chuyên gia về thanh toán và hoàn tiền",
system_message="""Bạn là billing specialist.
Xử lý refund requests, voucher distribution, payment issues.
Refund rules:
- Card refund: 5-7 business days
- Bank transfer: 3-5 business days
- Instant refund available for VIP customers""",
llm_config=get_llm_config("deepseek-v3.2") # Cost-effective cho routine tasks
)
5. Tech Support - Hỗ trợ kỹ thuật
tech_support = AssistantAgent(
name="TechSupport",
description="Hỗ trợ kỹ thuật và troubleshooting",
system_message="""Bạn là technical support agent.
Debug issues, guide users through troubleshooting steps,
escalate complex technical problems.
Response format:
1. Problem identified: ...
2. Steps to resolve: ...
3. Expected outcome: ...""",
llm_config=get_llm_config("gpt-4.1") # Best for technical reasoning
)
User Proxy cho interaction
user_proxy = UserProxyAgent(
name="Customer",
human_input_mode="NEVER" # Fully automated production mode
)
=== Team Definition ===
async def create_production_team():
"""Tạo team với termination conditions"""
team = Team(
name="CustomerServiceTeam",
agents=[
triage_agent,
product_expert,
order_manager,
billing_specialist,
tech_support
],
user_agent=user_proxy,
termination_conditions=[
TextMentionTermination("TERMINATE"),
MaxMessageTermination(max_messages=10)
]
)
return team
=== Production Pipeline ===
async def handle_customer_request(customer_message: str):
"""Xử lý request với model routing tự động"""
# Bước 1: Route đến đúng agent
task_type = await router.estimate_complexity(customer_message)
# Bước 2: Khởi tạo team
team = await create_production_team()
# Bước 3: Xử lý qua pipeline
if any(kw in customer_message.lower() for kw in ["đơn hàng", "order", "tracking", "hủy"]):
selected_agent = order_manager
elif any(kw in customer_message.lower() for kw in ["thanh toán", "hoàn tiền", "refund", "voucher"]):
selected_agent = billing_specialist
elif any(kw in customer_message.lower() for kw in ["sản phẩm", "product", "so sánh", "mua"]):
selected_agent = product_expert
elif any(kw in customer_message.lower() for kw in ["lỗi", "bug", "không hoạt động", "error"]):
selected_agent = tech_support
else:
selected_agent = triage_agent
# Bước 4: Chạy task
result = await team.run(
task=f"Customer query: {customer_message}\n\nHandle this request professionally.",
starting_agent=selected_agent
)
return {
"response": result.messages[-1].content,
"agent_used": selected_agent.name,
"model": selected_agent.llm_config.model,
"metrics": mcp_server.get_metrics()
}
=== Run Production ===
if __name__ == "__main__":
async def main():
# Test cases
test_queries = [
"Tôi muốn kiểm tra đơn hàng #12345",
"So sánh iPhone 15 Pro và Samsung S24 Ultra",
"Tôi cần hoàn tiền cho đơn hàng bị lỗi",
"Ứng dụng không mở được sau khi cập nhật"
]
for query in test_queries:
print(f"\n{'='*50}")
print(f"Query: {query}")
result = await handle_customer_request(query)
print(f"Agent: {result['agent_used']}")
print(f"Model: {result['model']}")
print(f"Response: {result['response'][:200]}...")
# Print cost summary
print(f"\n{'='*50}")
print("Cost Summary:")
print(mcp_server.get_metrics())
asyncio.run(main())Monitoring Và Cost Optimization
# monitoring.py - Production monitoring với real-time metrics
import asyncio
from datetime import datetime, timedelta
from collections import defaultdict
import json
class ProductionMonitor:
"""
Monitor hiệu suất và chi phí trong thời gian thực
"""
def __init__(self):
self.request_log = []
self.cost_by_model = defaultdict(float)
self.latency_by_model = defaultdict(list)
self.error_count = defaultdict(int)
self.model_pricing = {
"gpt-4.1": 8.0,
"claude-sonnet-4.5": 15.0,
"deepseek-v3.2": 0.42,
"gemini-2.5-flash": 2.50
}
def log_request(self, model: str, tokens: int, latency_ms: float, success: bool):
"""Log request để phân tích"""
self.request_log.append({
"timestamp": datetime.now().isoformat(),
"model": model,
"tokens": tokens,
"latency_ms": latency_ms,
"success": success
})
if success:
# Tính chi phí (input + output)
cost = (tokens / 1_000_000) * self.model_pricing.get(model, 8.0) * 2
self.cost_by_model[model] += cost
self.latency_by_model[model].append(latency_ms)
else:
self.error_count[model] += 1
def get_cost_report(self) -> dict:
"""Báo cáo chi phí chi tiết"""
total_cost = sum(self.cost_by_model.values())
return {
"total_cost_usd": round(total_cost, 4),
"by_model": {k: round(v, 4) for k, v in self.cost_by_model.items()},
"request_count": len(self.request_log),
"success_rate": round(
len([r for r in self.request_log if r["success"]]) /
max(len(self.request_log), 1) * 100, 2
),
"savings_vs_openai": self._calculate_savings()
}
def _calculate_savings(self) -> dict:
"""So sánh chi phí với OpenAI direct"""
# OpenAI pricing (approximate)
openai_pricing = {
"gpt-4.1": 15.0, # $15/MTok (actually higher)
}
# HolySheep pricing (your actual cost)
holysheep_total = sum(self.cost_by_model.values())
# Estimate if using OpenAI
model_map = {
"gpt-4.1": "gpt-4.1",
"claude-sonnet-4.5": "claude-sonnet-4-5",
"deepseek-v3.2": "deepseek-v3",
"gemini-2.5-flash": "gemini-1.5-flash"
}
estimated_openai = 0
for model, cost in self.cost_by_model.items():
# Rough estimate - actual would need token counts
multiplier = openai_pricing.get(model, 15.0) / self.model_pricing.get(model, 8.0)
estimated_openai += cost * multiplier
savings = estimated_openai - holysheep_total
savings_pct = (savings / estimated_openai * 100) if estimated_openai > 0 else 0
return {
"estimated_openai_cost": round(estimated_openai, 4),
"your_actual_cost": round(holysheep_total, 4),
"savings_usd": round(savings, 4),
"savings_percentage": round(savings_pct, 1)
}
def get_performance_report(self) -> dict:
"""Báo cáo hiệu suất theo model"""
report = {}
for model, latencies in self.latency_by_model.items():
if latencies:
sorted_lat = sorted(latencies)
report[model] = {
"avg_latency_ms": round(sum(latencies) / len(latencies), 2),
"p50_ms": round(sorted_lat[len(sorted_lat) // 2], 2),
"p95_ms": round(sorted_lat[int(len(sorted_lat) * 0.95)], 2),
"p99_ms": round(sorted_lat[int(len(sorted_lat) * 0.99)], 2),
"request_count": len(latencies),
"error_count": self.error_count.get(model, 0)
}
return report
def print_dashboard(self):
"""In dashboard tổng hợp"""
cost_report = self.get_cost_report()
perf_report = self.get_performance_report()
savings = cost_report["savings_vs_openai"]
print("\n" + "="*60)
print("📊 PRODUCTION MONITORING DASHBOARD")
print("="*60)
print("\n💰 COST ANALYSIS:")
print(f" Total Cost: ${cost_report['total_cost_usd']}")
print(f" By Model:")
for model, cost in cost_report['by_model'].items():
print(f" - {model}: ${cost}")
print(f"\n 💡 SAVINGS vs Direct API:")
print(f" OpenAI Est.: ${savings['estimated_openai_cost']}")
print(f" HolySheep: ${savings['your_actual_cost']}")
print(f" SAVINGS: ${savings['savings_usd']} ({savings['savings_percentage']}%)")
print("\n⚡ PERFORMANCE:")
print(f" Total Requests: {cost_report['request_count']}")
print(f" Success Rate: {cost_report['success_rate']}%")
print(f"\n By Model:")
for model, stats in perf_report.items():
print(f" {model}:")
print(f" Avg: {stats['avg_latency_ms']}ms | P95: {stats['p95_ms']}ms | P99: {stats['p99_ms']}ms")
print("\n" + "="*60)
Demo usage
if __name__ == "__main__":
monitor = ProductionMonitor()
# Simulate requests
test_data = [
("gpt-4.1", 1500, 450, True),
("deepseek-v3.2", 800, 120, True),
("gemini-2.5-flash", 2000, 180, True),
("claude-sonnet-4.5", 1200, 380, True),
("gpt-4.1", 3000, 520, True),
("deepseek-v3.2", 500, 95, True),
("gpt-4.1", 1800, 480, False), # Error case
]
for model, tokens, latency, success in test_data:
monitor.log_request(model, tokens, latency, success)
monitor.print_dashboard()Lỗi Thường Gặp Và Cách Khắc Phục
1. Lỗi Authentication - Invalid API Key
# ❌ Lỗi: "401 Unauthorized" hoặc "Invalid API key"
Nguyên nhân: API key không đúng hoặc chưa được set đúng cách
✅ Khắc phục:
import os
Method 1: Set trực tiếp trong code (không khuyến khích cho production)
os.environ["HOLYSHEEP_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
Method 2: Load từ .env file
from dotenv import load_dotenv
load_dotenv()
Method 3: Verify key trước khi sử dụng
import httpx
async def verify_api_key(api_key: str) -> bool:
"""Xác minh API key hợp lệ"""
async with httpx.AsyncClient() as client:
try:
response = await client.get(
"https://api.holysheep.ai/v1/models",
headers={"Authorization": f"Bearer {api_key}"}
)
return response.status_code == 200
except:
return False
Sử dụng:
API_KEY = os.getenv("HOLYSHEEP_API_KEY")
if not await verify_api_key(API_KEY):
raise ValueError("❌ API Key không hợp lệ. Vui lòng kiểm tra lại.")2. Lỗi Rate Limit - 429 Too Many Requests
# ❌ Lỗi: "429 Client Error: Too Many Requests"
Nguyên nhân: Vượt quá rate limit của API
✅ Khắc phục với exponential backoff:
import asyncio
import httpx
from datetime import datetime, timedelta
class RateLimitHandler:
def __init__(self, max_retries: int = 5):
self.max_retries = max_retries
self.base_delay = 1 # seconds
self.request_timestamps = []
self.window_size = 60 # 1 phút
async def call_with_retry(self, func, *args, **kwargs):
"""Gọi API với retry logic"""
for attempt in range(self.max_retries):
try:
result = await func(*args, **kwargs)
return result
except httpx.HTTPStatusError as e:
if e.response.status_code == 429:
# Extract retry-after header
retry_after = e.response.headers.get("retry-after", "60")
delay = int(retry_after) * (2 ** attempt) # Exponential backoff
print(f"⏳ Rate limited. Retrying in {delay}s (attempt {attempt + 1})")
await asyncio.sleep(delay)
else:
raise
raise Exception(f"❌ Failed after {self.max_retries} retries")
Sử dụng:
handler = RateLimitHandler()
async def safe_api_call(messages: list):
async def _call():
async with httpx.AsyncClient() as client:
return await client.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={"Authorization": f"Bearer {API_KEY}"},
json={"model": "gpt-4.1", "messages": messages}
)
return await handler.call_with_retry(_call)3. Lỗi Context Length Exceeded
# ❌ Lỗi: "context_length_exceeded" hoặc "maximum context length"
Nguyên nhân: Prompt + history vượt quá context window của model
✅ Khắc phục với intelligent truncation:
import tiktoken
class ConversationManager:
"""Quản lý context window thông minh"""
def __init__(self, model: str = "gpt-4.1"):
self.model = model
self.max_tokens = {
"gpt-4.1": 128000,
"claude-sonnet-4.5": 200000,
"deepseek-v3.2": 64000,
"gemini-2.5-flash": 1000000
}.get(model, 128000)
# Reserve tokens cho response
self.reserved_output = 2000
self.available_input = self.max_tokens - self.reserved_output
try:
self.encoding = tiktoken.encoding_for_model("gpt-4.1")
except:
self.encoding = tiktoken.get_encoding("cl100k_base")
def count_tokens(self, text: str) -> int:
"""Đếm số tokens trong text"""
return len(self.encoding.encode(text))
def truncate_history(self, messages: list, system_prompt: str = "") -> list:
"""Truncate history giữ ngữ cảnh quan trọng nhất"""
# Tính tokens cho system prompt
system_tokens = self.count_tokens(system_prompt) if system_prompt else 0
available_for_history = self.available_input - system_tokens
# Lọc messages, giữ newest
truncated = []
current_tokens = 0
# Duyệt ngược từ messages mới
Tài nguyên liên quan