LangGraph 90K Star กับการสร้าง Production-Grade AI Agent โดยใช้ Stateful Workflow Engine

ในโลกของ AI Agent การสร้างระบบที่จำสถานะการทำงานของ User ได้ตลอดเวลาเป็นสิ่งสำคัญมาก โดยเฉพาะเมื่อต้องทำงานซับซ้อนหลายขั้นตอน LangGraph ได้กลายเป็นเครื่องมือที่ Developer ทั่วโลกไว้วางใจ เพราะสามารถจัดการ Stateful Workflow ได้อย่างมีประสิทธิภาพ ในบทความนี้เราจะมาเรียนรู้วิธีการสร้าง Production-Grade AI Agent ตั้งแต่พื้นฐานจนถึงการนำไปใช้งานจริง

ปัญหาจริง: ทำไม AI Agent ถึงต้องการ Stateful Workflow

สมมติว่าคุณกำลังพัฒนา Customer Support Agent และพบว่าเมื่อ User ถามคำถามหลายข้อติดต่อกัน Agent ไม่สามารถจำได้ว่ากำลังอยู่ในขั้นตอนไหนของการแก้ปัญหา นี่คือตัวอย่างข้อผิดพลาดที่เกิดขึ้นจริงเมื่อไม่มี Stateful System:

# ปัญหาที่พบเมื่อไม่มี Stateful Workflow
class NaiveAgent:
    def __init__(self, api_key):
        self.api_key = api_key  # ไม่มี memory หรือ state
        
    def chat(self, message):
        # ทุกครั้งที่ถูกเรียกจะเริ่มใหม่เสมอ
        response = requests.post(
            "https://api.holysheep.ai/v1/chat/completions",
            headers={"Authorization": f"Bearer {self.api_key}"},
            json={"messages": [{"role": "user", "content": message}]}
        )
        return response.json()

ผลลัพธ์ที่ไม่ต่อเนื่อง
agent = NaiveAgent("YOUR_HOLYSHEEP_API_KEY")
print(agent.chat("ฉันต้องการสั่งซื้อสินค้า"))  # เริ่มใหม่ทุกครั้ง
print(agent.chat("สินค้าที่ฉันเลือกคือ item123"))  # ไม่รู้ว่า Item คืออะไร
print(agent.chat("ยืนยันการสั่งซื้อ"))  # ไม่มี Context เลย

ข้อผิดพลาดประเภทนี้ทำให้ User Experience แย่มาก เพราะต้องอธิบายบริบทซ้ำๆ กัน ซึ่ง LangGraph ช่วยแก้ปัญหานี้ได้ด้วยการออกแบบ Graph-Based Architecture ที่มี State Management ในตัว

พื้นฐาน LangGraph: เข้าใจ Stateful Graph

LangGraph สร้างขึ้นจากแนวคิด Graph-Based State Machine โดยมีองค์ประกอบหลักดังนี้:

• State - ข้อมูลที่ถูกส่งผ่านระหว่าง Node ทั้งหมด
• Node - Function ที่รับ State แล้ว Return State ใหม่
• Edge - เส้นทางการไหลของ State ระหว่าง Node
• Graph - การรวม Node และ Edge เข้าด้วยกัน

มาเริ่มต้นสร้าง Customer Support Agent แบบ Stateful กัน โดยใช้ HolySheep AI สำหรับ LLM API ซึ่งมีความเร็วต่ำกว่า 50ms และราคาประหยัดกว่า OpenAI ถึง 85%

# ติดตั้ง dependencies
pip install langgraph langchain-openai requests

import os
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
import requests

Configuration สำหรับ HolySheep AI
os.environ["HOLYSHEEP_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"

กำหนด State Schema สำหรับ Customer Support Agent
class CustomerSupportState(TypedDict):
    messages: list[dict]
    user_intent: str | None
    order_status: str | None
    collected_info: dict
    current_node: str

def create_holysheep_response(messages: list, model: str = "gpt-4o") -> str:
    """สร้าง response จาก HolySheep AI API"""
    response = requests.post(
        f"{HOLYSHEEP_BASE_URL}/chat/completions",
        headers={
            "Authorization": f"Bearer {os.environ['HOLYSHEEP_API_KEY']}",
            "Content-Type": "application/json"
        },
        json={
            "model": model,
            "messages": messages,
            "temperature": 0.7
        },
        timeout=30
    )
    response.raise_for_status()
    return response.json()["choices"][0]["message"]["content"]

Node แรก: รับข้อความจาก User
def receive_message(state: CustomerSupportState) -> CustomerSupportState:
    return {"current_node": "receive_message"}

Node ที่สอง: วิเคราะห์ Intent ของ User
def analyze_intent(state: CustomerSupportState) -> CustomerSupportState:
    system_prompt = """คุณคือ Customer Support AI วิเคราะห์ Intent ของ User 
    จากข้อความล่าสุด และ Return intent ที่เหมาะสม:
    - order: ต้องการสั่งซื้อสินค้า
    - inquiry: สอบถามข้อมูล
    - complaint: ร้องเรียนปัญหา
    - cancel: ต้องการยกเลิก
    - confirm: ยืนยันการดำเนินการ"""
    
    messages = [{"role": "system", "content": system_prompt}]
    messages.extend(state["messages"][-5:])
    
    intent = create_holysheep_response(messages)
    return {"user_intent": intent.strip(), "current_node": "analyze_intent"}

Node ที่สาม: ประมวลผลตาม Intent
def process_request(state: CustomerSupportState) -> CustomerSupportState:
    intent = state.get("user_intent", "inquiry")
    
    if "order" in intent.lower():
        new_status = "collecting_order_info"
    elif "cancel" in intent.lower():
        new_status = "processing_cancellation"
    else:
        new_status = "general_inquiry"
    
    return {
        "order_status": new_status,
        "current_node": "process_request"
    }

Node สุดท้าย: สร้าง Response
def generate_response(state: CustomerSupportState) -> CustomerSupportState:
    system_prompt = """คุณคือ Customer Support Agent ที่เป็นมิตร 
    ตอบคำถามลูกค้าอย่างเป็นธรรมชาติ โดยอ้างอิงจาก Order Status: """
    system_prompt += state.get("order_status", "unknown")
    
    messages = [{"role": "system", "content": system_prompt}]
    messages.extend(state["messages"][-5:])
    
    response = create_holysheep_response(messages)
    
    new_messages = state["messages"] + [{"role": "assistant", "content": response}]
    return {
        "messages": new_messages,
        "current_node": "generate_response"
    }

สร้าง Graph
workflow = StateGraph(CustomerSupportState)

เพิ่ม Nodes
workflow.add_node("receive_message", receive_message)
workflow.add_node("analyze_intent", analyze_intent)
workflow.add_node("process_request", process_request)
workflow.add_node("generate_response", generate_response)

กำหนด Flow
workflow.set_entry_point("receive_message")
workflow.add_edge("receive_message", "analyze_intent")
workflow.add_edge("analyze_intent", "process_request")
workflow.add_edge("process_request", "generate_response")
workflow.add_edge("generate_response", END)

Compile Graph
app = workflow.compile()

ทดสอบการทำงาน
initial_state = {
    "messages": [{"role": "user", "content": "สวัสดีครับ ผมต้องการสั่งซื้อ laptop"}],
    "user_intent": None,
    "order_status": None,
    "collected_info": {},
    "current_node": "start"
}

result = app.invoke(initial_state)
print(result["messages"][-1]["content"])

Multi-Agent Orchestration ด้วย LangGraph

ใน Production จริง คุณอาจต้องการหลาย Agent ทำงานร่วมกัน เช่น Order Agent, Complaint Agent, และ Billing Agent แต่ละตัวมีความเชี่ยวชาญเฉพาะด้าน LangGraph ช่วยให้การ Orchestrate Agent เหล่านี้ทำได้ง่าย

from typing import Literal
from langgraph.graph import MessagesState

Multi-Agent State
class MultiAgentState(MessagesState):
    primary_agent: str
    task_queue: list[str]
    completed_tasks: list[str]

def routing_node(state: MultiAgentState) -> MultiAgentState:
    """ตัดสินใจว่าจะส่งงานไป Agent ไหน"""
    last_message = state["messages"][-1]["content"]
    
    routing_prompt = f"""วิเคราะห์ข้อความนี้และบอกว่าควรส่งไป Agent ไหน:
    {last_message}
    
    Agent ที่มี:
    - order_agent: จัดการคำสั่งซื้อ
    - complaint_agent: จัดการเรื่องร้องเรียน
    - billing_agent: จัดการเรื่องบิลและการชำระเงิน
    - general_agent: ตอบคำถามทั่วไป
    
    Return เฉพาะชื่อ Agent เท่านั้น"""
    
    route = create_holysheep_response([{"role": "user", "content": routing_prompt}])
    route = route.strip().lower().replace("_agent", "") + "_agent"
    
    return {"primary_agent": route}

def order_agent_node(state: MultiAgentState) -> MultiAgentState:
    """Agent สำหรับจัดการคำสั่งซื้อ"""
    system = """คุณคือ Order Agent ผู้เชี่ยวชาญด้านการสั่งซื้อสินค้า
    ช่วยลูกค้าค้นหาสินค้า เปรียบเทียบราคา และดำเนินการสั่งซื้อ"""
    
    messages = [{"role": "system", "content": system}] + state["messages"]
    response = create_holysheep_response(messages)
    
    return {
        "messages": state["messages"] + [{"role": "assistant", "content": response}],
        "completed_tasks": state["completed_tasks"] + ["order"]
    }

def complaint_agent_node(state: MultiAgentState) -> MultiAgentState:
    """Agent สำหรับจัดการเรื่องร้องเรียน"""
    system = """คุณคือ Complaint Agent ผู้เชี่ยวชาญด้านการรับเรื่องร้องเรียน
    รับฟังปัญหาของลูกค้าอย่างใจเย็น และหาทางออกที่เหมาะสม"""
    
    messages = [{"role": "system", "content": system}] + state["messages"]
    response = create_holysheep_response(messages)
    
    return {
        "messages": state["messages"] + [{"role": "assistant", "content": response}],
        "completed_tasks": state["completed_tasks"] + ["complaint"]
    }

def create_multi_agent_graph():
    """สร้าง Multi-Agent Graph"""
    workflow = StateGraph(MultiAgentState)
    
    workflow.add_node("router", routing_node)
    workflow.add_node("order_agent", order_agent_node)
    workflow.add_node("complaint_agent", complaint_agent_node)
    
    def route_to_agent(state: MultiAgentState) -> str:
        return state.get("primary_agent", "order_agent")
    
    workflow.set_entry_point("router")
    workflow.add_conditional_edges("router", route_to_agent)
    workflow.add_edge("order_agent", END)
    workflow.add_edge("complaint_agent", END)
    
    return workflow.compile()

ทดสอบ Multi-Agent
multi_agent = create_multi_agent_graph()

test_state = {
    "messages": [{"role": "user", "content": "สินค้าที่สั่งไปยังไม่มาถึงเลย"}],
    "primary_agent": "",
    "task_queue": [],
    "completed_tasks": []
}

result = multi_agent.invoke(test_state)
print(f"Routing to: {result['primary_agent']}")
print(f"Response: {result['messages'][-1]['content']}")

Memory และ Persistence สำหรับ Stateful Conversation

ใน Production Environment คุณต้องการให้ State ถูกเก็บรักษาไว้ระหว่าง Session ซึ่ง LangGraph รองรับการต่อ Memory กับ Database ต่างๆ เช่น Redis, PostgreSQL หรือ SQLite

from langgraph.checkpoint.sqlite import SqliteSaver
import json

class PersistentCustomerSupport:
    def __init__(self, db_path: str = "./chat_history.db"):
        self.checkpointer = SqliteSaver.from_conn_string(db_path)
        self.app = self._build_graph()
    
    def _build_graph(self):
        workflow = StateGraph(CustomerSupportState)
        # ... เพิ่ม nodes ตามต้องการ ...
        workflow.add_node("analyze_intent", analyze_intent)
        workflow.add_node("generate_response", generate_response)
        workflow.set_entry_point("analyze_intent")
        workflow.add_edge("analyze_intent", "generate_response")
        workflow.add_edge("generate_response", END)
        return workflow.compile(checkpointer=self.checkpointer)
    
    def chat(self, user_id: str, message: str) -> str:
        """ส่งข้อความพร้อม Memory จาก Session ก่อนหน้า"""
        config = {"configurable": {"thread_id": user_id}}
        
        current_state = {
            "messages": [{"role": "user", "content": message}],
            "user_intent": None,
            "order_status": None,
            "collected_info": {},
            "current_node": "start"
        }
        
        result = self.app.invoke(current_state, config=config)
        return result["messages"][-1]["content"]
    
    def get_conversation_history(self, user_id: str) -> list:
        """ดึงประวัติการสนทนาทั้งหมด"""
        config = {"configurable": {"thread_id": user_id}}
        history = []
        
        for state in self.app.get_state_history(config):
            history.append({
                "messages": state.values.get("messages", []),
                "user_intent": state.values.get("user_intent"),
                "order_status": state.values.get("order_status")
            })
        
        return history

ใช้งาน Persistent Memory
support_system = PersistentCustomerSupport("./production_chats.db")

Session 1
print(support_system.chat("user123", "อยากสั่งซื้อ laptop ASUS"))
Session 2 - ระบบจำได้ว่า User ต้องการ ASUS laptop
print(support_system.chat("user123", "รุ่นไหนดี?"))  
print(support_system.chat("user123", "เอา ZenBook 14 นะ"))
Session 3 - วันถัดมา ยังจำได้
print(support_system.chat("user123", "ติดตามสถานะสั่งซื้อ"))

Production Deployment: Error Handling และ Monitoring

เมื่อนำไปใช้งานจริง คุณต้องมี Error Handling ที่ดี เพื่อให้ระบบทำงานต่อได้แม้เกิดข้อผิดพลาดบางอย่าง นี่คือ Pattern ที่แนะนำสำหรับ Production

import logging
from functools import wraps
from datetime import datetime

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ProductionLangGraphAgent:
    def __init__(self, api_key: str, max_retries: int = 3):
        self.api_key = api_key
        self.max_retries = max_retries
        self.base_url = "https://api.holysheep.ai/v1"
        self.workflow = self._build_resilient_graph()
    
    def _build_resilient_graph(self):
        """สร้าง Graph พร้อม Error Handling"""
        
        def safe_analyze_intent(state: CustomerSupportState) -> CustomerSupportState:
            try:
                return analyze_intent(state)
            except requests.exceptions.Timeout:
                logger.error("Timeout ในการเรียก analyze_intent")
                return {
                    **state, 
                    "user_intent": "timeout_error",
                    "messages": state["messages"] + [{
                        "role": "assistant", 
                        "content": "ขออภัยครับ เกิดปัญหาการเชื่อมต่อ กรุณาลองใหม่อีกครั้ง"
                    }]
                }
            except Exception as e:
                logger.error(f"Error ใน analyze_intent: {str(e)}")
                return {
                    **state,
                    "user_intent": "system_error",
                    "messages": state["messages"] + [{
                        "role": "assistant",
                        "content": "ระบบเกิดข้อผิดพลาด กรุณาติดต่อฝ่ายสนับสนุน"
                    }]
                }
        
        workflow = StateGraph(CustomerSupportState)
        workflow.add_node("analyze_intent", safe_analyze_intent)
        workflow.add_node("generate_response", generate_response)
        workflow.set_entry_point("analyze_intent")
        workflow.add_edge("analyze_intent", "generate_response")
        workflow.add_edge("generate_response", END)
        
        return workflow.compile()
    
    def process_with_retry(self, state: dict, thread_id: str) -> dict:
        """ประมวลผลพร้อม Retry Logic"""
        config = {"configurable": {"thread_id": thread_id}}
        
        for attempt in range(self.max_retries):
            try:
                result = self.workflow.invoke(state, config=config)
                logger.info(f"Success on attempt {attempt + 1} for thread {thread_id}")
                return result
            except requests.exceptions.ConnectionError as e:
                logger.warning(f"Connection error on attempt {attempt + 1}: {e}")
                if attempt == self.max_retries - 1:
                    raise
            except Exception as e:
                logger.error(f"Unexpected error: {e}")
                raise
        
        return None
    
    def health_check(self) -> dict:
        """ตรวจสอบสถานะระบบ"""
        try:
            test_response = requests.get(f"{self.base_url}/models", 
                headers={"Authorization": f"Bearer {self.api_key}"},
                timeout=5)
            return {
                "status": "healthy" if test_response.status_code == 200 else "degraded",
                "api_status": test_response.status_code,
                "timestamp": datetime.now().isoformat()
            }
        except Exception as e:
            return {
                "status": "unhealthy",
                "error": str(e),
                "timestamp": datetime.now().isoformat()
            }

ทดสอบ Production Agent
agent = ProductionLangGraphAgent("YOUR_HOLYSHEEP_API_KEY")
print(agent.health_check())

ทดสอบ Error Handling
result = agent.process_with_retry(
    {
        "messages": [{"role": "user", "content": "ทดสอบระบบ"}],
        "user_intent": None,
        "order_status": None,
        "collected_info": {},
        "current_node": "start"
    },
    "test_thread_001"
)
print(result["messages"][-1]["content"])

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

1. 401 Unauthorized Error

สถานการณ์ข้อผิดพลาด: เมื่อคุณเรียกใช้ API แล้วได้รับ Error 401 Unauthorized ซึ่งหมายความว่า API Key ไม่ถูกต้องหรือหมดอายุ

# ❌ วิธีที่ทำให้เกิด 401 Error
response = requests.post(
    "https://api.holysheep.ai/v1/chat/completions",
    headers={"Authorization": "Bearer YOUR_HOLYSHEEP_API_KEY"},  # Key ตรงๆ ไม่ดี
    json={"model": "gpt-4o", "messages": [...]}
)

✅ วิธีที่ถูกต้อง - ใช้ Environment Variable
import os
os.environ["HOLYSHEEP_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"  # ตั้งค่า ENV ก่อน

def create_api_client():
    api_key = os.environ.get("HOLYSHEEP_API_KEY")
    if not api_key:
        raise ValueError("HOLYSHEEP_API_KEY not found in environment")
    
    return api_key

ตรวจสอบ Key ก่อนเรียกใช้
api_key = create_api_client()
print(f"Using API Key starting with: {api_key[:7]}...")  # แสดงแค่ 7 ตัวอักษรแรก

เรียกใช้ API อย่างปลอดภัย
response = requests.post(
    "https://api.holysheep.ai/v1/chat/completions",
    headers={"Authorization": f"Bearer {api_key}"},
    json={"model": "gpt-4o", "messages": [{"role": "user", "content": "Hello"}]}
)

if response.status_code == 401:
    print("401 Error: กรุณาตรวจสอบ API Key ที่ https://www.holysheep.ai/api-keys")

2. Timeout Error และ ConnectionError

สถานการณ์ข้อผิดพลาด: เมื่อ Network ช้าหรือ Server โหลดมาก จะเกิด requests.exceptions.ReadTimeout หรือ ConnectionError

# ❌ โค้ดที่ไม่มี Timeout handling
response = requests.post(
    f"{HOLYSHEEP_BASE_URL}/chat/completions",
    headers={"Authorization": f"Bearer {api_key}"},
    json={"model": "gpt-4o", "messages": messages}  # ไม่มี timeout
)

✅ โค้ดที่มี Timeout และ Retry Logic
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def call_api_with_retry(session, url: str, headers: dict, payload: dict) -> dict:
    """เรียก API พร้อม Retry และ Timeout"""
    try:
        response = session.post(
            url,
            headers=headers,
            json=payload,
            timeout=30  # Timeout 30 วินาที
        )
        response.raise_for_status()
        return response.json()
    except requests.exceptions.Timeout:
        print("Request timeout - กำลัง retry...")
        raise
    except requests.exceptions.ConnectionError as e:
        print(f"Connection error: {e} - กำลัง retry...")
        raise
    except requests.exceptions.HTTPError as e:
        if response.status_code == 429:
            print("Rate limited - รอก่อน retry...")
            time.sleep(60)  # รอ 1 นาทีถ้าโดน rate limit
            raise
        raise

ใช้งาน
session = requests.Session()
result = call_api_with_retry(
    session,
    f"{HOLYSHEEP_BASE_URL}/chat/completions",
    {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
    {"model": "gpt-4o", "messages": messages}
)

3. State Management Error ใน LangGraph

สถานการณ์ข้อผิดพลาด: เมื่อ State ไม่ถูกอัปเดตอย่างถูกต้อง หรือ Graph State ไม่ตรงกับ Schema ที่กำหนด

# ❌ State Schema ไม่ตรงกับการใช้งานจริง
class WrongState(TypedDict):
    messages: list  # ไม่ระบุ type ของ messages

def bad_node(state: WrongState):
    # พยายามเข้าถึง field ที่ไม่มีใน State
    return {"user_id": "123"}  # ❌ Error: user_id ไม่อยู่ใน WrongState

✅ State Schema ที่ถูกต้อง
class CorrectState(TypedDict):
    messages: list[dict]  # ระบุ type ชัดเจน
    user_id: str | None
    session_data: dict

def good_node(state: CorrectState) -> CorrectState:
    # สามารถเข้าถึง field ที่มีใน State ได้อย่างปลอดภัย
    return {
        "user_id": "123",
        "session_data": {"last_action": "analyze"}
    }

ตรวจสอบ State ก่อน Return
def safe_node(state: CorrectState) -> CorrectState:
    # ตรวจสอบว่า field จำเป็นมีค่าหรือไม่
    if not state.get("user_id"):
        return {
แหล่งข้อมูลที่เกี่ยวข้อง
📚 บทช่วยสอน AI API
💰 ดูราคา
📖 เอกสารสำหรับนักพัฒนา
🚀 สมัครฟรี
บทความที่เกี่ยวข้อง
Cursor Agent โหมดภาคปฏิบัติ: การเปลี่ยนแปลงรูปแบบการพัฒนา AI
CrewAI กับ A2A Protocol: คู่มือฉบับสมบูรณ์สำหรับ Multi-Agent
AI短剧制作爆发：200部春节短剧背后的AI视频生成技术栈解析