LangGraph interrupt()로 구현하는 Human-in-the-Loop: 에이전트가 실수하기 전에 사람이 개입하는 인터럽트 설계

from langgraph.types import interrupt, Command
 
def sql_review_node(state):
    query = state["generated_query"]
    
    # 위험한 쿼리일 때만 인터럽트 — 조건부 동적 개입
    if any(keyword in query.upper() for keyword in ["DELETE", "DROP", "UPDATE"]):
        decision = interrupt({
            "query": query,
            "risk_level": "HIGH",
            "message": "위험한 쿼리입니다. 실행하시겠습니까?"
        })
        return {"approved": decision == "approved", "query": query}
    
    # SELECT 같은 안전한 쿼리는 auto-approve
    return {"approved": True, "query": query}

from langgraph.graph import StateGraph, END
from langgraph.types import interrupt, Command
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain_anthropic import ChatAnthropic
from typing import TypedDict, Optional
 
class SQLAgentState(TypedDict):
    user_request: str
    generated_query: str
    approved: bool
    result: Optional[str]
 
llm = ChatAnthropic(model="claude-sonnet-4-6")
 
def generate_query_node(state: SQLAgentState):
    """LLM으로 SQL 쿼리 생성"""
    response = llm.invoke(
        f"다음 요청에 맞는 SQL 쿼리를 생성해주세요: {state['user_request']}"
    )
    return {"generated_query": response.content.strip()}
 
def review_node(state: SQLAgentState):
    """위험한 쿼리는 DBA 검토 요청"""
    query = state["generated_query"]
    dangerous_keywords = ["DELETE", "DROP", "UPDATE", "TRUNCATE"]
    
    if any(kw in query.upper() for kw in dangerous_keywords):
        # interrupt()를 절대 try/except로 감싸면 안 됩니다
        # 내부적으로 예외를 던져서 실행을 멈추는 방식이라 잡아버리면 동작하지 않습니다
        decision = interrupt({
            "query": query,
            "risk": "HIGH",
            "tip": "영향받는 행 수를 먼저 SELECT로 확인해보시면 좋습니다"
        })
        return {"approved": decision == "approved"}
    
    return {"approved": True}
 
def execute_node(state: SQLAgentState):
    """승인된 쿼리만 실행"""
    if not state["approved"]:
        return {"result": "쿼리가 거부되었습니다. 수정 후 다시 시도해주세요."}
    return {"result": f"실행 완료: {state['generated_query']}"}
 
# 그래프 구성
builder = StateGraph(SQLAgentState)
builder.add_node("generate", generate_query_node)
builder.add_node("review", review_node)
builder.add_node("execute", execute_node)
builder.set_entry_point("generate")
builder.add_edge("generate", "review")
builder.add_edge("review", "execute")
builder.add_edge("execute", END)
 
# SQLite 체크포인터 (로컬 개발용 — 프로덕션은 AsyncPostgresSaver로 교체)
checkpointer = SqliteSaver.from_conn_string(":memory:")
graph = builder.compile(checkpointer=checkpointer)
 
# thread_id는 UUID로 생성하고 사용자 세션과 1:1 매핑하는 것을 권장합니다
# 재개 시 반드시 동일한 thread_id를 사용해야 합니다
import uuid
thread_id = str(uuid.uuid4())
config = {"configurable": {"thread_id": thread_id}}
 
result = graph.invoke({"user_request": "오래된 유저 정리"}, config=config)
 
# __interrupt__ 키가 있으면 사람의 판단이 필요한 상태
if "__interrupt__" in result:
    print("승인 대기 중:", result["__interrupt__"])
    
    # DBA가 승인하면 동일한 config(같은 thread_id!)로 재개합니다
    final = graph.invoke(Command(resume="approved"), config=config)
    print("최종 결과:", final["result"])

from langgraph.types import interrupt, Command
from typing import TypedDict, Optional
 
class TravelState(TypedDict):
    destination: str
    travel_plan: Optional[dict]
    booking_confirmed: bool
 
def research_node(state: TravelState):
    """항공편·호텔 조사 (읽기 전용 — 안전)"""
    plan = {
        "flight": "ICN → NRT, 2026-05-10, KRW 280,000",
        "hotel": "신주쿠 그랜드 호텔, 3박 KRW 450,000",
        "total": "KRW 730,000"
    }
    return {"travel_plan": plan}
 
def approval_node(state: TravelState):
    """사용자에게 플랜을 보여주고 승인·수정 대기"""
    response = interrupt({
        "message": "여행 플랜을 확인해 주세요. 수정하거나 승인해 주세요.",
        "plan": state["travel_plan"],
        "options": ["approve", "edit", "cancel"]
    })
    
    if response["action"] == "edit":
        return {
            "travel_plan": {**state["travel_plan"], **response["edits"]},
            "booking_confirmed": False
        }
    elif response["action"] == "approve":
        return {"booking_confirmed": True}
    else:
        return {"booking_confirmed": False}
 
def booking_node(state: TravelState):
    """승인된 플랜만 실제 예약 API 호출"""
    if not state["booking_confirmed"]:
        return {"result": "예약이 취소되었습니다."}
    return {"result": f"예약 완료! {state['travel_plan']}"}

# 사용자가 승인을 선택한 경우
final = graph.invoke(
    Command(resume={"action": "approve"}),
    config=config
)
 
# 사용자가 수정을 선택한 경우
final = graph.invoke(
    Command(resume={"action": "edit", "edits": {"hotel": "시부야 호텔, 3박 KRW 380,000"}}),
    config=config
)

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
import os
import asyncio
 
async def setup_production_graph():
    # 자격증명은 반드시 환경변수로 관리하세요. 코드에 하드코딩하면 안 됩니다
    conn_string = os.environ.get("DATABASE_URL")
    
    async with AsyncPostgresSaver.from_conn_string(conn_string) as checkpointer:
        # 처음 한 번 테이블 생성
        await checkpointer.setup()
        
        graph = builder.compile(checkpointer=checkpointer)
        
        config = {"configurable": {"thread_id": "prod-task-42"}}
        result = await graph.ainvoke(
            {"user_request": "..."},
            config=config
        )
        return result