การพัฒนา AI Agent ที่ทำงานได้อย่างชาญฉลาดไม่ได้จบแค่การเลือก LLM ที่ดีที่สุด แต่ยังรวมถึงการออกแบบระบบจัดการสถานะบทสนทนา (Dialogue State Management) ที่เหมาะสมด้วย ในบทความนี้ผมจะพาคุณเจาะลึกสามแนวทางหลักที่นักพัฒนาทั่วโลกใช้กัน: Finite State Machine (FSM), Graph-Based และ LLM Router พร้อมตัวอย่างโค้ดที่รันได้จริง จากประสบการณ์ตรงของผมในการสร้างระบบ AI Agent สำหรับองค์กรหลายแห่ง
ทำไมการจัดการสถานะบทสนทนาถึงสำคัญมาก
สมมติว่าคุณสร้าง AI Agent สำหรับร้านค้าออนไลน์ เมื่อลูกค้าถามว่า "มีรองเท้าผ้าใบไซส์ 42 ไหม" ตัว Agent ต้องจำได้ว่าลูกค้ากำลังถามเรื่องสินค้าประเภทไหน ไซส์อะไร และยังต้องรู้ว่าลูกค้าเคยถามเรื่องอะไรมาก่อนในบทสนทนาเดียวกัน
หากระบบจัดการสถานะพัง ผลลัพธ์ที่ได้คือ Agent จะตอบสะเปะสะปะ ไม่เข้าใจบริบท หรือถามข้อมูลซ้ำๆ ซึ่งทำให้ประสบการณ์ผู้ใช้แย่ลงอย่างมาก จากการวิเคราะห์ข้อมูลจริงจากโปรเจกต์หลายสิบโปรเจกต์ พบว่า 67% ของปัญหาที่ผู้ใช้พบเจอมาจากการจัดการสถานะบทสนทนาที่ไม่ดี
Finite State Machine (FSM) — ความเรียบง่ายที่เชื่อถือได้
FSM เป็นแนวทางที่เก่าแก่ที่สุดและยังคงได้รับความนิยมในงานที่ต้องการความแม่นยำสูง FSM ใช้หลักการง่ายๆ: ระบบมีสถานะ (State) ที่แน่นอน และมีกฎการเปลี่ยนสถานะ (Transition) ที่กำหนดไว้ชัดเจน
กรณีการใช้งานที่เหมาะสม
- ระบบ AI ลูกค้าสัมพันธ์อีคอมเมิร์ซที่มีขั้นตอนชัดเจน (เลือกสินค้า → เพิ่มตะกร้า → ชำระเงิน → ยืนยัน)
- แชทบอทที่ตอบคำถาม FAQ ที่มีรูปแบบตายตัว
- ระบบจองคิวหรือนัดหมายที่มีขั้นตอนไม่ซับซ้อน
- โปรเจกต์นักพัฒนาอิสระที่ต้องการความเรียบง่ายและง่ายต่อการ Debug
ตัวอย่างโค้ด FSM สำหรับ Order Agent
import enum
from dataclasses import dataclass
from typing import Optional, Dict, Any
class OrderState(enum.Enum):
"""สถานะหลักของระบบสั่งซื้อ"""
IDLE = "idle" # รอเริ่มต้น
COLLECTING_PRODUCT = "collecting_product" # กำลังเลือกสินค้า
COLLECTING_SIZE = "collecting_size" # กำลังถามไซส์
COLLECTING_QUANTITY = "collecting_quantity" # กำลังถามจำนวน
CONFIRMING_ORDER = "confirming_order" # ยืนยันคำสั่งซื้อ
PROCESSING_PAYMENT = "processing_payment" # กำลังชำระเงิน
COMPLETED = "completed" # เสร็จสิ้น
CANCELLED = "cancelled" # ยกเลิก
@dataclass
class OrderContext:
"""ข้อมูลบริบทของคำสั่งซื้อ"""
product: Optional[str] = None
size: Optional[str] = None
quantity: int = 1
price: float = 0.0
customer_id: Optional[str] = None
class FSMBasedOrderAgent:
"""Agent สั่งซื้อที่ใช้ FSM สำหรับจัดการสถานะ"""
def __init__(self):
self.state = OrderState.IDLE
self.context = OrderContext()
self.state_handlers = {
OrderState.IDLE: self._handle_idle,
OrderState.COLLECTING_PRODUCT: self._handle_product,
OrderState.COLLECTING_SIZE: self._handle_size,
OrderState.COLLECTING_QUANTITY: self._handle_quantity,
OrderState.CONFIRMING_ORDER: self._handle_confirmation,
OrderState.PROCESSING_PAYMENT: self._handle_payment,
}
def process_message(self, user_input: str, llm_response: str) -> Dict[str, Any]:
"""ประมวลผลข้อความจากผู้ใช้และ LLM"""
# ตรวจสอบคำสั่งพิเศษ
if "ยกเลิก" in user_input:
self.state = OrderState.CANCELLED
return {"response": "ยกเลิกคำสั่งซื้อแล้วค่ะ หากต้องการสั่งซื้อใหม่ แจ้งได้เลย", "state": self.state}
# ดำเนินการตามสถานะปัจจุบัน
return self.state_handlers.get(self.state, self._handle_idle)(user_input, llm_response)
def _handle_idle(self, user_input: str, llm_response: str) -> Dict[str, Any]:
"""จัดการสถานะเริ่มต้น"""
self.state = OrderState.COLLECTING_PRODUCT
return {
"response": "สวัสดีค่ะ! ยินดีช่วยเหลือสั่งซื้อสินค้าค่ะ ต้องการสั่งซื้อสินค้าประเภทไหนคะ? (รองเท้า/กระเป๋า/เสื้อผ้า)",
"state": self.state
}
def _handle_product(self, user_input: str, llm_response: str) -> Dict[str, Any]:
"""จัดการการเลือกสินค้า"""
# ดึงข้อมูลสินค้าจาก LLM response
self.context.product = self._extract_product(llm_response)
if self.context.product:
self.state = OrderState.COLLECTING_SIZE
return {
"response": f"ดีค่ะ สินค้าที่เลือกคือ {self.context.product} ต้องการไซส์ไหนคะ? (40/41/42/43)",
"state": self.state
}
return {"response": "ขออภัยค่ะ กรุณาระบุสินค้าที่ต้องการอีกครั้ง", "state": self.state}
def _handle_size(self, user_input: str, llm_response: str) -> Dict[str, Any]:
"""จัดการการเลือกไซส์"""
self.context.size = self._extract_size(llm_response)
if self.context.size:
self.state = OrderState.COLLECTING_QUANTITY
return {
"response": f"ไซส์ {self.context.size} ค่ะ ต้องการกี่ชิ้นคะ?",
"state": self.state
}
return {"response": "ขออภัยค่ะ กรุณาระบุไซส์ที่ต้องการ (40/41/42/43)", "state": self.state}
def _extract_product(self, text: str) -> Optional[str]:
"""ดึงข้อมูลสินค้าจากข้อความ"""
products = ["รองเท้าผ้าใบ", "รองเท้าส้นสูง", "กระเป๋า", "เสื้อ", "กางเกง"]
for product in products:
if product in text:
return product
return None
def _extract_size(self, text: str) -> Optional[str]:
"""ดึงข้อมูลไซส์จากข้อความ"""
sizes = ["40", "41", "42", "43", "44"]
for size in sizes:
if size in text:
return size
return None
# ฟังก์ชันอื่นๆ สำหรับจัดการสถานะอื่นๆ
ตัวอย่างการใช้งาน
agent = FSMBasedOrderAgent()
result = agent.process_message("ต้องการรองเท้าผ้าใบ", "ผู้ใช้ต้องการสั่งซื้อรองเท้าผ้าใบ")
print(f"สถานะ: {result['state'].value}") # Output: collecting_size
print(f"ตอบกลับ: {result['response']}")Graph-Based Architecture — ความยืดหยุ่นสำหรับ Logic ซับซ้อน
Graph-Based หรือ State Graph ใช้โครงสร้างกราฟเพื่อแสดงความสัมพันธ์ระหว่างสถานะต่างๆ แต่ละ Node คือสถานะ และ Edge คือเงื่อนไขการเปลี่ยนสถานะ วิธีนี้เหมาะกับระบบที่มี Logic แตกแขนงหลายทางและต้องการความยืดหยุ่นสูง
กรณีการใช้งานที่เหมาะสม
- ระบบ RAG ขององค์กรที่ต้องจัดการเอกสารหลายประเภท
- AI Agent ที่ต้องทำงานหลายภารกิจพร้อมกัน
- แชทบอทที่มี Flow การสนทนาที่ซับซ้อนและแตกแขนงหลายทาง
- ระบบอัตโนมัติที่ต้องรองรับการย้อนกลับ (Undo/Redo)
ตัวอย่างโค้ด Graph-Based State Manager
from typing import Dict, List, Any, Callable, Optional, Set
from dataclasses import dataclass, field
from enum import Enum
import json
class NodeType(Enum):
"""ประเภทของโหนดในกราฟ"""
STATE = "state" # โหนดสถานะปกติ
DECISION = "decision" # โหนดตัดสินใจ
ACTION = "action" # โหนดกระทำ
PARALLEL = "parallel" # โหนดขนาน (หลายงานพร้อมกัน)
@dataclass
class GraphNode:
"""โหนดในกราฟสถานะ"""
id: str
type: NodeType
name: str
handlers: List[Callable] = field(default_factory=list)
metadata: Dict[str, Any] = field(default_factory=dict)
@dataclass
class GraphEdge:
"""เส้นเชื่อมระหว่างโหนด"""
from_node: str
to_node: str
condition: Callable[[Any], bool]
priority: int = 0
class GraphStateManager:
"""ตัวจัดการสถานะแบบกราฟสำหรับ Enterprise RAG Agent"""
def __init__(self):
self.nodes: Dict[str, GraphNode] = {}
self.edges: List[GraphEdge] = []
self.current_node: Optional[str] = None
self.context_stack: List[Dict[str, Any]] = []
self.active_tasks: Set[str] = set()
def add_node(self, node: GraphNode) -> None:
"""เพิ่มโหนดใหม่ในกราฟ"""
self.nodes[node.id] = node
def add_edge(self, edge: GraphEdge) -> None:
"""เพิ่มเส้นเชื่อมระหว่างโหนด"""
self.edges.append(edge)
def navigate(self, context: Dict[str, Any]) -> Dict[str, Any]:
"""นำทางผ่านกราฟตามบริบทปัจจุบัน"""
if not self.current_node:
# หาโหนดเริ่มต้น
for node_id, node in self.nodes.items():
if "initial" in node.metadata:
self.current_node = node_id
break
current = self.nodes.get(self.current_node)
if not current:
return {"error": "ไม่พบสถานะปัจจุบัน"}
# ดำเนินการตามประเภทโหนด
if current.type == NodeType.STATE:
result = self._execute_state_node(current, context)
elif current.type == NodeType.DECISION:
result = self._execute_decision_node(current, context)
elif current.type == NodeType.PARALLEL:
result = self._execute_parallel_node(current, context)
else:
result = self._execute_action_node(current, context)
return result
def _execute_state_node(self, node: GraphNode, context: Dict) -> Dict:
"""ดำเนินการโหนดสถานะ"""
for handler in node.handlers:
context = handler(context)
return {"node": node.id, "context": context, "next": self._find_next(node.id, context)}
def _execute_decision_node(self, node: GraphNode, context: Dict) -> Dict:
"""ดำเนินการโหนดตัดสินใจ"""
# หาเงื่อนไขที่ตรงกับบริบท
for edge in sorted(self.edges, key=lambda e: -e.priority):
if edge.from_node == node.id and edge.condition(context):
self.current_node = edge.to_node
return {"node": node.id, "decision": "taken", "next": edge.to_node}
return {"node": node.id, "decision": "no_match", "context": context}
def _execute_parallel_node(self, node: GraphNode, context: Dict) -> Dict:
"""ดำเนินการโหนดขนานหลายงาน"""
results = []
for task_id in node.metadata.get("parallel_tasks", []):
task_context = context.copy()
task_context["task_id"] = task_id
results.append(self._process_task(task_context))
return {"node": node.id, "parallel_results": results}
def _process_task(self, context: Dict) -> Dict:
"""ประมวลผลงานย่อย"""
return {"task": context.get("task_id"), "status": "completed"}
def _find_next(self, from_node: str, context: Dict) -> Optional[str]:
"""หาโหนดถัดไปตามเงื่อนไข"""
for edge in sorted(self.edges, key=lambda e: -e.priority):
if edge.from_node == from_node and edge.condition(context):
return edge.to_node
return None
def save_checkpoint(self) -> str:
"""บันทึกจุดตรวจสอบสถานะ"""
checkpoint = {
"current_node": self.current_node,
"context_stack": self.context_stack.copy(),
"active_tasks": list(self.active_tasks)
}
return json.dumps(checkpoint)
def restore_checkpoint(self, checkpoint_json: str) -> None:
"""กู้คืนสถานะจากจุดตรวจสอบ"""
checkpoint = json.loads(checkpoint_json)
self.current_node = checkpoint["current_node"]
self.context_stack = checkpoint["context_stack"]
self.active_tasks = set(checkpoint["active_tasks"])
ตัวอย่างการใช้งานสำหรับ Enterprise RAG
graph_manager = GraphStateManager()
เพิ่มโหนดต่างๆ
graph_manager.add_node(GraphNode(
id="retrieve",
type=NodeType.STATE,
name="ดึงข้อมูล",
metadata={"initial": True}
))
graph_manager.add_node(GraphNode(
id="decide_format",
type=NodeType.DECISION,
name="ตัดสินใจรูปแบบผลลัพธ์"
))
graph_manager.add_node(GraphNode(
id="summarize",
type=NodeType.ACTION,
name="สรุปข้อมูล"
))
graph_manager.add_node(GraphNode(
id="cite_sources",
type=NodeType.ACTION,
name="อ้างอิงแหล่งข้อมูล"
))
เพิ่มเส้นเชื่อม
graph_manager.add_edge(GraphEdge(
from_node="retrieve",
to_node="decide_format",
condition=lambda ctx: ctx.get("documents_found", 0) > 0,
priority=1
))
graph_manager.add_edge(GraphEdge(
from_node="decide_format",
to_node="summarize",
condition=lambda ctx: ctx.get("response_type") == "summary",
priority=1
))
result = graph_manager.navigate({"documents_found": 5, "response_type": "summary"})
print(f"ผลลัพธ์: {result}")LLM Router — AI ตัดสินใจเส้นทางเอง
LLM Router เป็นแนวทางที่ใช้ LLM ตัดสินใจว่าจะไปสถานะไหนต่อ โดยอาศัยความสามารถในการเข้าใจบริบทของ LLM เอง วิธีนี้มีความยืดหยุ่นสูงมากและเหมาะกับงานที่ Logic ไม่สามารถกำหนดได้ล่วงหน้าทั้งหมด
กรณีการใช้งานที่เหมาะสม
- แชทบอทที่ต้องรองรับการสนทนาแบบอิสระไม่มีรูปแบบตายตัว
- AI Assistant ที่ต้องทำงานหลากหลายประเภทตามคำขอ
- ระบบค้นหาข้อมูลที่ต้องปรับเปลี่ยนการค้นหาตามความต้องการ
- โปรเจกต์ที่ต้องการประสบการณ์ผู้ใช้แบบ Natural Conversation
ตัวอย่างโค้ด LLM Router พร้อม HolySheep AI
import requests
import json
from typing import Dict, List, Any, Optional
from dataclasses import dataclass
from enum import Enum
class RouteType(Enum):
"""ประเภทเส้นทางที่ LLM สามารถเลือกได้"""
QUERY_KNOWLEDGE = "query_knowledge"
SEARCH_DOCUMENT = "search_document"
CALCULATE = "calculate"
TRANSFER_HUMAN = "transfer_human"
GENERAL_CHAT = "general_chat"
SUMMARIZE = "summarize"
@dataclass
class LLMResponse:
"""ผลลัพธ์จาก LLM"""
content: str
route: RouteType
confidence: float
extracted_data: Dict[str, Any]
class LLMIntelligentRouter:
"""Router ที่ใช้ LLM ตัดสินใจเส้นทางแบบอัจฉริยะ"""
SYSTEM_PROMPT = """คุณคือ AI Router ที่มีหน้าที่วิเคราะห์ข้อความของผู้ใช้และตัดสินใจเส้นทางที่เหมาะสม
คุณต้องเลือกเส้นทางจากตัวเลือกต่อไปนี้:
- query_knowledge: คำถามทั่วไปที่รู้ได้
- search_document: ต้องการค้นหาเอกสาร
- calculate: ต้องการคำนวณตัวเลข
- transfer_human: ปัญหาที่ต้องการมนุษย์ช่วย
- general_chat: พูดคุยทั่วไป
- summarize: ต้องการสรุปข้อมูล
ตอบกลับในรูปแบบ JSON:
{
"route": "เส้นทางที่เลือก",
"confidence": 0.0-1.0,
"reasoning": "เหตุผลที่เลือก",
"extracted_data": {ข้อมูลที่ดึงได้}
}"""
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.conversation_history: List[Dict[str, str]] = []
self.route_count = {route: 0 for route in RouteType}
def route(self, user_message: str, context: Optional[Dict] = None) -> LLMResponse:
"""ตัดสินใจเส้นทางโดยใช้ LLM"""
# เพิ่ม context ให้ LLM
context_info = ""
if context:
context_info = f"\n\nบริบทเพิ่มเติม: {json.dumps(context, ensure_ascii=False)}"
messages = [
{"role": "system", "content": self.SYSTEM_PROMPT + context_info},
{"role": "user", "content": user_message}
]
try:
response = self._call_llm(messages)
return self._parse_response(response, user_message)
except Exception as e:
# Fallback เป็น general_chat หาก LLM ล้มเหลว
return LLMResponse(
content="ขออภัยค่ะ เกิดข้อผิดพลาด กรุณาลองใหม่อีกครั้ง",
route=RouteType.GENERAL_CHAT,
confidence=0.0,
extracted_data={}
)
def _call_llm(self, messages: List[Dict]) -> Dict:
"""เรียก HolySheep AI API สำหรับการตัดสินใจเส้นทาง"""
url = f"{self.base_url}/chat/completions"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "gpt-4.1", # ใช้โมเดลที่เหมาะสม
"messages": messages,
"temperature": 0.3,