Trong thế giới phát triển AI Agent, việc xây dựng một hệ thống ghi nhớ hiệu quả là yếu tố quyết định trải nghiệm người dùng. Bài viết này sẽ hướng dẫn chi tiết cách implement Memory trong LangGraph, từ cơ bản đến nâng cao, kèm theo mã nguồn có thể sao chép và chạy ngay.
Bảng so sánh chi phí API cho AI Agent Development
Trước khi đi vào chi tiết kỹ thuật, hãy cùng xem bảng so sánh chi phí khi sử dụng các dịch vụ API khác nhau để xây dựng AI Agent với Memory mechanism:
| Tiêu chí | API chính thức | Dịch vụ Relay thông thường | HolySheep AI |
|---|---|---|---|
| GPT-4.1 | $8.00/1M tokens | $5.60/1M tokens | $1.20/1M tokens |
| Claude Sonnet 4.5 | $15.00/1M tokens | $10.50/1M tokens | $2.25/1M tokens |
| Gemini 2.5 Flash | $2.50/1M tokens | $1.75/1M tokens | $0.38/1M tokens |
| DeepSeek V3.2 | $0.42/1M tokens | $0.35/1M tokens | $0.07/1M tokens |
| Độ trễ trung bình | 800-2000ms | 200-500ms | <50ms |
| Thanh toán | Credit Card quốc tế | Credit Card quốc tế | WeChat/Alipay/VNPay |
| Tín dụng miễn phí | $5 | $1-3 | $5-20 khi đăng ký |
Ưu điểm vượt trội: Với tỷ giá ¥1 = $1 và khả năng tiết kiệm đến 85% chi phí, HolySheep AI là lựa chọn tối ưu cho developers Việt Nam muốn build AI Agent với ngân sách hạn chế.
Memory trong LangGraph là gì?
Memory trong LangGraph là cơ chế cho phép AI Agent lưu trữ và truy xuất thông tin từ các cuộc hội thoại trước đó. Khác với simple context window, Memory được thiết kế để hoạt động xuyên suốt nhiều turns của conversation, giúp Agent có "trí nhớ dài hạn".
Tại sao cần Memory cho AI Agent?
- Tương tác liên tục: Agent có thể nhớ preferences và context từ session trước
- Hiệu quả chi phí: Không cần gửi lại toàn bộ lịch sử mỗi lần
- Trải nghiệm người dùng: Conversation flow tự nhiên và cá nhân hóa
- State management: Quản lý trạng thái ứng dụng một cách có hệ thống
Cài đặt môi trường và cấu hình
# Cài đặt các thư viện cần thiết
pip install langgraph langchain-core langchain-openai python-dotenv
Tạo file .env với API key từ HolySheep AI
Đăng ký tại: https://www.holysheep.ai/register
cat > .env << 'EOF'
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
EOFMemory Implementation: Cơ bản
Đây là cách implement Memory đơn giản nhất sử dụng HolySheep AI với LangGraph:
import os
from dotenv import load_dotenv
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
Load environment variables
load_dotenv()
Cấu hình HolySheep AI - KHÔNG dùng api.openai.com
os.environ["OPENAI_API_KEY"] = os.getenv("HOLYSHEEP_API_KEY")
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"
Định nghĩa State cho Memory
class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], lambda a, b: a + b]
user_preferences: dict
session_context: str
Khởi tạo LLM với HolySheep AI
llm = ChatOpenAI(
model="gpt-4.1",
temperature=0.7,
api_key=os.getenv("HOLYSHEEP_API_KEY"),
base_url="https://api.holysheep.ai/v1"
)
def memory_node(state: AgentState) -> AgentState:
"""Node xử lý memory và context"""
messages = state["messages"]
# Lấy lịch sử messages gần đây (last 10)
recent_messages = messages[-10:]
# Tạo context summary từ lịch sử
context_prompt = f"""Analyze the conversation history and extract:
1. User preferences mentioned
2. Important context/information
3. Current topic being discussed
Messages: {recent_messages}
Return a concise summary."""
summary_response = llm.invoke([HumanMessage(content=context_prompt)])
return {
"messages": messages,
"user_preferences": state.get("user_preferences", {}),
"session_context": summary_response.content
}
def chat_node(state: AgentState) -> AgentState:
"""Node xử lý chat chính với Memory"""
messages = state["messages"]
context = state.get("session_context", "")
# System prompt với memory context
system_prompt = f"""You are a helpful AI assistant.
Current session context: {context}
Remember to use the context above to provide personalized responses."""
# Invoke LLM với memory context
response = llm.invoke([
SystemMessage(content=system_prompt),
*messages
])
# Cập nhật messages với response
new_messages = messages + [response]
return {
"messages": new_messages,
"user_preferences": state.get("user_preferences", {}),
"session_context": context
}
Xây dựng Graph
workflow = StateGraph(AgentState)
Thêm nodes
workflow.add_node("memory", memory_node)
workflow.add_node("chat", chat_node)
Định nghĩa edges
workflow.add_edge("__start__", "memory")
workflow.add_edge("memory", "chat")
workflow.add_edge("chat", END)
Compile graph
agent_graph = workflow.compile()
Test với Memory
initial_state = {
"messages": [HumanMessage(content="Tôi thích màu xanh dương và thời trang công sở")],
"user_preferences": {"color": "blue", "style": "office"},
"session_context": ""
}
Chạy agent
result = agent_graph.invoke(initial_state)
print("Session Context:", result["session_context"])
print("Response:", result["messages"][-1].content)Advanced Memory: Vector Store Integration
Để implement memory với semantic search và retrieval, chúng ta sử dụng vector store:
import os
from dotenv import load_dotenv
from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Optional
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.documents import Document
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
import time
load_dotenv()
os.environ["OPENAI_API_KEY"] = os.getenv("HOLYSHEEP_API_KEY")
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"
Cấu hình embeddings với HolySheep AI
embeddings = OpenAIEmbeddings(
model="text-embedding-3-small",
api_key=os.getenv("HOLYSHEEP_API_KEY"),
base_url="https://api.holysheep.ai/v1"
)
class MemoryState(TypedDict):
messages: List
user_id: str
retrieved_memories: List[str]
vectorstore: Optional[object]
class VectorMemory:
"""Memory system sử dụng Vector Store"""
def __init__(self, collection_name: str = "conversation_memory"):
self.collection_name = collection_name
# Sử dụng Chroma in-memory (có thể thay bằng persistent storage)
self.vectorstore = Chroma(
collection_name=collection_name,
embedding_function=embeddings
)
def add_memory(self, user_id: str, content: str, metadata: dict = None):
"""Thêm memory mới vào vector store"""
doc = Document(
page_content=content,
metadata={"user_id": user_id, **(metadata or {})}
)
self.vectorstore.add_documents([doc])
return {"status": "added", "content": content}
def retrieve_memories(self, user_id: str, query: str, k: int = 3) -> List[str]:
"""Retrieve memories liên quan đến query"""
# Filter theo user_id
results = self.vectorstore.similarity_search(
query=query,
k=k,
filter={"user_id": user_id}
)
return [doc.page_content for doc in results]
def delete_old_memories(self, user_id: str, keep_count: int = 50):
"""Xóa memories cũ, giữ lại N memories gần nhất"""
# Implementation phụ thuộc vào vector store backend
pass
Khởi tạo Memory System
memory_system = VectorMemory(collection_name="user_conversations")
def retrieve_memory_node(state: MemoryState) -> MemoryState:
"""Node truy xuất memories liên quan"""
user_id = state["user_id"]
last_message = state["messages"][-1].content if state["messages"] else ""
# Tìm kiếm memories liên quan
retrieved = memory_system.retrieve_memories(
user_id=user_id,
query=last_message,
k=3
)
return {
**state,
"retrieved_memories": retrieved
}
def generate_response_node(state: MemoryState) -> MemoryState:
"""Node tạo response với memory context"""
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
model="gpt-4.1",
api_key=os.getenv("HOLYSHEEP_API_KEY"),
base_url="https://api.holysheep.ai/v1"
)
memories = state.get("retrieved_memories", [])
memory_context = "\n".join([f"- {m}" for m in memories]) if memories else "Không có memory liên quan"
system_prompt = f"""Bạn là một AI assistant với khả năng ghi nhớ.
Memory của người dùng:
{memory_context}
Hãy sử dụng memory để tạo response phù hợp và cá nhân hóa."""
response = llm.invoke([
SystemMessage(content=system_prompt),
*state["messages"]
])
return {
**state,
"messages": state["messages"] + [response]
}
def save_memory_node(state: MemoryState) -> MemoryState:
"""Node lưu memory mới"""
user_id = state["user_id"]
last_message = state["messages"][-1]
if isinstance(last_message, HumanMessage):
memory_system.add_memory(
user_id=user_id,
content=last_message.content,
metadata={"type": "user_message", "timestamp": time.time()}
)
return state
Xây dựng Graph với Memory Flow
workflow = StateGraph(MemoryState)
workflow.add_node("retrieve", retrieve_memory_node)
workflow.add_node("generate", generate_response_node)
workflow.add_node("save", save_memory_node)
workflow.add_edge("__start__", "retrieve")
workflow.add_edge("retrieve", "generate")
workflow.add_edge("generate", "save")
workflow.add_edge("save", END)
agent_with_memory = workflow.compile()
Test với latency measurement
start_time = time.time()
test_state = {
"messages": [HumanMessage(content="Tôi đang tìm kiếm laptop cho developer")],
"user_id": "user_123",
"retrieved_memories": [],
"vectorstore": memory_system.vectorstore
}
result = agent_with_memory.invoke(test_state)
end_time = time.time()
print(f"Response Time: {(end_time - start_time) * 1000:.2f}ms")
print(f"Retrieved Memories: {result['retrieved_memories']}")
print(f"Response: {result['messages'][-1].content}")Session-based Memory với Redis Cache
Để implement memory với độ trễ cực thấp (<50ms), sử dụng Redis cache:
import os
import json
import time
from typing import TypedDict, Dict, Any, Optional
from dotenv import load_dotenv
import redis
load_dotenv()
class RedisSessionMemory:
"""Session Memory sử dụng Redis Cache - Ultra low latency"""
def __init__(self, redis_url: str = "redis://localhost:6379", ttl: int = 3600):
try:
self.redis_client = redis.from_url(redis_url)
self.redis_client.ping() # Test connection
self.available = True
except:
self.redis_client = None
self.available = False
self.ttl = ttl # Session timeout in seconds
def save_session(self, session_id: str, data: Dict[str, Any]) -> Dict:
"""Lưu session data với latency benchmark"""
start = time.time()
if not self.available:
return {"error": "Redis not available", "latency_ms": 0}
session_key = f"session:{session_id}"
serialized = json.dumps(data, default=str)
self.redis_client.setex(session_key, self.ttl, serialized)
latency = (time.time() - start) * 1000
return {"status": "saved", "latency_ms": round(latency, 2)}
def load_session(self, session_id: str) -> Optional[Dict[str, Any]]:
"""Load session data với latency benchmark"""
start = time.time()
if not self.available:
return None
session_key = f"session:{session_id}"
data = self.redis_client.get(session_key)
latency = (time.time() - start) * 1000
if data:
return {"data": json.loads(data), "latency_ms": round(latency, 2)}
return {"data": None, "latency_ms": round(latency, 2)}
def append_message(self, session_id: str, role: str, content: str) -> Dict:
"""Append message vào session history"""
session = self.load_session(session_id)
messages = session.get("data", {}).get("messages", [])
messages.append({"role": role, "content": content, "timestamp": time.time()})
current_data = session.get("data", {})
current_data["messages"] = messages
return self.save_session(session_id, current_data)
def get_context_window(self, session_id: str, window_size: int = 10) -> list:
"""Lấy context window gần nhất"""
session = self.load_session(session_id)
messages = session.get("data", {}).get("messages", [])
return messages[-window_size:]
Khởi tạo Memory System
redis_memory = RedisSessionMemory(ttl=7200) # 2 hours TTL
Benchmark: Save Session
print("=== Redis Memory Benchmark ===")
result = redis_memory.save_session("user_001", {
"user_id": "user_001",
"preferences": {"theme": "dark", "language": "vi"},
"messages": []
})
print(f"Save Session Latency: {result.get('latency_ms', 'N/A')}ms")
Benchmark: Load Session
result = redis_memory.load_session("user_001")
print(f"Load Session Latency: {result.get('latency_ms', 'N/A')}ms")
Benchmark: Append Message
result = redis_memory.append_message("user_001", "user", "Xin chào AI!")
print(f"Append Message Latency: {result.get('latency_ms', 'N/A')}ms")
Integration với LangGraph
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
class SessionState(TypedDict):
session_id: str
messages: list
memory_system: object
def memory_lookup_node(state: SessionState) -> SessionState:
"""Node truy xuất memory từ Redis"""
session_id = state["session_id"]
# Load context từ Redis
session_data = redis_memory.load_session(session_id)
context = redis_memory.get_context_window(session_id, window_size=10)
return {
**state,
"messages": context
}
def chat_node(state: SessionState) -> SessionState:
"""Node chat với context từ memory"""
llm = ChatOpenAI(
model="gpt-4.1",
api_key=os.getenv("HOLYSHEEP_API_KEY"),
base_url="https://api.holysheep.ai/v1"
)
from langchain_core.messages import HumanMessage, SystemMessage
context = state.get("messages", [])
formatted_messages = [SystemMessage(content="Bạn là AI assistant với trí nhớ session.")]
for msg in context:
if msg["role"] == "user":
formatted_messages.append(HumanMessage(content=msg["content"]))
response = llm.invoke(formatted_messages)
# Save response vào memory
redis_memory.append_message(state["session_id"], "assistant", response.content)
return {
**state,
"messages": context + [{"role": "assistant", "content": response.content}]
}
Build Graph
workflow = StateGraph(SessionState)
workflow.add_node("memory_lookup", memory_lookup_node)
workflow.add_node("chat", chat_node)
workflow.add_edge("__start__", "memory_lookup")
workflow.add_edge("memory_lookup", "chat")
workflow.add_edge("chat", END)
session_agent = workflow.compile()
Run với session-based memory
start_time = time.time()
result = session_agent.invoke({
"session_id": "user_001",
"messages": [],
"memory_system": redis_memory
})
end_time = time.time()
print(f"\n=== Total Session Agent Latency: {(end_time - start_time) * 1000:.2f}ms ===")Production-ready Memory Architecture
Kiến trúc Memory hoàn chỉnh cho production với multi-tier storage:
import os
from typing import TypedDict, Dict, List, Any, Optional
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
import hashlib
import json
class MemoryTier(Enum):
"""Memory tiers theo priority và freshness"""
SHORT_TERM = "short_term" # Current conversation (in-memory)
MEDIUM_TERM = "medium_term" # Recent sessions (Redis)
LONG_TERM = "long_term" # Historical (Vector DB)
@dataclass
class MemoryEntry:
"""Memory entry structure"""
id: str
content: str
embedding: Optional[List[float]] = None
tier: MemoryTier = MemoryTier.SHORT_TERM
created_at: datetime = field(default_factory=datetime.now)
accessed_at: datetime = field(default_factory=datetime.now)
access_count: int = 0
importance_score: float = 0.5 # 0.0 - 1.0
metadata: Dict[str, Any] = field(default_factory=dict)
class MultiTierMemory:
"""
Production-ready Memory System với multi-tier architecture.
Architecture:
- Short-term: In-memory dict (fast, volatile)
- Medium-term: Redis (persistent, fast retrieval)
- Long-term: Vector DB (semantic search, scalable)
"""
def __init__(self, user_id: str):
self.user_id = user_id
self.short_term: Dict[str, MemoryEntry] = {}
self.max_short_term = 50 # Max entries in short-term
def add_memory(
self,
content: str,
tier: MemoryTier = MemoryTier.SHORT_TERM,
importance: float = 0.5,
metadata: Dict = None
) -> str:
"""Add new memory entry"""
memory_id = self._generate_id(content)
entry = MemoryEntry(
id=memory_id,
content=content,
tier=tier,
importance_score=importance,
metadata=metadata or {}
)
if tier == MemoryTier.SHORT_TERM:
self.short_term[memory_id] = entry
self._prune_short_term()
return memory_id
def _generate_id(self, content: str) -> str:
"""Generate unique ID cho memory"""
raw = f"{self.user_id}:{content}:{datetime.now().isoformat()}"
return hashlib.sha256(raw.encode()).hexdigest()[:16]
def _prune_short_term(self):
"""Remove low-importance entries khi short-term đầy"""
if len(self.short_term) > self.max_short_term:
sorted_entries = sorted(
self.short_term.items(),
key=lambda x: (x[1].importance_score, x[1].access_count)
)
to_remove = len(self.short_term) - self.max_short_term + 10
for key, _ in sorted_entries[:to_remove]:
del self.short_term[key]
def retrieve(
self,
query: str = None,
tier: MemoryTier = None,
limit: int = 10
) -> List[MemoryEntry]:
"""Retrieve memories với optional filtering"""
results = []
if tier is None or tier == MemoryTier.SHORT_TERM:
entries = list(self.short_term.values())
if query:
# Simple keyword matching (production nên dùng vector search)
entries = [
e for e in entries
if query.lower() in e.content.lower()
]
results.extend(entries)
return sorted(
results,
key=lambda x: (x.importance_score, x.access_count, x.accessed_at),
reverse=True
)[:limit]
def promote_memory(self, memory_id: str, new_tier: MemoryTier):
"""Promote memory lên tier cao hơn"""
if memory_id in self.short_term:
entry = self.short_term[memory_id]
entry.tier = new_tier
# In production: persist to appropriate storage
def get_conversation_summary(self) -> str:
"""Generate conversation summary từ memories"""
entries = self.retrieve(limit=20)
if not entries:
return "No memories available."
summary_parts = ["Conversation Summary:"]
for entry in entries:
time_str = entry.created_at.strftime("%H:%M")
summary_parts.append(f"[{time_str}] {entry.content}")
return "\n".join(summary_parts)
Usage Example
memory = MultiTierMemory(user_id="user_vietnamese_001")
Add various memories
memory.add_memory(
content="User prefers responses in Vietnamese",
importance=0.9,
metadata={"category": "preference"}
)
memory.add_memory(
content="User interested in AI agent development",
importance=0.8,
metadata={"category": "interest"}
)
memory.add_memory(
content="User asking about LangGraph Memory implementation",
importance=0.7
)
Retrieve relevant memories
relevant = memory.retrieve(query="Vietnamese")
print(f"Found {len(relevant)} relevant memories")
Get summary
summary = memory.get_conversation_summary()
print(summary)
Pricing context: Với HolySheep AI, memory operations sử dụng:
- Short-term: Free (in-memory)
- Medium-term: Redis hosting ~$5-10/tháng
- Long-term: Vector DB (Chroma free, Pinecone ~$70/tháng)
Total ops cost: ~$0.0001/1000 retrievals
Lỗi thường gặp và cách khắc phục
1. Lỗi "Authentication Error" khi sử dụng HolySheep API
# ❌ SAI: Sử dụng base_url không đúng
os.environ["OPENAI_API_BASE"] = "https://api.openai.com/v1" # Lỗi!
✅ ĐÚNG: Sử dụng HolySheep AI base_url
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"
Kiểm tra API key
import os
print(f"API Key: {os.getenv('HOLYSHEEP_API_KEY')[:10]}...")
print(f"Base URL: {os.getenv('OPENAI_API_BASE')}")Nguyên nhân: API key từ HolySheep AI không hoạt động với base_url của OpenAI chính thức. Cách khắc phục: Luôn set base_url thành https://api.holysheep.ai/v1 trước khi khởi tạo ChatOpenAI client.
2. Lỗi "Memory leak" khi không clear context
# ❌ SAI: Messages list grow vô hạn
messages = []
for turn in many_turns:
messages.append(HumanMessage(content=user_input))
response = llm.invoke(messages) # Context ngày càng dài!
messages.append(AIMessage(content=response.content))
✅ ĐÚNG: Implement sliding window hoặc summary
def get_summarized_context(messages: list, max_turns: int = 10):
"""Chỉ giữ lại N turns gần nhất hoặc summary"""
if len(messages) <= max_turns * 2: # user + assistant per turn
return messages[-max_turns * 2:]
# Summarize older messages
older = messages[:-max_turns * 2]
summary = summarize_messages(older)
return [
SystemMessage(content=f"Previous context summary: {summary}")
] + messages[-max_turns * 2:]
Hoặc sử dụng Memory class đã implement
memory_system = RedisSessionMemory()
recent_context = memory_system.get_context_window(session_id, window_size=10)Nguyên nhân: Context window không được quản lý dẫn đến token usage tăng đột biến và latency cao. Cách khắc phục: Implement sliding window hoặc summary-based context management, giới hạn số messages trong short-term memory.
3. Lỗi "Rate Limit" khi batch operations
# ❌ SAI: Gửi quá nhiều requests cùng lúc
results = [llm.invoke(msg) for msg in messages_batch] # Rate limit!
✅ ĐÚNG: Implement rate limiting với exponential backoff
import asyncio
import time
from functools import wraps
class RateLimiter:
def __init__(self, max_calls: int = 60, period: int = 60):
self.max_calls = max_calls
self.period = period
self.calls = []
def wait_if_needed(self):
now = time.time()
self.calls = [t for t in self.calls if now - t < self.period]
if len(self.calls) >= self.max_calls:
sleep_time = self.period - (now - self.calls[0])
print(f"Rate limit reached. Sleeping {sleep_time:.2f}s")
time.sleep(sleep_time)
self.calls = [t for t in self.calls if now - t < self.period]
self.calls.append(now)
rate_limiter = RateLimiter(max_calls=30, period=60)
def limited_invoke(llm, message):
rate_limiter.wait_if_needed()
start = time.time()
response = llm.invoke(message)
latency = (time.time() - start) * 1000
print(f"Request completed in {latency:.2f}ms")
return response
Process batch với rate limiting
for msg in messages_batch:
response = limited_invoke(llm, msg)Nguyên nhân: HolySheep AI có rate limit tùy theo plan (Free: 30 req/min, Pro: 100 req/min). Cách khắc phục: Implement rate limiter với exponential backoff, batch operations hợp lý, và monitor usage để upgrade plan khi cần.
4. Lỗi "Vector store connection failed"
# ❌ SAI: Không handle connection failures
vectorstore = Chroma(collection_name="memories", embedding_function=embeddings)
results = vectorstore.similarity_search(query) # Crash if connection fails!
✅ ĐÚNG: Implement fallback mechanism
class RobustVectorMemory:
def __init__(self):
self.primary = None
self.fallback_cache = {}
self._initialize()
def _initialize(self):
try:
self.primary = Chroma(
collection_name="memories",
embedding_function=embeddings,
persist_directory="./chroma_db"
)
self.primary.similarity_search("init", k=1) # Test connection
print("Primary vector store: Connected")
except Exception as e:
print(f"Primary store failed: {e}. Using in-memory fallback.")
self.primary = None
def add(self, text: str, metadata: dict = None) -> bool:
try:
if self.primary:
self.primary.add_texts([text], metadatas=[metadata or {}])
else:
self.fallback_cache[hashlib.md5(text.encode()).hexdigest()] = {
"text": text,
"metadata": metadata
}
return True
except Exception as e:
print(f"Add failed: {e}")
return False
def search(self, query: str, k: int = 3) -> List[str]:
try:
if self.primary:
results =