คืนนั้นผมนั่ง deploy production system ที่ใช้ LangGraph สำหรับ customer service automation อยู่ดีๆ ก็เจอ error นี้โดยไม่ทันตั้งตัว:

RateLimitError: 429 Client Error: Too Many Requests for url: https://api.holysheep.ai/v1/chat/completions
Retry attempt 3/5 failed: gemini-2.0-flash models are currently overloaded
ConnectionError: HTTPSConnectionPool(host='api.holysheep.ai', port=443): Read timed out after 30.05s

ระบบล่มไป 2 ชั่วโมงเต็มๆ กว่าจะ restore ได้ นั่นคือจุดที่ผมเข้าใจว่า "การใช้ LLM API ใน production ไม่ใช่แค่เรียก function" แต่ต้องมี architecture ที่รัดกุม วันนี้ผมจะสอนทุกท่านวิธีสร้าง LangGraph State Machine Agent ที่รองรับ rate limit และ retry อย่างถูกต้อง โดยใช้ HolySheep AI ซึ่งมีราคาที่ประหยัดกว่า 85% เมื่อเทียบกับ OpenAI โดย Gemini 2.5 Flash ราคาเพียง $2.50 ต่อล้าน tokens

ทำไมต้อง LangGraph State Machine?

LangGraph เป็น library ที่ extends LangChain โดยมี concept เป็น "graph" ที่ node คือ state และ edge คือการเปลี่ยน state ต่างจาก prompt chaining แบบเดิมที่เป็น linear flow, LangGraph ทำให้เราสร้าง branching logic, loop, และ conditional routing ได้ง่าย

ข้อดีหลักๆ คือ:

การตั้งค่า Environment และ Dependencies

# requirements.txt
langgraph==0.2.45
langchain-core==0.3.24
langchain-holysheep==0.0.2
tenacity==8.3.0
pydantic==2.9.2
python-dotenv==1.0.1


.env

HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1
MODEL_NAME=gemini-2.5-flash


Rate limit configuration

MAX_RETRIES=5
RETRY_DELAY=2.0
RATE_LIMIT_THRESHOLD=0.8
import os
from dotenv import load_dotenv
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_holysheep import HolySheepChat
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import time

load_dotenv()

class AgentState(TypedDict):
    """State หลักของ Agent ที่ใช้ throughout the conversation"""
    messages: list
    current_step: str
    retry_count: int
    context: dict
    route_decision: str | None

class RateLimitError(Exception):
    """Custom exception สำหรับ rate limit errors"""
    def __init__(self, retry_after: int = 60):
        self.retry_after = retry_after
        super().__init__(f"Rate limit exceeded. Retry after {retry_after} seconds.")

class HolySheepLLMWrapper:
    """Wrapper class สำหรับ HolySheep API พร้อม rate limit handling"""
    
    def __init__(self, api_key: str, model: str = "gemini-2.5-flash"):
        self.api_key = api_key
        self.model = model
        self.base_url = "https://api.holysheep.ai/v1"
        self._rate_limit_calls = 0
        self._last_reset = time.time()
        self._calls_per_minute = 60  # Default rate limit
        
    def _check_rate_limit(self):
        """ตรวจสอบว่าเราใกล้ถึง rate limit หรือยัง"""
        current_time = time.time()
        if current_time - self._last_reset >= 60:
            self._rate_limit_calls = 0
            self._last_reset = current_time
            
        if self._rate_limit_calls >= self._calls_per_minute * 0.8:
            wait_time = 60 - (current_time - self._last_reset)
            if wait_time > 0:
                print(f"⏳ Approaching rate limit. Waiting {wait_time:.1f}s...")
                time.sleep(wait_time)
                self._rate_limit_calls = 0
                self._last_reset = time.time()
                
    def _make_request(self, messages: list, **kwargs):
        """ทำ request ไปยัง HolySheep API พร้อม retry logic"""
        self._check_rate_limit()
        
        llm = HolySheepChat(
            holysheep_api_key=self.api_key,
            holysheep_api_base=self.base_url,
            model=self.model,
            temperature=kwargs.get("temperature", 0.7),
            max_tokens=kwargs.get("max_tokens", 2048)
        )
        
        try:
            response = llm.invoke(messages)
            self._rate_limit_calls += 1
            return response
        except Exception as e:
            error_str = str(e).lower()
            if "429" in error_str or "rate limit" in error_str:
                retry_after = 60
                if "retry-after" in error_str:
                    try:
                        retry_after = int(error_str.split("retry-after")[-1].strip())
                    except:
                        pass
                raise RateLimitError(retry_after=retry_after)
            raise


