รูปแบบการออกแบบลูปและการแตกแขนงแบบมีเงื่อนไขใน LangGraph

การสร้าง Agent ที่ซับซ้อนด้วย LangGraph นั้น การจัดการลูป (Loop) และเงื่อนไข (Conditional Branching) เป็นหัวใจหลักที่ต้องเข้าใจอย่างลึกซึ้ง ในบทความนี้ผมจะพาทุกท่านไปสำรวจรูปแบบการออกแบบที่ใช้งานจริงใน Production พร้อมโค้ดตัวอย่างที่รันได้ทันที โดยใช้ HolySheep AI เป็น API Provider หลัก

ตารางเปรียบเทียบบริการ AI Relay

เกณฑ์	HolySheep AI	Official OpenAI	Official Anthropic	บริการ Relay อื่นๆ
อัตราแลกเปลี่ยน	¥1 = $1 (ประหยัด 85%+)	ราคาปกติ USD	ราคาปกติ USD	มีค่าธรรมเนียมต่างๆ
วิธีชำระเงิน	WeChat / Alipay	บัตรเครดิตระหว่างประเทศ	บัตรเครดิตระหว่างประเทศ	หลากหลาย
ความหน่วง (Latency)	< 50ms	100-300ms	150-400ms	80-200ms
เครดิตฟรี	✅ มีเมื่อลงทะเบียน	❌ ไม่มี	❌ ไม่มี	มีบ้างไม่มีบ้าง
GPT-4.1	$8/MTok	$8/MTok	-	$8.5-10/MTok
Claude Sonnet 4.5	$15/MTok	-	$15/MTok	$16-18/MTok
Gemini 2.5 Flash	$2.50/MTok	-	-	$3-4/MTok
DeepSeek V3.2	$0.42/MTok	-	-	$0.50-0.80/MTok

พื้นฐาน LangGraph State Graph

ก่อนจะเข้าสู่เรื่องลูปและเงื่อนไข มาทำความเข้าใจโครงสร้างพื้นฐานของ LangGraph กันก่อน LangGraph ใช้ StateGraph เป็นหัวใจหลัก โดยแต่ละ Node จะรับ State ปัจจุบันและคืนค่า State ใหม่กลับมา

การตั้งค่า Environment และการเชื่อมต่อ

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

ตั้งค่า API Key และ Base URL สำหรับ HolySheep AI
os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"

สร้าง LLM instance โดยใช้ HolySheep AI
llm = ChatOpenAI(
    model="gpt-4.1",
    temperature=0.7,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_API_BASE"]
)

กำหนด State Schema สำหรับ Graph
class AgentState(TypedDict):
    messages: list
    step: int
    result: str
    should_continue: bool

รูปแบบลูป (Loop Patterns) ใน LangGraph

1. ลูปแบบมีจำนวนรอบจำกัด (Fixed Iteration Loop)

รูปแบบแรกที่พบบ่อยคือการทำงานซ้ำๆ เป็นจำนวนรอบที่กำหนดไว้ เหมาะสำหรับงานที่ต้องการปรับปรุงผลลัพธ์ทีละขั้นตอน เช่น การ Refine ข้อความ หรือการวิเคราะห์ข้อมูลหลายมุมมอง

from langgraph.graph import StateGraph, END, START

def create_refinement_graph(max_iterations: int = 3):
    """สร้าง Graph สำหรับการปรับปรุงผลลัพธ์ซ้ำๆ"""
    
    # กำหนด State Schema
    class RefinementState(TypedDict):
        original_text: str
        refined_text: str
        iteration: int
        quality_score: float
    
    # Node สำหรับปรับปรุงข้อความ
    def refine_node(state: RefinementState) -> RefinementState:
        iteration = state["iteration"]
        
        prompt = f"""ปรับปรุงข้อความต่อไปนี้ให้ดีขึ้น (รอบที่ {iteration}/{max_iterations}):

ข้อความเดิม: {state.get('refined_text', state['original_text'])}

โปรดปรับปรุง:
1. ความชัดเจนของเนื้อหา
2. ความกระชับของประโยค
3. ความถูกต้องทางไวยากรณ์"""
        
        response = llm.invoke(prompt)
        
        return {
            "refined_text": response.content,
            "iteration": iteration + 1
        }
    
    # Node สำหรับประเมินคุณภาพ
    def evaluate_node(state: RefinementState) -> RefinementState:
        prompt = f"""ให้คะแนนคุณภาพข้อความนี้ 0-10:
        
{state['refined_text']}

ตอบกลับเฉพาะตัวเลขคะแนนเท่านั้น"""
        
        response = llm.invoke(prompt)
        try:
            score = float(response.content.strip()[0])
        except:
            score = 5.0
        
        return {"quality_score": score}
    
    # ฟังก์ชันตัดสินใจว่าจะทำต่อหรือหยุด
    def should_continue(state: RefinementState) -> str:
        if state["iteration"] >= max_iterations:
            return "end"
        return "refine"
    
    # สร้าง Graph
    graph = StateGraph(RefinementState)
    graph.add_node("refine", refine_node)
    graph.add_node("evaluate", evaluate_node)
    
    graph.add_edge(START, "refine")
    graph.add_edge("refine", "evaluate")
    graph.add_conditional_edges(
        "evaluate",
        should_continue,
        {
            "refine": "refine",
            "end": END
        }
    )
    
    return graph.compile()

ใช้งาน
graph = create_refinement_graph(max_iterations=3)
result = graph.invoke({
    "original_text": "วันนี้อากาศดีมากเลยนะ",
    "refined_text": "",
    "iteration": 0,
    "quality_score": 0.0
})
print(f"ผลลัพธ์สุดท้าย: {result['refined_text']}")
print(f"จำนวนรอบ: {result['iteration']}")

2. ลูปแบบมีเงื่อนไขการหยุด (Conditional Stop Loop)

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

from enum import Enum

class LoopMode(Enum):
    CONTINUE = "continue"
    END = "end"

def create_conditional_search_graph():
    """สร้าง Graph สำหรับการค้นหาที่หยุดเมื่อเจอคำตอบที่ดีพอ"""
    
    class SearchState(TypedDict):
        query: str
        search_results: list
        current_result: str
        attempts: int
        confidence: float
        found_answer: bool
    
    # Node สำหรับค้นหาข้อมูล
    def search_node(state: SearchState) -> SearchState:
        attempts = state["attempts"] + 1
        
        prompt = f"""ค้นหาข้อมูลเกี่ยวกับ: {state['query']}
        
รอบที่: {attempts}
ผลลัพธ์ก่อนหน้า: {state.get('current_result', 'ยังไม่มี')}

หากพบคำตอบที่น่าเชื่อถือ ให้ตอบกลับด้วยคำตอบนั้น
หากยังไม่แน่ใจ ให้บอกว่าต้องการข้อมูลเพิ่มเติม"""
        
        response = llm.invoke(prompt)
        
        return {
            "current_result": response.content,
            "attempts": attempts
        }
    
    # Node สำหรับประเมินความมั่นใจ
    def evaluate_confidence(state: SearchState) -> SearchState:
        prompt = f"""ประเมินความมั่นใจของคำตอบนี้ 0.0 - 1.0:

คำถาม: {state['query']}
คำตอบ: {state['current_result']}

ตอบกลับเฉพาะตัวเลขทศนิยม 1 ตำแหน่ง"""
        
        response = llm.invoke(prompt)
        try:
            confidence = float(response.content.strip()[:4])
        except:
            confidence = 0.5
        
        found = confidence >= 0.85 or state["attempts"] >= 5
        
        return {
            "confidence": confidence,
            "found_answer": found
        }
    
    # ฟังก์ชันตัดสินใจ - รองรับทั้ง str และ LoopMode enum
    def should_loop(state: SearchState) -> str:
        if state["found_answer"]:
            return "end"
        return "continue"
    
    # สร้าง Graph
    graph = StateGraph(SearchState)
    graph.add_node("search", search_node)
    graph.add_node("evaluate", evaluate_confidence)
    
    graph.add_edge(START, "search")
    graph.add_edge("search", "evaluate")
    graph.add_conditional_edges(
        "evaluate",
        should_loop,
        {
            "continue": "search",
            "end": END
        }
    )
    
    return graph.compile()

ใช้งาน
graph = create_conditional_search_graph()
result = graph.invoke({
    "query": "วิธีทำกาแฟในบ้านแบบง่ายๆ",
    "search_results": [],
    "current_result": "",
    "attempts": 0,
    "confidence": 0.0,
    "found_answer": False
})
print(f"ความมั่นใจ: {result['confidence']}")
print(f"จำนวนครั้งที่ค้นหา: {result['attempts']}")
print(f"คำตอบ: {result['current_result']}")

รูปแบบเงื่อนไขแตกแขนง (Conditional Branching)

3. เงื่อนไขแตกแขนงหลายทาง (Multi-branch Conditional)

ในบางกรณี เราต้องการแตกแขนงออกเป็นหลายเส้นทางตามเงื่อนไขที่ซับซ้อน เช่น การจัดการประเภทต่างๆ ของผู้ใช้หรือประเภทของงานที่แตกต่างกัน

from typing import Literal

def create_multi_branch_support_graph():
    """สร้าง Graph สำหรับระบบ Support ที่แตกแขนงตามประเภทปัญหา"""
    
    class SupportState(TypedDict):
        user_message: str
        issue_type: str
        priority: str
        response: str
        escalation_needed: bool
    
    # Node จำแนกประเภทปัญหา
    def classify_issue(state: SupportState) -> SupportState:
        prompt = f"""จำแนกปัญหาต่อไปนี้เป็นหมวดหมู่:
        
ข้อความ: {state['user_message']}

หมวดหมู่ที่เป็นไปได้:
- billing: ปัญหาเกี่ยวกับการเงิน/การชำระเงิน
- technical: ปัญหาทางเทคนิค
- general: คำถามทั่วไป
- complaint: ข้อร้องเรียน

ตอบกลับเฉพาะหมวดหมู่เท่านั้น"""
        
        response = llm.invoke(prompt)
        issue_type = response.content.strip().lower()
        
        # กำหนด Priority ตามเนื้อหา
        priority = "medium"
        if "ด่วน" in state["user_message"] or "เร่งด่วน" in state["user_message"]:
            priority = "high"
        elif "ฉุกเฉิน" in state["user_message"]:
            priority = "critical"
        
        return {
            "issue_type": issue_type,
            "priority": priority
        }
    
    # Node สำหรับจัดการปัญหาการเงิน
    def handle_billing(state: SupportState) -> SupportState:
        prompt = f"""คุณคือฝ่ายบริการลูกค้าเกี่ยวกับการเงิน
        
ข้อความลูกค้า: {state['user_message']}
Priority: {state['priority']}

ตอบกลับอย่างเป็นมิตรและให้ข้อมูลที่เป็นประโยชน์"""
        
        response = llm.invoke(prompt)
        
        return {
            "response": response.content,
            "escalation_needed": state["priority"] in ["high", "critical"]
        }
    
    # Node สำหรับจัดการปัญหาทางเทคนิค
    def handle_technical(state: SupportState) -> SupportState:
        prompt = f"""คุณคือฝ่ายสนับสนุนทางเทคนิค
        
ข้อความลูกค้า: {state['user_message']}
Priority: {state['priority']}

ให้คำตอบทางเทคนิคที่ละเอียดและเข้าใจง่าย"""
        
        response = llm.invoke(prompt)
        
        return {
            "response": response.content,
            "escalation_needed": state["priority"] == "critical"
        }
    
    # Node สำหรับจัดการคำถามทั่วไป
    def handle_general(state: SupportState) -> SupportState:
        prompt = f"""ตอบคำถามต่อไปนี้อย่างเป็นมิตร:
        
{state['user_message']}"""
        
        response = llm.invoke(prompt)
        
        return {
            "response": response.content,
            "escalation_needed": False
        }
    
    # Node สำหรับจัดการข้อร้องเรียน
    def handle_complaint(state: SupportState) -> SupportState:
        prompt = f"""คุณคือหัวหน้าฝ่ายบริการลูกค้า
        
ลูกค้ามีข้อร้องเรียน: {state['user_message']}
Priority: {state['priority']}

ตอบกลับด้วยความเข้าใจ แสดงความเสียใจ และเสนอแนวทางแก้ไข"""
        
        response = llm.invoke(prompt)
        
        return {
            "response": response.content,
            "escalation_needed": True  # ข้อร้องเรียนต้อง Escalate เสมอ
        }
    
    # Node สำหรับ Escalation
    def escalate_node(state: SupportState) -> SupportState:
        escalation_msg = f"\n\n[ESCALATED to Human Agent - Priority: {state['priority']}]"
        return {"response": state["response"] + escalation_msg}
    
    # ฟังก์ชันตัดสินใจเส้นทาง
    def route_issue(state: SupportState) -> Literal["billing", "technical", "general", "complaint"]:
        return state["issue_type"]
    
    # ฟังก์ชันตัดสินใจการ Escalate
    def should_escalate(state: SupportState) -> str:
        return "escalate" if state["escalation_needed"] else "end"
    
    # สร้าง Graph
    graph = StateGraph(SupportState)
    graph.add_node("classify", classify_issue)
    graph.add_node("billing", handle_billing)
    graph.add_node("technical", handle_technical)
    graph.add_node("general", handle_general)
    graph.add_node("complaint", handle_complaint)
    graph.add_node("escalate", escalate_node)
    
    graph.add_edge(START, "classify")
    graph.add_conditional_edges(
        "classify",
        route_issue,
        {
            "billing": "billing",
            "technical": "technical",
            "general": "general",
            "complaint": "complaint"
        }
    )
    graph.add_conditional_edges(
        "billing",
        should_escalate,
        {"escalate": "escalate", "end": END}
    )
    graph.add_conditional_edges(
        "technical",
        should_escalate,
        {"escalate": "escalate", "end": END}
    )
    graph.add_edge("general", END)
    graph.add_edge("complaint", "escalate")
    graph.add_edge("escalate", END)
    
    return graph.compile()

