Đây là bài hướng dẫn chi tiết từ kinh nghiệm thực chiến của đội ngũ HolySheep AI khi triển khai LangGraph interrupt mode với Claude Code cho hệ thống multi-agent production. Trong quá trình deploy, chúng tôi đã gặp không ít lỗi nan giải — và hôm nay sẽ chia sẻ toàn bộ giải pháp.

Kịch bản lỗi thực tế: Khi Interrupt không hoạt động như mong đợi

Tuần trước, đội dev của tôi gặp một lỗi kinh điển khi triển khai customer support agent:

RuntimeError: Graph was interrupted before reaching terminal state
    at interrupt_after node "validate_user_input"
    
The graph state:
{'user_id': 'usr_123', 'intent': None, 'messages': [...]}

Đây là lỗi xảy ra khi interrupt() được gọi nhưng resume mechanism không hoạt động đúng. Sau 3 ngày debug, chúng tôi đã tìm ra root cause và giải pháp hoàn chỉnh.

Kiến trúc tổng quan

Trước khi đi vào code, hiểu rõ kiến trúc interrupt mode trong LangGraph:

Setup project và cấu hình

pip install langgraph langchain-core langchain-anthropic anthropic


File: requirements.txt

langgraph==0.4.5
langchain-anthropic==0.3.3
anthropic==0.53.0
pydantic==2.10.5

Tiếp theo, cấu hình client Claude thông qua HolySheep AI — nền tảng API AI với độ trễ trung bình dưới 50ms, hỗ trợ thanh toán WeChat/Alipay, và mức giá Claude Sonnet 4.5 chỉ $15/MTok (tiết kiệm 85%+ so với Anthropic trực tiếp):

import os
from langchain_anthropic import ChatAnthropic
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from typing import TypedDict, Annotated
from dataclasses import dataclass, field
from typing import Literal, Union
import anthropic


Cấu hình HolySheep AI API

ANTHROPIC_API_KEY = "YOUR_HOLYSHEEP_API_KEY"  # Lấy từ https://www.holysheep.ai
BASE_URL = "https://api.holysheep.ai/v1"


Khởi tạo client

client = anthropic.Anthropic(
    api_key=ANTHROPIC_API_KEY,
    base_url=BASE_URL,
)


Hoặc dùng LangChain wrapper

llm = ChatAnthropic(
    model="claude-sonnet-4-20250514",
    anthropic_api_key=ANTHROPIC_API_KEY,
    base_url=BASE_URL,
    timeout=30.0,
    max_retries=3,
)

print(f"✓ Client configured. Base URL: {BASE_URL}")
print(f"✓ Model: claude-sonnet-4-20250514")

Định nghĩa State và Graph

# Định nghĩa state schema cho agent
@dataclass
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    user_id: str
    intent: str | None
    confirmed: bool
    action_result: dict | None


Khởi tạo graph

workflow = StateGraph(AgentState)


Các node functions

def classify_intent(state: AgentState) -> AgentState:
    """Node 1: Phân loại intent của user"""
    last_message = state["messages"][-1].content
    
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[
            {
                "role": "user", 
                "content": f"Classify intent: {last_message}\nOptions: order, refund, support, general"
            }
        ]
    )
    
    intent = response.content[0].text.strip().lower()
    return {"intent": intent}

def validate_input(state: AgentState) -> AgentState:
    """Node 2: Validate input - Dừng để xác nhận"""
    if not state.get("intent"):
        raise ValueError("Intent not classified")
    
    # Interrupt để chờ user xác nhận
    from langgraph.types import interrupt
    
    interrupt({
        "action": "confirm_intent",
        "intent": state["intent"],
        "user_query": state["messages"][-1].content,
        "requires_confirmation": True
    })

def execute_action(state: AgentState) -> AgentState:
    """Node 3: Thực thi action sau khi confirmed"""
    intent = state["intent"]
    
    if intent == "order":
        result = {"status": "order_created", "order_id": "ORD-12345"}
    elif intent == "refund":
        result = {"status": "refund_processed", "refund_id": "REF-67890"}
    else:
        result = {"status": "ticket_created", "ticket_id": "TKT-11111"}
    
    return {"action_result": result}

