การสร้างระบบ Multi-Agent ที่ซับซ้อนนั้น การกำหนดเส้นทาง (Routing) ตามเงื่อนไขเป็นหัวใจสำคัญ เพราะช่วยให้ Agent ตัดสินใจได้ว่าจะส่งต่อข้อมูลไปที่ใดต่อ หรือจบกระบวนการ ในบทความนี้เราจะสอนการใช้งาน LangGraph Conditional Edges อย่างละเอียด พร้อมตัวอย่างโค้ดที่ใช้งานได้จริงกับ HolySheep AI ซึ่งมีความหน่วงต่ำกว่า 50ms และราคาประหยัดกว่า 85% เมื่อเทียบกับ API ทางการ

TL;DR — สรุปเนื้อหาหลัก

Conditional Edges คืออะไร

ใน LangGraph เมื่อ Node ทำงานเสร็จแล้ว ปกติเราจะกำหนดเส้นทางตายตัว (Static Edges) แต่ในงานจริง เราต้องการให้เส้นทางเปลี่ยนไปตามผลลัพธ์ เช่น:

นี่คือหลักการของ Conditional Edges ที่ทำให้ระบบของเรามีความยืดหยุ่นและฉลาดขึ้น

ตารางเปรียบเทียบ API Provider สำหรับ LangGraph Multi-Agent

เกณฑ์ HolySheep AI OpenAI API Anthropic API Google Gemini
ราคา GPT-4.1 $8/MTok $15/MTok - -
ราคา Claude Sonnet 4.5 $15/MTok - $18/MTok -
ราคา Gemini 2.5 Flash $2.50/MTok - - $3.50/MTok
ราคา DeepSeek V3.2 $0.42/MTok - - -
ความหน่วง (Latency) <50ms 100-300ms 150-400ms 80-250ms
วิธีชำระเงิน WeChat/Alipay (¥) บัตรเครดิต บัตรเครดิต บัตรเครดิต
อัตราแลกเปลี่ยน ¥1=$1 (ประหยัด 85%+) USD ตรง USD ตรง USD ตรง
เครดิตฟรีเมื่อสมัคร ✓ มี $5 ฟรี $5 ฟรี $300 ฟรี
ทีมที่เหมาะสม Startup, นักพัฒนาไทย/จีน ทีมใหญ่, Enterprise ทีม Enterprise ทีม Google Ecosystem

พื้นฐาน: โครงสร้าง LangGraph Conditional Edges

ก่อนจะเข้าสู่ตัวอย่างจริง มาดูโครงสร้างพื้นฐานของ Conditional Edges กัน


from langgraph.graph import StateGraph, END
from typing import TypedDict, Literal


กำหนด State ของระบบ

class AgentState(TypedDict):
    query: str
    intent: str
    response: str
    confidence: float


สร้าง Graph

graph = StateGraph(AgentState)


เพิ่ม Node ต่างๆ

graph.add_node("classify", classify_node)
graph.add_node("answer", answer_node)
graph.add_node("escalate", escalate_node)


=== Static Edge: เส้นทางตายตัว ===

graph.add_edge("__start__", "classify")


=== Conditional Edge: เส้นทางตามเงื่อนไข ===


ฟังก์ชัน route จะตัดสินใจว่าจะไป edge ไหน

graph.add_conditional_edges(
    source_node="classify",
    path=route_after_classify,  # ฟังก์ชันที่คืนค่าชื่อ edge
    path_mapping={
        "answer": "answer",      # ถ้าคืน "answer" → ไป node answer
        "escalate": "escalate",  # ถ้าคืน "escalate" → ไป node escalate
        "end": END               # ถ้าคืน "end" → จบกระบวนการ
    }
)


คอมไพล์และรัน

app = graph.compile()
result = app.invoke({"query": "ราคา Enterprise plan รายเดือนเท่าไหร่?"})

ตัวอย่างที่ 1: Intent Classification Routing

ตัวอย่างนี้เป็นระบบ Chatbot ที่จำแนกความตั้งใจของผู้ใช้แล้วส่งต่อไปยัง Agent ที่เหมาะสม ใช้ HolySheep AI เป็น LLM backend ซึ่งมีความหน่วงต่ำกว่า 50ms ทำให้การ classify เร็วและแม่นยำ


import os
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator


============================================


ตั้งค่า HolySheep AI API


============================================

