ในโลกของ AI Agent ที่กำลังเติบโตอย่างรวดเร็ว หน่วยความจำคือหัวใจสำคัญที่ทำให้ตัวแทน AI สามารถจดจำบริบท เรียนรู้จากปฏิสัมพันธ์ก่อนหน้า และให้คำตอบที่สอดคล้องกับสถานการณ์ ในบทความนี้ ผมจะพาคุณไปสำรวจรูปแบบการออกแบบโมดูลหน่วยความจำที่ใช้งานได้จริงใน production พร้อมตัวอย่างโค้ดที่คุณสามารถนำไปประยุกต์ใช้ได้ทันที
ทำไมต้องมี Memory Module สำหรับ AI Agent
จากประสบการณ์ของผมในการพัฒนา AI Agent มากว่า 3 ปี ปัญหาที่พบบ่อยที่สุดคือ Agent ไม่สามารถจดจำสิ่งที่เกิดขึ้นก่อนหน้าได้ ทำให้ต้องอธิบายบริบทซ้ำๆ เหมือนกับคุยกับคนที่สูญเสียความจำระยะสั้นทุกครั้งที่ถามคำถามใหม่
กรณีศึกษา: ระบบตอบคำถามลูกค้าอีคอมเมิร์ซ
สมมติว่าคุณกำลังสร้างแชทบอทสำหรับร้านค้าออนไลน์ที่มีสินค้าหลายพันรายการ ลูกค้าถามเรื่องนโยบายการคืนสินค้า จากนั้นถามเรื่องสินค้าที่ต้องการสั่งซื้อ แล้วถามเรื่องการชำระเงินตามลำดับ หากไม่มี Memory Module ที่ดี Agent จะไม่รู้ว่าลูกค้าคนเดียวกันกำลังสนทนาเรื่องอะไร และอาจให้ข้อมูลที่ขัดแย้งกัน
รูปแบบหน่วยความจำหลัก 4 แบบ
1. Buffer Memory (ความจำระยะสั้น)
ความจำแบบนี้เก็บเฉพาะปฏิสัมพันธ์ล่าสุดในรอบการสนทนาปัจจุบัน เหมาะสำหรับการสนทนาที่ไม่ต้องการบริบทย้อนหลังมาก
"""
Buffer Memory Implementation สำหรับ AI Agent
เก็บเฉพาะข้อความล่าสุดในรอบการสนทนา
"""
from typing import List, Dict, Any
from collections import deque
import json
class BufferMemory:
"""ความจำระยะสั้นแบบ FIFO - เก็บเฉพาะ N ข้อความล่าสุด"""
def __init__(self, max_messages: int = 10):
self.max_messages = max_messages
self.buffer = deque(maxlen=max_messages)
def add_message(self, role: str, content: str, metadata: Dict = None):
"""เพิ่มข้อความเข้าสู่ buffer"""
message = {
"role": role,
"content": content,
"metadata": metadata or {}
}
self.buffer.append(message)
def get_messages(self) -> List[Dict[str, Any]]:
"""ดึงข้อความทั้งหมดใน buffer"""
return list(self.buffer)
def get_context_window(self) -> str:
"""สร้าง context string สำหรับส่งให้ LLM"""
context_parts = []
for msg in self.buffer:
role_label = "ผู้ใช้" if msg["role"] == "user" else "AI"
context_parts.append(f"{role_label}: {msg['content']}")
return "\n".join(context_parts)
def clear(self):
"""ล้าง buffer ทั้งหมด"""
self.buffer.clear()
def to_dict(self) -> Dict:
return {
"type": "buffer",
"max_messages": self.max_messages,
"messages": list(self.buffer)
}
ตัวอย่างการใช้งาน
memory = BufferMemory(max_messages=6)
memory.add_message("user", "สินค้านี้มีสีอะไรบ้าง?", {"product_id": "SKU-001"})
memory.add_message("assistant", "สินค้านี้มีสีดำ ขาว และน้ำเงินครับ", {"timestamp": "2025-01-20T10:30:00Z"})
memory.add_message("user", "จัดส่งกี่วัน?", {"product_id": "SKU-001"})
memory.add_message("assistant", "จัดส่งภายใน 3-5 วันทำการครับ", {"timestamp": "2025-01-20T10:31:00Z"})
print(memory.get_context_window())
print(f"จำนวนข้อความใน buffer: {len(memory.buffer)}")2. Summary Memory (ความจำแบบสรุป)
แทนที่จะเก็บข้อความทุกข้อความ เราจะใช้ LLM สรุปเนื้อหาสำคัญเป็นระยะ ลดการใช้ token และโฟกัสเฉพาะข้อมูลที่จำเป็น
"""
Summary Memory - ใช้ LLM สรุปความจำเป็นระยะ
ประหยัด token และโฟกัสเฉพาะข้อมูลสำคัญ
"""
import requests
import json
from typing import List, Dict, Optional
class SummaryMemory:
"""ความจำแบบสรุปที่ใช้ AI จับสาระสำคัญ"""
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.short_term = [] # ข้อความดิบล่าสุด
self.long_term = [] # ความจำที่สรุปแล้ว
self.summary_threshold = 8 # สรุปทุก 8 ข้อความ
def _call_llm_for_summary(self, messages: List[Dict]) -> str:
"""เรียก HolySheep API เพื่อสรุปข้อความ"""
prompt = """คุณคือผู้ช่วย AI ที่ต้องสรุปเนื้อหาการสนทนาต่อไปนี้:
เป็นข้อความสั้นไม่เกิน 3 ประโยค ระบุเฉพาะข้อมูลสำคัญที่ต้องจำ:
- ความต้องการ/ปัญหาของผู้ใช้
- ข้อมูลสินค้าหรือบริการที่กล่าวถึง
- ข้อตกลงหรือคำตอบที่ให้ไว้
การสนทนา:
"""
for msg in messages:
role = "ผู้ใช้" if msg["role"] == "user" else "AI"
prompt += f"{role}: {msg['content']}\n"
response = requests.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": "gpt-4o-mini",
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 200,
"temperature": 0.3
}
)
return response.json()["choices"][0]["message"]["content"]
def add_message(self, role: str, content: str, metadata: Dict = None):
"""เพิ่มข้อความและสรุปเมื่อถึง threshold"""
self.short_term.append({
"role": role,
"content": content,
"metadata": metadata or {}
})
# สรุปเมื่อมีข้อความครบ threshold
if len(self.short_term) >= self.summary_threshold:
summary = self._call_llm_for_summary(self.short_term)
self.long_term.append({
"summary": summary,
"message_count": len(self.short_term),
"timestamp": metadata.get("timestamp") if metadata else None
})
self.short_term = [] # ล้าง short-term หลังสรุป
def get_full_context(self) -> str:
"""ดึงบริบทความจำทั้งหมด"""
context = "## ความจำระยะยาว:\n"
for item in self.long_term:
context += f"- {item['summary']}\n"
context += "\n## การสนทนาล่าสุด:\n"
for msg in self.short_term:
role = "ผู้ใช้" if msg["role"] == "user" else "AI"
context += f"{role}: {msg['content']}\n"
return context
ตัวอย่างการใช้งาน
memory = SummaryMemory(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
เพิ่มข้อความหลายรายการ
for i in range(10):
memory.add_message("user", f"สอบถามเรื่องสินค้ารายการที่ {i+1}", {"product_id": f"SKU-{i:03d}"})
print("บริบทความจำทั้งหมด:")
print(memory.get_full_context())3. Vector Memory (ความจำแบบเวกเตอร์)
ใช้ embedding เพื่อค้นหาความจำที่เกี่ยวข้องกับคำถามปัจจุบัน เหมาะสำหรับ RAG (Retrieval-Augmented Generation) ที่ต้องการดึงข้อมูลจากฐานความรู้ขนาดใหญ่
"""
Vector Memory - ค้นหาความจำที่เกี่ยวข้องด้วย semantic search
ใช้ embedding เพื่อหาข้อมูลที่สอดคล้องกับคำถามปัจจุบัน
"""
import requests
import numpy as np
from typing import List, Dict, Tuple, Optional
from datetime import datetime
class VectorMemory:
"""ความจำแบบ vector search สำหรับ semantic retrieval"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1", embedding_model: str = "text-embedding-3-small"):
self.api_key = api_key
self.base_url = base_url
self.embedding_model = embedding_model
self.memory_store = [] # List of (embedding, metadata)
def _get_embedding(self, text: str) -> np.ndarray:
"""สร้าง embedding vector จาก HolySheep API"""
response = requests.post(
f"{self.base_url}/embeddings",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": self.embedding_model,
"input": text
}
)
data = response.json()
return np.array(data["data"][0]["embedding"])
def _cosine_similarity(self, vec1: np.ndarray, vec2: np.ndarray) -> float:
"""คำนวณ cosine similarity ระหว่าง vector สองตัว"""
dot_product = np.dot(vec1, vec2)
norm_product = np.linalg.norm(vec1) * np.linalg.norm(vec2)
return dot_product / norm_product if norm_product > 0 else 0
def add_memory(self, content: str, metadata: Dict):
"""เพิ่มความจำพร้อม embedding"""
embedding = self._get_embedding(content)
self.memory_store.append({
"embedding": embedding,
"content": content,
"metadata": metadata,
"created_at": datetime.now().isoformat()
})
def search(self, query: str, top_k: int = 5, threshold: float = 0.7) -> List[Dict]:
"""ค้นหาความจำที่เกี่ยวข้องกับ query"""
query_embedding = self._get_embedding(query)
results = []
for item in self.memory_store:
similarity = self._cosine_similarity(query_embedding, item["embedding"])
if similarity >= threshold:
results.append({
"content": item["content"],
"metadata": item["metadata"],
"similarity": float(similarity),
"created_at": item["created_at"]
})
# เรียงลำดับตาม similarity และดึง top_k
results.sort(key=lambda x: x["similarity"], reverse=True)
return results[:top_k]
def get_context_for_query(self, query: str, max_items: int = 3) -> str:
"""สร้าง context string สำหรับ query"""
relevant_memories = self.search(query, top_k=max_items)
if not relevant_memories:
return "ไม่พบความจำที่เกี่ยวข้อง"
context = "## ข้อมูลจากความจำที่เกี่ยวข้อง:\n"
for i, mem in enumerate(relevant_memories, 1):
context += f"{i}. {mem['content']} (ความเหมือน: {mem['similarity']:.2%})\n"
return context
ตัวอย่างการใช้งาน - ระบบ RAG สำหรับฐานความรู้สินค้า
memory = VectorMemory(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
เพิ่มข้อมูลสินค้าลงในความจำ vector
products = [
("รองเท้าผ้าใบ Nike Air Max สีดำ ขนาด 42 ราคา 4,500 บาท มีโปรโมชั่นลด 15%", {"type": "product", "category": "รองเท้า"}),
("เสื้อยืด Adidas ผ้า cotton 100% ราคา 890 บาท มีสีขาว ดำ เทา", {"type": "product", "category": "เสื้อผ้า"}),
("นโยบายการคืนสินค้า: คืนได้ภายใน 30 วัน ไม่ต้องกล่อง สินค้าต้องไม่ผ่านการใช้", {"type": "policy", "category": "การคืนสินค้า"}),
("วิธีการชำระเงิน: บัตรเครดิต พร้อมเพย์ ชำระปลายทาง ได้ทุกช่องทาง", {"type": "payment", "category": "การชำระเงิน"}),
]
for content, metadata in products:
memory.add_memory(content, metadata)
ค้นหาความจำที่เกี่ยวข้อง
query = "รองเท้าสีดำราคาเท่าไหร่"
context = memory.get_context_for_query(query, max_items=2)
print(f"Query: {query}\n")
print(f"Context:\n{context}")
query2 = "ถ้าไม่พอใจสินค้าคืนยังไง"
context2 = memory.get_context_for_query(query2, max_items=1)
print(f"\nQuery: {query2}\n")
print(f"Context:\n{context2}")สถาปัตยกรรม Memory Module แบบครบวงจร
สำหรับ production system ที่ต้องรองรับ use cases หลากหลาย ผมแนะนำให้ใช้สถาปัตยกรรมแบบ Layered Memory ที่รวมทุกรูปแบบเข้าด้วยกัน
"""
Layered Memory Architecture - รวม Buffer + Summary + Vector เป็นหนึ่งเดียว
สถาปัตยกรรม production-ready สำหรับ AI Agent ทุกประเภท
"""
import requests
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, field
from datetime import datetime
from collections import deque
@dataclass
class MemoryConfig:
"""การตั้งค่าสำหรับ Memory Module"""
api_key: str
base_url: str = "https://api.holysheep.ai/v1"
buffer_size: int = 10
summary_threshold: int = 8
vector_top_k: int = 3
vector_threshold: float = 0.7
embedding_model: str = "text-embedding-3-small"
llm_model: str = "gpt-4o-mini"
class HybridMemorySystem:
"""
ระบบหน่วยความจำแบบลำดับชั้นที่รวม:
- Buffer: ความจำระยะสั้น (การสนทนาปัจจุบัน)
- Summary: ความจำสรุป (บริบทย้อนหลัง)
- Vector: ความจำเวกเตอร์ (ค้นหาตามความหมาย)
"""
def __init__(self, config: MemoryConfig):
self.config = config
# Layer 1: Buffer Memory
self.buffer = deque(maxlen=config.buffer_size)
# Layer 2: Summary Memory
self.summaries: List[Dict] = []
self.pending_messages: List[Dict] = []
# Layer 3: Vector Memory
self.vector_store: List[Dict] = []
def _call_llm(self, prompt: str, max_tokens: int = 300) -> str:
"""เรียก HolySheep API สำหรับ LLM operations"""
response = requests.post(
f"{self.config.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.config.api_key}",
"Content-Type": "application/json"
},
json={
"model": self.config.llm_model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": max_tokens,
"temperature": 0.3
}
)
return response.json()["choices"][0]["message"]["content"]
def _get_embedding(self, text: str) -> List[float]:
"""สร้าง embedding vector"""
response = requests.post(
f"{self.config.base_url}/embeddings",
headers={
"Authorization": f"Bearer {self.config.api_key}",
"Content-Type": "application/json"
},
json={
"model": self.config.embedding_model,
"input": text
}
)
return response.json()["data"][0]["embedding"]
def add_interaction(self, user_message: str, assistant_message: str, metadata: Dict = None):
"""เพิ่มปฏิสัมพันธ์ใหม่เข้าสู่ทุก layer"""
timestamp = datetime.now().isoformat()
# Layer 1: เพิ่มลง Buffer
self.buffer.append({
"user": user_message,
"assistant": assistant_message,
"timestamp": timestamp,
"metadata": metadata or {}
})
# Layer 2: เพิ่มลง Pending และสรุปเมื่อครบ threshold
self.pending_messages.append({"user": user_message, "assistant": assistant_message})
if len(self.pending_messages) >= self.config.summary_threshold:
self._create_summary()
def _create_summary(self):
"""สร้าง summary จาก pending messages"""
prompt = f"""สรุปการสนทนาต่อไปนี้เป็นภาษาไทย 2-3 ประโยค:
เฉพาะข้อมูลสำคัญเท่านั้น (ความต้องการ ปัญหา ข้อตกลง)
"""
for i, msg in enumerate(self.pending_messages, 1):
prompt += f"{i}. ถาม: {msg['user']}\n ตอบ: {msg['assistant']}\n"
summary_text = self._call_llm(prompt, max_tokens=150)
self.summaries.append({
"summary": summary_text,
"message_count": len(self.pending_messages),
"created_at": datetime.now().isoformat()
})
self.pending_messages = []
def add_to_vector_store(self, content: str, metadata: Dict):
"""เพิ่มข้อมูลลง vector store สำหรับ RAG"""
embedding = self._get_embedding(content)
self.vector_store.append({
"embedding": embedding,
"content": content,
"metadata": metadata,
"created_at": datetime.now().isoformat()
})
def _semantic_search(self, query: str, top_k: int) -> List[Dict]:
"""ค้นหาแบบ semantic ใน vector store"""
if not self.vector_store:
return []
query_embedding = self._get_embedding(query)
results = []
import numpy as np
for item in self.vector_store:
similarity = np.dot(query_embedding, item["embedding"]) / (
np.linalg.norm(query_embedding) * np.linalg.norm(item["embedding"])
)
if similarity >= self.config.vector_threshold:
results.append({
"content": item["content"],
"metadata": item["metadata"],
"similarity": float(similarity)
})
results.sort(key=lambda x: x["similarity"], reverse=True)
return results[:top_k]
def get_context_for_agent(self, current_query: str = "") -> str:
"""สร้าง context string สำหรับส่งให้ Agent"""
context_parts = []
# 1. Vector search results (ถ้ามี query)
if current_query:
vector_results = self._semantic_search(current_query, self.config.vector_top_k)
if vector_results:
context_parts.append("## ข้อมูลจากฐานความรู้:")
for r in vector_results:
context_parts.append(f"- {r['content']} (ความเหมือน: {r['similarity']:.2%})")
# 2. Summary memories
if self.summaries:
context_parts.append("\n## บริบทจากการสนทนาก่อนหน้า:")
for s in self.summaries[-3:]: # เอาแค่ 3 summary ล่าสุด
context_parts.append(f"- {s['summary']}")
# 3. Recent buffer
if self.buffer:
context_parts.append("\n## การสนทนาล่าสุด:")
for item in list(self.buffer)[-3:]:
context_parts.append(f"ถาม: {item['user']}")
context_parts.append(f"ตอบ: {item['assistant']}")
return "\n".join(context_parts) if context_parts else "ไม่มีข้อมูลบริบท"
def get_memory_stats(self) -> Dict:
"""ดึงสถิติหน่วยความจำ"""
return {
"buffer_count": len(self.buffer),
"summary_count": len(self.summaries),
"vector_store_count": len(self.vector_store),
"pending_messages": len(self.pending_messages)
}
========== ตัวอย่างการใช้งานจริง ==========
ตั้งค่า config
config = MemoryConfig(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1",
buffer_size=10,
summary_threshold=4, # สรุปทุก 4 ข้อความเพื่อ demo
vector_top_k=2,
vector_threshold=0.6
)
สร้างระบบหน่วยความจำ
memory_system = HybridMemorySystem(config)
เพิ่มข้อมูลสินค้าลง vector store
memory_system.add_to_vector_store(
"รองเท้าผ้าใบ Nike Air Max 90 ราคา 5,900 บาท มีสีขาว ดำ และเขียว",
{"type": "product", "category": "รองเท้า", "brand": "Nike"}
)
memory_system.add_to_vector_store(
"นโยบายการส่งสินค้า: สั่งก่อน 14:00 จัดส่งวันเดียว ในกรุงเทพฯ",
{"type": "policy", "category": "การจัดส่ง"}
)
เพิ่มปฏิสัมพันธ์การสนทนา
interactions = [
("สวัสดีครับ สนใจรองเท้าวิ่ง", "สวัสดีครับ! ย