Initialize LLM wrapper

llm_wrapper = HolySheepLLMWrapper(
    api_key=os.getenv("HOLYSHEEP_API_KEY"),
    model=os.getenv("MODEL_NAME", "gemini-2.5-flash")
)

สร้าง LangGraph State Machine Agent

ต่อไปเราจะสร้าง state machine ที่มีหลาย states และ transitions ซึ่งเหมาะสำหรับ workflow ที่ซับซ้อน เช่น customer service หรือ data processing pipeline

from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_core.tools import tool

@tool
def search_knowledge_base(query: str) -> str:
    """ค้นหาข้อมูลใน knowledge base"""
    # Implement knowledge base search logic
    return f"Found information about: {query}"

@tool
def escalate_to_human(ticket_id: str, reason: str) -> str:
    """ส่งต่อ case ไปให้ human agent"""
    # Implement escalation logic
    return f"Ticket {ticket_id} escalated. Reason: {reason}"

@tool
def generate_response(state: AgentState, response_type: str) -> str:
    """สร้าง response ตาม type ที่กำหนด"""
    if response_type == "apologize":
        return "ขออภัยในความไม่สะดวกที่เกิดขึ้น"
    elif response_type == "confirm":
        return "ยืนยันว่าเราได้รับข้อมูลแล้ว"
    elif response_type == "solution":
        return "นี่คือวิธีแก้ปัญหาสำหรับคุณ"
    return "กรุณารอสักครู่"


Define the nodes of our state machine

def intent_classifier(state: AgentState) -> AgentState:
    """Node 1: Classify user intent"""
    messages = state["messages"]
    last_message = messages[-1].content if messages else ""
    
    # Use LLM to classify intent
    classification_prompt = f"""Classify the user's intent into one of:
    - PRODUCT_INQUIRY
    - TECHNICAL_SUPPORT
    - COMPLAINT
    - BILLING
    - ESCALATE
    
    User message: {last_message}
    
    Return only the intent category."""
    
    try:
        response = llm_wrapper._make_request(
            messages=[SystemMessage(content=classification_prompt)],
            temperature=0.3,
            max_tokens=50
        )
        intent = response.content.strip()
        
        return {
            **state,
            "current_step": "route_decision",
            "route_decision": intent,
            "retry_count": 0
        }
    except RateLimitError as e:
        print(f"Rate limit hit during classification: {e}")
        raise

def route_handler(state: AgentState) -> AgentState:
    """Node 2: Route to appropriate handler based on intent"""
    intent = state["route_decision"]
    
    route_map = {
        "PRODUCT_INQUIRY": "search_knowledge_base",
        "TECHNICAL_SUPPORT": "search_knowledge_base",
        "COMPLAINT": "apology_generator",
        "BILLING": "billing_handler",
        "ESCALATE": "escalation_handler"
    }
    
    next_step = route_map.get(intent, "search_knowledge_base")
    
    return {
        **state,
        "current_step": next_step,
        "context": {**state["context"], "next_handler": next_step}
    }

