ในการพัฒนา AI Agent หรือแชทบอทที่ซับซ้อน การจัดการสถานะการสนทนา (Dialogue State Management) เป็นหัวใจสำคัญที่กำหนดความสำเร็จของระบบ ในบทความนี้ผมจะเปรียบเทียบ 3 วิธีการหลักที่นิยมใช้กัน ได้แก่ FSM (Finite State Machine), Graph-based และ LLM Router พร้อมวิเคราะห์จากประสบการณ์ตรงในการใช้งานจริง

ทำความรู้จัก 3 วิธีการจัดการสถานะการสนทนา

1. Finite State Machine (FSM)

FSM เป็นวิธีดั้งเดิมที่ใช้หลักการ state และ transition กำหนดสถานะชัดเจน มีการเปลี่ยนแปลงตามเงื่อนไขที่กำหนดไว้ล่วงหน้า เหมาะกับระบบที่มี flow ตายตัว

2. Graph-based Architecture

ใช้โครงสร้างกราฟในการเชื่อมต่อสถานะต่างๆ แต่ละโหนดสามารถมีหลายทางออกและรองรับการ branch ได้ดีกว่า FSM

3. LLM Router

ใช้ Large Language Model ในการตัดสินใจว่าควรเปลี่ยนไปสถานะใด มีความยืดหยุ่นสูงและปรับตัวตามบริบทของการสนทนาได้

การเปรียบเทียบประสิทธิภาพตามเกณฑ์ที่ชัดเจน

เกณฑ์การประเมิน FSM Graph-based LLM Router
ความหน่วง (Latency) ~5-15ms ~10-30ms ~150-500ms
อัตราสำเร็จ (Success Rate) 95%+ (flow ตายตัว) 90%+ 75-85%
ความสะดวกในการพัฒนา ง่ายมาก ปานกลาง ยาก (ต้อง prompt engineering)
ความครอบคลุมของโมเดล ไม่ต้องใช้ LLM บางส่วน ต้องใช้ LLM เต็มรูปแบบ
ค่าใช้จ่าย (ต่อ 1M tokens) $0 $0-2.50 $2.50-15
ความยืดหยุ่น (Flexibility) ต่ำ ปานกลาง สูงมาก
Debug และ Testing ง่ายมาก ปานกลาง ยาก

ตัวอย่างโค้ดแต่ละวิธีการ

FSM Implementation

from enum import Enum
from typing import Callable, Dict, Optional

class DialogueState(Enum):
    GREETING = "greeting"
    MENU_SELECTION = "menu_selection"
    ORDER_CONFIRM = "order_confirm"
    PAYMENT = "payment"
    COMPLETED = "completed"
    FAILED = "failed"

class FSMDialogueManager:
    def __init__(self):
        self.current_state = DialogueState.GREETING
        self.user_data = {}
        self.transitions: Dict[DialogueState, Dict[str, DialogueState]] = {
            DialogueState.GREETING: {
                "hi": DialogueState.MENU_SELECTION,
                "hello": DialogueState.MENU_SELECTION
            },
            DialogueState.MENU_SELECTION: {
                "select_item": DialogueState.ORDER_CONFIRM,
                "cancel": DialogueState.GREETING
            },
            DialogueState.ORDER_CONFIRM: {
                "confirm": DialogueState.PAYMENT,
                "modify": DialogueState.MENU_SELECTION,
                "cancel": DialogueState.GREETING
            },
            DialogueState.PAYMENT: {
                "success": DialogueState.COMPLETED,
                "failed": DialogueState.FAILED
            }
        }
    
    def process_input(self, user_input: str) -> DialogueState:
        intent = self.classify_intent(user_input)
        
        if self.current_state in self.transitions:
            next_state = self.transitions[self.current_state].get(
                intent, 
                self.current_state
            )
            self.current_state = next_state
        
        return self.current_state
    
    def classify_intent(self, text: str) -> str:
        text_lower = text.lower()
        if any(word in text_lower for word in ["hi", "hello", "สวัสดี"]):
            return "hi"
        elif "confirm" in text_lower or "ยืนยัน" in text_lower:
            return "confirm"
        return "unknown"
    
    def reset(self):
        self.current_state = DialogueState.GREETING
        self.user_data = {}



ทดสอบ FSM

fsm = FSMDialogueManager()
print(f"Initial state: {fsm.current_state.value}")

fsm.process_input("hello")
print(f"After 'hello': {fsm.current_state.value}")

fsm.process_input("select item")
print(f"After 'select item': {fsm.current_state.value}")

LLM Router Implementation พร้อม HolySheep API

import requests
import json
from typing import List, Dict, Optional

