LangGraph Studio 可视化调试：复杂 Agent 工作流完整指南

การพัฒนา Multi-Agent System ที่ซับซ้อนนั้นยากมากพอ แต่การ debug ยิ่งยากกว่าเป็นไกล เมื่อคุณมี 5-10 sub-agents ทำงานพร้อมกัน แต่ละตัวมี state เป็นของตัวเอง และ message วิ่งข้ามไปมาระหว่างกัน การหาว่า "bug อยู่ตรงไหน" กลายเป็นฝันร้าย

เมื่อวานผมเจอปัญหาหนัก: ConnectionError: timeout หลังจากรัน 30 วินาที ใน production แต่ dev environment ทำงานได้ปกติ เลยต้องมานั่ง trace ทีละ node ใน LangGraph เพื่อหาว่า agent ไหนที่ทำให้เกิด timeout

LangGraph Studio คืออะไร และทำไมต้องใช้

LangGraph Studio เป็นเครื่องมือ visualization จาก LangChain ที่ช่วยให้เราเห็นภาพรวมของ workflow ทั้งหมด ดู state ณ แต่ละ step, ตรวจสอบ message ที่ส่งระหว่าง nodes, และที่สำคัญคือ replay execution ย้อนกลับไป debug ทีละ step

การตั้งค่า LangGraph Studio กับ HolySheep AI

ก่อนจะเริ่ม debug ต้องตั้งค่า environment ให้เรียบร้อยก่อน เราจะใช้ HolySheep AI เพราะราคาถูกกว่า API อื่นถึง 85%+ และมี latency ต่ำกว่า 50ms สมัครที่นี่ เพื่อรับเครดิตฟรี

# ติดตั้ง dependencies ที่จำเป็น
pip install langgraph langchain-core langchain-holysheep python-dotenv

สร้างไฟล์ .env
cat > .env << 'EOF'
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1
EOF

ตรวจสอบการตั้งค่า
python -c "from langchain_holysheep import ChatHolySheep; print('HolySheep SDK พร้อมใช้งาน')"

สร้าง Complex Agent Workflow พร้อม Visualization

มาสร้างตัวอย่างจริงที่มีหลาย agents ทำงานร่วมกัน โดยจะมี supervisor agent, researcher agent, และ writer agent ที่ทำงานตามลำดับ

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import create_react_agent
from typing import TypedDict, Annotated
import operator

โหลด environment variables
load_dotenv()

สร้าง LLM สำหรับทุก agents โดยใช้ HolySheep
ราคา HolySheep: GPT-4.1 $8/MTok, Claude Sonnet 4.5 $15/MTok, 
Gemini 2.5 Flash $2.50/MTok, DeepSeek V3.2 $0.42/MTok (ประหยัด 85%+)
llm = ChatOpenAI(
    model="gpt-4.1",
    base_url="https://api.holysheep.ai/v1",  # ใช้ HolySheep เท่านั้น
    api_key=os.getenv("HOLYSHEEP_API_KEY"),
    default_headers={"Connection": "keep-alive"}
)

กำหนด state structure สำหรับ workflow
class AgentState(TypedDict):
    task: str
    research: str
    draft: str
    review: str
    iteration: int
    messages: Annotated[list, operator.add]

def create_researcher_agent():
    """สร้าง researcher agent ที่ค้นหาข้อมูล"""
    tools = [
        # ใส่ tools สำหรับค้นหาข้อมูลจริงๆ
    ]
    return create_react_agent(llm, tools)

def create_writer_agent():
    """สร้าง writer agent ที่เขียนบทความ"""
    return create_react_agent(llm, tools=[])

สร้าง agents
researcher = create_researcher_agent()
writer = create_writer_agent()

def research_node(state: AgentState):
    """Node สำหรับงานวิจัย"""
    result = researcher.invoke({"messages": [("user", f"ค้นหาข้อมูลเกี่ยวกับ: {state['task']}")]})
    return {"research": str(result), "iteration": state.get("iteration", 0)}

def write_node(state: AgentState):
    """Node สำหรับเขียนบทความ"""
    result = writer.invoke({
        "messages": [("user", f"เขียนบทความจากข้อมูล: {state['research']}")]
    })
    return {"draft": str(result)}

สร้าง graph
workflow = StateGraph(AgentState)
workflow.add_node("research", research_node)
workflow.add_node("write", write_node)
workflow.set_entry_point("research")
workflow.add_edge("research", "write")
workflow.add_edge("write", END)

compile เป็น app
app = workflow.compile()

ตรวจสอบ graph structure
print("Graph nodes:", app.get_graph().nodes.keys())