def response_generator(state: AgentState) -> AgentState:
    """Node 3: Generate final response"""
    handler = state["context"].get("next_handler", "search_knowledge_base")
    
    response_prompt = f"""Based on the conversation context, generate a helpful response in Thai.
    
    Current handler: {handler}
    Messages: {state['messages']}
    
    Respond in a helpful and professional manner."""
    
    try:
        response = llm_wrapper._make_request(
            messages=[SystemMessage(content=response_prompt)] + state["messages"],
            temperature=0.7,
            max_tokens=1024
        )
        
        new_messages = state["messages"] + [AIMessage(content=response.content)]
        
        return {
            **state,
            "messages": new_messages,
            "current_step": "end"
        }
    except RateLimitError as e:
        # Fallback to rule-based response if rate limited
        fallback = "ขออภัยในความล่าช้า ทางเรากำลังดำเนินการ กรุณารอสักครู่"
        new_messages = state["messages"] + [AIMessage(content=fallback)]
        
        return {
            **state,
            "messages": new_messages,
            "current_step": "end",
            "context": {**state["context"], "fallback_used": True}
        }


Build the state machine graph

def create_agent_graph():
    """สร้าง LangGraph state machine"""
    
    workflow = StateGraph(AgentState)
    
    # Add nodes
    workflow.add_node("intent_classifier", intent_classifier)
    workflow.add_node("route_handler", route_handler)
    workflow.add_node("search_knowledge_base", search_knowledge_base)
    workflow.add_node("apology_generator", generate_response)
    workflow.add_node("billing_handler", generate_response)
    workflow.add_node("escalation_handler", escalate_to_human)
    workflow.add_node("response_generator", response_generator)
    
    # Define edges
    workflow.set_entry_point("intent_classifier")
    
    workflow.add_edge("intent_classifier", "route_handler")
    
    # Conditional routing from route_handler
    def should_process(state: AgentState) -> str:
        return state["context"].get("next_handler", "response_generator")
    
    workflow.add_conditional_edges(
        "route_handler",
        should_process,
        {
            "search_knowledge_base": "search_knowledge_base",
            "apology_generator": "apology_generator",
            "billing_handler": "billing_handler",
            "escalation_handler": "escalation_handler"
        }
    )
    
    workflow.add_edge("search_knowledge_base", "response_generator")
    workflow.add_edge("apology_generator", "response_generator")
    workflow.add_edge("billing_handler", "response_generator")
    workflow.add_edge("escalation_handler", END)
    workflow.add_edge("response_generator", END)
    
    return workflow.compile()


Initialize the agent

agent = create_agent_graph()

ระบบ Retry ที่แข็งแกร่งด้วย Exponential Backoff

นี่คือหัวใจสำคัญของ production system ที่ดี การ retry ต้องทำอย่างชาญฉลาดไม่ใช่แค่ loop ไปเรื่อยๆ

from tenacity import (
    retry, 
    stop_after_attempt, 
    wait_exponential, 
    retry_if_exception_type,
    before_sleep_log,
    after_retry
)
import logging
import asyncio

logger = logging.getLogger(__name__)

class RetryableLLMWrapper:
    """Enhanced LLM wrapper พร้อม intelligent retry"""
    
    def __init__(
        self,
        api_key: str,
        model: str = "gemini-2.5-flash",
        max_retries: int = 5,
        base_delay: float = 1.0,
        max_delay: float = 60.0
    ):
        self.api_key = api_key
        self.model = model
        self.base_delay = base_delay
        self.max_delay = max_delay
        self.max_retries = max_retries
        self.wrapper = HolySheepLLMWrapper(api_key=api_key, model=model)
        
    def _calculate_backoff(self, attempt: int) -> float:
        """คำนวณ delay time ด้วย exponential backoff + jitter"""
        import random
        import math
        
        exponential_delay = self.base_delay * (2 ** attempt)
        jitter = random.uniform(0, 1)
        delay = min(exponential_delay + jitter, self.max_delay)
        
        return delay
    
    @retry(
        retry=retry_if_exception_type(RateLimitError),
        stop=stop_after_attempt(5),
        wait=wait_exponential(multiplier=1, min=2, max=60),
        before_sleep=before_sleep_log(logger, logging.WARNING),
        after=after_retry(logger=logger, logging_func=logging.info)
    )
    def invoke_with_retry(self, messages: list, **kwargs):
        """เรียก LLM พร้อม automatic retry"""
        
        attempt = self.retry.retry_state.attempt_number
        logger.info(f"🔄 LLM request attempt {attempt}/{self.max_retries}")
        
        try:
            response = self.wrapper._make_request(messages, **kwargs)
            if attempt > 1:
                logger.info(f"✅ Request succeeded on attempt {attempt}")
            return response
            
        except RateLimitError as e:
            logger.warning(f"⚠️ Rate limit hit: {e}")
            raise
            
        except Exception as e:
            error_msg = str(e).lower()
            
            # 429 - Rate limit
            if "429" in error_msg:
                raise RateLimitError(retry_after=60)
            
            # 500, 502, 503 - Server errors - retryable
            if any(code in error_msg for code in ["500", "502", "503", "timeout"]):
                if attempt < self.max_retries:
                    logger.warning(f"⚠️ Server error, will retry: {e}")
                    raise
                
            # 401, 403 - Auth errors - not retryable
            if any(code in error_msg for code in ["401", "403", "unauthorized"]):
                logger.error(f"❌ Authentication error - not retrying: {e}")
                raise
            
            # Other errors - might be retryable
            if attempt < self.max_retries:
                logger.warning(f"⚠️ Unexpected error: {e}")
                raise
            
            raise


