Trong thế giới AI Agent ngày nay, việc thiết kế hệ thống nhớ hiệu quả là yếu tố quyết định giữa một agent "ngớ ngẩn" và một agent thông minh có khả năng học hỏi từ tương tác. Bài viết này sẽ hướng dẫn bạn xây dựng memory module từ con số 0, đồng thời chia sẻ case study thực tế từ một startup AI tại Hà Nội đã tối ưu chi phí lên tới 83% với HolySheep AI.
Case Study: Startup AI Việt Nam Tối Ưu Memory Module
Bối cảnh kinh doanh
Một startup AI ở Hà Nội chuyên xây dựng chatbot chăm sóc khách hàng cho các sàn thương mại điện tử Việt Nam đã gặp khó khăn nghiêm trọng với hệ thống memory của họ. Với 50,000+ người dùng hoạt động hàng ngày, chi phí API OpenAI lên tới $4,200/tháng trong khi độ trễ trung bình đạt 420ms - quá chậm cho trải nghiệm chat real-time.
Điểm đau của nhà cung cấp cũ
- Chi phí context window quá cao khi lưu trữ conversation history dài
- Không có cơ chế memory tiering - mọi thứ đều được đẩy vào prompt
- Độ trễ 420ms làm giảm trải nghiệm người dùng đáng kể
- Không hỗ trợ persistent memory giữa các session
Lý do chọn HolySheep
Sau khi benchmark nhiều nhà cung cấp, đội ngũ kỹ thuật đã chọn HolySheep AI vì:
- Tỷ giá ¥1=$1 - tiết kiệm 85%+ so với các provider phương Tây
- Độ trễ trung bình dưới 50ms (thực đo 38ms)
- Hỗ trợ WeChat/Alipay thanh toán dễ dàng
- Tín dụng miễn phí khi đăng ký để test đầu tiên
Các bước di chuyển cụ thể
Đội ngũ đã thực hiện migration theo 3 giai đoạn: đổi base_url sang https://api.holysheep.ai/v1, xoay key sang YOUR_HOLYSHEEP_API_KEY, và áp dụng canary deploy 5% → 25% → 100% traffic.
Kết quả sau 30 ngày go-live
- Độ trễ: 420ms → 180ms (giảm 57%)
- Chi phí hàng tháng: $4,200 → $680 (tiết kiệm 83.8%)
- Số lượng token xử lý tăng 40% nhưng chi phí giảm rò rã
Kiến Trúc Memory Module Cho AI Agent
Một memory module hiệu quả cho AI Agent cần được thiết kế theo nguyên tắc tiering - phân loại thông tin theo tầm quan trọng và tần suất truy cập.
1. Sensory Memory - Bộ Nhớ Cảm Giác
Đây là lớp nhớ ngắn hạn nhất, lưu trữ thông tin từ conversation buffer hiện tại. Dữ liệu được tự động evicted sau khi context window đầy.
import json
from datetime import datetime, timedelta
from collections import deque
from typing import List, Dict, Any, Optional
import hashlib
class SensoryMemory:
"""
Lớp nhớ cảm giác - lưu trữ conversation buffer
Chỉ giữ lại N tin nhắn gần nhất trong rolling window
"""
def __init__(self, max_messages: int = 50, max_age_seconds: int = 3600):
self.buffer = deque(maxlen=max_messages)
self.max_age = timedelta(seconds=max_age_seconds)
self.timestamps = deque(maxlen=max_messages)
def add_message(self, role: str, content: str, metadata: Optional[Dict] = None):
"""Thêm message vào sensory buffer"""
message = {
"role": role,
"content": content,
"timestamp": datetime.now().isoformat(),
"metadata": metadata or {}
}
self.buffer.append(message)
self.timestamps.append(datetime.now())
# Tự động cleanup messages cũ
self._cleanup_old_messages()
def _cleanup_old_messages(self):
"""Loại bỏ messages quá hạn"""
cutoff = datetime.now() - self.max_age
while self.timestamps and self.timestamps[0] < cutoff:
self.buffer.popleft()
self.timestamps.popleft()
def get_context(self) -> List[Dict[str, Any]]:
"""Trả về toàn bộ context cho agent"""
return list(self.buffer)
def get_last_n(self, n: int) -> List[Dict[str, Any]]:
"""Lấy N messages gần nhất"""
return list(self.buffer)[-n:]
def get_token_count(self) -> int:
"""Ước tính số tokens trong buffer"""
total_chars = sum(len(m["content"]) for m in self.buffer)
return total_chars // 4 # Rough estimate
Demo sử dụng
sensory = SensoryMemory(max_messages=20, max_age_seconds=1800)
sensory.add_message("user", "Xin chào, tôi muốn tìm laptop gaming dưới 20 triệu")
sensory.add_message("assistant", "Bạn có thể tham khảo các dòng laptop ASUS ROG, MSI Katana, hoặc Lenovo Legion")
sensory.add_message("user", "Cảm ơn, vậy còn Dell G15 thì sao?")
print(f"Số messages: {len(sensory.get_context())}")
print(f"Ước tính tokens: {sensory.get_token_count()}")2. Working Memory - Bộ Nhớ Làm Việc
Working memory là lớp trung gian, lưu trữ thông tin quan trọng cần giữ lâu hơn sensory nhưng không cần persist vĩnh viễn. Đây là nơi agent suy nghĩ và reasoning.
import sqlite3
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, asdict
from datetime import datetime
import json
@dataclass
class MemoryItem:
"""Cấu trúc một memory item"""
id: str
content: str
importance: float # 0.0 - 1.0
category: str
created_at: str
last_accessed: str
access_count: int
embedding_key: Optional[str] = None
class WorkingMemory:
"""
Lớp nhớ làm việc - PostgreSQL-backed
Lưu trữ facts, preferences, context quan trọng
"""
def __init__(self, db_path: str = "working_memory.db"):
self.db_path = db_path
self._init_db()
def _init_db(self):
"""Khởi tạo database schema"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS memories (
id TEXT PRIMARY KEY,
content TEXT NOT NULL,
importance REAL DEFAULT 0.5,
category TEXT DEFAULT 'general',
created_at TEXT,
last_accessed TEXT,
access_count INTEGER DEFAULT 0,
embedding_key TEXT
)
''')
cursor.execute('''
CREATE INDEX IF NOT EXISTS idx_importance
ON memories(importance DESC)
''')
cursor.execute('''
CREATE INDEX IF NOT EXISTS idx_category
ON memories(category)
''')
conn.commit()
conn.close()
def store(self, content: str, category: str = "general",
importance: float = 0.5) -> str:
"""Lưu một memory item mới"""
import uuid
memory_id = hashlib.md5(
f"{content}{datetime.now().isoformat()}".encode()
).hexdigest()[:16]
now = datetime.now().isoformat()
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
INSERT INTO memories
(id, content, importance, category, created_at, last_accessed, access_count)
VALUES (?, ?, ?, ?, ?, ?, ?)
''', (memory_id, content, importance, category, now, now, 0))
conn.commit()
conn.close()
return memory_id
def retrieve(self, query: str, limit: int = 10,
min_importance: float = 0.3) -> List[MemoryItem]:
"""Retrieve memories liên quan đến query"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
# Simple keyword matching - trong production nên dùng vector search
keywords = query.lower().split()
search_pattern = f"%{'%'.join(keywords)}%"
cursor.execute('''
SELECT id, content, importance, category,
created_at, last_accessed, access_count
FROM memories
WHERE (content LIKE ? OR category LIKE ?)
AND importance >= ?
ORDER BY importance DESC, last_accessed DESC
LIMIT ?
''', (search_pattern, search_pattern, min_importance, limit))
results = []
for row in cursor.fetchall():
item = MemoryItem(*row)
results.append(item)
# Update access statistics
cursor.execute('''
UPDATE memories
SET access_count = access_count + 1,
last_accessed = ?
WHERE id = ?
''', (datetime.now().isoformat(), item.id))
conn.commit()
conn.close()
return results
def update_importance(self, memory_id: str, new_importance: float):
"""Cập nhật importance score"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
UPDATE memories SET importance = ? WHERE id = ?
''', (new_importance, memory_id))
conn.commit()
conn.close()
def get_summary(self, category: Optional[str] = None) -> Dict[str, Any]:
"""Lấy tóm tắt statistics"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
if category:
cursor.execute('''
SELECT COUNT(*), AVG(importance), MAX(access_count)
FROM memories WHERE category = ?
''', (category,))
else:
cursor.execute('''
SELECT COUNT(*), AVG(importance), MAX(access_count)
FROM memories
''')
count, avg_imp, max_access = cursor.fetchone()
conn.close()
return {
"total_memories": count,
"avg_importance": avg_imp or 0,
"max_access_count": max_access or 0
}
Demo sử dụng
working = WorkingMemory("demo_working.db")
Lưu các facts quan trọng
working.store(
"User thích laptop gaming với budget 20 triệu VNĐ",
category="preference",
importance=0.8
)
working.store(
"User quan tâm đến Dell G15 và ASUS ROG",
category="product_interest",
importance=0.7
)
Retrieve
results = working.retrieve("laptop gaming budget", limit=5)
for r in results:
print(f"- {r.content} (importance: {r.importance})")
print(f"\nSummary: {working.get_summary()}")3. Long-Term Memory - Bộ Nhớ Dài Hạn
Đây là lớp lưu trữ vĩnh viễn các kiến thức, facts, và patterns mà agent đã học được qua thời gian. Sử dụng vector database để enable semantic search hiệu quả.
import hashlib
import json
from typing import List, Dict, Any, Optional, Tuple
from datetime import datetime, timedelta
from abc import ABC, abstractmethod
class VectorStore(ABC):
"""Abstract interface cho vector storage"""
@abstractmethod
def upsert(self, id: str, vector: List[float], metadata: Dict[str, Any]):
pass
@abstractmethod
def search(self, query_vector: List[float], top_k: int) -> List[Dict]:
pass
class LongTermMemory:
"""
Lớp nhớ dài hạn - Vector-backed persistent storage
Sử dụng HolySheep AI API cho embeddings
"""
def __init__(self, api_key: str, vector_store: VectorStore):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.vector_store = vector_store
self.consolidation_threshold = 10 # Số lần access trước khi consolidate
def _get_embedding(self, text: str, model: str = "embedding-v3") -> List[float]:
"""Gọi HolySheep API để lấy embedding vector"""
import requests
response = requests.post(
f"{self.base_url}/embeddings",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": model,
"input": text
}
)
if response.status_code != 200:
raise Exception(f"Embedding API error: {response.text}")
data = response.json()
return data["data"][0]["embedding"]
def add_knowledge(self, content: str, metadata: Dict[str, Any]) -> str:
"""Thêm knowledge vào long-term memory"""
knowledge_id = hashlib.sha256(
f"{content}{datetime.now().isoformat()}".encode()
).hexdigest()[:16]
# Get embedding vector
vector = self._get_embedding(content)
# Store in vector database
self.vector_store.upsert(
id=knowledge_id,
vector=vector,
metadata={
**metadata,
"content": content,
"created_at": datetime.now().isoformat(),
"access_count": 0
}
)
return knowledge_id
def query_knowledge(self, query: str, top_k: int = 5,
time_decay: bool = True) -> List[Dict[str, Any]]:
"""Query knowledge với semantic search"""
query_vector = self._get_embedding(query)
results = self.vector_store.search(query_vector, top_k * 2)
# Apply time decay nếu enabled
if time_decay:
results = self._apply_time_decay(results)
return results[:top_k]
def _apply_time_decay(self, memories: List[Dict]) -> List[Dict]:
"""Áp dụng exponential time decay cho old memories"""
now = datetime.now()
def decay_score(item: Dict) -> float:
created = datetime.fromisoformat(item["metadata"]["created_at"])
days_old = (now - created).days
# Exponential decay: score *= e^(-0.1 * days)
base_score = 1.0
decayed = base_score * (2.71828 ** (-0.1 * days_old))
# Boost by access count
access_count = item["metadata"].get("access_count", 0)
access_boost = 1.0 + (access_count * 0.1)
return decayed * access_boost
return sorted(memories, key=decay_score, reverse=True)
def consolidate_memories(self, working_memory_ids: List[str]):
"""
Đẩy memories quan trọng từ working memory
sang long-term memory
"""
# Implementation chi tiết phụ thuộc vào working memory integration
pass
Example usage với HolySheep API
class SimpleInMemoryVectorStore(VectorStore):
"""Simple vector store cho demo - production nên dùng Pinecone/Milvus"""
def __init__(self):
self.store = {}
self.vectors = {}
def upsert(self, id: str, vector: List[float], metadata: Dict[str, Any]):
self.store[id] = metadata
self.vectors[id] = vector
def search(self, query_vector: List[float], top_k: int) -> List[Dict]:
# Cosine similarity đơn giản
def cosine_sim(v1, v2):
dot = sum(a * b for a, b in zip(v1, v2))
norm1 = sum(a * a for a in v1) ** 0.5
norm2 = sum(a * a for a in v2) ** 0.5
return dot / (norm1 * norm2 + 1e-10)
scores = [
(id, cosine_sim(query_vector, vec))
for id, vec in self.vectors.items()
]
scores.sort(key=lambda x: x[1], reverse=True)
return [
{"id": id, "score": score, "metadata": self.store[id]}
for id, score in scores[:top_k]
]
Khởi tạo LongTermMemory
api_key = "YOUR_HOLYSHEEP_API_KEY"
vector_store = SimpleInMemoryVectorStore()
ltm = LongTermMemory(api_key, vector_store)
Thêm knowledge
ltm.add_knowledge(
"ASUS ROG Strix G16 có CPU Intel i9-14900HX, RTX 4070, 32GB RAM",
metadata={"category": "product", "brand": "ASUS", "price_range": "25-30tr"}
)
print("Long-term memory initialized successfully")Tích Hợp HolySheep AI Cho Memory Module
HolySheep AI cung cấp API endpoint tương thích OpenAI với chi phí thấp hơn 85%. Dưới đây là cách tích hợp vào memory module của bạn:
import requests
from typing import List, Dict, Any, Optional
import time
class HolySheepAIClient:
"""
Client cho HolySheep AI API
Endpoint: https://api.holysheep.ai/v1
"""
# Pricing 2026 (USD per 1M tokens)
PRICING = {
"gpt-4.1": {"input": 8.0, "output": 8.0},
"claude-sonnet-4.5": {"input": 15.0, "output": 15.0},
"gemini-2.5-flash": {"input": 2.50, "output": 2.50},
"deepseek-v3.2": {"input": 0.42, "output": 0.42}
}
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
def chat_completion(
self,
messages: List[Dict[str, str]],
model: str = "deepseek-v3.2",
temperature: float = 0.7,
max_tokens: Optional[int] = None,
**kwargs
) -> Dict[str, Any]:
"""
Gọi chat completion API
Model recommended: deepseek-v3.2 (giá rẻ nhất, chất lượng tốt)
"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
**kwargs
}
if max_tokens:
payload["max_tokens"] = max_tokens
start_time = time.time()
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
)
latency = (time.time() - start_time) * 1000 # ms
if response.status_code != 200:
raise Exception(f"API Error: {response.status_code} - {response.text}")
result = response.json()
result["latency_ms"] = round(latency, 2)
return result
def estimate_cost(
self,
model: str,
input_tokens: int,
output_tokens: int
) -> Dict[str, float]:
"""Ước tính chi phí theo model"""
pricing = self.PRICING.get(model, self.PRICING["deepseek-v3.2"])
input_cost = (input_tokens / 1_000_000) * pricing["input"]
output_cost = (output_tokens / 1_000_000) * pricing["output"]
return {
"input_cost": round(input_cost, 4),
"output_cost": round(output_cost, 4),
"total_cost": round(input_cost + output_cost, 4)
}
def get_embedding(self, text: str, model: str = "embedding-v3") -> List[float]:
"""Lấy embedding vector cho text"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
response = requests.post(
f"{self.base_url}/embeddings",
headers=headers,
json={
"model": model,
"input": text
}
)
if response.status_code != 200:
raise Exception(f"Embedding API Error: {response.text}")
return response.json()["data"][0]["embedding"]
Demo sử dụng
client = HolySheepAIClient("YOUR_HOLYSHEEP_API_KEY")
So sánh chi phí giữa các model
test_tokens_input = 5000
test_tokens_output = 1000
print("=== So Sánh Chi Phí (5000 input + 1000 output tokens) ===\n")
for model, pricing in client.PRICING.items():
costs = client.estimate_cost(model, test_tokens_input, test_tokens_output)
print(f"{model}:")
print(f" - Input: ${costs['input_cost']}")
print(f" - Output: ${costs['output_cost']}")
print(f" - Total: ${costs['total_cost']}")
print()
Test chat completion
messages = [
{"role": "system", "content": "Bạn là assistant chuyên về laptop"},
{"role": "user", "content": "So sánh Dell G15 và ASUS ROG Strix G16"}
]
result = client.chat_completion(
messages,
model="deepseek-v3.2",
temperature=0.7
)
print(f"Response latency: {result['latency_ms']}ms")
print(f"Model: {result['model']}")
print(f"Usage: {result['usage']}")Chiến Lược Tối Ưu Memory Cho Production
Memory Consolidation Strategy
Để tránh context overflow và tối ưu chi phí, bạn cần implement chiến lược consolidation thông minh:
from enum import Enum
from typing import List, Dict, Any, Callable
import json
class MemoryPriority(Enum):
CRITICAL = 1.0
HIGH = 0.8
MEDIUM = 0.5
LOW = 0.2
DISCARD = 0.0
class MemoryConsolidator:
"""
Quản lý memory consolidation giữa các tier
Strategy: Move up/down based on access patterns
"""
def __init__(
self,
sensory_threshold: int = 100, # tokens
working_threshold: int = 1000, # tokens
consolidation_interval: int = 300 # seconds
):
self.sensory_threshold = sensory_threshold
self.working_threshold = working_threshold
self.interval = consolidation_interval
def should_consolidate(
self,
memory: Dict[str, Any],
current_tier: str,
target_tier: str
) -> bool:
"""Quyết định có nên consolidate memory không"""
access_count = memory.get("access_count", 0)
importance = memory.get("importance", 0.5)
age_seconds = memory.get("age_seconds", 0)
# Rules-based decision
rules = [
# Promote: High importance + frequent access
(access_count >= 10 and importance >= 0.7),
# Promote: Critical info regardless of access
(importance >= 0.9),
# Demote: Low importance + old
(importance < 0.3 and age_seconds > 86400), # 24 hours
# Demote: Never accessed + medium age
(access_count == 0 and age_seconds > 3600 and importance < 0.5)
]
return any(rules)
def calculate_memory_score(self, memory: Dict[str, Any]) -> float:
"""Tính composite score cho memory prioritization"""
importance = memory.get("importance", 0.5)
access_count = memory.get("access_count", 0)
recency = memory.get("recency_score", 0.5)
# Weighted scoring
score = (
importance * 0.5 +
min(access_count / 20, 1.0) * 0.3 +
recency * 0.2
)
return score
def get_tokens_for_context(
self,
memories: List[Dict[str, Any]],
max_tokens: int
) -> List[Dict[str, Any]]:
"""
Chọn memories phù hợp để đưa vào context
Sử dụng greedy selection với score-based ordering
"""
scored_memories = [
{**m, "score": self.calculate_memory_score(m)}
for m in memories
]
# Sort by score descending
scored_memories.sort(key=lambda x: x["score"], reverse=True)
selected = []
current_tokens = 0
for memory in scored_memories:
mem_tokens = memory.get("token_count", len(memory["content"]) // 4)
if current_tokens + mem_tokens <= max_tokens:
selected.append(memory)
current_tokens += mem_tokens
elif memory["score"] >= 0.7: # Force include high-value memories
# Truncate to fit
selected.append(memory)
break
return selected
def generate_summary(
self,
memories: List[Dict[str, Any]],
client, # HolySheepAIClient
model: str = "deepseek-v3.2"
) -> str:
"""
Tạo summary cho một nhóm memories
Để giảm token usage khi memories quá nhiều
"""
if len(memories) <= 5:
return "\n".join(m["content"] for m in memories)
# Summarize using AI
memory_texts = "\n".join(
f"- {m['content']}" for m in memories[:20]
)
prompt = f"""Summarize the following memories into a concise paragraph:
{memory_texts}
Focus on key facts and important details. Keep it under 200 words."""
response = client.chat_completion(
messages=[{"role": "user", "content": prompt}],
model=model,
max_tokens=300
)
return response["choices"][0]["message"]["content"]
Usage example
consolidator = MemoryConsolidator()
memories = [
{"content": "User prefers gaming laptops", "importance": 0.8,
"access_count": 15, "recency_score": 0.9, "token_count": 8},
{"content": "Budget is 20 million VND", "importance": 0.9,
"access_count": 20, "recency_score": 1.0, "token_count": 10},
{"content": "Weather today is sunny", "importance": 0.1,
"access_count": 1, "recency_score": 0.5, "token_count": 7}
]
Chọn memories cho context 50 tokens
selected = consolidator.get_tokens_for_context(memories, max_tokens=50)
print("Selected memories:")
for m in selected:
print(f" - {m['content']} (score: {m['score']:.2f})")Lỗi Thường Gặp Và Cách Khắc Phục
1. Lỗi Context Overflow Khi Memory Quá Lớn
Mô tả lỗi: Khi conversation history quá dài, API trả về lỗi context_length_exceeded hoặc chi phí tăng vọt.
# ❌ SAI: Không kiểm tra token limit
def add_to_context(messages, new_message):
messages.append(new_message) # Có thể overflow!
return messages
✅ ĐÚNG: Luôn check token limit trước
MAX_TOKENS = 128000 # Ví dụ cho context window
def add_to_context_safe(messages, new_message, memory_module):
"""Thêm message với automatic summarization"""
# Check current token count
current_tokens = sum(
len(m["content"]) // 4 for m in messages
)
# Nếu sắp đầy context
if current_tokens > MAX_TOKENS * 0.8:
# Consolidate older messages
older_messages = messages[:-10] # Giữ 10 messages gần nhất
# Tạo summary cho older messages
summary = memory_module.consolidator.generate_summary(
older_messages,
memory_module.client
)
# Thay thế older messages bằng summary
messages = [
{"role": "system", "content": f"Previous conversation summary: {summary}"}
] + messages[-10:]
messages.append(new_message)
return messages
Implement với try-except
def chat_with_memory(memory_module, user_message):
"""Chat với automatic memory management"""
try:
messages = memory_module.sensory.get_context()
messages.append({"role": "user", "content": user_message})
# Check trước khi call API
messages = add_to_context_safe(
messages,
user_message,
memory_module
)
response = memory_module.client.chat_completion(messages)
return response["choices"][0]["message"]["content"]
except Exception as e:
if "context_length" in str(e):
# Emergency: Force consolidate
memory_module.force_consolidate()
return "Tôi đang gặp vấn đề với bộ nhớ, xin vui lòng chờ..."
raise2. Lỗi Duplicate Memories Khi Consolidate
Mô tả lỗi: Cùng một fact được lưu nhiều lần ở các tier khác nhau, gây redundant storage và confused responses.
# ❌ SAI: Không deduplicate khi thêm memory
class BrokenMemoryManager:
def add_memory(self, content):
# Lưu trực tiếp không check trùng lặp
self.long_term.add_knowledge(content, {})
self.working.store(content, importance=0.5)
# -> Duplicate!
✅ ĐÚNG: Implement deduplication
import hashlib
class MemoryDeduplicator:
"""Deduplicate memories dựa trên content hash"""
def __init__(self):
self.content_hashes = set()
def is_duplicate(self, content: str) -> bool:
"""Check xem content đã tồn tại chưa"""
content_hash = hashlib.md5(content.lower().strip().encode()).hexdigest()
return content_hash in self.content_hashes
def add_hash(self, content: str):
"""Register một content hash"""
content_hash = hashlib.md5(content.lower().strip().encode()).hexdigest()
self.content_hashes.add(content_hash)
def find_similar(
self,
content: str,
existing_memories: List[Dict],
similarity_threshold: float = 0.85
) -> List[Dict]:
"""Tìm memories tương tự để merge hoặc skip"""
import difflib
content_lower = content.lower().strip()
similar = []
for mem in existing_memories:
existing_content = mem.get("content", "").lower().strip()
ratio = difflib.SequenceMatcher(