LangGraph State Machine: สร้าง AI Agent ที่ควบคุมง่ายแม้ไม่มีประสบการณ์เขียนโค้ด

คุณเคยสงสัยไหมว่า AI Agent ที่ฉลาดๆ อย่าง ChatGPT หรือ Claude ทำงานเป็นขั้นตอนได้อย่างไร? คำตอบคือ State Machine — เทคนิคที่ช่วยให้ AI รู้ว่าตอนนี้อยู่ขั้นตอนไหน และควรไปขั้นตอนถัดไปอย่างไร

ในบทความนี้ ผมจะสอนคุณสร้าง AI Agent ด้วย LangGraph ตั้งแต่เริ่มต้น ไม่ต้องมีพื้นฐานการเขียนโค้ดมาก่อนก็ทำได้ โดยใช้ HolySheep AI เป็น API หลักที่ราคาประหยัดกว่า 85% และเร็วต่ำกว่า 50 มิลลิวินาที

State Machine คืออะไร — อธิบายแบบเข้าใจง่าย

ลองนึกภาพเครื่องขายของอัตโนมัติ:

เมื่อใส่เงิน → เครื่องรอเลือกสินค้า
เมื่อเลือกสินค้า → เครื่องจ่ายของ
เมื่อได้ของแล้ว → เครื่องคืนเงินทอน

แต่ละสถานะมีเงื่อนไขเฉพาะที่ทำให้เปลี่ยนไปสถานะถัดไป AI Agent ก็ทำงานคล้ายกัน:

State (สถานะ): คือ "งานที่ AI กำลังทำอยู่" เช่น รอคำถาม, กำลังค้นหา, กำลังตอบ
Transition (การเปลี่ยนสถานะ): คือ "เงื่อนไขที่ทำให้เปลี่ยนไปทำอย่างอื่น"

ทำไมต้องใช้ LangGraph?

LangGraph เป็นไลบรารีที่ช่วยสร้าง State Machine สำหรับ AI โดยเฉพาะ:

มอง AI เป็นกราฟของสถานะ ต่อกันเป็นลำดับ
เพิ่มความสามารถให้ AI ทำงานหลายขั้นตอนติดต่อกัน
มีระบบ Memory เก็บสิ่งที่เกิดขึ้นก่อนหน้า
รองรับการวนลูป (กลับไปทำขั้นตอนเดิม)

เริ่มต้นติดตั้งและตั้งค่า

1. ติดตั้งโปรแกรมที่จำเป็น

เปิด Terminal (หรือ Command Prompt) แล้วพิมพ์คำสั่งนี้:

pip install langgraph langchain-openai python-dotenv

2. สร้างไฟล์ .env เก็บ API Key

สร้างไฟล์ชื่อ .env ในโฟลเดอร์เดียวกับโค้ดของคุณ:

HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY

3. ตั้งค่า HolySheep API

คุณสามารถสมัคร HolySheep AI ได้ที่ สมัครที่นี่ ซึ่งให้อัตราแลกเปลี่ยน ¥1=$1 (ประหยัดกว่า 85%) และเครดิตฟรีเมื่อลงทะเบียน โดยราคาของ GPT-4.1 อยู่ที่ $8/ล้าน Token, Claude Sonnet 4.5 อยู่ที่ $15/ล้าน Token และ Gemini 2.5 Flash เพียง $2.50/ล้าน Token

สร้าง State Machine แรกของคุณ

โครงสร้างพื้นฐาน

import os
from dotenv import load_dotenv
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

โหลด API Key
load_dotenv()

กำหนดประเภทข้อมูล State (สถานะ)
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    current_step: str
    context: dict

def create_agent():
    """สร้าง Agent แบบง่ายที่มี 3 ขั้นตอน"""
    
    # สร้างกราฟ
    graph = StateGraph(AgentState)
    
    # เพิ่มโหนด (ขั้นตอนการทำงาน)
    graph.add_node("greeting", greeting_node)
    graph.add_node("process", process_node)
    graph.add_node("response", response_node)
    
    # กำหนดเส้นทางการทำงาน
    graph.add_edge(START, "greeting")
    graph.add_edge("greeting", "process")
    graph.add_edge("process", "response")
    graph.add_edge("response", END)
    
    return graph.compile()

def greeting_node(state):
    """ขั้นตอนที่ 1: ทักทาย"""
    return {"current_step": "greeting", "context": {"status": "done"}}

def process_node(state):
    """ขั้นตอนที่ 2: ประมวลผล"""
    return {"current_step": "process"}

def response_node(state):
    """ขั้นตอนที่ 3: ตอบกลับ"""
    return {"current_step": "response"}

ทดสอบ
agent = create_agent()
result = agent.invoke({"messages": [], "current_step": "start", "context": {}})
print(result)