Async version for better performance

class AsyncRetryableLLMWrapper(RetryableLLMWrapper):
    """Async version of the retry wrapper"""
    
    async def invoke_with_retry_async(self, messages: list, **kwargs):
        """Async invoke with retry"""
        
        for attempt in range(1, self.max_retries + 1):
            logger.info(f"🔄 Async LLM request attempt {attempt}/{self.max_retries}")
            
            try:
                # In real implementation, use httpx.AsyncClient
                # For now, wrapping sync call
                response = await asyncio.to_thread(
                    self.wrapper._make_request, messages, **kwargs
                )
                
                if attempt > 1:
                    logger.info(f"✅ Async request succeeded on attempt {attempt}")
                return response
                
            except RateLimitError as e:
                if attempt == self.max_retries:
                    logger.error(f"❌ Max retries ({self.max_retries}) reached")
                    raise
                    
                delay = self._calculate_backoff(attempt)
                logger.warning(f"⏳ Waiting {delay:.1f}s before retry...")
                await asyncio.sleep(delay)
                
            except Exception as e:
                if attempt == self.max_retries:
                    logger.error(f"❌ Max retries reached, last error: {e}")
                    raise
                    
                delay = self._calculate_backoff(attempt)
                logger.warning(f"⚠️ Error: {e}, retrying in {delay:.1f}s...")
                await asyncio.sleep(delay)
        
        raise Exception("Should not reach here")


Usage example

async def run_agent_with_full_retry():
    """Run agent with comprehensive error handling"""
    
    async_wrapper = AsyncRetryableLLMWrapper(
        api_key=os.getenv("HOLYSHEEP_API_KEY"),
        model="gemini-2.5-flash",
        max_retries=5,
        base_delay=2.0,
        max_delay=120.0
    )
    
    initial_state = AgentState(
        messages=[HumanMessage(content="ฉันมีปัญหาเรื่องการสั่งซื้อสินค้า")],
        current_step="start",
        retry_count=0,
        context={},
        route_decision=None
    )
    
    try:
        result = await async_wrapper.invoke_with_retry_async(
            messages=[SystemMessage(content="Classify this: " + initial_state["messages"][0].content)],
            temperature=0.3
        )
        return result
    except Exception as e:
        logger.error(f"❌ All retries exhausted: {e}")
        return None

Production-Ready Circuit Breaker Pattern

นอกจาก retry แล้ว เราต้องมี circuit breaker เพื่อป้องกัน cascade failure การทำ circuit breaker จะช่วยให้ระบบ "หยุดพยายาม" ชั่วคราวเมื่อ API มีปัญหาต่อเนื่อง

from enum import Enum
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import Callable
import threading

class CircuitState(Enum):
    CLOSED = "closed"      # Normal operation
    OPEN = "open"          # Failing, reject requests
    HALF_OPEN = "half_open"  # Testing if recovered