class LLMStateRouter:
    def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
        self.api_key = api_key
        self.base_url = base_url
        self.available_states = [
            "greeting", "menu_selection", "order_confirm", 
            "payment", "completed", "failed", "transfer_to_agent"
        ]
    
    def determine_next_state(self, conversation_history: List[Dict]) -> Dict:
        """
        ใช้ LLM ตัดสินใจว่าควรไป state ใดต่อ
        คืนค่า dict ที่มี next_state, confidence, reasoning
        """
        prompt = self._build_routing_prompt(conversation_history)
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": "gpt-4.1",
            "messages": [
                {"role": "system", "content": "You are a dialogue state router."},
                {"role": "user", "content": prompt}
            ],
            "temperature": 0.3,
            "max_tokens": 200
        }
        
        response = requests.post(
            f"{self.base_url}/chat/completions",
            headers=headers,
            json=payload,
            timeout=30
        )
        
        if response.status_code != 200:
            raise Exception(f"API Error: {response.status_code} - {response.text}")
        
        result = response.json()
        return self._parse_llm_response(result["choices"][0]["message"]["content"])
    
    def _build_routing_prompt(self, history: List[Dict]) -> str:
        context = "\n".join([
            f"{msg['role']}: {msg['content']}" 
            for msg in history[-10:]
        ])
        
        return f"""Based on this conversation, determine the next dialogue state.

Available states: {', '.join(self.available_states)}

Conversation:
{context}

Respond in JSON format:
{{"next_state": "state_name", "confidence": 0.0-1.0, "reasoning": "brief explanation"}}
"""
    
    def _parse_llm_response(self, response_text: str) -> Dict:
        try:
            return json.loads(response_text)
        except json.JSONDecodeError:
            return {
                "next_state": "greeting",
                "confidence": 0.5,
                "reasoning": "Failed to parse, defaulting to greeting"
            }



ตัวอย่างการใช้งาน

router = LLMStateRouter(api_key="YOUR_HOLYSHEEP_API_KEY")

conversation = [
    {"role": "user", "content": "สวัสดีครับ อยากสั่งซื้อสินค้า"},
    {"role": "assistant", "content": "สวัสดีครับ ยินดีให้บริการ กรุณาเลือกรายการสินค้า"},
    {"role": "user", "content": "ขอสั่งโทรศัพท์ iPhone 16 Pro"}
]

result = router.determine_next_state(conversation)
print(f"Next state: {result['next_state']}")
print(f"Confidence: {result['confidence']}")
print(f"Reasoning: {result['reasoning']}")

Graph-based Architecture

from dataclasses import dataclass, field
from typing import Dict, List, Optional, Callable
import networkx as nx

@dataclass
class DialogueNode:
    state_id: str
    prompt: str
    handlers: List[Callable] = field(default_factory=list)
    validators: List[Callable] = field(default_factory=list)
    
@dataclass
class DialogueEdge:
    from_state: str
    to_state: str
    condition: Callable[[str], bool]
    priority: int = 0

class GraphDialogueManager:
    def __init__(self):
        self.graph = nx.DiGraph()
        self.current_state: Optional[str] = None
        self.context: Dict = {}
        self.nodes: Dict[str, DialogueNode] = {}
    
    def add_node(self, state_id: str, prompt: str) -> "GraphDialogueManager":
        node = DialogueNode(state_id=state_id, prompt=prompt)
        self.nodes[state_id] = node
        self.graph.add_node(state_id)
        if self.current_state is None:
            self.current_state = state_id
        return self
    
    def add_edge(
        self, 
        from_state: str, 
        to_state: str, 
        condition: Callable[[str], bool],
        priority: int = 0
    ) -> "GraphDialogueManager":
        edge = DialogueEdge(
            from_state=from_state,
            to_state=to_state,
            condition=condition,
            priority=priority
        )
        self.graph.add_edge(from_state, to_state)
        self._edges[(from_state, to_state)] = edge
        return self
    
    def process_input(self, user_input: str) -> str:
        edges_from_current = [
            (u, v, data) 
            for u, v, data in self.graph.out_edges(self.current_state, data=True)
        ]
        
        edges_from_current.sort(key=lambda x: x[2].get("priority", 0), reverse=True)
        
        for _, to_state, _ in edges_from_current:
            edge_key = (self.current_state, to_state)
            if edge_key in self._edges:
                edge = self._edges[edge_key]
                if edge.condition(user_input):
                    self.current_state = to_state
                    return to_state
        
        return self.current_state
    
    def get_reachable_states(self) -> List[str]:
        if self.current_state:
            return list(nx.descendants(self.graph, self.current_state))
        return []


สร้างกราฟสำหรับระบบสั่งซื้อ

dialogue_graph = GraphDialogueManager()

