Trong bài viết này, tôi sẽ chia sẻ kinh nghiệm thực chiến khi thiết kế hệ thống AI Agent từ cấp độ prototype đến production. Sau 2 năm xây dựng các agent phục vụ hàng triệu request mỗi ngày tại HolySheep AI, tôi đã rút ra được nhiều bài học quý giá về cách phối hợp hiệu quả giữa system prompt, tools và skills.
Tổng quan kiến trúc Agent Orchestration
Một AI Agent hoàn chỉnh cần ba thành phần cốt lõi hoạt động đồng bộ:
- System Prompt: Định nghĩa vai trò, ngữ cảnh và quy tắc hành vi của agent
- Tools: Các hàm có thể gọi để tương tác với external systems (API, database, file system)
- Skills: Các kỹ năng chuyên biệt được đóng gói để xử lý các tác vụ cụ thể
Điểm mấu chốt nằm ở cách ba thành phần này giao tiếp với nhau thông qua một orchestration layer. Dưới đây là kiến trúc tôi đã triển khai thành công trong production.
Xây dựng Base Agent với HolySheep AI
Trước tiên, hãy thiết lập kết nối đến HolySheep AI. Với mức giá DeepSeek V3.2 chỉ $0.42/MTok (so với $8 của GPT-4.1), bạn có thể tiết kiệm đến 85%+ chi phí khi phát triển và test agent. Ngoài ra, HolySheep hỗ trợ thanh toán qua WeChat/Alipay và đạt độ trễ trung bình dưới 50ms.
"""
AI Agent Orchestration Framework
Base Agent với System Prompt + Tools + Skills
"""
import json
import asyncio
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
from openai import AsyncOpenAI
Kết nối đến HolySheep AI
client = AsyncOpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1" # LUÔN dùng endpoint này
)
@dataclass
class Tool:
"""Định nghĩa tool có thể gọi"""
name: str
description: str
parameters: Dict[str, Any]
handler: callable
@dataclass
class Skill:
"""Định nghĩa skill chuyên biệt"""
name: str
system_prompt_addition: str
tools: List[Tool]
enabled: bool = True
@dataclass
class AgentConfig:
"""Cấu hình agent"""
model: str = "deepseek-chat"
temperature: float = 0.7
max_tokens: int = 4096
tools: List[Tool] = field(default_factory=list)
skills: List[Skill] = field(default_factory=list)
class BaseAgent:
"""Agent orchestration cơ bản"""
def __init__(self, config: AgentConfig):
self.config = config
self.conversation_history = []
def build_system_prompt(self) -> str:
"""Xây dựng system prompt động từ skills"""
base_prompt = """Bạn là một AI Agent thông minh.
Bạn có khả năng sử dụng tools để hoàn thành tác vụ.
Luôn suy nghĩ trước khi hành động và giải thích reasoning của mình.
"""
# Thêm system prompt từ các skills đang enabled
for skill in self.config.skills:
if skill.enabled:
base_prompt += f"\n\n## Skill: {skill.name}\n{skill.system_prompt_addition}"
return base_prompt
async def execute(self, user_message: str) -> Dict[str, Any]:
"""Thực thi agent với message từ user"""
self.conversation_history.append({"role": "user", "content": user_message})
# Build messages với system prompt
messages = [
{"role": "system", "content": self.build_system_prompt()}
] + self.conversation_history
# Gọi API
response = await client.chat.completions.create(
model=self.config.model,
messages=messages,
temperature=self.config.temperature,
max_tokens=self.config.max_tokens,
tools=self._build_tools_schema()
)
return self._process_response(response)
def _build_tools_schema(self) -> List[Dict]:
"""Build OpenAI tools schema từ config"""
tools = []
for tool in self.config.tools:
tools.append({
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": tool.parameters
}
})
# Thêm tools từ skills
for skill in self.config.skills:
if skill.enabled:
for tool in skill.tools:
tools.append({
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": tool.parameters
}
})
return tools
def _process_response(self, response) -> Dict[str, Any]:
"""Xử lý response từ API"""
choice = response.choices[0]
result = {
"content": choice.message.content,
"tool_calls": []
}
if choice.message.tool_calls:
for tool_call in choice.message.tool_calls:
result["tool_calls"].append({
"id": tool_call.id,
"name": tool_call.function.name,
"arguments": json.loads(tool_call.function.arguments)
})
return result
print("✅ Agent framework khởi tạo thành công")Triển khai Tools với Error Handling
Tools là cầu nối giữa agent và thế giới thực. Một tool được thiết kế tốt cần có error handling chặt chẽ, retry logic và graceful degradation.
"""
Tool Implementation với Robust Error Handling
"""
import httpx
from typing import Any
import asyncio
from functools import wraps
def retry_on_failure(max_retries: int = 3, backoff: float = 1.0):
"""Decorator cho retry logic với exponential backoff"""
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(max_retries):
try:
return await func(*args, **kwargs)
except Exception as e:
last_exception = e
if attempt < max_retries - 1:
await asyncio.sleep(backoff * (2 ** attempt))
raise last_exception
return wrapper
return decorator
class ToolRegistry:
"""Registry quản lý tất cả tools"""
def __init__(self):
self._tools: Dict[str, Tool] = {}
def register(self, tool: Tool):
self._tools[tool.name] = tool
async def execute(self, tool_name: str, arguments: Dict[str, Any]) -> Any:
if tool_name not in self._tools:
raise ValueError(f"Tool '{tool_name}' không tồn tại")
tool = self._tools[tool_name]
return await tool.handler(arguments)
Ví dụ: Tool tra cứu thông tin sản phẩm
async def get_product_info_handler(args: Dict) -> Dict:
"""Handler cho tool tra cứu sản phẩm"""
product_id = args.get("product_id")
@retry_on_failure(max_retries=3, backoff=0.5)
async def fetch():
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.get(
f"https://api.example.com/products/{product_id}",
headers={"Authorization": "Bearer YOUR_API_KEY"}
)
response.raise_for_status()
return response.json()
try:
data = await fetch()
return {
"success": True,
"data": {
"id": data.get("id"),
"name": data.get("name"),
"price": data.get("price"),
"stock": data.get("stock_quantity")
}
}
except httpx.TimeoutException:
return {"success": False, "error": "Request timeout"}
except httpx.HTTPStatusError as e:
return {"success": False, "error": f"HTTP {e.response.status_code}"}
except Exception as e:
return {"success": False, "error": str(e)}
Đăng ký tool
registry = ToolRegistry()
registry.register(Tool(
name="get_product_info",
description="Lấy thông tin sản phẩm theo ID. Trả về tên, giá và số lượng tồn kho.",
parameters={
"type": "object",
"properties": {
"product_id": {"type": "string", "description": "ID của sản phẩm"}
},
"required": ["product_id"]
},
handler=get_product_info_handler
))
print("✅ Tool registry đã sẵn sàng")Skills Orchestration - Điều phối đa tác vụ
Trong thực tế, một agent cần xử lý nhiều loại tác vụ khác nhau. Skills cho phép agent "mặc" different "personas" tùy theo yêu cầu. Dưới đây là hệ thống orchestration nâng cao với skill routing thông minh.
"""
Advanced Skill Orchestration với Task Routing
"""
from enum import Enum
from typing import List, Optional
import re
class TaskType(Enum):
"""Phân loại tác vụ"""
GENERAL = "general"
DATA_ANALYSIS = "data_analysis"
CODE_GENERATION = "code_generation"
CUSTOMER_SUPPORT = "customer_support"
RESEARCH = "research"
class SkillOrchestrator:
"""Orchestrator quản lý skills và routing"""
def __init__(self):
self.skills: Dict[TaskType, Skill] = {}
self._init_default_skills()
def _init_default_skills(self):
"""Khởi tạo skills mặc định"""
# Skill: Data Analysis
self.skills[TaskType.DATA_ANALYSIS] = Skill(
name="Data Analyst",
system_prompt_addition="""Bạn là chuyên gia phân tích dữ liệu.
Khi được yêu cầu phân tích:
1. Xác định loại dữ liệu và nguồn gốc
2. Áp dụng phương pháp thống kê phù hợp
3. Trình bày kết quả với visualization
4. Đưa ra insights có thể hành động
Luôn sử dụng format table cho dữ liệu dạng bảng.""",
tools=[],
enabled=True
)
# Skill: Code Generation
self.skills[TaskType.CODE_GENERATION] = Skill(
name="Senior Developer",
system_prompt_addition="""Bạn là senior developer với 10 năm kinh nghiệm.
Nguyên tắc code:
1. Clean Code: tên biến có ý nghĩa, hàm ngắn gọn
2. Type hints đầy đủ cho Python
3. Docstring cho public APIs
4. Error handling chặt chẽ
5. Unit tests cho critical functions
Luôn giải thích WHY đằng sau design decisions.""",
tools=[],
enabled=True
)
# Skill: Customer Support
self.skills[TaskType.CUSTOMER_SUPPORT] = Skill(
name="Support Agent",
system_prompt_addition="""Bạn là agent chăm sóc khách hàng chuyên nghiệp.
Nguyên tắc:
1. Empathetic: Thể hiện sự thấu hiểu
2. Professional: Ngôn ngữ lịch sự, chuẩn mực
3. Solution-oriented: Tập trung giải pháp
4. Escalation: Biết khi nào cần chuyển cấp
Không bao giờ đổ lỗi cho khách hàng.""",
tools=[],
enabled=True
)
def classify_task(self, message: str) -> TaskType:
"""Phân loại tác vụ từ message"""
message_lower = message.lower()
# Keywords mapping
patterns = {
TaskType.DATA_ANALYSIS: [
r"phân tích", r"thống kê", r"biểu đồ", r"dashboard",
r"analytics", r"insights", r"xu hướng"
],
TaskType.CODE_GENERATION: [
r"viết code", r"function", r"class", r"api",
r"implement", r"debug", r"lỗi", r"bug"
],
TaskType.CUSTOMER_SUPPORT: [
r"hỗ trợ", r"khiếu nại", r"hoàn tiền", r"refund",
r"problem", r"issue", r"complaint"
],
TaskType.RESEARCH: [
r"tìm hiểu", r"nghiên cứu", r"so sánh", r"đánh giá",
r"research", r"compare", r"review"
]
}
scores = {}
for task_type, keywords in patterns.items():
score = sum(1 for pattern in keywords if re.search(pattern, message_lower))
scores[task_type] = score
if max(scores.values()) == 0:
return TaskType.GENERAL
return max(scores, key=scores.get)
def get_active_prompt(self, message: str) -> str:
"""Build prompt với skill phù hợp"""
task_type = self.classify_task(message)
skill = self.skills.get(task_type)
base = "Bạn là một AI Assistant thông minh và hữu ích.\n"
if skill and skill.enabled:
base += f"\n{skill.system_prompt_addition}"
return base
Demo
orchestrator = SkillOrchestrator()
test_messages = [
"Phân tích doanh thu tháng này so với tháng trước",
"Viết function sort array trong Python",
"Tôi muốn khiếu nại về đơn hàng bị trễ"
]
for msg in test_messages:
task = orchestrator.classify_task(msg)
print(f"Message: '{msg[:30]}...' -> Task: {task.value}")Concurrent Execution với Async Control
Một trong những thách thức lớn nhất khi xây dựng multi-agent system là kiểm soát đồng thời. Tôi đã triển khai semaphore-based concurrency control để tránh quá tải hệ thống.
"""
Concurrent Agent Execution với Semaphore Control
"""
import asyncio
from typing import List, Dict, Any
from dataclasses import dataclass
import time
@dataclass
class ExecutionResult:
"""Kết quả thực thi agent"""
agent_id: str
success: bool
result: Any
duration_ms: float
error: Optional[str] = None
class ConcurrentAgentRunner:
"""Runner hỗ trợ thực thi đồng thời nhiều agents"""
def __init__(self, max_concurrent: int = 5):
self.semaphore = asyncio.Semaphore(max_concurrent)
self.max_concurrent = max_concurrent
async def run_single(
self,
agent: BaseAgent,
message: str,
agent_id: str
) -> ExecutionResult:
"""Thực thi một agent với semaphore control"""
async with self.semaphore:
start_time = time.time()
try:
result = await agent.execute(message)
duration = (time.time() - start_time) * 1000
return ExecutionResult(
agent_id=agent_id,
success=True,
result=result,
duration_ms=round(duration, 2)
)
except Exception as e:
duration = (time.time() - start_time) * 1000
return ExecutionResult(
agent_id=agent_id,
success=False,
result=None,
duration_ms=round(duration, 2),
error=str(e)
)
async def run_batch(
self,
agents: List[tuple], # List of (agent, message, agent_id)
timeout: float = 30.0
) -> List[ExecutionResult]:
"""Thực thi batch với timeout"""
tasks = [
self.run_single(agent, message, agent_id)
for agent, message, agent_id in agents
]
try:
results = await asyncio.wait_for(
asyncio.gather(*tasks, return_exceptions=True),
timeout=timeout
)
# Convert exceptions to ExecutionResult
processed_results = []
for i, result in enumerate(results):
if isinstance(result, Exception):
processed_results.append(ExecutionResult(
agent_id=f"agent_{i}",
success=False,
result=None,
duration_ms=0,
error=str(result)
))
else:
processed_results.append(result)
return processed_results
except asyncio.TimeoutError:
return [ExecutionResult(
agent_id="timeout",
success=False,
result=None,
duration_ms=0,
error=f"Batch timeout sau {timeout}s"
)]
Benchmark function
async def benchmark_concurrency():
"""Benchmark để so sánh hiệu suất"""
runner = ConcurrentAgentRunner(max_concurrent=3)
# Tạo mock agents cho demo
config = AgentConfig(model="deepseek-chat")
agents = [BaseAgent(config) for _ in range(5)]
test_tasks = [
(agents[i % 5], f"Tác vụ {i+1}", f"agent_{i+1}")
for i in range(10)
]
start = time.time()
results = await runner.run_batch(test_tasks, timeout=60.0)
total_time = time.time() - start
# Thống kê
successful = sum(1 for r in results if r.success)
failed = len(results) - successful
avg_duration = sum(r.duration_ms for r in results) / len(results) if results else 0
print(f"📊 Benchmark Results:")
print(f" Tổng tasks: {len(test_tasks)}")
print(f" Thành công: {successful}")
print(f" Thất bại: {failed}")
print(f" Max concurrent: {runner.max_concurrent}")
print(f" Tổng thời gian: {total_time:.2f}s")
print(f" Avg duration/task: {avg_duration:.2f}ms")
Chạy benchmark
asyncio.run(benchmark_concurrency())Cost Optimization với HolySheep AI
Đây là phần quan trọng mà nhiều kỹ sư bỏ qua. Tôi đã tiết kiệm được 85%+ chi phí API bằng cách áp dụng chiến lược tiered model selection trên HolySheep AI.
- DeepSeek V3.2 ($0.42/MTok): Cho các tác vụ đơn giản, classification, routing
- Gemini 2.5 Flash ($2.50/MTok): Cho tác vụ trung bình cần context dài
- Claude Sonnet 4.5 ($15/MTok): Cho tác vụ phức tạp cần reasoning cao
- GPT-4.1 ($8/MTok): Khi cần compatibility với existing codebase
"""
Tiered Model Selection cho Cost Optimization
"""
from enum import Enum
from typing import Optional
import tiktoken
class ModelTier(Enum):
"""Các tier model với chi phí khác nhau"""
BUDGET = "budget" # DeepSeek V3.2
STANDARD = "standard" # Gemini 2.5 Flash
PREMIUM = "premium" # Claude Sonnet 4.5
ENTERPRISE = "enterprise" # GPT-4.1
MODEL_CONFIG = {
ModelTier.BUDGET: {
"model": "deepseek-chat",
"cost_per_1k_tokens": 0.00042, # $0.42/MTok
"max_tokens": 8192,
"use_cases": ["classification", "routing", "simple_qa", "embedding"]
},
ModelTier.STANDARD: {
"model": "gemini-2.0-flash",
"cost_per_1k_tokens": 0.00250, # $2.50/MTok
"max_tokens": 32768,
"use_cases": ["summarization", "translation", "moderation"]
},
ModelTier.PREMIUM: {
"model": "claude-sonnet-4-20250514",
"cost_per_1k_tokens": 0.015, # $15/MTok
"max_tokens": 200000,
"use_cases": ["complex_reasoning", "code_generation", "analysis"]
},
ModelTier.ENTERPRISE: {
"model": "gpt-4.1",
"cost_per_1k_tokens": 0.008, # $8/MTok
"max_tokens": 128000,
"use_cases": ["compatibility", "fine_tuned_tasks"]
}
}
class CostAwareRouter:
"""Router thông minh với cost awareness"""
def __init__(self, client: AsyncOpenAI):
self.client = client
self.encoding = tiktoken.encoding_for_model("gpt-4")
self.cost_tracker = {"total_tokens": 0, "total_cost": 0}
def estimate_cost(self, text: str, tier: ModelTier) -> float:
"""Ước tính chi phí cho một text"""
tokens = len(self.encoding.encode(text))
config = MODEL_CONFIG[tier]
cost = (tokens / 1000) * config["cost_per_1k_tokens"]
return cost
def select_tier(self, task_complexity: str, context_length: int) -> ModelTier:
"""Chọn tier phù hợp dựa trên độ phức tạp"""
# Logic đơn giản - có thể mở rộng với ML model
if task_complexity == "simple" and context_length < 1000:
return ModelTier.BUDGET
elif task_complexity == "medium" and context_length < 8000:
return ModelTier.STANDARD
elif task_complexity == "complex":
return ModelTier.PREMIUM
else:
return ModelTier.STANDARD
async def execute_with_tier(
self,
messages: List[Dict],
tier: ModelTier,
save_cost: bool = True
) -> tuple:
"""Execute với tier đã chọn"""
config = MODEL_CONFIG[tier]
# Tính tokens để track chi phí
total_text = " ".join(m["content"] for m in messages if "content" in m)
input_tokens = len(self.encoding.encode(total_text))
response = await self.client.chat.completions.create(
model=config["model"],
messages=messages,
max_tokens=config["max_tokens"] // 4
)
output_tokens = response.usage.completion_tokens
total_tokens = input_tokens + output_tokens
cost = (total_tokens / 1000) * config["cost_per_1k_tokens"]
if save_cost:
self.cost_tracker["total_tokens"] += total_tokens
self.cost_tracker["total_cost"] += cost
return response, {
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"cost_usd": cost,
"tier": tier.value
}
def get_cost_report(self) -> Dict:
"""Báo cáo chi phí"""
return {
**self.cost_tracker,
"estimated_savings_vs_gpt4": (
self.cost_tracker["total_cost"] * 19 # So với GPT-4
)
}
Demo cost comparison
router = CostAwareRouter(client)
sample_text = "Hãy phân tích dữ liệu bán hàng tháng này và đưa ra insights"
print("💰 Cost Comparison:")
for tier in ModelTier:
cost = router.estimate_cost(sample_text, tier)
config = MODEL_CONFIG[tier]
print(f" {tier.value:12}: ${cost:.6f} ({config['model']})")
Giả sử xử lý 1000 requests với DeepSeek thay vì GPT-4
saved = 1000 * router.estimate_cost(sample_text, ModelTier.ENTERPRISE)
actual = 1000 * router.estimate_cost(sample_text, ModelTier.BUDGET)
print(f"\n📈 Với 1000 requests: GPT-4 = ${saved:.2f}, DeepSeek = ${actual:.2f}")
print(f" Tiết kiệm: ${saved - actual:.2f} ({(1 - actual/saved)*100:.1f}%)")Lỗi thường gặp và cách khắc phục
1. Lỗi "Invalid API Key" hoặc Authentication Error
Nguyên nhân: API key không đúng format hoặc chưa được set đúng biến môi trường.
# ❌ SAI - Key bị thiếu hoặc sai
client = AsyncOpenAI(
api_key="sk-xxxxx", # Key không hợp lệ
base_url="https://api.holysheep.ai/v1"
)
✅ ĐÚNG - Sử dụng environment variable
import os
from dotenv import load_dotenv
load_dotenv() # Load .env file
API_KEY = os.getenv("HOLYSHEEP_API_KEY")
if not API_KEY:
raise ValueError("HOLYSHEEP_API_KEY không được tìm thấy trong .env")
client = AsyncOpenAI(
api_key=API_KEY,
base_url="https://api.holysheep.ai/v1"
)
Verify connection
async def verify_connection():
try:
await client.models.list()
print("✅ Kết nối HolySheep AI thành công")
except Exception as e:
print(f"❌ Lỗi kết nối: {e}")2. Lỗi "Tool calling loop detected" - Agent bị stuck
Nguyên nhân: Agent liên tục gọi tool mà không có kết quả, tạo infinite loop.
"""
Giải quyết: Thêm max iterations và tool call tracking
"""
MAX_TOOL_ITERATIONS = 10
class SafeAgent(BaseAgent):
"""Agent với protection chống infinite loop"""
def __init__(self, config: AgentConfig):
super().__init__(config)
self.tool_call_history = []
async def execute_safe(self, user_message: str) -> Dict[str, Any]:
"""Execute với safety checks"""
self.conversation_history.append(
{"role": "user", "content": user_message}
)
iteration = 0
final_response = None
while iteration < MAX_TOOL_ITERATIONS:
response = await self._call_llm()
# Nếu không có tool call, đây là response cuối
if not response.get("tool_calls"):
final_response = response
break
# Kiểm tra duplicate calls
current_calls = tuple(
tc["name"] for tc in response.get("tool_calls", [])
)
if current_calls in self.tool_call_history[-3:]:
# Detected loop - stop và return reasoning
final_response = {
"content": (
"Agent đã phát hiện vòng lặp tool calls. "
"Kết thúc để tránh infinite loop."
),
"tool_calls": [],
"loop_detected": True
}
break
self.tool_call_history.append(current_calls)
# Execute tools
tool_results = await self._execute_tools(
response.get("tool_calls", [])
)
# Add results to conversation
self.conversation_history.extend(tool_results)
iteration += 1
return final_response or {
"content": "Đạt max iterations",
"tool_calls": []
}
Kết quả: Agent sẽ không bao giờ vượt quá 10 tool calls
3. Lỗi Context Overflow khi xử lý long conversations
Nguyên nhân: Conversation history quá dài vượt quá context window của model.
"""
Giải quyết: Smart context truncation với summarization
"""
MAX_CONTEXT_TOKENS = {
"deepseek-chat": 8192,
"gemini-2.0-flash": 32768,
"claude-sonnet-4-20250514": 200000
}
class ContextManager:
"""Quản lý context với truncation thông minh"""
def __init__(self, model: str):
self.max_tokens = MAX_CONTEXT_TOKENS.get(model, 4096)
self.encoding = tiktoken.encoding_for_model("gpt-4")
def count_tokens(self, messages: List[Dict]) -> int:
"""Đếm tokens trong messages"""
text = ""
for msg in messages:
if isinstance(msg, dict) and "content" in msg:
text += msg["content"]
return len(self.encoding.encode(text))
def truncate_history(
self,
messages: List[Dict],
preserve_system: bool = True
) -> List[Dict]:
"""Truncate history giữ lại system prompt và messages gần nhất"""
if self.count_tokens(messages) <= self.max_tokens:
return messages
result = []
system_msg = None
# Tách system message
if messages and messages[0].get("role") == "system":
system_msg = messages[0]
messages = messages[1:]
# Giữ lại messages gần nhất cho đến khi fit
for msg in reversed(messages):
msg_tokens = len(self.encoding.encode(msg.get("content", "")))
current_total = self.count_tokens(result)
if current_total + msg_tokens + 500 <= self.max_tokens:
result.insert(0, msg)
else:
# Tạo summary thay thế
summary = {
"role": "system",
"content": (
f"[{len(result)} messages đã được truncated "
f"để tiết kiệm context]"
)
}
result.insert(0, summary)
break
if preserve_system and system_msg:
result.insert(0, system_msg)
return result
async def summarize_old_messages(
self,
client: AsyncOpenAI,
messages: List[Dict]
) -> List[Dict]:
"""Summarize messages cũ để tiết kiệm tokens"""
# Lấy 10 messages đầu tiên (không phải system)
old_messages = [m for m in messages if m.get("role") != "system"][:10]
if len(old_messages) < 5:
return messages
# Tạo summary prompt
summary_prompt = f"""Hãy tóm tắt ngắn gọn cuộc trò chuyện sau:
{chr(10).join(f'{m["role"]}: {m["content"]}' for m in old_messages)}
Format: JSON với keys: summary, key_points (array)"""
response = await client.chat.completions.create(
model="deepseek-chat",
messages=[{"role": "user", "content": summary_prompt}],
max_tokens=500
)
summary_text = response.choices[0].message.content
# Replace old messages với summary
system = [messages[0]] if messages and messages[0].get("role") == "system" else []
recent = [m for m in messages if m.get("role") != "system"][-10:]
return system + [
{"role": "system", "content": f"[Earlier conversation summary: {