@dataclass
class CircuitBreaker:
    """Circuit breaker implementation for API calls"""
    
    name: str
    failure_threshold: int = 5
    recovery_timeout: int = 60  # seconds
    expected_exception: type = Exception
    
    state: CircuitState = field(default=CircuitState.CLOSED, init=False)
    failure_count: int = field(default=0, init=False)
    last_failure_time: datetime = field(default=None, init=False)
    _lock: threading.Lock = field(default_factory=threading.Lock, init=False)
    
    def call(self, func: Callable, *args, **kwargs):
        """Execute function through circuit breaker"""
        
        with self._lock:
            # Check if we should transition to half-open
            if self.state == CircuitState.OPEN:
                if self.last_failure_time:
                    time_since_failure = (datetime.now() - self.last_failure_time).total_seconds()
                    if time_since_failure >= self.recovery_timeout:
                        self.state = CircuitState.HALF_OPEN
                        print(f"🔄 Circuit breaker {self.name}: OPEN -> HALF_OPEN")
                    else:
                        raise Exception(f"Circuit breaker {self.name} is OPEN. Retry after {int(self.recovery_timeout - time_since_failure)}s")
            
            # Execute the call
            try:
                result = func(*args, **kwargs)
                self._on_success()
                return result
            except self.expected_exception as e:
                self._on_failure()
                raise
    
    def _on_success(self):
        """Handle successful call"""
        if self.state == CircuitState.HALF_OPEN:
            self.state = CircuitState.CLOSED
            print(f"✅ Circuit breaker {self.name}: HALF_OPEN -> CLOSED")
        self.failure_count = 0
    
    def _on_failure(self):
        """Handle failed call"""
        self.failure_count += 1
        self.last_failure_time = datetime.now()
        
        if self.failure_count >= self.failure_threshold:
            if self.state != CircuitState.OPEN:
                self.state = CircuitState.OPEN
                print(f"🚨 Circuit breaker {self.name}: CLOSED -> OPEN")

class ResilientAgentSystem:
    """Complete resilient agent with circuit breaker and retry"""
    
    def __init__(self, api_key: str):
        self.llm_wrapper = RetryableLLMWrapper(
            api_key=api_key,
            model="gemini-2.5-flash",
            max_retries=5
        )
        self.circuit_breaker = CircuitBreaker(
            name="holysheep_llm",
            failure_threshold=3,
            recovery_timeout=120
        )
        self.agent = create_agent_graph()
    
    def process_message(self, user_message: str) -> dict:
        """Process user message with full resilience"""
        
        initial_state = AgentState(
            messages=[HumanMessage(content=user_message)],
            current_step="start",
            retry_count=0,
            context={},
            route_decision=None
        )
        
        def execute_agent():
            return self.agent.invoke(initial_state)
        
        try:
            # Execute through circuit breaker
            result = self.circuit_breaker.call(execute_agent)
            return {"status": "success", "result": result}
            
        except RateLimitError as e:
            logger.error(f"Rate limited: {e}")
            return {
                "status": "rate_limited",
                "message": "Service is busy. Please try again later.",
                "retry_after": e.retry_after
            }
            
        except Exception as e:
            logger.error(f"System error: {e}")
            return {
                "status": "error",
                "message": "An error occurred. Please try again.",
                "error": str(e)
            }


Initialize production system

production_agent = ResilientAgentSystem(
    api_key=os.getenv("HOLYSHEEP_API_KEY")
)


Example usage

if __name__ == "__main__":
    response = production_agent.process_message(
        "สวัสดีครับ ผมต้องการสอบถามเรื่องการจัดส่งสินค้า"
    )
    print(response)

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

1. Error 401 Unauthorized — API Key ไม่ถูกต้อง

สถานการณ์: เริ่มต้นระบบแล้วเจอ error นี้ทันที

# ❌ ผิดพลาด - ลืมใส่ API key หรือ format ผิด
llm = HolySheepChat(
    holysheep_api_key="YOUR_HOLYSHEEP_API_KEY",  # ยังไม่ได้แทนที่
    holysheep_api_base="https://api.holysheep.ai/v1",
    model="gemini-2.5-flash"
)