dialogue_graph.add_node("start", "สวัสดีครับ ยินดีให้บริการ")
dialogue_graph.add_node("browse", "กรุณาเลือกสินค้า")
dialogue_graph.add_node("cart", "สินค้าในตะกร้า")
dialogue_graph.add_node("payment", "ชำระเงิน")
dialogue_graph.add_node("complete", "คำสั่งซื้อเสร็จสมบูรณ์")

dialogue_graph._edges = {}

dialogue_graph.add_edge("start", "browse", 
    lambda x: "เลือก" in x or "ดู" in x, priority=10)
dialogue_graph.add_edge("browse", "cart",
    lambda x: "เพิ่ม" in x or "สั่ง" in x, priority=10)
dialogue_graph.add_edge("cart", "payment",
    lambda x: "ชำระ" in x or "จ่าย" in x, priority=10)
dialogue_graph.add_edge("payment", "complete",
    lambda x: "เสร็จ" in x or "สำเร็จ" in x, priority=10)

print(f"Current state: {dialogue_graph.current_state}")
dialogue_graph.process_input("ต้องการเลือกสินค้า")
print(f"After input: {dialogue_graph.current_state}")

ราคาและ ROI

วิธีการ ค่าใช้จ่ายต่อเดือน (approx) ROI เมื่อเทียบกับ FSM ความคุ้มค่า
FSM $0 (ไม่ต้องใช้ LLM) Baseline เหมาะกับงานที่ flow ตายตัว
Graph-based $5-50 (ขึ้นกับโมเดล) ประหยัดเวลา development 30-50% เหมาะกับงานปานกลาง
LLM Router $50-500 (ขึ้นกับ traffic) ปรับตัวได้ดี แต่ค่าใช้จ่ายสูง เหมาะกับงานที่ต้องการความยืดหยุ่นสูง

เหมาะกับใคร / ไม่เหมาะกับใคร

✅ FSM เหมาะกับ:

❌ FSM ไม่เหมาะกับ:

✅ Graph-based เหมาะกับ:

✅ LLM Router เหมาะกับ:

ทำไมต้องเลือก HolySheep

จากการทดสอบทั้ง 3 วิธีการ พบว่า LLM Router ให้ความยืดหยุ่นสูงสุด แต่มีค่าใช้จ่ายสูงตามไปด้วย สมัครที่นี่ เพื่อรับ API ที่คุ้มค่าที่สุดสำหรับงานนี้

โมเดล ราคา (ต่อ 1M tokens) เหมาะกับงาน
DeepSeek V3.2 $0.42 Routing พื้นฐาน, งานที่ต้องประหยัด
Gemini 2.5 Flash $2.50 Fast routing, real-time
GPT-4.1 $8.00 Complex routing, high accuracy
Claude Sonnet 4.5 $15.00 Best reasoning, enterprise

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

ข้อผิดพลาดที่ 1: LLM Router ตอบสถานะที่ไม่มีอยู่ในระบบ

# ❌ วิธีที่ผิด - ไม่ validate output
def determine_next_state(self, user_input: str) -> str:
    response = self.call_llm(user_input)
    return response["next_state"]  # อาจไม่ตรงกับ states ที่มี


✅ วิธีที่ถูกต้อง - validate output

def determine_next_state(self, user_input: str) -> str:
    response = self.call_llm(user_input)
    next_state = response.get("next_state", "greeting")
    
    # Map to valid state if LLM returns unknown
    state_mapping = {
        "greeting": "greeting",
        "hello": "greeting",
        "menu": "menu_selection",
        "select": "menu_selection",
        "order": "order_confirm",
        "pay": "payment"
    }
    
    return state_mapping.get(next_state.lower(), "greeting")

ข้อผิดพลาดที่ 2: Context ล้น (Context Overflow) ในการสนทนายาว

# ❌ วิธีที่ผิด - ส่ง history ทั้งหมด
def call_llm(self, history: List[Dict]) -> str:
    return self.client.chat.completions.create(
        model="gpt-4.1",
        messages=history  # อาจมี thousands of tokens
    )


✅ วิธีที่ถูกต้อง - summarize และ truncate

from holysheep import HolySheepClient