เวอร์ชันเต็มที่เชื่อมต่อ HolySheep API

import os
from dotenv import load_dotenv
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
import openai

โหลด API Key
load_dotenv()

ตั้งค่า HolySheep เป็น OpenAI Compatible
client = openai.OpenAI(
    api_key=os.getenv("HOLYSHEEP_API_KEY"),
    base_url="https://api.holysheep.ai/v1"  # ต้องใช้ URL นี้เท่านั้น
)

กำหนด State
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    current_step: str
    user_request: str

def call_holysheep(prompt: str) -> str:
    """เรียก HolySheep API เพื่อประมวลผล"""
    response = client.chat.completions.create(
        model="gpt-4.1",  # หรือเลือก claude-sonnet-4.5, gemini-2.5-flash
        messages=[{"role": "user", "content": prompt}],
        temperature=0.7
    )
    return response.choices[0].message.content

def receive_node(state):
    """รับคำถามจากผู้ใช้"""
    return {"current_step": "waiting_input"}

def analyze_node(state):
    """วิเคราะห์คำถาม"""
    messages = state["messages"]
    if messages:
        last_message = messages[-1].content
        analysis = call_holysheep(f"จำแนกประเภทคำถามนี้: {last_message}")
        return {"current_step": "analyzed", "context": {"analysis": analysis}}
    return {"current_step": "analyze_failed"}

def answer_node(state):
    """ตอบคำถาม"""
    messages = state["messages"]
    if messages:
        last_message = messages[-1].content
        answer = call_holysheep(f"ตอบคำถามนี้อย่างละเอียด: {last_message}")
        new_messages = messages + [{"role": "assistant", "content": answer}]
        return {"current_step": "completed", "messages": new_messages}
    return {"current_step": "answer_failed"}

def create_smart_agent():
    """สร้าง Agent ที่เชื่อมต่อกับ HolySheep"""
    graph = StateGraph(AgentState)
    
    graph.add_node("receive", receive_node)
    graph.add_node("analyze", analyze_node)
    graph.add_node("answer", answer_node)
    
    # กำหนดเส้นทาง
    graph.add_edge(START, "receive")
    graph.add_edge("receive", "analyze")
    graph.add_edge("analyze", "answer")
    graph.add_edge("answer", END)
    
    return graph.compile()

ทดสอบ Agent
agent = create_smart_agent()
test_state = {
    "messages": [{"role": "user", "content": "อธิบายเรื่อง AI ให้ฟังหน่อย"}],
    "current_step": "start",
    "user_request": "อธิบายเรื่อง AI"
}
result = agent.invoke(test_state)
print("ผลลัพธ์:", result)

เพิ่มความฉลาดด้วย Conditional Branching

Conditional Branching คือการตัดสินใจว่าจะไปขั้นตอนไหนต่อ ขึ้นอยู่กับสถานการณ์:

from typing import Literal

def route_decision(state) -> Literal["handle_question", "handle_complaint", "end"]:
    """ตัดสินใจเส้นทางตามประเภทคำถาม"""
    messages = state["messages"]
    if messages:
        last_msg = messages[-1].content.lower()
        
        if any(word in last_msg for word in ["ซื้อ", "ราคา", "สั่งซื้อ"]):
            return "handle_question"
        elif any(word in last_msg for word in ["ไม่พอใจ", "ร้องเรียน", "แจ้งปัญหา"]):
            return "handle_complaint"
        else:
            return "end"
    return "end"

def create_routing_agent():
    """สร้าง Agent ที่มีการแยกเส้นทาง"""
    graph = StateGraph(AgentState)
    
    graph.add_node("classify", analyze_node)
    graph.add_node("handle_question", answer_node)
    graph.add_node("handle_complaint", handle_complaint_node)
    graph.add_node("end", end_node)
    
    graph.add_edge(START, "classify")
    graph.add_conditional_edges(
        "classify",
        route_decision,
        {
            "handle_question": "handle_question",
            "handle_complaint": "handle_complaint",
            "end": "end"
        }
    )
    graph.add_edge("handle_question", END)
    graph.add_edge("handle_complaint", END)
    graph.add_edge("end", END)
    
    return graph.compile()

def handle_complaint_node(state):
    """จัดการเรื่องร้องเรียนด้วยความเข้าใจ"""
    messages = state["messages"]
    last_msg = messages[-1].content if messages else ""
    
    response = call_holysheep(
        f"รับฟังปัญหานี้ด้วยความเห็นอกเห็นใจ และช่วยหาทางออก:\n{last_msg}"
    )
    return {
        "current_step": "complaint_handled",
        "messages": messages + [{"role": "assistant", "content": response}]
    }

def end_node(state):
    """จบการทำงาน"""
    return {"current_step": "ended"}

