ในการพัฒนา AI Agent ที่ซับซ้อน การจัดการความจำเป็นสิ่งสำคัญที่สุดประการหนึ่ง LangGraph มอบสถาปัตยกรรม Memory ที่ยืดหยุ่น ช่วยให้ Agent สามารถจดจำข้อมูลสำคัญในระยะยาว พร้อมทั้งรักษาบริบทของการสนทนาปัจจุบันได้อย่างมีประสิทธิภาพ บทความนี้จะพาคุณเจาะลึกการใช้งาน LangGraph Memory ผ่าน HolySheep AI ซึ่งให้บริการ API ราคาประหยัดกว่า 85% พร้อมความเร็วในการตอบสนองต่ำกว่า 50 มิลลิวินาที
ตารางเปรียบเทียบบริการ AI API
| เกณฑ์ | HolySheep AI | API อย่างเป็นทางการ | บริการ Relay อื่นๆ |
|---|---|---|---|
| ราคา GPT-4.1 | $8/MTok | $60/MTok | $30-45/MTok |
| ราคา Claude Sonnet 4.5 | $15/MTok | $75/MTok | $40-60/MTok |
| ราคา Gemini 2.5 Flash | $2.50/MTok | $17.50/MTok | $10-15/MTok |
| ราคา DeepSeek V3.2 | $0.42/MTok | ไม่มี | $1.5-3/MTok |
| ความเร็ว (Latency) | <50ms | 100-300ms | 80-200ms |
| การชำระเงิน | ¥1=$1, WeChat/Alipay | บัตรเครดิตเท่านั้น | บัตรเครดิต/PayPal |
| เครดิตฟรี | มีเมื่อลงทะเบียน | $5 ฟรี | ขึ้นอยู่กับผู้ให้บริการ |
ทำความเข้าใจ LangGraph Memory Architecture
LangGraph แบ่ง Memory ออกเป็น 2 ประเภทหลักที่ทำงานร่วมกัน:
- Short-term Memory (Checkpointer): จัดเก็บสถานะของการสนทนาปัจจุบัน ช่วยให้ Agent ตอบสนองต่อผู้ใช้อย่างต่อเนื่องในเซสชันเดียวกัน
- Long-term Memory (Store): จัดเก็บข้อมูลถาวรที่ Agent สามารถค้นหาและนำกลับมาใช้ในเซสชันต่างๆ ช่วยให้มีความรู้เกี่ยวกับผู้ใช้หรืองานเฉพาะ
การติดตั้งและตั้งค่าโครงสร้างพื้นฐาน
ก่อนเริ่มต้น คุณต้องติดตั้ง dependencies ที่จำเป็นและตั้งค่า HolySheep API:
# ติดตั้ง packages ที่จำเป็น
pip install langgraph langchain-core langchain-holysheep redis
หรือใช้ environment variables
export HOLYSHEEP_API_KEY="YOUR_HOLYSHEEP_API_KEY"
export HOLYSHEEP_BASE_URL="https://api.holysheep.ai/v1"การสร้าง Chatbot พื้นฐานพร้อม Short-term Memory
เริ่มต้นด้วยการสร้าง Chatbot ที่มีความจำระยะสั้น ซึ่งจะจดจำประวัติการสนทนาในเซสชันปัจจุบัน:
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_holysheep import HolySheepChat
from langchain_core.messages import HumanMessage, AIMessage
import os
ตั้งค่า HolySheep Chat ผ่าน LangChain
llm = HolySheepChat(
holysheep_api_key=os.getenv("HOLYSHEEP_API_KEY"),
model="gpt-4.1", # หรือ "claude-sonnet-4.5", "gemini-2.5-flash"
temperature=0.7
)
กำหนด system prompt
SYSTEM_PROMPT = """คุณเป็นผู้ช่วย AI ที่เป็นมิตร จดจำรายละเอียดสำคัญจากการสนทนา
เพื่อให้บริการที่ต่อเนื่องและเป็นส่วนตัวมากขึ้น"""
def should_continue(state: MessagesState) -> str:
messages = state["messages"]
last_message = messages[-1]
if last_message.content.lower() in ["exit", "quit", "หยุด"]:
return END
return "continue"
def call_model(state: MessagesState):
messages = state["messages"]
response = llm.invoke(
[{"role": "system", "content": SYSTEM_PROMPT}] + messages
)
return {"messages": [response]}
สร้าง Graph พร้อม Checkpointer
builder = StateGraph(MessagesState)
builder.add_node("chatbot", call_model)
builder.add_edge(START, "chatbot")
builder.add_conditional_edges("chatbot", should_continue)
builder.add_edge("chatbot", END)
ใช้ MemorySaver สำหรับ short-term memory
checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)
เริ่มการสนทนา
config = {"configurable": {"thread_id": "user-123-session-1"}}
print("เริ่มสนทนากับ AI Agent...")
result = graph.invoke(
{"messages": [HumanMessage(content="สวัสดีครับ ผมชื่อสมชาย")]}
config=config
)
print(f"AI: {result['messages'][-1].content}")การเพิ่ม Long-term Memory ด้วย In-memory Store
สำหรับการจัดเก็บข้อมูลถาวรที่ Agent สามารถเข้าถึงได้ในทุกเซสชัน เราใช้ Store API:
from langgraph.store.memory import MemoryStore
from langgraph.store.base import BaseStore
from typing import Optional
import uuid
สร้าง Long-term Memory Store
store = MemoryStore()
ฟังก์ชันสำหรับบันทึกข้อมูลผู้ใช้
def save_user_preference(user_id: str, key: str, value: str):
"""บันทึกความชอบของผู้ใช้ลงใน Long-term Memory"""
namespace = ("user_preferences", user_id)
store.put(
namespace=namespace,
key=key,
value={
"data": value,
"updated_at": str(uuid.uuid4()) # timestamp
}
)
def get_user_preference(user_id: str, key: str) -> Optional[dict]:
"""ดึงความชอบของผู้ใช้จาก Long-term Memory"""
namespace = ("user_preferences", user_id)
items = store.get(namespace=namespace, key=key)
return items[0].value if items else None
ตัวอย่าง: บันทึกและดึงข้อมูล
user_id = "user-456"
save_user_preference(user_id, "favorite_topic", "การเขียนโปรแกรม Python")
save_user_preference(user_id, "language", "ภาษาไทย")
favorite = get_user_preference(user_id, "favorite_topic")
print(f"ความชอบของผู้ใช้: {favorite}")การรวม Short-term และ Long-term Memory เข้าด้วยกัน
นี่คือตัวอย่างที่สมบูรณ์ซึ่งรวมทั้งสองระบบ Memory เข้าด้วยกัน ทำให้ Agent มีความจำทั้งระยะสั้นและระยะยาว:
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.store.memory import MemoryStore
from langchain_holysheep import HolySheepChat
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
import os
ตั้งค่า LLM
llm = HolySheepChat(
holysheep_api_key=os.getenv("HOLYSHEEP_API_KEY"),
model="gpt-4.1",
temperature=0.7
)
สร้าง Memory ทั้งสองประเภท
checkpointer = MemorySaver() # Short-term
store = MemoryStore() # Long-term
def retrieve_memory(state: MessagesState, config: dict) -> dict:
"""ค้นหาข้อมูลจาก Long-term Memory"""
user_id = config.get("configurable", {}).get("user_id", "anonymous")
namespace = ("user_preferences", user_id)
# ดึงข้อมูลทั้งหมดของผู้ใช้
memories = store.search(namespace=namespace, query="")
if memories:
memory_context = "\n".join([
f"- {item.key}: {item.value['data']}"
for item in memories
])
return {"memory_context": f"ข้อมูลผู้ใช้:\n{memory_context}"}
return {"memory_context": "ไม่มีข้อมูลในความจำระยะยาว"}
def chat_node(state: MessagesState, config: dict) -> dict:
"""Node หลักสำหรับการสนทนา"""
messages = state["messages"]
memory_context = state.get("memory_context", "")
system_message = SystemMessage(
content=f"""คุณเป็นผู้ช่วย AI อัจฉริยะ
{memory_context}
กรุณาใช้ข้อมูลจากความจำระยะยาว (ถ้ามี) เพื่อตอบสนองได้อย่างเป็นส่วนตัว"""
)
response = llm.invoke([system_message] + messages)
return {"messages": [response]}
def save_memory_node(state: MessagesState, config: dict) -> dict:
"""บันทึกข้อมูลสำคัญลง Long-term Memory"""
user_id = config.get("configurable", {}).get("user_id", "anonymous")
messages = state["messages"]
last_message = messages[-1].content
# ตรวจจับข้อมูลสำคัญ (simplified)
if "ชอบ" in last_message or "ชื่นชอบ" in last_message:
namespace = ("user_preferences", user_id)
store.put(namespace=namespace, key="interest", value={"data": last_message})
return state
สร้าง Graph
builder = StateGraph(MessagesState)
builder.add_node("retrieve", retrieve_memory)
builder.add_node("chat", chat_node)
builder.add_node("save_memory", save_memory_node)
builder.add_edge(START, "retrieve")
builder.add_edge("retrieve", "chat")
builder.add_edge("chat", "save_memory")
builder.add_edge("save_memory", END)
คอมไพล์ Graph
graph = builder.compile(checkpointer=checkpointer, store=store)
ทดสอบการทำงาน
config = {"configurable": {"thread_id": "session-1", "user_id": "user-789"}}
result = graph.invoke(
{"messages": [HumanMessage(content="สวัสดีครับ ผมชื่อมาร์ค ชอบอ่านหนังสือ sci-fi และเล่นเกม RPG")]}
config=config
)
print(f"AI: {result['messages'][-1].content}")การใช้ Redis สำหรับ Production Memory
สำหรับการใช้งานจริงในระดับ Production ควรใช้ Redis แทน In-memory Store เพื่อความคงทนและประสิทธิภาพ:
from langgraph.store.redis import RedisStore
from langgraph.checkpoint.redis import RedisSaver
import redis
ตั้งค่า Redis connection
redis_client = redis.Redis(host='localhost', port=6379, db=0)
สร้าง Redis-based memory stores
checkpointer = RedisSaver(redis_client) # Short-term
store = RedisStore(
client=redis_client,
index_name="langgraph_memory"
) # Long-term
ใช้งานเหมือนเดิม
graph = builder.compile(checkpointer=checkpointer, store=store)
ข้อมูลจะถูกจัดเก็บใน Redis อย่างถาวร
config = {"configurable": {"thread_id": "prod-session-1", "user_id": "prod-user"}}
result = graph.invoke(
{"messages": [HumanMessage(content="ทดสอบการจัดเก็บความจำใน Redis")]},
config=config
)