ใช้งาน
graph = create_multi_branch_support_graph()
result = graph.invoke({
    "user_message": "ฉันต้องการขอคืนเงินค่าบริการเดือนที่แล้ว เพราะระบบไม่ทำงานเลย",
    "issue_type": "",
    "priority": "",
    "response": "",
    "escalation_needed": False
})
print(f"ประเภทปัญหา: {result['issue_type']}")
print(f"Priority: {result['priority']}")
print(f"ต้อง Escalate: {result['escalation_needed']}")
print(f"คำตอบ: {result['response']}")

รูปแบบผสม: ลูปซ้อนเงื่อนไข (Nested Loop with Conditionals)

4. Research Agent แบบครอบคลุม

ตัวอย่างนี้เป็นรูปแบบที่ใช้งานจริงใน Production มากที่สุด เป็นการผสมผสานระหว่างลูปและเงื่อนไขเข้าด้วยกัน สร้าง Research Agent ที่ทำงานซ้ำจนกว่าจะได้ข้อมูลครบถ้วน

from datetime import datetime

def create_research_agent():
    """สร้าง Research Agent ที่มีลูปและเงื่อนไขซับซ้อน"""
    
    class ResearchState(TypedDict):
        topic: str
        current_phase: str
        findings: list
        gaps: list
        verification_count: int
        final_report: str
        is_complete: bool
        confidence_score: float
    
    # Phase 1: รวบรวมข้อมูลเบื้องต้น
    def collect_initial_data(state: ResearchState) -> ResearchState:
        prompt = f"""วิจัยเรื่อง: {state['topic']}

รวบรวมข้อมูลเบื้องต้น:
1. คำนิยามและความเข้าใจพื้นฐาน
2. ประเด็นสำคัญ 3-5 ข้อ
3. ผู้เชี่ยวชาญหรือแหล่งข้อมูลที่น่าเชื่อถือ

ใช้ภาษาไทยในการตอบกลับ"""
        
        response = llm.invoke(prompt)
        
        return {
            "current_phase": "collection",
            "findings": [response.content],
            "gaps": []
        }
    
    # Phase 2: ค้นหาช่องว่างของข้อมูล
    def identify_gaps(state: ResearchState) -> ResearchState:
        prompt = f"""จากข้อมูลที่รวบรวมได้:
        
{state['findings'][-1]}

ระบุช่องว่างของข้อมูลที่ต้องการค้นหาเพิ่มเติม:
1. ระบุ 2-3 ประเด็นที่ยังไม่ชัดเจน
2. ระบุข้อมูลที่ขัดแย้งกัน (ถ้ามี)
3. ระบุแง่มุมที่ควรศึกษาเพิ่มเติม

ใช้ภาษาไทยในการตอบกลับ"""
        
        response = llm.invoke(prompt)
        
        return {
            "current_phase": "gap_analysis",
            "gaps": response.content.split("\n")
        }
    
    # Phase 3: เจาะลึกข้อมูลที่ยังขาด
    def deep_dive(state: ResearchState) -> ResearchState:
        prompt = f"""เจาะลึกเรื่อง: {state['topic']}

ช่องว่างที่ต้องเติม:
{state['gaps']}

ค้นหาข้อมูลเพิ่มเติมเพื่อเติมเต็มช่องว่างเหล่านี้

ใช้ภาษาไทยในการตอบกลับ"""
        
        response = llm.invoke(prompt)
        new_findings = state["findings"] + [response.content]
        
        return {
            "current_phase": "deep_dive",
            "findings": new_findings,
            "verification_count": state["verification_count"] + 1
        }
    
    # Phase 4: ตรวจสอบความน่าเชื่อถือ
    def verify_findings(state: ResearchState) -> ResearchState:
        prompt = f"""ตรวจสอบความน่าเชื่อถือของข้อมูล:
        
{state['findings'][-1]}

ให้คะแนนความมั่นใจ 0.0 - 1.0 และระบุเหตุผล:
- คะแนนสูง: ข้อมูลมาจากแหล่งที่น่าเชื่อถือ สอดคล้องกัน
- คะแนนต่ำ: ข้อมูลขัดแย้ง ไม่มีแหล่งอ้างอิง

ตอบกลับในรูปแบบ: SCORE: [คะแนน] REASON: [เหตุผล]"""
        
        response = llm.invoke(prompt)
        
        try:
            score_text = response.content.split("SCORE:")[1].split("REASON:")[0].strip()
            confidence = float(score_text[:4])
        except:
            confidence = 0.5
        
        return {
            "current_phase": "verification",
            "confidence_score": confidence
        }
    
    # Phase 5: สร้างรายงานสุดท้าย
    def generate_report(state: ResearchState) -> ResearchState:
        all_findings = "\n\n---\n\n".join(state["findings"])
        
        prompt = f"""สร้างรายงานวิจัยเรื่อง: {state['topic']}

จากข้อมูลที่รวบรวมได้:
{all_findings}

จำนวนแหล่งข้อมูล: {len(state['findings'])}
คะแนนความมั่นใจ: {state['confidence_score']}

สร้างรายงานที่:
1. มีโครงสร้างชัดเจน
2. อ้างอิงข้อมูลจากแหล่งต่างๆ
3. ระบุข้อจำกัดของการวิจัย
4. ใช้ภาษาไทยที่เป็นทางการ

รายงานควรมีความยาว 500-1000 คำ"""
        
        response = llm.invoke(prompt)
        
        return {
            "current_phase": "complete",
            "final_report": response.content,
            "is_complete": True
        }
    
    # ตัดสินใจว่าจะทำลูปต่อหรือไม่
    def decision_loop(state: ResearchState) -> str:
        # เงื่อนไขการหยุด
        max_verifications = 3
        min_confidence = 0.75
        
        if state["verification_count"] >= max_verifications:
            return "generate_report"
        if state["confidence_score"] >= min_confidence and state["verification_count"] >= 1:
            return "generate_report"
        return "deep_dive"
    
    # สร้า�
แหล่งข้อมูลที่เกี่ยวข้อง
📚 บทช่วยสอน AI API
💰 ดูราคา
📖 เอกสารสำหรับนักพัฒนา
🚀 สมัครฟรี
บทความที่เกี่ยวข้อง
การปรับปรุงประสิทธิภาพการประมวลผลข้อมูล Parquet แบบเข้ารหัสใ
DeepSeek-TUI เชื่อมต่อ中转API สู่ Production — คู่มือฉบับวิศวก
การตั้งค่า Rate Limiting สำหรับ AI API ตามระดับ Subscription