ทดสอบ
routing_agent = create_routing_agent()
test_complaint = {
    "messages": [{"role": "user", "content": "ฉันไม่พอใจกับบริการ"}],
    "current_step": "start",
    "user_request": "ร้องเรียน"
}
result = routing_agent.invoke(test_complaint)

เพิ่ม Memory ให้ Agent จดจำสนทนา

from langgraph.checkpoint.memory import MemorySaver

def create_persistent_agent():
    """สร้าง Agent ที่จดจำสนทนาได้"""
    memory = MemorySaver()
    
    graph = StateGraph(AgentState)
    
    graph.add_node("receive", receive_node)
    graph.add_node("analyze", analyze_node)
    graph.add_node("answer", answer_node)
    
    graph.add_edge(START, "receive")
    graph.add_edge("receive", "analyze")
    graph.add_edge("analyze", "answer")
    graph.add_edge("answer", END)
    
    # ส่ง memory เข้าไปใน compile
    return graph.compile(checkpointer=memory)

สร้าง Agent
agent = create_persistent_agent()

กำหนด thread_id สำหรับจดจำสนทนา
config = {"configurable": {"thread_id": "user-001"}}

ครั้งที่ 1
state1 = {
    "messages": [{"role": "user", "content": "ฉันชื่อมิน"}],
    "current_step": "start",
    "user_request": "แนะนำตัว"
}
result1 = agent.invoke(state1, config=config)

ครั้งที่ 2 - Agent จะจำได้ว่าชื่อมิน
state2 = {
    "messages": [{"role": "user", "content": "ฉันชอบกินอะไร"}],
    "current_step": "start",
    "user_request": "ถามความชอบ"
}
result2 = agent.invoke(state2, config=config)
Agent จะตอบว่า "มิน" ตามที่บอกไว้ในครั้งก่อน

สร้าง Multi-Agent System

เมื่อโปรเจกต์ใหญ่ขึ้น เราอาจแบ่ง Agent ออกเป็นหลายตัว ทำงานร่วมกัน:

# Agent สำหรับงานเฉพาะทาง
research_agent = create_smart_agent()  # ค้นหาข้อมูล
writer_agent = create_smart_agent()    # เขียนบทความ
editor_agent = create_smart_agent()    # แก้ไข

def research_node(state):
    """รับหัวข้อมา ค้นหาข้อมูล"""
    topic = state.get("user_request", "")
    result = research_agent.invoke({
        "messages": [{"role": "user", "content": f"ค้นหาข้อมูลเกี่ยวกับ: {topic}"}],
        "current_step": "start",
        "user_request": topic
    })
    return {"context": {"research": result}}

def write_node(state):
    """เขียนบทความจากข้อมูลที่ได้"""
    research = state.get("context", {}).get("research", "")
    result = writer_agent.invoke({
        "messages": [{"role": "user", "content": f"เขียนบทความจากข้อมูลนี้:\n{research}"}],
        "current_step": "start",
        "user_request": "เขียนบทความ"
    })
    return {"context": {"draft": result}}

def edit_node(state):
    """แก้ไขบทความ"""
    draft = state.get("context", {}).get("draft", "")
    result = editor_agent.invoke({
        "messages": [{"role": "user", "content": f"แก้ไขบทความนี้ให้ดีขึ้น:\n{draft}"}],
        "current_step": "start",
        "user_request": "แก้ไขบทความ"
    })
    return {"messages": [{"role": "assistant", "content": result}]}

def create_content_pipeline():
    """สร้าง pipeline สำหรับสร้างเนื้อหา"""
    graph = StateGraph(AgentState)
    
    graph.add_node("research", research_node)
    graph.add_node("write", write_node)
    graph.add_node("edit", edit_node)
    
    graph.add_edge(START, "research")
    graph.add_edge("research", "write")
    graph.add_edge("write", "edit")
    graph.add_edge("edit", END)
    
    return graph.compile()

ทดสอบ Pipeline
pipeline = create_content_pipeline()
content_result = pipeline.invoke({
    "messages": [],
    "current_step": "start",
    "user_request": "AI คืออะไร"
})

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

1. ไม่สามารถเชื่อมต่อ API ได้

อาการ: เกิดข้อผิดพลาด ConnectionError หรือ AuthenticationError

สาเหตุ: base_url ไม่ถูกต้อง หรือ API Key ไม่ถูกต้อง

# ❌ ผิด - อย่าใช้ URL เหล่านี้
client = openai.OpenAI(
    api_key="sk-xxx",
    base_url="https://api.openai.com/v1"  # ห้ามใช้
)

✅ ถูก - ใช้ HolySheep URL เท่านั้น
client = openai.OpenAI(
    api_key=os.getenv("HOLYSHEEP_API_KEY"),
    base_url="https://api.holysheep.ai/v1"
)

ตรวจสอบว่า API Key ถูกโหลดหรือไม่
print(f"API Key loaded: {bool(os.getenv('HOLYSHEEP_API_KEY'))}")

