การออกแบบระบบ Memory สำหรับ AI Agent เป็นหัวใจสำคัญในการสร้าง LLM Application ที่มีความ "Intelligent" จริง ๆ Agent ที่มี Memory ดีสามารถจดจำบทสนทนาก่อนหน้า ประมวลผลความรู้เชิงลึก และตัดสินใจได้อย่างมีบริบท ในบทความนี้เราจะพาทุกท่านเจาะลึกสถาปัตยกรรม Memory 3 ชั้น พร้อมโค้ด Production-ready และ Benchmark จริง
ทำไม AI Agent ต้องมี Memory System
LLM โดยธรรมชาติเป็น Stateless Model — ไม่มีความจำระหว่าง Request แต่ใน Application จริง เราต้องการ:
- Context Continuity: Agent ต้องติดตามสิ่งที่เกิดขึ้นใน Session ปัจจุบัน
- Knowledge Persistence: ข้อมูลสำคัญต้องถูกเก็บรักษาระหว่าง Session
- Semantic Retrieval: สามารถค้นหาข้อมูลที่เกี่ยวข้องด้วยความหมาย
- Information Synthesis: รวมข้อมูลจากหลายแหล่งมาตอบคำถาม
ระบบ Memory ที่ดีจะช่วยให้ Agent มีความรู้สึกเหมือนมี "สติปัญญา" จริง ๆ
สถาปัตยกรรม Memory 3 ชั้น (Three-Tier Memory Architecture)
ชั้นที่ 1: Working Memory (Short-term)
Working Memory ใช้เก็บข้อมูลที่กำลังถูกประมวลผลใน Session ปัจจุบัน เป็นพื้นที่ทำงานชั่วคราวที่มีความเร็วสูงสุด โดยใช้ Python Dict/List หรือ Redis ก็ได้
import json
from datetime import datetime, timedelta
from typing import Any, Optional
from collections import OrderedDict
class WorkingMemory:
"""
Short-term Memory สำหรับเก็บ Conversation ใน Session ปัจจุบัน
ใช้ LRU Cache จำกัดขนาดอัตโนมัติ
"""
def __init__(self, max_items: int = 50, ttl_seconds: int = 3600):
self.max_items = max_items
self.ttl_seconds = ttl_seconds
self._cache: OrderedDict[str, dict] = OrderedDict()
self._timestamps: dict[str, datetime] = {}
def add(self, key: str, value: Any, metadata: Optional[dict] = None) -> None:
"""เพิ่ม item เข้า Working Memory"""
if key in self._cache:
self._cache.move_to_end(key)
else:
if len(self._cache) >= self.max_items:
self._cache.popitem(last=False)
self._cache[key] = {
"value": value,
"metadata": metadata or {},
"created_at": datetime.now().isoformat()
}
self._timestamps[key] = datetime.now()
def get(self, key: str) -> Optional[Any]:
"""ดึงข้อมูลจาก Working Memory"""
if key not in self._cache:
return None
# ตรวจสอบ TTL
if self._is_expired(key):
self.delete(key)
return None
self._cache.move_to_end(key)
return self._cache[key]["value"]
def add_message(self, role: str, content: str, metadata: Optional[dict] = None) -> str:
"""เพิ่มข้อความ Conversation"""
msg_id = f"msg_{len(self._cache)}_{datetime.now().timestamp()}"
self.add(msg_id, {"role": role, "content": content}, metadata)
return msg_id
def get_conversation(self, limit: int = 10) -> list[dict]:
"""ดึง Conversation ล่าสุด"""
items = list(self._cache.items())[-limit:]
return [{"role": v["value"]["role"], "content": v["value"]["content"]}
for _, v in items if "role" in v["value"]]
def _is_expired(self, key: str) -> bool:
"""ตรวจสอบว่า item หมดอายุหรือยัง"""
if key not in self._timestamps:
return True
elapsed = (datetime.now() - self._timestamps[key]).total_seconds()
return elapsed > self.ttl_seconds
def delete(self, key: str) -> None:
"""ลบ item ออกจาก Memory"""
self._cache.pop(key, None)
self._timestamps.pop(key, None)
def clear(self) -> None:
"""ล้าง Memory ทั้งหมด"""
self._cache.clear()
self._timestamps.clear()
def cleanup_expired(self) -> int:
"""ทำความสะอาด item ที่หมดอายุ"""
expired_keys = [k for k in self._cache.keys() if self._is_expired(k)]
for key in expired_keys:
self.delete(key)
return len(expired_keys)
ตัวอย่างการใช้งาน
working_mem = WorkingMemory(max_items=50, ttl_seconds=3600)
working_mem.add_message("user", "ช่วยสรุปรายงานของบริษัท ABC ให้หน่อย")
working_mem.add_message("assistant", "รายงานของบริษัท ABC แสดงรายได้รวม 50 ล้านบาท...")
conversation = working_mem.get_conversation(limit=5)
print(f"มี {len(conversation)} ข้อความใน Working Memory")ชั้นที่ 2: Semantic Memory (Long-term)
Long-term Memory ใช้เก็บข้อมูลถาวรที่ต้องการ Vector Embedding สำหรับ Semantic Search ซึ่งเหมาะกับการค้นหาด้วยความหมาย ที่นิยมใช้กันคือ FAISS, ChromaDB หรือ Qdrant
import numpy as np
from typing import List, Tuple, Optional
from dataclasses import dataclass
from datetime import datetime
import hashlib
@dataclass
class MemoryItem:
"""โครงสร้างข้อมูลสำหรับ Long-term Memory"""
id: str
content: str
embedding: np.ndarray
metadata: dict
created_at: str
access_count: int = 0
importance_score: float = 0.5
class SemanticMemory:
"""
Long-term Memory พร้อม Vector Search ใช้ FAISS สำหรับ ANN Search
รองรับ Semantic Retrieval, Importance Weighting และ Temporal Decay
"""
def __init__(
self,
dimension: int = 1536,
index_type: str = "IVF",
nlist: int = 100,
nprobe: int = 10,
use_importance_weighting: bool = True
):
self.dimension = dimension
self.use_importance = use_importance_weighting
self._items: dict[str, MemoryItem] = {}
# Initialize FAISS Index
if index_type == "IVF":
quantizer = faiss.IndexFlatIP(dimension)
self.index = faiss.IndexIVFFlat(quantizer, dimension, nlist, faiss.METRIC_INNER_PRODUCT)
elif index_type == "HNSW":
self.index = faiss.IndexHNSWFlat(dimension, 32)
else:
self.index = faiss.IndexFlatIP(dimension)
self._is_trained = False
self._id_to_idx: dict[str, int] = {}
self._idx_to_id: dict[int, str] = {}
def add(
self,
content: str,
metadata: Optional[dict] = None,
importance: float = 0.5,
embedding: Optional[np.ndarray] = None,
openai_api_key: str = None
) -> str:
"""เพิ่ม Memory Item พร้อมสร้าง Embedding"""
# สร้าง ID และ Embedding
item_id = hashlib.md5(f"{content}{datetime.now().isoformat()}".encode()).hexdigest()[:16]
if embedding is None and openai_api_key:
embedding = self._create_embedding(content, openai_api_key)
elif embedding is None:
embedding = np.random.randn(self.dimension).astype('float32')
faiss.normalize_L2(embedding)
# สร้าง Memory Item
item = MemoryItem(
id=item_id,
content=content,
embedding=embedding.reshape(1, -1),
metadata=metadata or {},
created_at=datetime.now().isoformat(),
importance_score=importance
)
# เพิ่มเข้า Index
idx = len(self._items)
self._id_to_idx[item_id] = idx
self._idx_to_id[idx] = item_id
self._items[item_id] = item
if self._is_trained:
self.index.add(np.array([embedding], dtype='float32'))
else:
self.index.add(np.array([embedding], dtype='float32'))
return item_id
def search(
self,
query: str,
top_k: int = 5,
openai_api_key: str = None,
time_decay_factor: float = 0.1
) -> List[Tuple[MemoryItem, float]]:
"""
ค้นหา Memory ด้วย Semantic Search
รวม Importance Weighting และ Temporal Decay
"""
# สร้าง Query Embedding
if openai_api_key:
query_emb = self._create_embedding(query, openai_api_key)
else:
query_emb = np.random.randn(self.dimension).astype('float32')
faiss.normalize_L2(query_emb)
# Vector Search
faiss.normalize_L2(query_emb.reshape(1, -1))
scores, indices = self.index.search(
np.array([query_emb], dtype='float32'),
min(top_k * 2, len(self._items))
)
results = []
now = datetime.now()
for score, idx in zip(scores[0], indices[0]):
if idx < 0:
continue
item_id = self._idx_to_id.get(int(idx))
if not item_id:
continue
item = self._items[item_id]
# คำนวณ Temporal Decay
age_days = (now - datetime.fromisoformat(item.created_at)).days
temporal_weight = np.exp(-time_decay_factor * age_days)
# คำนวณ Final Score
if self.use_importance:
final_score = score * (0.5 + 0.5 * item.importance_score) * temporal_weight
else:
final_score = score * temporal_weight
# อัพเดท Access Count
item.access_count += 1
results.append((item, final_score))
# เรียงลำดับตาม Final Score
results.sort(key=lambda x: x[1], reverse=True)
return results[:top_k]
def _create_embedding(self, text: str, api_key: str) -> np.ndarray:
"""สร้าง Embedding ผ่าน API"""
import requests
response = requests.post(
"https://api.holysheep.ai/v1/embeddings",
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
},
json={
"model": "text-embedding-3-small",
"input": text