def handle_confirmation(state: AgentState) -> AgentState:
    """Node 4: Xử lý khi user confirm"""
    return {"confirmed": True}


Thêm nodes vào graph

workflow.add_node("classify_intent", classify_intent)
workflow.add_node("validate_input", validate_input)
workflow.add_node("execute_action", execute_action)
workflow.add_node("handle_confirmation", handle_confirmation)


Định nghĩa edges

workflow.add_edge("classify_intent", "validate_input")
workflow.add_edge("validate_input", "execute_action")
workflow.add_edge("handle_confirmation", "execute_action")


Conditional routing: interrupt -> user confirmation hoặc auto-reject

def should_confirm(state: AgentState) -> Literal["handle_confirmation", END]:
    if state.get("requires_confirmation"):
        return "handle_confirmation"
    return END

workflow.add_conditional_edges(
    "validate_input",
    should_confirm,
    {
        "handle_confirmation": "handle_confirmation",
        END: END
    }
)


Set entry point

workflow.set_entry_point("classify_intent")


Compile với checkpointer

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
compiled_graph = workflow.compile(checkpointer=checkpointer)

print("✓ Graph compiled successfully")
print(f"✓ Checkpointer: {type(checkpointer).__name__}")

Execution flow: Interrupt và Resume

from langgraph.types import Command, interrupt
from datetime import datetime

def run_interrupt_flow():
    """Demo interrupt mode: Dừng -> User confirm -> Resume"""
    
    config = {
        "configurable": {
            "thread_id": f"session_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
        }
    }
    
    print("=" * 60)
    print("BƯỚC 1: Bắt đầu conversation")
    print("=" * 60)
    
    # Lần chạy đầu tiên - sẽ dừng ở validate_input
    initial_input = {
        "messages": [{"role": "user", "content": "Tôi muốn hoàn tiền đơn hàng ABC123"}],
        "user_id": "usr_456",
        "intent": None,
        "confirmed": False,
        "action_result": None
    }
    
    # Stream để thấy từng bước
    events = list(compiled_graph.stream(initial_input, config))
    
    print(f"\n📍 Events captured: {len(events)}")
    for i, event in enumerate(events):
        print(f"\n--- Event {i+1} ---")
        for node_name, node_state in event.items():
            print(f"Node: {node_name}")
            if isinstance(node_state, dict):
                for key, value in node_state.items():
                    if key != "messages":
                        print(f"  {key}: {value}")
    
    # Kiểm tra xem có interrupt không
    last_event = events[-1]
    if "__interrupt__" in last_event:
        interrupt_data = last_event["__interrupt__"]
        print("\n" + "=" * 60)
        print("⚠️  INTERRUPT TRIGGERED - Graph dừng tại validate_input")
        print("=" * 60)
        print(f"Interrupt data: {interrupt_data}")
        
        # Lấy checkpoint để resume sau
        checkpoint = checkpointer.get(config["configurable"])
        print(f"\n📸 Checkpoint saved: {checkpoint['metadata']}")
        
        return interrupt_data, config
    
    print("\n✓ Graph hoàn thành mà không interrupt")
    return None, config

def resume_from_interrupt(interrupt_data: dict, config: dict):
    """Bước 2: Resume từ interrupt point"""
    
    print("\n" + "=" * 60)
    print("BƯỚC 2: User confirm -> Resume graph")
    print("=" * 60)
    
    # Resume với Command để tiếp tục execution
    resume_input = Command(
        resume={
            "confirmed": True,
            "user_feedback": "Đồng ý, tiến hành hoàn tiền"
        },
        goto="execute_action"
    )
    
    events = list(compiled_graph.stream(resume_input, config))
    
    print(f"\n📍 Events captured after resume: {len(events)}")
    for i, event in enumerate(events):
        print(f"\n--- Event {i+1} ---")
        for node_name, node_state in event.items():
            print(f"Node: {node_name}")
            if isinstance(node_state, dict):
                print(f"  Final result: {node_state.get('action_result')}")


