การพัฒนา Multi-Agent System ด้วย LangGraph นั้นมีความซับซ้อนอย่างยิ่ง โดยเฉพาะเมื่อต้องจัดการกับ เงื่อนไขการตัดสินใจ (Conditional Branching) และ การวนซ้ำ (Loop Control) ที่ถูกต้อง ในบทความนี้ผมจะแบ่งปันประสบการณ์ตรงจากการแก้ไขข้อผิดพลาดที่เกิดขึ้นจริงในโปรเจกต์
สถานการณ์ข้อผิดพลาดจริง: "TypeError: state is not a StateGraph"
ในการพัฒนาระบบ Customer Support Agent ผมเจอข้อผิดพลาดที่ทำให้แอปพลิเคชันล่มทั้งระบบ:
TypeError: state is not a StateGraph, cannot add conditional edges from a state graph to another state graphปัญหานี้เกิดจากการที่ผมเข้าใจผิดว่าสามารถเพิ่ม add_conditional_edges ได้โดยตรงจาก subgraph ไปยัง graph หลัก ซึ่งในความเป็นจริงต้องผ่าน END state หรือใช้ compile ก่อน
พื้นฐาน Conditional Branching ใน LangGraph
การสร้างเงื่อนไขใน LangGraph ทำได้โดยการกำหนด route function ที่คืนค่าชื่อ edge ที่ต้องการ:
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict, Literal
กำหนด State Schema
class AgentState(TypedDict):
messages: list
intent: str
confidence: float
สร้าง route function สำหรับตัดสินใจ
def route_intent(state: AgentState) -> Literal["handle_complaint", "handle_inquiry", "escalate"]:
"""
ฟังก์ชันสำหรับตัดสินใจเส้นทางตาม intent
"""
intent = state.get("intent", "")
confidence = state.get("confidence", 0.0)
if confidence < 0.5:
return "escalate"
elif "complaint" in intent.lower():
return "handle_complaint"
elif "inquiry" in intent.lower():
return "handle_inquiry"
else:
return "escalate"
สร้าง Graph
workflow = StateGraph(AgentState)
เพิ่ม node
workflow.add_node("analyze", analyze_intent)
workflow.add_node("handle_complaint", handle_complaint_node)
workflow.add_node("handle_inquiry", handle_inquiry_node)
workflow.add_node("escalate", escalate_node)
กำหนด entry point
workflow.set_entry_point("analyze")
เพิ่ม conditional edge
workflow.add_conditional_edges(
"analyze",
route_intent,
{
"handle_complaint": "handle_complaint",
"handle_inquiry": "handle_inquiry",
"escalate": "escalate"
}
)
compile graph
app = workflow.compile(checkpointer=MemorySaver())การใช้ Loop Control ใน LangGraph
ในบางกรณีเราต้องการให้ Agent ทำงานวนซ้ำจนกว่าจะได้ผลลัพธ์ที่ต้องการ ผมใช้ loop ด้วย iteration counter ซึ่งช่วยแก้ปัญหา infinite loop ได้:
from langgraph.graph import StateGraph, START, END
from typing import Annotated
import operator
Extended State สำหรับ loop control
class LoopState(TypedDict):
messages: list
iteration: int
status: str
def should_continue(state: LoopState) -> Literal["refine", END]:
"""
ตรวจสอบว่าควรจบการทำงานหรือทำต่อ
"""
iteration = state.get("iteration", 0)
messages = state.get("messages", [])
# จำกัดการวนซ้ำสูงสุด 5 ครั้ง
if iteration >= 5:
return END
# ตรวจสอบว่าผลลัพธ์น่าเชื่อถือหรือไม่
if messages and len(messages) > 3:
last_message = messages[-1].get("content", "")
if len(last_message) > 100: # มีเนื้อหาเพียงพอ
return END
return "refine"
สร้าง workflow พร้อม loop
loop_graph = StateGraph(LoopState)
loop_graph.add_node("initial", initial_process)
loop_graph.add_node("refine", refine_process)
loop_graph.add_node("finalize", finalize_process)
loop_graph.add_edge(START, "initial")
loop_graph.add_edge("initial", "refine")
loop_graph.add_conditional_edges(
"refine",
should_continue,
{
"refine": "refine",
END: "finalize"
}
)
loop_graph.add_edge("finalize", END)
loop_app = loop_graph.compile()การรวม Conditional Routing กับ Tool Calling
ในการใช้งานจริงผมต้องการให้ Agent เลือกใช้ Tool ตามสถานการณ์ โดยใช้ HolySheep AI สำหรับ LLM API ที่มีความเร็วสูงและราคาประหยัด สมัครที่นี่ เพื่อเริ่มต้นใช้งาน:
import httpx
from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.tools import tool
กำหนด Tools สำหรับ Agent
@tool
def search_knowledge_base(query: str) -> str:
"""ค้นหาข้อมูลในฐานความรู้"""
# implementation
return f"Found: {query} in KB"
@tool
def calculate_price(items: list) -> str:
"""คำนวณราคาสินค้า"""
total = sum(items)
return f"Total: ${total}"
Function สำหรับเรียก HolySheep AI
def call_holysheep(prompt: str, system: str = None) -> str:
"""เรียก HolySheep AI API สำหรับ LLM inference"""
from openai import OpenAI
client = OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
messages = []
if system:
messages.append({"role": "system", "content": system})
messages.append({"role": "user", "content": prompt})
response = client.chat.completions.create(
model="gpt-4.1",
messages=messages,
temperature=0.7,
max_tokens=500
)
return response.choices[0].message.content
Route function สำหรับเลือกใช้ tool
def route_based_on_intent(state: AgentState) -> Literal["search_knowledge_base", "calculate_price", "general_response"]:
"""เลือกเส้นทางตามประเภทของคำถาม"""
last_message = state["messages"][-1].content.lower()
if "ราคา" in last_message or "price" in last_message or "คำนวณ" in last_message:
return "calculate_price"
elif "ค้นหา" in last_message or "search" in last_message or "หา" in last_message:
return "search_knowledge_base"
else:
return "general_response"การจัดการ Error Handling ใน Conditional Flow
สิ่งสำคัญคือการจัดการข้อผิดพลาดในแต่ละ branch ผมใช้ try-except block และ retry pattern:
from tenacity import retry, stop_after_attempt, wait_exponential
class ConditionalWorkflow:
def __init__(self):
self.graph = self._build_graph()
def _build_graph(self):
workflow = StateGraph(AgentState)
# เพิ่ม nodes พร้อม error handling
workflow.add_node("process", self._safe_process)
workflow.add_node("fallback", self._fallback_handler)
workflow.add_node("retry", self._retry_node)
workflow.set_entry_point("process")
# Conditional edges พร้อม error state
workflow.add_conditional_edges(
"process",
self._determine_outcome,
{
"success": END,
"error": "fallback",
"retry": "retry"
}
)
workflow.add_conditional_edges(
"retry",
self._check_retry_status,
{
"continue": "process",
"max_reached": END
}
)
return workflow.compile(checkpointer=MemorySaver())
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10))
def _safe_process(self, state: AgentState):
try:
# ลองเรียก HolySheep AI
result = call_holysheep(state["messages"][-1].content)
return {"messages": state["messages"] + [AIMessage(content=result)]}
except httpx.TimeoutException:
raise # ให้ retry decorator จัดการ
except Exception as e:
return {"status": "error", "error": str(e)}
def _fallback_handler(self, state: AgentState):
return {
"messages": state["messages"] + [
AIMessage(content="ขออภัย เกิดข้อผิดพลาด กรุณาลองใหม่อีกครั้ง")
]
}
def _determine_outcome(self, state: AgentState):
if state.get("status") == "error":
return "error"
elif state.get("retry_needed"):
return "retry"
return "success"ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
1. KeyError: 'state' เมื่อใช้ Conditional Edge
สาเหตุ: ฟังก์ชัน route คืนค่า key ที่ไม่ตรงกับที่กำหนดใน edge mapping
# ❌ วิธีที่ผิด - key ไม่ตรงกัน
def bad_route(state):
if state["value"] > 10:
return "high" # ไม่มี "high" ใน mapping
workflow.add_conditional_edges(
"node1",
bad_route,
{
"low": "node2", # รอ "low" แต่ได้ "high"
"medium": "node3" # รอ "medium" แต่ได้ "high"
}
)
✅ วิธีที่ถูกต้อง - การันตีว่าทุกเส้นทางมี key ตรงกัน
def good_route(state):
value = state["value"]
if value > 10:
return "high"
elif value > 5:
return "medium"
else:
return "low" # ครอบคลุมทุกกรณี
workflow.add_conditional_edges(
"node1",
good_route,
{
"low": "node2",
"medium": "node3",
"high": "node4"
}
)2. Infinite Loop เมื่อใช้ Conditional Edge กลับไปยังตัวเอง
สาเหตุ: ไม่มี condition ที่จะหยุดการวนซ้ำ
# ❌ วิธีที่ผิด - จะวนไม่รู้จบถ้า condition เป็น True ตลอด
def infinite_route(state):
if state["needs_refinement"]: # ถ้าเป็น True ตลอด = loop ตลอดไป
return "refine"
return END
✅ วิธีที่ถูกต้อง - เพิ่ม iteration counter และ maximum limit
def safe_loop_route(state: AgentState):
iteration = state.get("iteration", 0)
# จำกัดจำนวนครั้งสูงสุด
if iteration >= 5:
return END
# ตรวจสอบเงื่อนไขอื่นๆ
if state["needs_refinement"] and iteration < 5:
return "refine"
return END
อย่าลืม increment iteration ใน node
def refine_node(state: AgentState):
current_iteration = state.get("iteration", 0)
return {"iteration": current_iteration + 1}3. AttributeError: 'dict' object has no attribute 'messages'
สาเหตุ: State เป็น dict ไม่ใช่ object ต้องเข้าถึงด้วย bracket notation
# ❌ วิธีที่ผิด - ใช้ dot notation กับ dict
def bad_node(state: AgentState):
messages = state.messages # AttributeError!
return {"result": messages[-1]}
✅ วิธีที่ถูกต้อง - ใช้ bracket notation
def good_node(state: AgentState):
messages = state.get("messages", []) # หรือ state["messages"]
if not messages:
return {"result": "No messages"}
return {"result": messages[-1].content if hasattr(messages[-1], 'content') else str(messages[-1])}
หรือใช้ type annotation แบบถูกต้อง
class ProperState(TypedDict):
messages: list
data: dict
def proper_node(state: ProperState) -> ProperState:
messages = state["messages"] # TypedDict อนุญาตให้ใช้ bracket
return {"data": {"count": len(messages)}}4. ConnectionError: Timeout เมื่อเรียก External API
สาเหตุ: ไม่มี timeout configuration หรือ retry mechanism
# ❌ วิธีที่ผิด - ไม่มี timeout
def unsafe_api_call(prompt: str) -> str:
client = OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
response = client.chat.completions.create(
model