class OptimizedStateRouter:
    def __init__(self, api_key: str):
        self.client = HolySheepClient(api_key)
        self.max_history_tokens = 4000
    
    def call_llm(self, history: List[Dict], current_input: str) -> str:
        # Summarize old messages
        summarized = self._summarize_history(history[:-5])
        
        # Keep only recent messages
        recent = history[-5:]
        
        messages = summarized + recent
        messages.append({"role": "user", "content": current_input})
        
        # Trim if exceeds limit
        messages = self._trim_to_token_limit(messages, self.max_history_tokens)
        
        return self.client.chat.completions.create(
            model="gpt-4.1",
            messages=messages
        )
    
    def _summarize_history(self, old_messages: List[Dict]) -> List[Dict]:
        if len(old_messages) <= 2:
            return old_messages
        
        summary_prompt = f"""Summarize this conversation briefly (under 100 words):
        {old_messages}"""
        
        # Use cheap model for summarization
        summary = self.client.chat.completions.create(
            model="deepseek-v3.2",
            messages=[{"role": "user", "content": summary_prompt}]
        )
        
        return [{"role": "system", "content": f"Summary: {summary}"}]
    
    def _trim_to_token_limit(self, messages: List[Dict], limit: int) -> List[Dict]:
        # Rough estimate: 1 token ≈ 4 characters
        estimated_tokens = sum(len(m["content"]) for m in messages) // 4
        
        while estimated_tokens > limit and len(messages) > 2:
            messages.pop(0)
            estimated_tokens = sum(len(m["content"]) for m in messages) // 4
        
        return messages

ข้อผิดพลาดที่ 3: Rate Limit เมื่อ Scale

import time
from collections import defaultdict
from threading import Lock


❌ วิธีที่ผิด - ไม่จัดการ rate limit

class BadStateRouter:
    def call_llm(self, user_id: str, message: str):
        return requests.post(
            "https://api.holysheep.ai/v1/chat/completions",
            headers={"Authorization": f"Bearer {self.api_key}"},
            json={"model": "gpt-4.1", "messages": [{"role": "user", "content": message}]}
        )


✅ วิธีที่ถูกต้อง - implement rate limiting

class RateLimitedStateRouter:
    def __init__(self, api_key: str, requests_per_minute: int = 60):
        self.api_key = api_key
        self.rpm = requests_per_minute
        self.request_times = defaultdict(list)
        self.lock = Lock()
    
    def call_llm(self, user_id: str, message: str) -> dict:
        with self.lock:
            current_time = time.time()
            # Remove requests older than 1 minute
            self.request_times[user_id] = [
                t for t in self.request_times[user_id]
                if current_time - t < 60
            ]
            
            # Wait if at limit
            if len(self.request_times[user_id]) >= self.rpm:
                sleep_time = 60 - (current_time - self.request_times[user_id][0])
                if sleep_time > 0:
                    time.sleep(sleep_time)
                self.request_times[user_id].pop(0)
            
            self.request_times[user_id].append(current_time)
        
        # Make request
        response = requests.post(
            "https://api.holysheep.ai/v1/chat/completions",
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            },
            json={
                "model": "gpt-4.1",
                "messages": [{"role": "user", "content": message}],
                "max_tokens": 200
            },
            timeout=30
        )
        
        if response.status_code == 429:
            time.sleep(5)
            return self.call_llm(user_id, message)  # Retry
        
        response.raise_for_status()
        return response.json()
    
    def batch_process(self, user_id: str, messages: List[str]) -> List[dict]:
        results = []
        for msg in messages:
            result = self.call_llm(user_id, msg)
            results.append(result)
            time.sleep(0.1)  # Small delay between requests
        return results

ข้อผิดพลาดที่ 4: ไม่จัดการ Fallback สถานะ

# ❌ วิธีที่ผิด - ไม่มี fallback
def process(self, user_input: str) -> str:
    return self.llm_router.determine_state(user_input)


✅ วิธีที่ถูกต้อง - มี fallback และ escalation

class RobustStateRouter:
    def __init__(self, api_key: str):
        self.llm_router = LLMStateRouter(api_key)
        self.fsm_fallback = FSMDialogueManager()
        self.confidence_threshold = 0.7
    
    def process(self, conversation_history: List[Dict], user_input: str) -> str:
        try:
            result = self.llm_router.determine_next_state(conversation_history)
            
            # High confidence - use LLM decision
            if result["confidence"] >= self.confidence_threshold:
                return result["next_state"]
            
            # Medium confidence - combine with FSM
            fsm_state = self.fsm_fallback.process_input(user_input)
            return self._blend_decisions(result, fsm_state)
            
        except Exception as e:
            # Low confidence or error - escalate to human
            print(f"Routing error: {e}, escalating to human agent")
            return "transfer_to_agent"
    
    def _blend_decisions(self, llm_result: Dict, fsm_state: str) -> str:
        # Prefer FSM for simple commands
        simple_keywords = ["help", "cancel", "reset", "stop"]
        if any(kw in fsm_state for kw in simple_keywords):
            return fsm_state
        
        # Use LLM for complex intent
        if llm_result["confidence"] > 0.5:
            return llm_result["next_state"]
        
        # Default to safer option (FSM)
        return fsm_state

สรุปและคำแนะนำ

จากการทดสอบเชิงลึกทั้ง 3 วิธีการ พบว่าการเลือกวิธีที่เหมาะสมขึ้นอยู่กับ: