หากคุณกำลังพัฒนา AI Agent ที่ต้องการความซับซ้อนในการจัดการสถานะ (Stateful) และ workflow ที่ยืดหยุ่น LangGraph คือเครื่องมือที่ GitHub มี stars เกิน 90,000 ครั้ง และถูกใช้งานจริงใน production ขององค์กรชั้นนำทั่วโลก ในบทความนี้ผมจะพาคุณเจาะลึกการทำงาน พร้อมตัวอย่างโค้ดที่ใช้งานได้จริง และวิเคราะห์ต้นทุนอย่างละเอียด

ทำไมต้อง LangGraph? ทำความเข้าใจ Stateful vs Stateless Agent

AI Agent พื้นฐานที่ใช้ prompt ธรรมดาเป็น Stateless — ทุกครั้งที่ส่ง request ใหม่ ระบบจะ "ลืม" ทุกอย่างที่เคยคุยมา แต่ในโลกจริง AI Agent ต้องการ:

LangGraph สร้างมาเพื่อแก้ปัญหานี้โดยเฉพาะ ด้วยสถาปัตยกรรม Graph-based ที่ทำให้การจัดการ state สะอาดและ predictable

ตารางเปรียบเทียบต้นทุน LLM APIs ปี 2026

ก่อนจะเข้าสู่โค้ด มาดูต้นทุนที่สำคัญสำหรับการ deploy AI Agent ใน production กันก่อน:

ModelOutput Price ($/MTok)10M Tokens/เดือนDeepSeek ประหยัด
GPT-4.1$8.00$80.0095%
Claude Sonnet 4.5$15.00$150.0097%
Gemini 2.5 Flash$2.50$25.0083%
DeepSeek V3.2$0.42$4.20—

จะเห็นได้ว่า DeepSeek V3.2 มีราคาถูกกว่า GPT-4.1 ถึง 95% ซึ่งสำหรับ workflow ที่ต้องเรียก LLM หลายครั้งต่อการทำงานหนึ่งครั้ง (เช่น multi-step agent) ต้นทุนนี้จะส่งผลต่อ ROI อย่างมาก

สถาปัตยกรรมของ LangGraph: Graph, Nodes, และ Edges

LangGraph ใช้ concept ง่ายๆ คือทุกอย่างคือ Graph ที่ประกอบด้วย:

┌─────────────────────────────────────────────────────────────┐
│                      LangGraph Flow                          │
│                                                             │
│   ┌──────────┐    ┌──────────┐    ┌──────────┐             │
│   │  START   │───▶│  Node 1  │───▶│  Node 2  │             │
│   └──────────┘    └────┬─────┘    └────┬─────┘             │
│                        │               │                    │
│                        ▼               ▼                    │
│                  ┌──────────┐    ┌──────────┐               │
│                  │Condition │    │  Node 3  │               │
│                  └────┬─────┘    └────┬─────┘               │
│                       │               │                    │
│              ┌────────┴────────┐      │                    │
│              ▼                 ▼      ▼                    │
│        ┌────────┐         ┌────────┐│ END                 │
│        │ Node A │         │ Node B ││                     │
│        └────┬───┘         └────┬───┘│                     │
│             └────────┬────────┬─┘                           │
│                      ▼                                      │
│                 ┌────────┐                                  │
│                 │  END   │                                  │
│                 └────────┘                                  │
└─────────────────────────────────────────────────────────────┘

การติดตั้งและ Setup Environment

pip install langgraph langchain-core langchain-holysheep

ตัวอย่างที่ 1: Customer Support Agent พื้นฐาน

มาเริ่มกันง่ายๆ ด้วย Customer Support Agent ที่จำสถานะการสนทนาและตัดสินใจ route ปัญหาไปยัง department ที่เหมาะสม:

import os
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_holysheep import ChatHolySheep


Setup HolySheep API - ราคาถูกกว่า 85%+ เมื่อเทียบกับ official API