✅ ถูกต้อง - ใช้ environment variable

import os
llm = HolySheepChat(
    holysheep_api_key=os.getenv("HOLYSHEEP_API_KEY"),
    holysheep_api_base="https://api.holysheep.ai/v1",
    model="gemini-2.5-flash"
)


ตรวจสอบว่า API key ถูกต้อง

print(f"API Key loaded: {bool(os.getenv('HOLYSHEEP_API_KEY'))}")

วิธีแก้: ตรวจสอบว่า .env file ถูกสร้างและอยู่ใน working directory และลอง print API key ออกมาดูว่าถูก load หรือไม่ หากยังไม่ได้สมัคร สามารถ สมัครที่นี่ เพื่อรับ API key ฟรี

2. Error 429 Rate Limit — เรียก API บ่อยเกินไป

สถานการณ์: ระบบทำงานได้สักพักแล้วอยู่ๆ ก็เจอ 429 error

# ❌ ผิดพลาด - ไม่มีการจัดการ rate limit
def bad_example():
    while True:
        response = llm.invoke(messages)  # เรียกต่อเนื่องโดยไม่สนใจ rate limit
        process(response)


✅ ถูกต้อง - ใช้ rate limiter และ exponential backoff

from collections import defaultdict
from datetime import datetime, timedelta
import time

class TokenBucketRateLimiter:
    def __init__(self, calls_per_minute: int = 60):
        self.calls_per_minute = calls_per_minute
        self.calls = defaultdict(list)
    
    def acquire(self) -> bool:
        """Return True if request is allowed, False if should wait"""
        now = datetime.now()
        window_start = now - timedelta(minutes=1)
        
        # Clean old calls
        self.calls[threading.get_ident()] = [
            t for t in self.calls[threading.get_ident()]
            if t > window_start
        ]
        
        if len(self.calls[threading.get_ident()]) >= self.calls_per_minute:
            return False
        
        self.calls[threading.get_ident()].append(now)
        return True
    
    def wait_if_needed(self):
        """Block until request is allowed"""
        while not self.acquire():
            time.sleep(1)


ใช้งาน

rate_limiter = TokenBucketRateLimiter(calls_per_minute=50)  # เผื่อ buffer

def good_example():
    rate_limiter.wait_if_needed()
    response = llm.invoke(messages)
    process(response)

วิธีแก้: ใช้ rate limiter class ด้านบน หรือ configure client ของ HolySheep ให้รอ automatic retry เมื่อเจอ 429 error ซึ่ง HolySheep มี latency เพียง <50ms ทำให้สามารถรองรับ traffic ได้ดี

3. Timeout Error — Request ใช้เวลานานเกินไป

สถานการณ์: API call ใช้เวลานานจน timeout

# ❌ ผิดพลาด - ไม่มี timeout configuration
llm = HolySheepChat(
    holysheep_api_key=api_key,
    holysheep_api_base="https://api.holysheep.ai/v1"
)
response = llm.invoke(messages)  # รอนานมากโดยไม่มี timeout


✅ ถูกต้อง - ตั้งค่า timeout และใช้ asyncio

import httpx
import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

async def async_llm_call_with_timeout():
    timeout = httpx.Timeout(30.0, connect=10.0)  # 30s total, 10s connect
    
    async with httpx.AsyncClient(timeout=timeout) as client:
        @retry(
            stop=stop_after_attempt(3),
            wait=wait_exponential(multiplier=1, min=2, max=30)
        )
        async def call_with_retry():
            try:
                response = await client.post(
                    "https://api.holysheep.ai/v1/chat/completions",
                    json={
                        "model": "gemini-2.5-flash",
                        "messages": [{"role": "user", "content": "Hello"}],
                        "max_tokens": 100
                    },
                    headers={"Authorization": f"Bearer {api_key}"}
                )
                return response.json()
            except httpx.TimeoutException:
                print("⏰ Request timed out, retrying...")
                raise
            except httpx.HTTPStatusError as e:
                if e.response.status_code == 429:
                    raise RateLimitError()
                raise
        
        return await call_with

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