จากประสบการณ์การพัฒนา AI Agent มากว่า 3 ปี ผมเชื่อว่าหลายคนเคยเจอปัญหาเดียวกัน — สร้าง Chatbot ง่ายได้ แต่พอต้องทำให้ระบบจดจำสถานะการสนทนา ตัดสินใจแบบมีเงื่อนไข และรองรับเวิร์กโฟลว์ที่ซับซ้อน กลับกลายเป็นฝันร้าย วันนี้ผมจะมาแชร์วิธีที่ผมใช้ LangGraph ตั้งแต่ต้นจนถึง Production รวมถึงการเชื่อมต่อกับ HolySheep AI ที่ราคาถูกกว่า 85% และ Latency ต่ำกว่า 50ms
ทำไมต้อง LangGraph? ปัญหาที่ AI Agent แบบเดิมแก้ไม่ได้
สมัยก่อนเวลาสร้าง AI Agent เรามักใช้ if-else หรือ switch-case เพื่อควบคุม logic ซึ่งมีข้อจำกัดหลายอย่าง:
- ไม่มี State Management: ไม่สามารถจดจำสถานะระหว่าง Turn ของการสนทนา
- Logic ซับซ้อน: เมื่อมี Branch หลายระดับ โค้ดจะกลายเป็น Spaghetti Code
- ไม่รองรับ Long-Running Task: ระบบล้มเหลวกลางคันแล้วไม่รู้จะเริ่มต่อจากไหน
- Debug ยาก: ไม่มี Visualization ของ Workflow ที่ทำงานอยู่
LangGraph มาแก้ปัญหาเหล่านี้ด้วย Graph-based Architecture ที่ทำให้เรามอง AI Agent เป็น "กราฟของสถานะ" แทนที่จะเป็น "โค้ดคำสั่งเชิงเส้น" ปัจจุบัน LangGraph มี 90K+ Stars บน GitHub ถือเป็น Standard ของ Industry แล้ว
กรณีศึกษา: AI ลูกค้าสัมพันธ์อีคอมเมิร์ซแบบ Stateful
ผมจะใช้โปรเจกต์จริงที่เคยทำ — ระบบ Chatbot ช่วยเลือกสินค้าสำหรับร้านค้าออนไลน์ ซึ่งมีความต้องการดังนี้:
- จดจำประวัติการสนทนาและสินค้าที่ดู
- รองรับการตอบคำถามเรื่องสินค้า สถานะคำสั่งซื้อ การคืนสินค้า
- Escalate ไปพนักงานคนเมื่อ AI ไม่แน่ใจ
- สรุปบทสนทนาส่งให้ CRM หลังจบ
เริ่มต้น: ติดตั้งและ Setup Environment
# สร้าง Virtual Environment
python -m venv langgraph-env
source langgraph-env/bin/activate # Windows: langgraph-env\Scripts\activate
ติดตั้ง Dependencies
pip install langgraph langchain-core langchain-holysheep
pip install langchain-community # สำหรับ Tool integrations
pip install python-dotenv redis # สำหรับ State persistence
ตรวจสอบ version
python -c "import langgraph; print(f'LangGraph version: {langgraph.__version__}')"
Output: LangGraph version: 0.2.x หรือสูงกว่า
เขียนโค้ด: AI Customer Service Agent ด้วย LangGraph
import os
from typing import Annotated, TypedDict, Literal
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_holysheep import HolySheep
========== Configuration ==========
os.environ["HOLYSHEEP_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
Initialize HolySheep LLM - ราคาถูกกว่า 85%, Latency <50ms
llm = HolySheep(
base_url=HOLYSHEEP_BASE_URL,
api_key=os.environ["HOLYSHEEP_API_KEY"],
model="gpt-4.1" # $8/MTok - ประหยัดมากสำหรับ Customer Service
)
========== Define State Schema ==========
class CustomerServiceState(TypedDict):
"""สถานะทั้งหมดของระบบ Customer Service"""
messages: Annotated[list, add_messages]
customer_id: str
cart_items: list[str]
order_history: list[dict]
current_intent: str | None
escalation_needed: bool
session_summary: str
========== Define Intent Detection Node ==========
def detect_intent_node(state: CustomerServiceState) -> CustomerServiceState:
"""วิเคราะห์ Intent ของลูกค้าจากข้อความล่าสุด"""
# ดึงข้อความล่าสุด
last_message = state["messages"][-1].content
# Prompt สำหรับ Intent Classification
intent_prompt = f"""คุณคือ AI Intent Classifier สำหรับร้านค้าอีคอมเมิร์ซ
จงวิเคราะห์ข้อความต่อไปนี้และระบุ Intent:
ข้อความ: {last_message}
Intent ที่เป็นไปได้:
- product_inquiry: สอบถามเรื่องสินค้า
- order_status: สอบถามสถานะคำสั่งซื้อ
- return_request: ขอคืนสินค้า
- recommendation: ขอคำแนะนำสินค้า
- general: คำถามทั่วไป
- escalation: ต้องการพูดคุยกับคน
ตอบกลับเฉพาะ Intent เท่านั้น"""
response = llm.invoke([HumanMessage(content=intent_prompt)])
detected_intent = response.content.strip().lower()
# ตรวจสอบว่าต้อง Escalate หรือไม่
escalation_keywords = ["คน", "พนักงาน", "ผู้จัดการ", "เป็นห่วง", "กังวล"]
needs_escalation = any(keyword in last_message for keyword in escalation_keywords)
return {
**state,
"current_intent": detected_intent,
"escalation_needed": needs_escalation or "escalation" in detected_intent
}
========== Define Route Function ==========
def route_based_on_intent(
state: CustomerServiceState
) -> Literal["product_inquiry_node", "order_status_node",
"return_request_node", "recommendation_node", "general_node",
"escalation_node"]:
"""Route ไปยัง Node ที่เหมาะสมตาม Intent"""
intent = state.get("current_intent", "general")
if state["escalation_needed"]:
return "escalation_node"
intent_mapping = {
"product_inquiry": "product_inquiry_node",
"order_status": "order_status_node",
"return_request": "return_request_node",
"recommendation": "recommendation_node",
}
return intent_mapping.get(intent, "general_node")
========== Define Intent-Specific Nodes ==========
def product_inquiry_node(state: CustomerServiceState) -> CustomerServiceState:
"""จัดการคำถามเรื่องสินค้า"""
messages = state["messages"]
cart = state["cart_items"]
product_prompt = f"""คุณคือ Product Specialist ของร้าน TechMart
ข้อมูลสินค้าในตะกร้า: {cart}
ประวัติการสนทนา:
{self._format_messages(messages)}
จงตอบคำถามลูกค้าอย่างเป็นมิตร และแนะนำสินค้าที่เกี่ยวข้อง
ถ้าลูกค้าถามเรื่องสินค้าเฉพาะ ให้ค้นหาข้อมูลจาก Product Database"""
response = llm.invoke(messages + [HumanMessage(content=product_prompt)])
return {
**state,
"messages": messages + [AIMessage(content=response.content)]
}
def order_status_node(state: CustomerServiceState) -> CustomerServiceState:
"""ตรวจสอบสถานะคำสั่งซื้อ"""
messages = state["messages"]
order_history = state.get("order_history", [])
status_prompt = f"""คุณคือ Order Status Assistant
ประวัติคำสั่งซื้อของลูกค้า:
{order_history}
จงตรวจสอบสถานะคำสั่งซื้อและแจ้งลูกค้า
รวมถึง ETA ของการจัดส่งด้วย"""
response = llm.invoke(messages + [HumanMessage(content=status_prompt)])
return {
**state,
"messages": messages + [AIMessage(content=response.content)]
}
def escalation_node(state: CustomerServiceState) -> CustomerServiceState:
"""ส่งต่อไปยังพนักงานคน"""
messages = state["messages"]
customer_id = state["customer_id"]
escalation_prompt = f"""คุณคือ AI ที่กำลังส่งต่อบทสนทนาไปยังพนักงาน
รหัสลูกค้า: {customer_id}
จงแจ้งลูกค้าว่ากำลังเชื่อมต่อกับพนักงาน และขอข้อมูลเพิ่มเติม
พร้อมทั้งสรุปบริบทให้พนักงานทราบ"""
response = llm.invoke(messages + [HumanMessage(content=escalation_prompt)])
return {
**state,
"messages": messages + [AIMessage(content=response.content)],
"session_summary": self._generate_summary(state)
}
def recommendation_node(state: CustomerServiceState) -> CustomerServiceState:
"""แนะนำสินค้าตามพฤติกรรมลูกค้า"""
messages = state["messages"]
cart = state["cart_items"]
rec_prompt = f"""คุณคือ Personal Shopper AI
สินค้าที่ลูกค้าดู/ใส่ตะกร้า: {cart}
จงแนะนำสินค้าที่เกี่ยวข้อง 5 รายการ พร้อมเหตุผล
คำนึงถึง:
- สินค้าที่ดูบ่อย
- สินค้าในหมวดเดียวกัน
- โปรโมชั่นปัจจุบัน"""
response = llm.invoke(messages + [HumanMessage(content=rec_prompt)])
return {
**state,
"messages": messages + [AIMessage(content=response.content)]
}
def general_node(state: CustomerServiceState) -> CustomerServiceState:
"""ตอบคำถามทั่วไป"""
messages = state["messages"]
general_prompt = """คุณคือ Customer Service Bot ของ TechMart
ตอบคำถามลูกค้าอย่างเป็นมิตรและให้ข้อมูลที่ถูกต้อง
ถ้าไม่แน่ใจ ให้แนะนำให้ติดต่อพนักงาน"""
response = llm.invoke(
messages + [SystemMessage(content=general_prompt)]
)
return {
**state,
"messages": messages + [AIMessage(content=response.content)]
}
========== Build the Graph ==========
def build_customer_service_graph():
"""สร้าง StateGraph สำหรับ Customer Service"""
# Initialize Graph
workflow = StateGraph(CustomerServiceState)
# เพิ่ม Nodes
workflow.add_node("intent_detector", detect_intent_node)
workflow.add_node("product_inquiry_node", product_inquiry_node)
workflow.add_node("order_status_node", order_status_node)
workflow.add_node("return_request_node", return_request_node)
workflow.add_node("recommendation_node", recommendation_node)
workflow.add_node("escalation_node", escalation_node)
workflow.add_node("general_node", general_node)
# Set Entry Point
workflow.set_entry_point("intent_detector")
# เพิ่ม Conditional Edges
workflow.add_conditional_edges(
"intent_detector",
route_based_on_intent,
{
"product_inquiry_node": "product_inquiry_node",
"order_status_node": "order_status_node",
"return_request_node": "return_request_node",
"recommendation_node": "recommendation_node",
"general_node": "general_node",
"escalation_node": "escalation_node",
}
)
# เพิ่ม Edges กลับไปยัง Intent Detector (Loop)
for node in ["product_inquiry_node", "order_status_node",
"return_request_node", "recommendation_node",
"general_node", "escalation_node"]:
workflow.add_edge(node, END)
return workflow.compile()
========== Helper Methods ==========
@staticmethod
def _format_messages(messages: list) -> str:
return "\n".join([
f"{'ลูกค้า' if isinstance(m, HumanMessage) else 'AI'}: {m.content}"
for m in messages[-5:] # 5 ข้อความล่าสุด
])
@staticmethod
def _generate_summary(state: CustomerServiceState) -> str:
return f"สรุป Session - Intent: {state['current_intent']}, " \
f"สินค้าในตะกร้า: {len(state['cart_items'])} รายการ"
========== Usage Example ==========
if __name__ == "__main__":
# สร้าง Graph
graph = build_customer_service_graph()
# Initial State
initial_state = {
"messages": [HumanMessage(content="อยากทราบสถานะคำสั่งซื้อ #12345")],
"customer_id": "CUST_001",
"cart_items": ["iPhone 15", "AirPods Pro"],
"order_history": [
{"order_id": "12345", "status": "shipping", "eta": "2 วัน"}
],
"current_intent": None,
"escalation_needed": False,
"session_summary": ""
}
# Run Graph
result = graph.invoke(initial_state)
print("=" * 50)
print("ผลลัพธ์จาก AI Agent:")
print("=" * 50)
print(result["messages"][-1].content)
print(f"\nDetected Intent: {result['current_intent']}")
print(f"Escalation: {result['escalation_needed']}")Advanced: เพิ่ม Persistence Layer ด้วย Redis
สำหรับ Production จริง เราต้องการให้ระบบจดจำสถานะข้าม Session ได้ ผมใช้ Redis เป็น Persistence Store
import json
from langgraph.checkpoint.redis import RedisSaver
import redis
class StatefulCustomerService:
"""Customer Service Agent พร้อม State Persistence"""
def __init__(self, redis_url: str = "redis://localhost:6379"):
# เชื่อมต่อ Redis
self.redis_client = redis.from_url(redis_url)
# สร้าง Checkpointer
self.checkpointer = RedisSaver(self.redis_client)
# Build Graph พร้อม Checkpointer
self.graph = self._build_graph_with_checkpointer()
# System Prompt
self.system_prompt = """คุณคือ Customer Service Bot ของ TechMart
คุณต้อง:
1. ตอบคำถามลูกค้าอย่างเป็นมิตร
2. จดจำสินค้าที่ลูกค้าสนใจ
3. แนะนำสินค้าที่เหมาะสม
4. ส่งต่อพนักงานเมื่อจำเป็น"""
def _build_graph_with_checkpointer(self):
"""Build Graph พร้อม Memory"""
# (โค้ด Graph เหมือนด้านบน)
workflow = StateGraph(CustomerServiceState)
# ... เพิ่ม nodes และ edges ...
return workflow.compile(checkpointer=self.checkpointer)
def chat(self, customer_id: str, user_message: str, thread_id: str = None):
"""ส่งข้อความและรับ Response"""
# สร้าง Thread ID ถ้าไม่มี
if thread_id is None:
thread_id = f"thread_{customer_id}_{int(time.time())}"
# Config สำหรับ Memory
config = {
"configurable": {
"thread_id": thread_id,
"customer_id": customer_id
}
}
# Load State ก่อนหน้าจาก Redis
existing_state = None
try:
existing_state = self.graph.get_state(config)
except:
pass
# Initial State ถ้ายังไม่มี Session
if existing_state is None:
initial_state = CustomerServiceState(
messages=[HumanMessage(content=user_message)],
customer_id=customer_id,
cart_items=self._load_cart_from_redis(customer_id),
order_history=self._load_orders_from_redis(customer_id),
current_intent=None,
escalation_needed=False,
session_summary=""
)
else:
# เพิ่มข้อความใหม่ต่อจาก State เดิม
initial_state = existing_state.values
initial_state["messages"] = initial_state["messages"] + \
[HumanMessage(content=user_message)]
# Run Graph
result = self.graph.invoke(initial_state, config=config)
# Cache ข้อมูลลง Redis
self._cache_cart_to_redis(customer_id, result["cart_items"])
return {
"response": result["messages"][-1].content,
"thread_id": thread_id,
"intent": result["current_intent"],
"needs_escalation": result["escalation_needed"]
}
def _load_cart_from_redis(self, customer_id: str) -> list:
"""โหลดตะกร้าสินค้าจาก Redis"""
cart_data = self.redis_client.get(f"cart:{customer_id}")
if cart_data:
return json.loads(cart_data)
return []
def _cache_cart_to_redis(self, customer_id: str, cart: list):
"""บันทึกตะกร้าสินค้าลง Redis"""
self.redis_client.setex(
f"cart:{customer_id}",
86400 * 7, # 7 วัน
json.dumps(cart)
)
def _load_orders_from_redis(self, customer_id: str) -> list:
"""โหลดประวัติคำสั่งซื้อจาก Redis"""
orders_data = self.redis_client.get(f"orders:{customer_id}")
if orders_data:
return json.loads(orders_data)
return []
========== Flask API Example ==========
from flask import Flask, request, jsonify
app = Flask(__name__)
customer_service = StatefulCustomerService(redis_url="redis://localhost:6379")
@app.route("/chat", methods=["POST"])
def chat():
data = request.json
result = customer_service.chat(
customer_id=data["customer_id"],
user_message=data["message"],
thread_id=data.get("thread_id")
)
return jsonify(result)
@app.route("/cart", methods=["GET"])
def get_cart():
customer_id = request.args.get("customer_id")
cart = customer_service._load_cart_from_redis(customer_id)
return jsonify({"cart": cart})
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5000, debug=True)Production Deployment: Docker + Kubernetes
สำหรับการ Deploy ขึ้น Production ผมแนะนำให้ใช้ Docker Container และ Kubernetes
# Dockerfile
FROM python:3.11-slim
WORKDIR /app
Install Dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
Copy Application
COPY . .
Environment Variables
ENV HOLYSHEEP_API_KEY=${HOLYSHEEP_API_KEY}
ENV REDIS_URL=redis://redis:6379
Expose Port
EXPOSE 5000
Health Check
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s \
CMD curl -f http://localhost:5000/health || exit 1
Run Application
CMD ["gunicorn", "--bind", "0.0.0.0:5000", "--workers", "4", "--threads", "2", "app:app"]ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
กรณีที่ 1: "Invalid base_url format" Error
สาเหตุ: ลืมใส่ slash ท้าย URL หรือใช้ URL ผิด
# ❌ ผิด - จะ Error
llm = HolySheep(
base_url="https://api.holysheep.ai/v1/",
api_key="YOUR_HOLYSHEEP_API_KEY"
)
✅ ถูกต้อง
llm = HolySheep(
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY"
)กรณีที่ 2: State ไม่ถูกจดจำระหว่าง Turn
สาเหตุ: ไม่ได้ส่ง config ที่มี thread_id ซ้ำ
# ❌ ผิด - State จะไม่ถูกจดจำ
result = graph.invoke({"messages": [HumanMessage(content="สวัสดี")]})
✅ ถูกต้อง - ส่ง thread_id เดิม
config = {"configurable": {"thread_id": "customer_123_session_1"}}
result = graph.invoke({"messages": [HumanMessage(content="สวัสดี")]}, config=config)
ข้อความต่อไปใช้ thread_id เดิม
result2 = graph.invoke(
{"messages": [HumanMessage(content="อยากดู iPhone")]},
config=config # ต้องส่ง config เดิม
)กรณีที่ 3: Token Limit Exceeded ใน Session ยาว
สาเหตุ: ส่ง messages ทั้งหมดให้ LLM จนเกิน context window
# ❌ ผิด - ส่ง messages ทั้งหมด
all_messages = state["messages"]
response = llm.invoke(all_messages)
✅ ถูกต้อง - ส่งเฉพาะ N ข้อความล่าสุด
def trim_messages(messages: list, max_turns: int = 10) -> list:
"""ตัดข้อความเก่าออกเหลือเฉพาะ max_turns"""
# 1 turn = 1 human + 1 AI
return messages[-(max_turns * 2):]
trimmed_messages = trim_messages(state["messages"], max_turns=10)
response = llm.invoke(trimmed_messages)
หรือใช้ Summarization
if len(state["messages"]) > 20:
summary = llm.invoke([
SystemMessage(content="สรุปบทสนทนาต่อไปนี้ให้กระชับ"),
HumanMessage(content=str(state["messages"][:-10]))
])
state["messages"] = [
SystemMessage(content=f"สรุปการสนทนาก่อนหน้า: {summary.content}")
] + state["messages"][-10:]กรณีที่ 4: Redis Connection Refused
สาเหตุ: Redis Server ไม่ได้ทำงานหรือ URL ผิด
# ✅ วิธีแก้: ตรวจสอบ Redis Connection
import redis
try:
r = redis.from_url("redis://localhost:6379", decode_responses=True)
r.ping()
print("Redis connection OK")
except redis.ConnectionError as e:
print(f"Redis not available: {e}")
# Fallback ไปใช้ Memory Saver แทน
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
✅ หรือใช้ Memory Saver สำหรับ Development
from langgraph.checkpoint.memory import MemorySaver
memory