การ Debug ด้วย LangGraph Studio Visualization

หลังจากสร้าง workflow แล้ว มาดูวิธีการ visualize และ debug กัน

from langchain_core.callbacks import get_openai_callback
from langgraph.store.memory import InMemoryStore

สร้าง memory store สำหรับเก็บ history
store = InMemoryStore()

ตั้งค่า config สำหรับ debug
config = {
    "configurable": {
        "thread_id": "debug-session-001",
        "store": store
    },
    "recursion_limit": 100,
    "max_iterations": 10
}

รัน workflow พร้อม tracking
initial_state = {
    "task": "เทคนิคการ optimize LangGraph performance",
    "research": "",
    "draft": "",
    "review": "",
    "iteration": 0,
    "messages": []
}

print("=" * 60)
print("เริ่ม Debug Session")
print("=" * 60)

รันแบบ stream เพื่อดูแต่ละ step
for step in app.stream(initial_state, config=config):
    node_name = list(step.keys())[0]
    node_output = step[node_name]
    print(f"\n[Node: {node_name}]")
    print(f"Output keys: {node_output.keys()}")
    
    # แสดง state ปัจจุบัน
    if "research" in node_output:
        print(f"Research length: {len(node_output['research'])} chars")
    if "draft" in node_output:
        print(f"Draft length: {len(node_output['draft'])} chars")

ดึงข้อมูล history สำหรับ visualization
with get_openai_callback() as cb:
    history = store.search(["debug-session-001"])
    print(f"\nTotal tokens: {cb.total_tokens}")
    print(f"Total cost: ${cb.total_cost:.4f}")

สร้าง visualization data
def generate_debug_report(app, initial_state, config):
    """สร้างรายงาน debug สำหรับ LangGraph Studio"""
    debug_data = {
        "workflow_name": "multi_agent_research_pipeline",
        "nodes": [],
        "edges": [],
        "execution_trace": []
    }
    
    # ดึง graph structure
    graph = app.get_graph()
    for node in graph.nodes:
        debug_data["nodes"].append({
            "id": node,
            "type": graph.nodes[node].__class__.__name__
        })
    
    # ดึง edges
    for edge in graph.edges:
        debug_data["edges"].append({
            "source": edge[0],
            "target": edge[1]
        })
    
    return debug_data

report = generate_debug_report(app, initial_state, config)
print("\nDebug Report:")
print(f"Nodes: {len(report['nodes'])}")
print(f"Edges: {len(report['edges'])}")

การใช้ Breakpoints และ Checkpointing

นี่คือเทคนิคสำคัญในการ debug: ใช้ breakpoints เพื่อหยุด execution ที่จุดที่ต้องการตรวจสอบ

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.errors import NodeInterrupt

สร้าง checkpoint saver
checkpointer = SqliteSaver.from_conn_string(":memory:")

def conditional_breakpoint(state: AgentState) -> bool:
    """หยุดถ้า iteration เกิน 3 ครั้ง"""
    if state.get("iteration", 0) > 3:
        raise NodeInterrupt(
            f"เกินจำนวน iteration สูงสุด: {state.get('iteration')}"
        )
    return True

เพิ่ม conditional breakpoint ใน workflow
workflow_with_breakpoints = StateGraph(AgentState, checkpointer=checkpointer)
workflow_with_breakpoints.add_node("research", research_node)
workflow_with_breakpoints.add_node("write", write_node)
workflow_with_breakpoints.set_entry_point("research")
workflow_with_breakpoints.add_edge("research", "write")

เพิ่ม conditional edge
workflow_with_breakpoints.add_conditional_edges(
    "write",
    lambda state: END if state.get("iteration", 0) >= 3 else "research",
    {
        END: END,
        "research": "research"
    }
)

app_with_breakpoints = workflow_with_breakpoints.compile()

Debug ด้วย checkpoint
config_with_checkpoint = {
    "configurable": {
        "thread_id": "breakpoint-debug-001",
        "checkpoint_ns": "iteration_check"
    }
}

print("ทดสอบ breakpoint...")
try:
    result = app_with_breakpoints.invoke(
        {**initial_state, "iteration": 4},
        config=config_with_checkpoint
    )
except NodeInterrupt as e:
    print(f"✓ Breakpoint triggered: {e}")

ดึง checkpoints ทั้งหมด
checkpoints = list(checkpointer.list({"configurable": {"thread_id": "breakpoint-debug-001"}}))
print(f"พบ {len(checkpoints)} checkpoints")
for cp in checkpoints:
    print(f"  - {cp['config']['configurable']['checkpoint_id']}")

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

1. ConnectionError: timeout หลังจาก 30 วินาที