os.environ["HOLYSHEEP_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"  # ห้ามใช้ api.openai.com

from openai import OpenAI
client = OpenAI(
    api_key=os.environ["HOLYSHEEP_API_KEY"],
    base_url=HOLYSHEEP_BASE_URL
)


============================================


กำหนด State Schema


============================================

class ChatState(TypedDict):
    user_message: str
    intent: str
    agent_response: str
    needs_escalation: bool
    conversation_history: list


============================================


Node: จำแนกความตั้งใจ (Intent Classification)


============================================

def classify_intent(state: ChatState) -> ChatState:
    """จำแนกว่าผู้ใช้ต้องการอะไร"""
    
    classification_prompt = f"""จำแนกความตั้งใจของข้อความต่อไปนี้:

ข้อความ: {state['user_message']}

ตอบกลับเฉพาะหมวดหมู่เดียว:
- pricing: ถามเกี่ยวกับราคา, แพ็กเกจ, ค่าบริการ
- technical: ถามเกี่ยวกับวิธีใช้งาน, ปัญหาทางเทคนิค
- billing: ถามเกี่ยวกับการเรียกเก็บเงิน, วิธีชำระเงิน
- complaint: บ่น, แจ้งปัญหา, ต้องการความช่วยเหลือเร่งด่วน
- general: คำถามทั่วไป, สนทนาธรรมดา

ตอบเฉพาะชื่อหมวดหมู่เท่านั้น:"""

    response = client.chat.completions.create(
        model="gpt-4.1",  # หรือเลือก claude-sonnet-4.5, gemini-2.5-flash, deepseek-v3.2
        messages=[{"role": "user", "content": classification_prompt}],
        temperature=0.1,
        max_tokens=20
    )
    
    intent = response.choices[0].message.content.strip().lower()
    
    # ตรวจสอบว่าต้องส่งต่อผู้เชี่ยวชาญหรือไม่
    needs_escalation = intent in ["complaint", "billing"] and len(state.get('conversation_history', [])) > 2
    
    return {
        **state,
        "intent": intent,
        "needs_escalation": needs_escalation
    }


============================================


Node: ตอบคำถามราคา


============================================

def pricing_agent(state: ChatState) -> ChatState:
    """ตัวแทนตอบคำถามเกี่ยวกับราคา"""
    
    system_prompt = """คุณคือที่ปรึกษาด้านราคาของ HolySheep AI 
ราคาโมเดล 2026:
- GPT-4.1: $8/MTok
- Claude Sonnet 4.5: $15/MTok  
- Gemini 2.5 Flash: $2.50/MTok
- DeepSeek V3.2: $0.42/MTok (ประหยัดที่สุด)
- อัตราแลกเปลี่ยน: ¥1=$1 (ประหยัด 85%+)
- วิธีชำระเงิน: WeChat, Alipay"""

    response = client.chat.completions.create(
        model="deepseek-v3.2",  # ใช้โมเดลราคาถูกสำหรับงานง่าย
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": state['user_message']}
        ],
        temperature=0.3,
        max_tokens=500
    )
    
    return {
        **state,
        "agent_response": response.choices[0].message.content
    }


============================================


Node: ตอบคำถามเทคนิค


============================================

def technical_agent(state: ChatState) -> ChatState:
    """ตัวแทนตอบคำถามทางเทคนิค"""
    
    system_prompt = """คุณคือวิศวกรฝ่ายเทคนิคของ HolySheep AI
ช่วยตอบคำถามเกี่ยวกับ:
- การใช้งาน API
- การตั้งค่า base_url (https://api.holysheep.ai/v1)
- การแก้ปัญหา error
- ความสามารถของแต่ละโมเดล"""

    response = client.chat.completions.create(
        model="gpt-4.1",  # ใช้โมเดลดีสำหรับงานเทคนิค
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": state['user_message']}
        ],
        temperature=0.2,
        max_tokens=800
    )
    
    return {
        **state,
        "agent_response": response.choices[0].message.content
    }


============================================


Node: Escalation - ส่งต่อผู้เชี่ยวชาญ


============================================

def escalate_to_human(state: ChatState) -> ChatState:
    """ส่งต่อให้ผู้เชี่ยวชาญดูแล"""
    
    escalation_message = f"""📞 ระบบได้ส่งต่อข้อความของคุณไปยังฝ่ายบริการลูกค้าแล้ว

ประวัติการสนทนา:
{chr(10).join([f"- {msg}" for msg in state.get('conversation_history', [])])}

คำถามล่าสุด: {state['user_message']}
หมวดหมู่: {state['intent']}

ทีมงานจะติดต่อกลับภายใน 24 ชั่วโมง"""
    
    return {
        **state,
        "agent_response": escalation_message
    }


============================================


CONDITIONAL ROUTING FUNCTION


============================================

def route_based_on_intent(state: ChatState) -> str:
    """
    ฟังก์ชันนี้เป็นหัวใจของ Conditional Edges
    คืนค่าเป็นชื่อ edge ที่ต้องการไป
    
    Returns:
        - "to_pricing" → ไป pricing_agent
        - "to_technical" → ไป technical_agent
        - "to_general" → ไป general_agent
        - "escalate" → ส่งต่อคน
    """
    intent = state.get("intent", "")
    needs_escalation = state.get("needs_escalation", False)
    
    # ถ้าต้องการ escalation ให้ส่งต่อคนเสมอ
    if needs_escalation:
        return "escalate"
    
    # แมป intent กับ edge
    intent_mapping = {
        "pricing": "to_pricing",
        "technical": "to_technical",
        "billing": "escalate",
        "complaint": "escalate",
        "general": "to_general"
    }
    
    return intent_mapping.get(intent, "to_general")


============================================


สร้าง LangGraph


============================================

def create_chat_graph():
    graph = StateGraph(ChatState)
    
    # เพิ่ม Nodes
    graph.add_node("classify", classify_intent)
    graph.add_node("pricing", pricing_agent)
    graph.add_node("technical", technical_agent)
    graph.add_node("escalate", escalate_to_human)
    
    # Entry point
    graph.set_entry_point("classify")
    
    # Conditional Edges - นี่คือหัวใจของระบบ!
    graph.add_conditional_edges(
        source_node="classify",
        path=route_based_on_intent,
        path_mapping={
            "to_pricing": "pricing",
            "to_technical": "technical",
            "escalate": "escalate",
            "to_general": "pricing"  # fallback to pricing
        }
    )
    
    # ทุก node จบที่ END
    graph.add_edge("pricing", END)
    graph.add_edge("technical", END)
    graph.add_edge("escalate", END)
    
    return graph.compile()


============================================


ทดสอบระบบ


============================================

if __name__ == "__main__":
    app = create_chat_graph()
    
    test_queries = [
        "ราคา Enterprise plan เท่าไหร่?",
        "API timeout แก้ยังไง?",
        "ผมโอนเงินไปแล้วแต่เครดิตยังไม่เข้า ช่วยดูให้หน่อย",
        "ทดสอบระบบ"
    ]
    
    for query in test_queries:
        print(f"\n{'='*50}")
        print(f"คำถาม: {query}")
        print(f"{'='*50}")
        
        result = app.invoke({
            "user_message": query,
            "intent": "",
            "agent_response": "",
            "needs_escalation": False,
            "conversation_history": []
        })
        
        print(f"Intent: {result['intent']}")
        print(f"Escalation: {result['needs_escalation']}")
        print(f"คำตอบ: {result['agent_response'][:200]}...")

ตัวอย่างที่ 2: Multi-Turn Conversation with Loop

ตัวอย่างนี้สาธิตการใช้ Conditional Edges ร่วมกับ Loop เพื่อให้ระบบถามตอบซ้ำจนกว่าจะได้คำตอบที่ถูกต้อง ใช้ได้กับงานเช่น การเก็บข้อมูลลูกค้า, การยืนยันตัวตน, หรือการค้นหาข้อมูลแบบ Interactive


from langgraph.graph import StateGraph, END
from typing import TypedDict, Literal

class LeadGenState(TypedDict):
    customer_name: str | None
    customer_email: str | None
    interest_topic: str | None
    budget_range: str | None
    collected_info: list[str]
    current_question: str
    attempts: int
    final_response: str

def greet_node(state: LeadGenState) -> LeadGenState:
    """Node เริ่มต้น - ทักทายลูกค้า"""
    return {
        **state,
        "current_question": "สวัสดีครับ! ขอทราบชื่อของท่านได้ไหมครับ?",
        "collected_info": []
    }

def collect_name(state: LeadGenState) -> LeadGenState:
    """เก็บข้อมูลชื่อ"""
    name = state.get("customer_name", "ลูกค้า")
    return {
        **state,
        "current_question": f"สวัสดีครับ {name}! มีอีเมลที่สามารถติดต่อได้ไหมครับ?",
        "collected_info": state.get("collected_info", []) + ["name"]
    }

def collect_email(state: LeadGenState) -> LeadGenState:
    """เก็บข้อมูลอีเมล"""
    email = state.get("customer_email", "")
    return {
        **state,
        "current_question": f"ขอบคุณครับ! สนใจบริการเกี่ยวกับเรื่องอะไรเป็นพิเศษครับ (AI, Data, Cloud)?",
        "collected_info": state.get("collected_info", []) + ["email"]
    }

def collect_interest(state: LeadGenState) -> LeadGenState:
    """เก็บข้อมูลความสนใจ"""
    topic = state.get("interest_topic", "")
    return {
        **state,
        "current_question": "มีงบประมาณในการลงทุนเท่าไหร่ครับ?",
        "collected_info": state.get("collected_info", []) + ["interest"]
    }

def collect_budget(state: LeadGenState) -> LeadGenState:
    """เก็บข้อมูลงบประมาณ"""
    return {
        **state,
        "current_question": "ขอบคุณครับ! รอสักครู่นะครับ...",
        "collected_info": state.get("collected_info", []) + ["budget"]
    }

def generate_response(state: LeadGenState) -> LeadGenState:
    """สร้างคำตอบสรุปและเสนอบริการ"""
    info = state.get("collected_info", [])
    
    # เรียกใช้ HolySheep AI เพื่อสร้างคำตอบ
    os.environ["HOLYSHEEP_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
    client = OpenAI(
        api_key=os.environ["HOLYSHEEP_API_KEY"],
        base_url="https://api.holysheep.ai/v1"
    )
    
    summary_prompt = f"""สรุปข้อมูลลูกค้า:
- ชื่อ: {state.get('customer_name')}
- อีเมล: {state.get('customer_email')}
- ความสนใจ: {state.get('interest_topic')}
- งบประมาณ: {state.get('budget_range')}

แนะนำแพ็กเกจที่เหมาะสมจาก HolySheep AI และสร้างข้อความตอบกลับที่เป็นมิตร"""

    response = client.chat.completions.create(
        model="gemini-2.5-flash",  # โมเดลเร็วและถูก
        messages=[{"role": "user", "content": summary_prompt}],
        temperature=0.7
    )
    
    return {
        **state,
        "final_response": response.choices[0].message.content,
        "current_question": ""
    }


============================================


CONDITIONAL ROUTING: ควบคุมการไหลของข้อมูล


============================================

def route_collection_flow(state: LeadGenState) -> Literal["collect_name", "collect_email", "collect_interest", "collect_budget", "generate_response"]:
    """
    กำหนดเส้นทางตามข้อมูลที่เก็บได้
    
    Logic:
    - ยังไม่มีชื่อ → ไปเก็บชื่อ
    - มีชื่อแต่ไม่มีอีเมล → ไปเก็บอีเมล
    - มีอีเมลแต่ไม่มีความสนใจ → ไปเก็บความสนใจ
    - มีความสนใจแต่ไม่มีงบ → ไปเก็บงบ
    - เก็บครบแล้ว → ไปสร้าง response
    """
    collected = state.get("collected_info", [])
    attempts = state.get("attempts", 0)
    
    # ป้องกัน infinite loop
    if attempts >= 10:
        return "generate_response"
    
    if "name" not in collected:
        return "collect_name"
    elif "email" not in collected:
        return "collect_email"
    elif "interest" not in collected:
        return "collect_interest"
    elif "budget" not in collected:
        return "collect_budget"
    else:
        return "generate_response"

def create_lead_generation_graph():
    graph = StateGraph(LeadGenState)
    
    # เพิ่ม Nodes
    graph.add_node("greet", greet_node)
    graph.add_node("collect_name", collect_name)
    graph.add_node("collect_email", collect_email)
    graph.add_node("collect_interest", collect_interest)
    graph.add_node("collect_budget", collect_budget)
    graph.add_node("generate_response", generate_response)
    
    # Entry
    graph.set_entry_point("greet")
    
    # Conditional Edges: greet → ตรวจสอบว่าต้องเก็บอะไรต่อ
    graph.add_conditional_edges(
        source_node="greet",
        path=route_collection_flow,
        path_mapping={
            "collect_name": "collect_name",
            "collect_email": "collect_email",
            "collect_interest": "collect_interest",
            "collect_budget": "collect_budget",
            "generate_response": "generate_response"
        }
    )
    
    # Collection nodes → วนกลับไปตรวจสอบว่าเก็บครบหรือยัง
    for node in ["collect_name", "collect_email", "collect_interest", "collect_budget"]:
        graph.add_conditional_edges(
            source_node=node,
            path=route_collection_flow,
            path_mapping={
                "collect_name": "collect_name",
                "collect_email": "collect_email",
                "collect_interest": "collect_interest",
                "collect_budget": "collect_budget",
                "generate_response": "generate_response"
            }
        )
    
    # จบกระบวนการ
    graph.add_edge("generate_response", END)
    
    return graph.compile()


============================================


ทดสอบ


============================================

if __name__ == "__main__":
    app = create_lead_generation_graph()
    
    # Simulate user inputs
    initial_state = {
        "customer_name": "สมชาย มั่นคง",
        "customer_email": "[email protected]",
        "interest_topic": "AI Automation",
        "budget

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