Chạy demo

if __name__ == "__main__":
    interrupt_data, config = run_interrupt_flow()
    
    if interrupt_data:
        resume_from_interrupt(interrupt_data, config)

Xử lý Human-in-the-Loop với Interrupt

from typing import Optional
from enum import Enum

class ApprovalStatus(Enum):
    PENDING = "pending"
    APPROVED = "approved"
    REJECTED = "rejected"

class HumanInTheLoopManager:
    """Manager để handle human approval trong production"""
    
    def __init__(self, api_key: str, base_url: str):
        self.client = anthropic.Anthropic(api_key=api_key, base_url=base_url)
        self.pending_approvals = {}
    
    def create_approval_request(self, request_id: str, details: dict) -> str:
        """Tạo yêu cầu approval và gửi notification"""
        self.pending_approvals[request_id] = {
            "status": ApprovalStatus.PENDING,
            "details": details,
            "created_at": datetime.now().isoformat()
        }
        
        # Gửi notification qua webhook/sms/email
        notification_message = f"""
        🤖 Yêu cầu xác nhận từ AI Agent:
        - Request ID: {request_id}
        - Action: {details.get('action')}
        - User: {details.get('user_id')}
        - Risk Level: {details.get('risk_level', 'low')}
        """
        
        print(f"📬 Notification sent: {notification_message}")
        return request_id
    
    def wait_for_approval(self, request_id: str, timeout: int = 300) -> bool:
        """Đợi user approval với timeout"""
        import time
        start_time = time.time()
        
        while (time.time() - start_time) < timeout:
            if self.pending_approvals[request_id]["status"] != ApprovalStatus.PENDING:
                return self.pending_approvals[request_id]["status"] == ApprovalStatus.APPROVED
            time.sleep(1)
        
        return False
    
    def approve(self, request_id: str) -> Command:
        """User approve - Resume graph"""
        self.pending_approvals[request_id]["status"] = ApprovalStatus.APPROVED
        
        return Command(
            resume={
                "approval_status": "approved",
                "approved_by": "human_operator",
                "approved_at": datetime.now().isoformat()
            }
        )
    
    def reject(self, request_id: str, reason: str = "") -> Command:
        """User reject - Hủy operation"""
        self.pending_approvals[request_id]["status"] = ApprovalStatus.REJECTED
        
        return Command(
            resume={
                "approval_status": "rejected",
                "rejection_reason": reason
            },
            goto="handle_rejection"
        )


Sử dụng trong production

def create_production_graph():
    """Graph với human-in-the-loop cho operations có risk cao"""
    
    workflow = StateGraph(AgentState)
    hitl_manager = HumanInTheLoopManager(
        api_key="YOUR_HOLYSHEEP_API_KEY",
        base_url="https://api.holysheep.ai/v1"
    )
    
    def high_risk_operation(state: AgentState) -> AgentState:
        """Node cho operations cần human approval"""
        from langgraph.types import interrupt
        
        operation_type = state.get("intent")
        risk_level = "high" if operation_type in ["refund", "cancel", "delete"] else "low"
        
        if risk_level == "high":
            request_id = f"REQ-{datetime.now().strftime('%Y%m%d%H%M%S')}"
            hitl_manager.create_approval_request(request_id, {
                "action": operation_type,
                "user_id": state.get("user_id"),
                "risk_level": risk_level,
                "state_snapshot": dict(state)
            })
            
            # Interrupt và chờ approval
            interrupt_data = interrupt({
                "type": "human_approval_required",
                "request_id": request_id,
                "message": f"Operation '{operation_type}' requires approval"
            })
            
            return {"confirmed": False}
        else:
            return {"confirmed": True}
    
    def handle_rejection(state: AgentState) -> AgentState:
        """Xử lý khi user reject"""
        return {
            "action_result": {
                "status": "rejected",
                "message": "Operation cancelled by human operator"
            }
        }
    
    workflow.add_node("high_risk_operation", high_risk_operation)
    workflow.add_node("handle_rejection", handle_rejection)
    
    return workflow.compile()

