ในโลกของ AI Agent ที่ทำงานซับซ้อน การจัดการ State (สถานะ) ถือเป็นหัวใจสำคัญที่หลายคนมองข้าม ในฐานะ Senior AI Engineer ที่เคยพัฒนา AI Chatbot สำหรับระบบ E-commerce ขนาดใหญ่ ฉันเคยเจอปัญหาที่ผู้ใช้ถามเรื่องสินค้าในหน้า A แต่พอไปหน้า B ระบบกลับลืมทุกอย่าง นี่คือจุดที่ LangGraph State Management เข้ามาแก้ปัญหาได้อย่างตรงจุด

ทำไม LangGraph ถึงต่างจาก LangChain ในเรื่อง State Management

LangChain เป็นเครื่องมือที่ดีสำหรับงานง่ายๆ แต่เมื่อต้องจัดการ State ที่ซับซ้อน มักจะต้องเขียน Code เพิ่มเติมจำนวนมาก LangGraph ใช้ Graph-based Architecture ที่ออกแบบมาสำหรับ State Management โดยเฉพาะ ทำให้การ Debug และ Scale ทำได้ง่ายกว่ามาก

การติดตั้ง LangGraph และการใช้งานเบื้องต้น

ก่อนจะเข้าสู่เนื้อหาหลัก มาเริ่มติดตั้ง Package ที่จำเป็นกันก่อน:

pip install langgraph langchain-openai langchain-core


หรือใช้ Poetry

poetry add langgraph langchain-openai

พื้นฐาน State Management ใน LangGraph

State ใน LangGraph คือ Dictionary ที่เก็บข้อมูลทุกอย่างที่ต้องการคงไว้ตลอด Conversation โดยสามารถกำหนด Schema ได้อย่างยืดหยุ่น:

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from operator import itemgetter


กำหนด State Schema

class ConversationState(TypedDict):
    messages: list[dict]
    user_profile: dict | None
    cart_items: list[dict]
    conversation_turn: int
    context_summary: str


สร้าง Graph Builder

builder = StateGraph(ConversationState)


เพิ่ม Node สำหรับจัดการ State

def process_message(state: ConversationState) -> ConversationState:
    """Process ข้อความและอัพเดต State"""
    new_turn = state.get("conversation_turn", 0) + 1
    
    return {
        "messages": state["messages"],
        "user_profile": state.get("user_profile", {}),
        "cart_items": state.get("cart_items", []),
        "conversation_turn": new_turn,
        "context_summary": f"Turn {new_turn}: {len(state['messages'])} messages"
    }


คอมไพล์ Graph พร้อม Memory Checkpointing

graph = builder.compile(checkpointer=MemorySaver())

การ Integrate กับ LLM Provider (HolySheep AI)

สำหรับการเรียกใช้ LLM ในโปรเจกต์จริง ฉันแนะนำ HolySheep AI ที่มีความเร็วตอบสนองน้อยกว่า 50ms และราคาประหยัดกว่า 85% เมื่อเทียบกับ OpenAI:

from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent


ใช้ HolySheep AI แทน OpenAI

llm = ChatOpenAI(
    base_url="https://api.holysheep.ai/v1",
    api_key="YOUR_HOLYSHEEP_API_KEY",  # เปลี่ยนเป็น API Key จริง
    model="gpt-4.1",
    temperature=0.7,
)


สร้าง Agent พร้อม Tool

agent = create_react_agent(
    llm,
    tools=[search_products, add_to_cart, get_order_status],
    state_schema=ConversationState
)


เริ่ม Conversation

config = {"configurable": {"thread_id": "user_12345"}}
initial_state = {
    "messages": [],
    "user_profile": {"user_id": "12345", "tier": "gold"},
    "cart_items": [],
    "conversation_turn": 0,
    "context_summary": "Start"
}


Run Agent

result = agent.invoke(
    {"messages": [{"role": "user", "content": "ฉันอยากดูรองเท้าผ้าใบสีขาว"}]},
    config=config
)

Persistence Strategy สำหรับ Production

สำหรับระบบ Production จริง MemorySaver เดียวไม่เพียงพอ ต้องใช้ Database Persistence เพื่อรองรับ Horizontal Scaling:

from langgraph.checkpoint.postgres import PostgresSaver
from sqlalchemy import create_engine
import json


PostgreSQL Checkpointer สำหรับ Production

engine = create_engine("postgresql://user:pass@localhost/conversations")
checkpointer = PostgresSaver(engine)


Custom Serializer สำหรับ Complex Objects

class EnhancedPostgresSaver(PostgresSaver):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
    
    def serialize(self, state: dict) -> str:
        """Serialize State พร้อม Handle Complex Types"""
        # แปลง objects ที่ไม่สามารถ JSON serialize ได้
        cleaned_state = self._clean_state(state)
        return json.dumps(cleaned_state, default=str)
    
    def _clean_state(self, state: dict) -> dict:
        """ลบ fields ที่ไม่จำเป็นออก"""
        exclude_fields = {"__root__", "__parent__", "__id__"}
        return {
            k: v for k, v in state.items() 
            if k not in exclude_fields
        }


Compile Graph ด้วย Enhanced Checkpointer

production_graph = builder.compile(checkpointer=EnhancedPostgresSaver(engine))

การกู้คืน Conversation State

หนึ่งในฟีเจอร์ที่สำคัญที่สุดคือการกู้คืน Session เก่า ซึ่งมีประโยชน์มากสำหรับ E-commerce Chatbot ที่ลูกค้าอาจกลับมาหลังจากหลายวัน:

from datetime import datetime, timedelta

class ConversationRecovery:
    """Class สำหรับจัดการการกู้คืน Conversation"""
    
    def __init__(self, checkpointer):
        self.checkpointer = checkpointer
    
    def get_last_session(self, user_id: str, days_back: int = 30) -> dict | None:
        """ดึง Session ล่าสุดของ User"""
        cutoff_date = datetime.now() - timedelta(days=days_back)
        
        # ค้นหา Thread ID จาก Database
        thread_id = self._find_thread_by_user(user_id, cutoff_date)
        
        if not thread_id:
            return None
        
        # ดึง State ล่าสุด
        config = {"configurable": {"thread_id": thread_id}}
        
        # อ่าน checkpoint ล่าสุด
        checkpoint = self.checkpointer.get(config)
        
        if checkpoint and "channel_values" in checkpoint:
            return checkpoint["channel_values"]
        
        return None
    
    def resume_conversation(self, user_id: str, new_message: str) -> dict:
        """กู้คืนและสานต่อ Conversation"""
        old_state = self.get_last_session(user_id)
        
        if old_state is None:
            # สร้าง Session ใหม่
            return self._create_new_session(user_id, new_message)
        
        # Resume ด้วย State เดิม
        config = {"configurable": {"thread_id": old_state.get("thread_id")}}
        resumed_state = {
            "messages": old_state["messages"] + [{"role": "user", "content": new_message}],
            "user_profile": old_state.get("user_profile"),
            "cart_items": old_state.get("cart_items", []),
            "conversation_turn": old_state.get("conversation_turn", 0) + 1,
            "context_summary": f"Resumed from previous session"
        }
        
        return resumed_state


ใช้งาน

recovery = ConversationRecovery(production_graph.checkpointer)
resumed = recovery.resume_conversation("user_12345", "รองเท้าที่ฉันดูเมื่อวานยังมีไหม?")

แนวทางปฏิบัติที่ดีที่สุด (Best Practices)

ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข

ข้อผิดพลาด สาเหตุ วิธีแก้ไข
State หายหลังจาก Restart ใช้ MemorySaver ซึ่งเก็บใน RAM เท่านั้น เปลี่ยนเป็น PostgresSaver หรือ RedisSaver สำหรับ Production
thread_id ซ้ำกัน ไม่ได้กำหนด Unique Thread ID ต่อ User ใช้ Format: f"{user_id}_{session_timestamp}" หรือ UUID
State โตเกิน Memory messages สะสมไม่หยุดโดยไม่มีการ truncate กำหนด max_messages และใช้ Summarization หรือ sliding window
Serialization Error State มี Object ที่ไม่สามารถ JSON serialize ได้ ใช้ Custom Serializer หรือแปลง Object เป็น dict ก่อนเก็บ
Checkpoint Conflict หลาย Processes เขียน Checkpoint พร้อมกัน ใช้ Database Transaction หรือ Locking Mechanism 
# ตัวอย่าง: การ truncate messages อัตโนมัติ
def truncate_messages(state: ConversationState, max_messages: int = 20) -> ConversationState:
    """ตัด messages เก่าออกแต่เก็บ context summary"""
    messages = state.get("messages", [])
    if len(messages) > max_messages:
        # เก็บ system prompt และ messages ล่าสุด
        system_msg = [m for m in messages if m.get("role") == "system"]
        recent = messages[-max_messages:]
        
        state["messages"] = system_msg + recent
        state["context_summary"] = f"Summarized from {len(messages)} messages"
    
    return state