สาเหตุ: ปัญหานี้มักเกิดจาก HolySheep API ที่ใช้ timeout default 30 วินาที โดยเฉพาะเมื่อ agent รอผลลัพธ์จาก tool execution ที่ใช้เวลานาน

# วิธีแก้ไข: เพิ่ม timeout parameter และ retry logic
from openai import OpenAI
from tenacity import retry, stop_after_attempt, wait_exponential
import httpx

สร้าง client ที่รองรับ timeout
client = OpenAI(
    api_key=os.getenv("HOLYSHEEP_API_KEY"),
    base_url="https://api.holysheep.ai/v1",
    timeout=httpx.Timeout(120.0, connect=30.0),  # 120 วินาที total, 30 วินาที connect
    max_retries=3
)

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=30)
)
def call_with_retry(messages, model="gpt-4.1"):
    """เรียก API พร้อม retry logic"""
    try:
        response = client.chat.completions.create(
            model=model,
            messages=messages,
            temperature=0.7
        )
        return response
    except httpx.TimeoutException as e:
        print(f"Timeout occurred: {e}")
        raise
    except httpx.ConnectError as e:
        print(f"Connection error: {e}")
        raise

ใช้งาน
llm_with_timeout = ChatOpenAI(
    model="gpt-4.1",
    base_url="https://api.holysheep.ai/v1",
    api_key=os.getenv("HOLYSHEEP_API_KEY"),
    default_headers={"timeout": "120"}
)

หรือเพิ่มใน LangGraph config
config_with_timeout = {
    "configurable": {
        "thread_id": "timeout-debug-001"
    },
    "recursion_limit": 100,
    "execution_timeout": 120  # เพิ่ม timeout สำหรับทั้ง workflow
}

2. ValueError: invalid literal for int() with base 10

สาเหตุ: API key ที่ใส่มีช่องว่างหรือ format ผิด หรือบางครั้งเกิดจากการ parse response ที่ไม่ตรง format

# วิธีแก้ไข: ตรวจสอบและ clean API key
def validate_api_key(key: str) -> str:
    """ตรวจสอบความถูกต้องของ API key"""
    if not key:
        raise ValueError("API key ห้ามว่าง")
    
    # ลบช่องว่างและ newline
    cleaned_key = key.strip()
    
    # ตรวจสอบว่าเป็น format ที่ถูกต้อง
    if not cleaned_key.startswith("sk-"):
        raise ValueError(f"API key ต้องขึ้นต้นด้วย 'sk-' แต่ได้: {cleaned_key[:10]}...")
    
    if len(cleaned_key) < 40:
        raise ValueError(f"API key สั้นเกินไป: {len(cleaned_key)} chars")
    
    return cleaned_key

ใช้งาน
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "")
validated_key = validate_api_key(HOLYSHEEP_API_KEY)

สร้าง LLM ด้วย key ที่ validated
llm = ChatOpenAI(
    model="gpt-4.1",
    base_url="https://api.holysheep.ai/v1",
    api_key=validated_key
)

ทดสอบ connection
try:
    response = llm.invoke("ทดสอบ")
    print("✓ Connection สำเร็จ")
except Exception as e:
    print(f"✗ Connection failed: {e}")

3. AttributeError: 'NoneType' object has no attribute 'get'

สาเหตุ: State บาง field เป็น None เพราะไม่ได้ initialize ค่าเริ่มต้น หรือ node ก่อนหน้าไม่ได้ return field นั้น

# วิธีแก้ไข: ใช้ Pydantic model หรือ default values
from typing import Optional
from pydantic import BaseModel, Field

class AgentState(BaseModel):
    """ใช้ Pydantic เพื่อบังคับ type และ default values"""
    task: str = Field(default="")
    research: Optional[str] = Field(default="")
    draft: Optional[str] = Field(default="")
    review: Optional[str] = Field(default="")
    iteration: int = Field(default=0)
    messages: list = Field(default_factory=list)
    
    # เพิ่ม validator เพื่อ ensure non-null
    @model_validator(mode='after')
    def check_required_fields(self):
        if not self.task:
            raise
แหล่งข้อมูลที่เกี่ยวข้อง
📚 บทช่วยสอน AI API
💰 ดูราคา
📖 เอกสารสำหรับนักพัฒนา
🚀 สมัครฟรี
บทความที่เกี่ยวข้อง
```json
LLM Guard เฟรมเวิร์กโอเพนซอร์สสำหรับกรองเนื้อหา AI อย่างครบว
คู่มือฉบับสมบูรณ์: ต้นไม้ตัดสินใจเลือกโมเดล LLM — จากประสบกา