os.environ["HOLYSHEEP_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"

llm = ChatHolySheep(
    model="deepseek-v3.2",
    base_url="https://api.holysheep.ai/v1",
    api_key=os.environ["HOLYSHEEP_API_KEY"],
    temperature=0.7
)


Define State Schema - กำหนดโครงสร้างข้อมูลที่ agent จะจำ

class SupportAgentState(TypedDict):
    messages: list
    intent: str | None
    department: str | None
    resolution_status: str

def classify_intent(state: SupportAgentState) -> SupportAgentState:
    """Node 1: วิเคราะห์ว่าลูกค้าต้องการอะไร"""
    messages = state["messages"]
    last_message = messages[-1].content
    
    prompt = f"""Classify customer intent into one of these categories:
    - billing: ปัญหาเกี่ยวกับบิลหรือการชำระเงิน
    - technical: ปัญหาทางเทคนิคหรือ bug
    - sales: สอบถามเกี่ยวกับ product ใหม่
    - complaint: ร้องเรียนหรือแจ้งปัญหาบริการ
    
    Customer message: {last_message}
    
    Return only the category name."""
    
    response = llm.invoke([HumanMessage(content=prompt)])
    intent = response.content.strip().lower()
    
    return {"intent": intent}

def route_to_department(state: SupportAgentState) -> str:
    """Edge: ตัดสินใจว่าจะ route ไป department ไหน"""
    intent = state.get("intent", "unknown")
    
    routes = {
        "billing": "handle_billing",
        "technical": "handle_technical", 
        "sales": "handle_sales",
        "complaint": "handle_complaint"
    }
    
    return routes.get(intent, "handle_general")

def handle_billing(state: SupportAgentState) -> SupportAgentState:
    """Node 2a: จัดการปัญหา billing"""
    messages = state["messages"]
    system_msg = SystemMessage(content="""You are a billing specialist. 
    Help customers with invoice issues, payment problems, or refund requests.
    Be clear about policies and timelines.""")
    
    response = llm.invoke([system_msg] + messages)
    
    return {
        "messages": messages + [response],
        "department": "billing",
        "resolution_status": "in_progress"
    }

def handle_technical(state: SupportAgentState) -> SupportAgentState:
    """Node 2b: จัดการปัญหาทางเทคนิค"""
    messages = state["messages"]
    system_msg = SystemMessage(content="""You are a technical support specialist.
    Help troubleshoot technical issues, bugs, or integration problems.
    Ask clarifying questions to pinpoint the exact issue.""")
    
    response = llm.invoke([system_msg] + messages)
    
    return {
        "messages": messages + [response],
        "department": "technical",
        "resolution_status": "in_progress"
    }

def handle_sales(state: SupportAgentState) -> SupportAgentState:
    """Node 2c: จัดการคำถามขาย"""
    messages = state["messages"]
    system_msg = SystemMessage(content="""You are a sales representative.
    Provide product information, pricing details, and help with purchasing decisions.
    Be enthusiastic but not pushy.""")
    
    response = llm.invoke([system_msg] + messages)
    
    return {
        "messages": messages + [response],
        "department": "sales",
        "resolution_status": "completed"
    }

def handle_complaint(state: SupportAgentState) -> SupportAgentState:
    """Node 2d: จัดการร้องเรียน"""
    messages = state["messages"]
    system_msg = SystemMessage(content="""You are a customer relations specialist.
    Handle complaints with empathy. Acknowledge the issue, apologize sincerely,
    and propose solutions. Escalate if customer remains unsatisfied.""")
    
    response = llm.invoke([system_msg] + messages)
    
    return {
        "messages": messages + [response],
        "department": "customer_relations",
        "resolution_status": "escalated" if "unsatisfied" in response.content.lower() else "resolved"
    }

def handle_general(state: SupportAgentState) -> SupportAgentState:
    """Node 2e: จัดการกรณีทั่วไป"""
    messages = state["messages"]
    system_msg = SystemMessage(content="""You are a general customer support agent.
    Help with any questions or concerns. If you cannot resolve the issue,
    collect details and promise a callback.""")
    
    response = llm.invoke([system_msg] + messages)
    
    return {
        "messages": messages + [response],
        "department": "general",
        "resolution_status": "pending"
    }


Build the Graph

workflow = StateGraph(SupportAgentState)


เพิ่ม nodes

workflow.add_node("classify_intent", classify_intent)
workflow.add_node("handle_billing", handle_billing)
workflow.add_node("handle_technical", handle_technical)
workflow.add_node("handle_sales", handle_sales)
workflow.add_node("handle_complaint", handle_complaint)
workflow.add_node("handle_general", handle_general)


เพิ่ม edges

workflow.set_entry_point("classify_intent")
workflow.add_conditional_edges(
    "classify_intent",
    route_to_department,
    {
        "handle_billing": "handle_billing",
        "handle_technical": "handle_technical",
        "handle_sales": "handle_sales",
        "handle_complaint": "handle_complaint",
        "handle_general": "handle_general"
    }
)


ทุก department node จบที่ END

for node in ["handle_billing", "handle_technical", "handle_sales", 
             "handle_complaint", "handle_general"]:
    workflow.add_edge(node, END)


Compile

app = workflow.compile()


Run the agent

initial_state = {
    "messages": [HumanMessage(content="ผมได้รับบิลเกินมา 500 บาท ช่วยตรวจสอบให้หน่อยได้ไหมครับ")],
    "intent": None,
    "department": None,
    "resolution_status": ""
}

result = app.invoke(initial_state)
print(f"Department: {result['department']}")
print(f"Status: {result['resolution_status']}")
print(f"Last Response: {result['messages'][-1].content}")

ตัวอย่างที่ 2: Multi-Agent Collaboration ด้วย LangGraph

ใน production จริง คุณอาจต้องการหลาย specialized agents ทำงานร่วมกัน มาดูตัวอย่าง Research & Writing Team:

import os
from typing import TypedDict, List
from langgraph.graph import StateGraph, END, START
from langgraph.prebuilt import ToolNode
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, BaseMessage
from langchain_holysheep import ChatHolySheep

os.environ["HOLYSHEEP_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"


Initialize multiple specialized LLMs via HolySheep

research_llm = ChatHolySheep(
    model="deepseek-v3.2",
    base_url="https://api.holysheep.ai/v1",
    api_key=os.environ["HOLYSHEEP_API_KEY"]
)

writer_llm = ChatHolySheep(
    model="gpt-4.1",  # ใช้ GPT-4.1 สำหรับงานเขียนที่ต้องการคุณภาพสูง
    base_url="https://api.holysheep.ai/v1",
    api_key=os.environ["HOLYSHEEP_API_KEY"]
)

reviewer_llm = ChatHolySheep(
    model="claude-sonnet-4.5",  # Claude สำหรับงาน review ที่ต้องการ critical thinking
    base_url="https://api.holysheep.ai/v1",
    api_key=os.environ["HOLYSHEEP_API_KEY"]
)

class TeamResearchState(TypedDict):
    topic: str
    research_notes: str
    draft: str
    feedback: str
    final_version: str
    iteration: int
    messages: List[BaseMessage]

def researcher(state: TeamResearchState) -> TeamResearchState:
    """Agent 1: ทำหน้าที่ค้นคว้าและรวบรวมข้อมูล"""
    topic = state["topic"]
    
    system_msg = SystemMessage(content="""You are a thorough research analyst.
    Your task is to gather comprehensive information about the given topic.
    Structure your findings with:
    1. Key concepts and definitions
    2. Important facts and statistics
    3. Different perspectives or approaches
    4. Notable examples or case studies
    
    Be objective and cite sources conceptually (not real URLs).""")
    
    prompt = f"""Research the following topic and provide structured notes:
    Topic: {topic}
    
    Focus on recent developments and practical applications."""
    
    response = research_llm.invoke([system_msg, HumanMessage(content=prompt)])
    
    return {
        "research_notes": response.content,
        "iteration": state.get("iteration", 0),
        "messages": state.get("messages", []) + [
            {"agent": "researcher", "content": response.content}
        ]
    }

def writer(state: TeamResearchState) -> TeamResearchState:
    """Agent 2: เขียนบทความจากข้อมูลที่ได้รับ"""
    research = state.get("research_notes", "")
    previous_feedback = state.get("feedback", "")
    
    system_msg = SystemMessage(content="""You are a skilled technical writer.
    Transform research notes into a well-structured, engaging article.
    Use appropriate headings, examples, and maintain consistent tone.
    If there's previous feedback, incorporate it into the revision.""")
    
    prompt = f"""Write an article based on the following research:
    
    Research Notes:
    {research}
    
    Previous Feedback (if any):
    {previous_feedback}
    
    Write a comprehensive, publication-ready article."""

    response = writer_llm.invoke([system_msg, HumanMessage(content=prompt)])
    
    return {
        "draft": response.content,
        "messages": state.get("messages", []) + [
            {"agent": "writer", "content": response.content}
        ]
    }

def reviewer(state: TeamResearchState) -> TeamResearchState:
    """Agent 3: ตรวจสอบและให้ feedback"""
    draft = state["draft"]
    iteration = state.get("iteration", 0)
    
    system_msg = SystemMessage(content="""You are a critical reviewer and editor.
    Evaluate the draft for:
    1. Accuracy and factual correctness
    2. Logical flow and structure
    3. Clarity and readability
    4. Completeness of coverage
    
    Provide specific, actionable feedback. If the draft is good enough,
    say 'APPROVED - ready for publication'.""")
    
    prompt = f"""Review this draft and provide detailed feedback:
    
    Draft:
    {draft}
    
    Iteration: {iteration}"""

    response = reviewer_llm.invoke([system_msg, HumanMessage(content=prompt)])
    
    needs_revision = "APPROVED" not in response.content.upper()
    
    return {
        "feedback": response.content,
        "iteration": iteration + 1 if needs_revision else iteration,
        "final_version": draft if not needs_revision else state["draft"]
    }

def should_revise(state: TeamResearchState) -> str:
    """Decision edge: ตรวจสอบว่าต้องแก้ไขหรือจบ"""
    feedback = state["feedback"]
    iteration = state.get("iteration", 0)
    
    if "APPROVED" in feedback.upper():
        return "publish"
    elif iteration >= 3:  # Max 3 iterations
        return "publish"
    else:
        return "revise"


Build Team Workflow

workflow = StateGraph(TeamResearchState)

workflow.add_node("researcher", researcher)
workflow.add_node("writer", writer)
workflow.add_node("reviewer", reviewer)

workflow.add_edge(START, "researcher")
workflow.add_edge("researcher", "writer")
workflow.add_edge("writer", "reviewer")

workflow.add_conditional_edges(
    "reviewer",
    should_revise,
    {
        "revise": "writer",  # ส่งกลับไปให้ writer แก้ไข
        "publish": END
    }
)

app = workflow.compile()


Run the team

initial_state = {
    "topic": "ความแตกต่างระหว่าง Stateless และ Stateful AI Agents",
    "research_notes": "",
    "draft": "",
    "feedback": "",
    "final_version": "",
    "iteration": 0,
    "messages": []
}

result = app.invoke(initial_state)
print(f"Final Version:\n{result['final_version']}")
print(f"\nTotal Iterations: {result['iteration']}")
print(f"Team Messages: {len(result['messages'])}")

Memory Persistence: เก็บ State ลง Database

ใน production คุณต้องการเก็บ state ไว้ใช้งานต่อเนื่อง มาดูวิธีใช้ Redis หรือ PostgreSQL:

from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.checkpoint.memory import MemorySaver
import json
from datetime import datetime


Option 1: In-memory (เหมาะสำหรับ dev)

checkpointer = MemorySaver()


Option 2: PostgreSQL (เหมาะสำหรับ production)


checkpointer = PostgresSaver.from_conn_string("postgresql://user:pass@localhost/db")


class PersistentAgentState(TypedDict):
    user_id: str
    session_id: str
    conversation_history: List[dict]
    user_preferences: dict
    current_task: str | None
    task_progress: dict

def create_session(user_id: str, session_id: str) -> PersistentAgentState:
    """สร้าง session ใหม่"""
    return {
        "user_id": user_id,
        "session_id": session_id,
        "conversation_history": [],
        "user_preferences": {},
        "current_task": None,
        "task_progress": {}
    }

def process_message(state: PersistentAgentState, new_message: str) -> PersistentAgentState:
    """ประมวลผลข้อความใหม่พร้อมจดจำ context"""
    # ดึง system prompt ที่รวม conversation history
    history_text = "\n".join([
        f"{'User' if m['role']=='user' else 'Assistant'}: {m['content']}"
        for m in state["conversation_history"][-10:]  # last 10 messages
    ])
    
    system_msg = SystemMessage(content=f"""You are an AI assistant with long-term memory.
    
    Conversation History (recent):
    {history_text}
    
    User Preferences:
    {json.dumps(state.get('user_preferences', {}), ensure_ascii=False)}
    
    Current Task: {state.get('current_task', 'None')}
    
    Always consider the context from previous conversations.""")

    response = llm.invoke([system_msg, HumanMessage(content=new_message)])
    
    # อัพเดท state
    updated_history = state["conversation_history"] + [
        {"role": "user", "content": new_message, "timestamp": datetime.now().isoformat()},
        {"role": "assistant", "content": response.content, "timestamp": datetime.now().isoformat()}
    ]
    
    return {
        "conversation_history": updated_history,
        "messages": state.get("messages", []) + [response]
    }


Recompile with checkpointer

app = workflow.compile(checkpointer=checkpointer)


สร้าง config สำหรับ thread/session

config = {
    "configurable": {
        "thread_id": "user_123_session_456"
    }
}


ส่งข้อความแรก

state1 = {"messages": [HumanMessage(content="ผมชอบอ่านบทความเกี่ยวกับ AI")]}
app.invoke(state1, config)


ส่งข้อความที่สอง - agent จะจำได้ว่าผู้ใช้ชอบ AI

state2 = {"messages": [HumanMessage(content="แนะนำบทความให้หน่อย")]}
app.invoke(state2, config)  # จะใช้ context จากข้อความก่อนหน้า

Error Handling และ Retry Logic

Production agent ต้องจัดการ error อย่าง graceful ให้ดูวิธีเพิ่ม retry และ fallback:

from tenacity import retry, stop_after_attempt, wait_exponential
from langchain_core.outputs import LLMResult
import time

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def robust_llm_call(messages, model="deepseek-v3.2", temperature=0.7):
    """LLM call ที่มี retry logic ในตัว"""
    try:
        llm = ChatHolySheep(
            model=model,
            base_url="https://api.holysheep.ai/v1",
            api_key=os.environ["HOLYSHEEP_API_KEY"],
            temperature=temperature,
            max_retries=0  # disable langchain's default retry
        )
        
        response = llm.invoke(messages)
        return response
        
    except Exception as e:
        print(f"Error occurred: {type(e).__name__}")
        if "rate_limit" in str(e).lower():
            print("Rate limited - waiting before retry...")
            time.sleep(5)
        raise  # Re-raise to trigger retry

def process_with_fallback(state: AgentState) -> AgentState:
    """Process พร้อม fallback ไป model อื่นหาก fail"""
    messages = state["messages"]
    
    models_to_try = ["deepseek-v3.2", "gemini-2.5-flash", "gpt-4.1"]
    last_error = None
    
    for model in models_to_try:
        try:
            response = robust_llm_call(messages, model=model)
            return {"messages": messages + [response], "model_used": model}
        except Exception as e:
            last_error = e
            print(f"Model {model} failed: {e}")
            continue
    
    # ทุก model ล้มเหลว - return graceful error message
    error_msg = AIMessage(content="ขออภัยครับ ระบบไม่สามารถประมวลผลได้ในขณะนี้ กรุณาลองใหม่อีกครั้งในอีกสักครู่")
    return {"messages": messages + [error_msg], "error": str(last_error)}

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

กรณีที่ 1: "Connection timeout" เมื่อเรียก HolySheep API

# ❌ วิธีผิด - ไม่มี timeout handling
response = llm.invoke(messages)


✅ วิธีถูก - เพิ่ม timeout และ retry

from langchain_holysheep import ChatHolySheep
from langchain_core.utils import get_from_dict_or_env

class TimeoutHolySheep(ChatHolySheep):
    def _generate(self, messages, stop=None, run_manager=None, **kwargs):
        import signal
        
        def timeout_handler(signum, frame):
            raise TimeoutError("API call exceeded 30 seconds")
        
        # Set timeout for 30 seconds
        signal.signal(signal.SIGALRM, timeout_handler)
        signal.alarm(30)
        
        try:
            result = super()._generate(messages, stop, run_manager, **kwargs)
            signal.alarm(0)  # Cancel alarm
            return result
        except TimeoutError:
            signal.alarm(0)
            raise


Usage

llm = TimeoutHolySheep(
    model="deepseek-v3.2",
    base_url="https://api.holysheep.ai/v1",
    api_key="YOUR_HOLYSHEEP_API_KEY"
)

กรณีที่ 2: "State not persisting" - State หายระหว่าง nodes

# ❌ วิธีผิด - return state ไม่ครบ
def bad_node(state):
    new_content = process(state["messages"][-1])
    return {"result": new_content}  # ไม่ได้ return messages กลับไป!


✅ วิธีถูก - ต้อง merge state อย่างถูกต้อง

def correct_node(state: AgentState) -> AgentState:
    # สร้าง state ใหม่โดย copy ทุก field
    new_state = dict(state)
    
    new_content = process(state["messages"][-1])
    
    # Option 1: Append to existing list
    new_state["messages"] = state["messages"] + [AIMessage(content=new_content)]
    
    # Option 2: ใช้ state.update() สำหรับ partial update
    new_state.update({
        "last_result": new_content,
        "processed": True
    })
    
    return new_state


Alternative: ใช้ Annotated สำหรับ automatic state merging

from typing import Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]  # automatic append
    context: dict

def auto_append_node(state: AgentState) -> AgentState:
    return {"messages": [AIMessage(content="new message")]}  # will auto-merge!

กรณีที่ 3: "Maximum iterations exceeded" - Infinite loop ใน conditional edges

# ❌ วิธีผิด - ไม่มี exit condition
def bad_router(state):
    if some_condition:
        return "continue_processing"  # infinite loop!
    return "end"


✅ วิธีถูก - เพิ่ม iteration counter และ max limit

class SafeAgentState(TypedDict):
    iteration: int
    max_iterations: int
    
def safe_router(state: SafeAgentState) -> str:
    iteration = state.get("iteration", 0)
    max_iter = state.get("max_iterations", 5)
    
    # Force exit if max iterations reached
    if iteration >= max_iter:
        return "end"
    
    if some_condition:
        return "continue_processing"
    return "end"

def increment_iteration(state: SafeAgentState) -> SafeAgentState:
    return {"iteration": state.get("iteration", 0) + 1}


Add iteration node in the loop

workflow.add_edge("continue_processing", "increment_iteration")
workflow.add_edge("increment_iteration", "safe_router")  # loop back

กรณีที่ 4: "Invalid base_url" - ใช้ URL ผิด

# ❌ วิธีผิด - URL จาก official API (ห้ามใช้!)
llm = ChatHolySheep(
    model="deepseek-v3.2",
    base_url="https://api.openai.com/v1",  # ❌ ผิด!
    api_key="sk-xxx"
)


❌


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