Chào các bạn, mình là Minh, Tech Lead tại một startup AI ở Hà Nội. Hôm nay mình sẽ chia sẻ hành trình 6 tháng xây dựng hệ thống Agent Memory cho dự án của team — từ việc đối mặt với chi phí OpenAI đội lên 300%, đến việc tìm ra giải pháp tối ưu và cuối cùng là migration sang HolySheep AI giúp tiết kiệm 85% chi phí API.
Tại sao Agent Memory lại quan trọng?
Trong các ứng dụng AI Agent thực tế, bạn cần hệ thống nhớ để Agent có thể:
- Ghi nhớ lịch sử hội thoại của user
- Truy xuất thông tin từ knowledge base
- Duy trì context xuyên suốt session
- Học hỏi từ các tương tác trước đó
Mình đã thử qua 3 phương án: chỉ dùng conversation history (quá tốn token), Redis đơn giản (không đủ cho knowledge retrieval), và cuối cùng là hybrid approach với PostgreSQL + Vector Search. Kết hợp với việc chuyển sang HolySheep, team mình giảm chi phí từ $2,400/tháng xuống còn $360/tháng.
Kiến trúc Agent Memory tổng quan
Hệ thống Agent Memory cần 2 tầng xử lý riêng biệt:
Tầng 1: Short-term Memory (Bộ nhớ ngắn hạn)
- Mục đích: Lưu trữ conversation context của session hiện tại
- Công nghệ: Redis, PostgreSQL (JSONB), hoặc in-memory cache
- TTL: 24-72 giờ tùy use case
- Ví dụ: Lưu lịch sử chat, temporary variables, user preferences
Tầng 2: Long-term Memory (Bộ nhớ dài hạn)
- Mục đích: Knowledge base, facts, learned information
- Công nghệ: Vector database (pgvector, Pinecone, Qdrant)
- Duration: Vĩnh viễn hoặc semi-permanent
- Ví dụ: Product docs, FAQ, user profiles, historical interactions
Triển khai Short-term Memory với Redis
Short-term memory cần tốc độ đọc/ghi nhanh. Mình dùng Redis với cấu trúc sorted set để lưu message history:
import redis
import json
from datetime import datetime, timedelta
from typing import List, Dict, Optional
class ShortTermMemory:
"""Short-term memory sử dụng Redis với TTL 24h"""
def __init__(self, redis_url: str = "redis://localhost:6379"):
self.redis = redis.from_url(redis_url)
self.default_ttl = 86400 # 24 hours
def _get_key(self, session_id: str) -> str:
return f"memory:session:{session_id}"
def add_message(self, session_id: str, role: str, content: str,
metadata: Optional[Dict] = None) -> bool:
"""Thêm message vào conversation history"""
key = self._get_key(session_id)
message = {
"role": role,
"content": content,
"timestamp": datetime.now().isoformat(),
"metadata": metadata or {}
}
# Sử dụng sorted set với timestamp làm score
self.redis.zadd(key, {json.dumps(message): datetime.now().timestamp()})
self.redis.expire(key, self.default_ttl)
return True
def get_conversation(self, session_id: str,
limit: int = 50) -> List[Dict]:
"""Lấy conversation history, sorted theo thời gian"""
key = self._get_key(session_id)
messages = self.redis.zrange(key, 0, -1)
return [json.loads(m.decode('utf-8')) for m in messages[-limit:]]
def get_recent_context(self, session_id: str,
last_n: int = 10) -> str:
"""Format recent messages thành context string cho LLM"""
messages = self.get_conversation(session_id, limit=last_n)
formatted = []
for msg in messages:
formatted.append(f"{msg['role']}: {msg['content']}")
return "\n".join(formatted)
def clear_session(self, session_id: str) -> bool:
"""