บทที่ 9: LangGraph — Multi-Agent Workflows

LangGraph คืออะไร?

LangGraph คือไลบรารีสำหรับสร้าง AI workflow แบบมีสถานะ (stateful) ที่ซับซ้อน โดยเหมาะกับงานที่มี:

การวนลูป (loops)
การแตกแขนงตามเงื่อนไข (conditional branching)
การทำงานหลาย agent ร่วมกัน (multi-agent)
การหยุดรอมนุษย์อนุมัติ (human-in-the-loop)

คำอธิบาย: แผนภาพด้านล่างสรุปโฟลว์การทำงานให้เห็นภาพรวมของหัวข้อนี้อย่างชัดเจน

graph TD
    LCEL["LangChain LCEL"] --> Linear["Input -> Process -> Output"]

    subgraph LG["LangGraph"]
        A["Node A"] --> B["Node B"]
        B --> C{"Condition"}
        C -->|Path 1| D["Node D"]
        C -->|Path 2| E["Node E"]
        D --> B
        E --> F["End"]
    end

เมื่อไหร่ควรใช้ LangGraph?

ใช้ LCEL	ใช้ LangGraph
Pipeline เส้นตรง	มี loop/branch
งานต่อเนื่องไม่ซับซ้อน	ต้องตัดสินใจตามเงื่อนไข
ไม่ต้องเก็บ state ซับซ้อน	ต้องจัดการ state หลายขั้น
single-agent	multi-agent

Core Concepts

1. Graph

คำอธิบาย: โค้ดตัวอย่างด้านล่างแสดงวิธีใช้งานด้วย Python ตามหัวข้อนี้แบบทีละขั้นตอน

from langgraph.graph import StateGraph

graph = StateGraph(State)

2. State

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]  # บทสนทนาสะสม
    current_step: str                         # ขั้นตอนปัจจุบัน
    data: dict                                # ข้อมูลธุรกิจที่ใช้ร่วมกัน

3. Node

def my_node(state: State) -> dict:
    return {"current_step": "done"}

4. Edge

graph.add_edge("node_a", "node_b")

graph.add_conditional_edges(
    "node_a",
    router,
    {"path_1": "node_b", "path_2": "node_c"}
)

5. START / END

from langgraph.graph import START, END

graph.add_edge(START, "first_node")
graph.add_edge("last_node", END)

6. Checkpointing

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

Graph แรกของคุณ

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI

class State(TypedDict):
    messages: Annotated[list, add_messages]

llm = ChatOpenAI(model="gpt-4o-mini")

def chatbot(state: State):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

graph = StateGraph(State)
graph.add_node("chatbot", chatbot)
graph.add_edge(START, "chatbot")
graph.add_edge("chatbot", END)

app = graph.compile()

result = app.invoke({
    "messages": [("human", "สวัสดี! LangGraph คืออะไร?")]
})

print(result["messages"][-1].content)

Conditional Edges

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import ToolMessage

class State(TypedDict):
    messages: Annotated[list, add_messages]
    needs_tool: bool

def assistant(state: State):
    response = llm_with_tools.invoke(state["messages"])
    return {
        "messages": [response],
        "needs_tool": len(response.tool_calls) > 0,
    }

def tool_executor(state: State):
    last_message = state["messages"][-1]
    outputs = []
    for tool_call in last_message.tool_calls:
        result = tool_map[tool_call["name"]].invoke(tool_call["args"])
        outputs.append(ToolMessage(content=str(result), tool_call_id=tool_call["id"]))
    return {"messages": outputs}

def router(state: State):
    return "tools" if state["needs_tool"] else "end"

graph = StateGraph(State)
graph.add_node("assistant", assistant)
graph.add_node("tools", tool_executor)

graph.add_edge(START, "assistant")
graph.add_conditional_edges("assistant", router, {"tools": "tools", "end": END})
graph.add_edge("tools", "assistant")

app = graph.compile()

graph TD
    START([START]) --> assistant[assistant]
    assistant --> decision{needs_tool?}
    decision -->|yes| tools[tools]
    tools --> assistant
    decision -->|no| END([END])

Pattern: Router + Worker Nodes

โครงสร้างที่ใช้บ่อยใน production:

node แรกทำหน้าที่วิเคราะห์งาน
router ตัดสินใจเส้นทาง
ส่งไป worker node ที่เหมาะสม
รวมผลลัพธ์กลับมาตอบ