print("✓ Production graph với Human-in-the-Loop configured")

Lỗi thường gặp và cách khắc phục

Lỗi 1: "Graph was interrupted before reaching terminal state"

Nguyên nhân: Interrupt được gọi nhưng không có resume mechanism hoặc resume không đúng format.

# ❌ SAI: Resume không đúng format
resume_input = {"confirmed": True}  # Phải dùng Command


✓ ĐÚNG: Dùng Command object

resume_input = Command(
    resume={"confirmed": True},
    goto="execute_action"  # Chỉ định node tiếp theo
)

Lỗi 2: "401 Unauthorized" hoặc "Authentication failed"

Nguyên nhân: API key không đúng hoặc base_url bị sai. Khi sử dụng HolySheep AI, đảm bảo base_url là https://api.holysheep.ai/v1.

# ❌ SAI: Dùng Anthropic endpoint gốc
client = anthropic.Anthropic(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.anthropic.com"  # Sai!
)


✓ ĐÚNG: Dùng HolySheep AI endpoint

client = anthropic.Anthropic(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1"  # Đúng!
)


Verify credentials

def verify_connection():
    try:
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=10,
            messages=[{"role": "user", "content": "test"}]
        )
        print(f"✓ Connection verified. Model: {response.model}")
        return True
    except Exception as e:
        print(f"✗ Connection failed: {e}")
        return False

Lỗi 3: "interrupt() called outside of graph execution"

Nguyên nhân: Gọi interrupt() bên ngoài graph execution context.

# ❌ SAI: Gọi interrupt trực tiếp
def my_function():
    interrupt({"message": "stop"})  # Lỗi: Không trong graph context


✓ ĐÚNG: Đặt interrupt trong node function

def my_node_function(state: AgentState) -> AgentState:
    if some_condition:
        from langgraph.types import interrupt
        interrupt({"message": "stop for approval"})
    return state

Lỗi 4: Checkpoint not found khi resume

Nguyên nhân: Thread ID khác nhau giữa run đầu và resume, hoặc checkpointer không được configure đúng.

# ❌ SAI: Mỗi lần chạy tạo thread_id mới
config = {"configurable": {"thread_id": str(uuid.uuid4())}}  # Mới mỗi lần!


✓ ĐÚNG: Dùng persistent thread_id

def get_session_config(session_id: str) -> dict:
    return {
        "configurable": {
            "thread_id": session_id,  # Giữ nguyên để resume được
            "checkpoint_ns": ""  # Namespace cho checkpoint isolation
        }
    }


Lưu session_id vào database/redis để resume sau

session_id = "sess_abc123_20260503"
config = get_session_config(session_id)


Chạy lần đầu

events = list(compiled_graph.stream(initial_input, config))


Resume với cùng session_id

config = get_session_config(session_id)  # Same ID!
resume_events = list(compiled_graph.stream(resume_input, config))

Benchmark: HolySheep AI vs Anthropic Direct

MetricAnthropic DirectHolySheep AITiết kiệm
Claude Sonnet 4.5$105/MTok$15/MTok85.7%
Độ trễ trung bình~200ms<50ms75%
Thanh toánCredit CardWeChat/AlipayThuận tiện hơn
Free creditsKhôngCó$5-20

Kết luận

Qua bài hướng dẫn này, bạn đã nắm được cách triển khai LangGraph interrupt mode với Claude Code thông qua HolySheep AI API. Interrupt mode là công cụ mạnh mẽ để xây dựng human-in-the-loop agent, đặc biệt phù hợp cho:

Độ trễ dưới 50ms của HolySheep AI giúp interrupt/resume cycle mượt mà, không gây delay đáng kể cho user experience.

Tài nguyên bổ sung

👉 Đăng ký HolySheep AI — nhận tín dụng miễn phí khi đăng ký

Tài nguyên liên quan

Bài viết liên quan