2. State ไม่ถูกอัพเดต

อาการ: ค่าใน state ไม่เปลี่ยนแปลงหลังจากผ่าน node

สาเหตุ: ลืม return state ใหม่กลับมา

# ❌ ผิด - ไม่ return state
def process_node(state):
    state["current_step"] = "processed"
    # ลืม return!

✅ ถูก - return state ใหม่
def process_node(state):
    new_state = state.copy()
    new_state["current_step"] = "processed"
    return new_state

หรือใช้วิธีง่ายๆ
def process_node(state):
    return {"current_step": "processed"}

3. Memory ไม่จดจำสนทนา

อาการ: Agent ไม่จำสิ่งที่คุยก่อนหน้า

สาเหตุ: ลืมส่ง config ทุกครั้งที่ invoke

# ❌ ผิด - ไม่ใช้ config
result = agent.invoke(state)

✅ ถูก - ใช้ config ทุกครั้ง
config = {"configurable": {"thread_id": "unique-user-id"}}
result = agent.invoke(state, config=config)

ตรวจสอบประวัติ
print(agent.get_state(config))

4. Conditional Edge ไม่ทำงาน

อาการ: Agent ไป node ที่ไม่ควรไป

สาเหตุ: function route ไม่ได้ return ค่าตรงตามที่กำหนด

# ❌ ผิด - return string โดยตรง
def route(state):
    if condition:
        return "node_a"  # ผิด!
    return "node_b"

✅ ถูก - ใช้ Literal type
from typing import Literal

def route(state) -> Literal["node_a", "node_b"]:
    if condition:
        return "node_a"
    return "node_b"

graph.add_conditional_edges(
    "decision_node",
    route,
    {
        "node_a": "node_a",
        "node_b": "node_b"
    }
)

สรุป

ในบทความนี้ คุณได้เรียนรู้พื้นฐาน LangGraph State Machine:

State คือสถานะปัจจุบันของ Agent
Node คือการทำงานในแต่ละขั้นตอน
Edge คือเส้นทางเชื่อมระหว่าง Node
Conditional Edge ช่วยให้ Agent ตัดสินใจได้
Memory ช่วยให้ Agent จดจำสนทนา

การใช้ HolySheep AI ทำให้ค่าใช้จ่ายลดลงมากกว่า 85% พร้อมความเร็วต่ำกว่า 50 มิลลิวินาที ทำให้การพัฒนา AI Agent คุ้มค่าและรวดเร็ว

เริ่มต้นสร้าง State Machine วันนี้ แล้วคุณจะเห็นว่า AI Agent ไม่ใช่เรื่องยากอีกต่อไป!

👉 สมัคร HolySheep AI — รับเครดิตฟรีเมื่อลงทะเบียน

State Machine คืออะไร — อธิบายแบบเข้าใจง่าย

ทำไมต้องใช้ LangGraph?

เริ่มต้นติดตั้งและตั้งค่า

1. ติดตั้งโปรแกรมที่จำเป็น

2. สร้างไฟล์ .env เก็บ API Key

3. ตั้งค่า HolySheep API

สร้าง State Machine แรกของคุณ

โครงสร้างพื้นฐาน

โหลด API Key

กำหนดประเภทข้อมูล State (สถานะ)

ทดสอบ

เวอร์ชันเต็มที่เชื่อมต่อ HolySheep API

โหลด API Key

ตั้งค่า HolySheep เป็น OpenAI Compatible

กำหนด State

ทดสอบ Agent

เพิ่มความฉลาดด้วย Conditional Branching

ทดสอบ

เพิ่ม Memory ให้ Agent จดจำสนทนา

สร้าง Agent

กำหนด thread_id สำหรับจดจำสนทนา

ครั้งที่ 1

ครั้งที่ 2 - Agent จะจำได้ว่าชื่อมิน

Agent จะตอบว่า "มิน" ตามที่บอกไว้ในครั้งก่อน

สร้าง Multi-Agent System

ทดสอบ Pipeline

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

1. ไม่สามารถเชื่อมต่อ API ได้

✅ ถูก - ใช้ HolySheep URL เท่านั้น

ตรวจสอบว่า API Key ถูกโหลดหรือไม่

2. State ไม่ถูกอัพเดต

✅ ถูก - return state ใหม่

หรือใช้วิธีง่ายๆ

3. Memory ไม่จดจำสนทนา

✅ ถูก - ใช้ config ทุกครั้ง

ตรวจสอบประวัติ

4. Conditional Edge ไม่ทำงาน

✅ ถูก - ใช้ Literal type

สรุป

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

บทความที่เกี่ยวข้อง

🔥 ลอง HolySheep AI

`Agent จะตอบว่า "มิน" ตามที่บอกไว้ในครั้งก่อน`