ตารางเปรียบเทียบ Checkpointer Options

Checkpointer ข้อดี ข้อเสีย เหมาะกับ
MemorySaver เร็ว, ง่าย, ไม่ต้องตั้งค่า สูญหายเมื่อ restart, ไม่ Scale Development, Prototype
PostgresSaver Persistent, Scalable, Transaction Support ต้องตั้งค่า PostgreSQL, ช้ากว่า Memory Production ขนาดกลาง-ใหญ่
RedisSaver เร็วมาก, Persistent, TTL Support ต้องตั้งค่า Redis, RAM-dependent High-traffic, Real-time Applications
SQLiteSaver ไม่ต้องตั้งค่า Server, File-based ไม่เหมาะกับ Multi-instance Small Production, Edge Deployment

Performance Optimization Tips

จากประสบการณ์ในโปรเจกต์ E-commerce ที่มี Traffic หลายพัน Requests ต่อวินาที มีเทคนิคที่ช่วยเพิ่ม Performance:

import asyncio
from functools import lru_cache

class OptimizedStateManager:
    """State Manager ที่ Optimize สำหรับ High-throughput"""
    
    def __init__(self, checkpointer, cache_size: int = 1000):
        self.checkpointer = checkpointer
        self._cache = {}
        self._max_cache = cache_size
    
    def _get_cache_key(self, thread_id: str) -> str:
        return f"state:{thread_id}"
    
    async def get_state_cached(self, config: dict) -> dict | None:
        """ดึง State พร้อม Caching"""
        thread_id = config["configurable"]["thread_id"]
        cache_key = self._get_cache_key(thread_id)
        
        # ลองดึงจาก Cache ก่อน
        if cache_key in self._cache:
            return self._cache[cache_key]
        
        # ดึงจาก Database
        state = await self._async_get_checkpoint(config)
        
        if state:
            # เพิ่มเข้า Cache
            if len(self._cache) >= self._max_cache:
                # Remove oldest
                self._cache.pop(next(iter(self._cache)))
            self._cache[cache_key] = state
        
        return state
    
    async def _async_get_checkpoint(self, config: dict) -> dict | None:
        """Async wrapper สำหรับ Checkpointer"""
        loop = asyncio.get_event_loop()
        return await loop.run_in_executor(
            None, 
            self.checkpointer.get, 
            config
        )


ใช้งาน

manager = OptimizedStateManager(production_graph.checkpointer)
state = await manager.get_state_cached(config)

สรุป

LangGraph State Management เป็นหัวใจสำคัญสำหรับการสร้าง AI Agent ที่ทรงพลังและน่าเชื่อถือ การเลือก Persistence Strategy ที่เหมาะสมกับ Scale ของโปรเจกต์จะช่วยประหยัดเวลาในการ Debug และเพิ่มประสิทธิภาพในการ Scale ระบบได้อย่างมาก

สำหรับใครที่กำลังมองหา LLM Provider ที่คุ้มค่าและเร็ว ลองพิจารณา HolySheep AI ที่มีความเร็วตอบสนองน้อยกว่า 50ms พร้อมราคาที่ประหยัดกว่า OpenAI ถึง 85% รองรับทั้ง WeChat และ Alipay พร้อมเครดิตฟรีเมื่อลงทะเบียน

👉 สมัคร HolySheep AI — รับเครดิตฟรีเมื่อลงทะเบียน

แหล่งข้อมูลที่เกี่ยวข้อง

บทความที่เกี่ยวข้อง