graph LR
    A[Analyzer] --> B{Route}
    B -->|faq| C[FAQ Worker]
    B -->|search| D[Search Worker]
    B -->|calc| E[Math Worker]
    C --> F[Response]
    D --> F
    E --> F

Human-in-the-Loop

ให้มนุษย์อนุมัติก่อนขั้นตอนที่มีความเสี่ยง

from typing import TypedDict, Annotated
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import AIMessage

checkpointer = MemorySaver()

class State(TypedDict):
    messages: Annotated[list, add_messages]

def propose_action(state: State):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def execute_action(state: State):
    return {"messages": [AIMessage(content="ดำเนินการเรียบร้อย")]}

graph = StateGraph(State)
graph.add_node("propose", propose_action)
graph.add_node("execute", execute_action)

graph.add_edge(START, "propose")
graph.add_edge("propose", "execute")
graph.add_edge("execute", END)

app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["execute"],
)

Multi-Agent System

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI

class State(TypedDict):
    messages: Annotated[list, add_messages]
    task: str
    current_agent: str
    result: str

llm = ChatOpenAI(model="gpt-4o-mini")

def researcher(state: State):
    prompt = f"คุณเป็นนักวิจัย ค้นข้อมูลเกี่ยวกับ: {state['task']}"
    response = llm.invoke([("system", prompt)] + state["messages"])
    return {"messages": [response], "current_agent": "researcher"}

def writer(state: State):
    prompt = "คุณเป็นนักเขียน เขียนจากข้อมูลที่นักวิจัยรวบรวม"
    response = llm.invoke([("system", prompt)] + state["messages"])
    return {"messages": [response], "current_agent": "writer"}

def editor(state: State):
    prompt = "คุณเป็นบรรณาธิการ ปรับปรุงบทความให้สมบูรณ์"
    response = llm.invoke([("system", prompt)] + state["messages"])
    return {"messages": [response], "current_agent": "editor", "result": response.content}

graph = StateGraph(State)
graph.add_node("researcher", researcher)
graph.add_node("writer", writer)
graph.add_node("editor", editor)

graph.add_edge(START, "researcher")
graph.add_edge("researcher", "writer")
graph.add_edge("writer", "editor")
graph.add_edge("editor", END)

app = graph.compile()

graph LR
    START([START]) --> R[Researcher]
    R --> W[Writer]
    W --> E[Editor]
    E --> END([END])

สถาปัตยกรรม Multi-Agent ที่ใช้บ่อย

Planner Agent แยกงานเป็น subtasks
Specialist Agents ทำงานเฉพาะด้าน
Reviewer Agent ตรวจคุณภาพ
Finalizer Agent รวมคำตอบสุดท้าย

Streaming กับ LangGraph

for event in app.stream(
    {
        "messages": [("human", "เขียนบทความเรื่อง AI")],
        "task": "AI",
        "current_agent": "",
        "result": "",
    },
    stream_mode="values",
):
    if "current_agent" in event:
        print(f"Agent: {event['current_agent']}")

Debugging และ Monitoring

แนวทางที่ควรทำในระบบจริง:

เก็บ input/output ของแต่ละ node
บันทึก route ที่ตัดสินใจในทุกคำขอ
วัด latency ต่อ node
จำกัดจำนวนรอบ loop สูงสุด

LangSmith ช่วย trace workflow และหาจุดคอขวดได้ง่ายมาก

LangGraph vs LCEL

Feature	LCEL	LangGraph
Linear chains	เหมาะมาก	ใช้ได้แต่เกินความจำเป็น
Loops/Cycles	จำกัด	รองรับ
State management	จำกัด	ยืดหยุ่น
Multi-agent	ทำได้ยาก	ออกแบบมาเพื่อสิ่งนี้
Human-in-the-loop	จำกัด	รองรับ

Best Practices

ออกแบบ State ให้เล็กและชัดเจน
แยก node ให้มีความรับผิดชอบเดียว
เขียน router ให้ deterministic
ใส่ timeout/retry สำหรับ node ที่เรียก API ภายนอก
เขียน test ทั้งระดับ node และระดับ graph

Common Pitfalls

state ใหญ่เกินไปจน graph ช้า
loop ไม่มีเงื่อนไขจบที่ชัดเจน
router พึ่งข้อความอิสระมากเกินไป
ไม่มี error path แยกสำหรับ debug