Skip to main content

​
Human-in-the-Loop

Human-in-the-Loop(HITL)는 Agent의 실행 중간에 사람이 개입하여 확인, 승인, 수정할 수 있는 패턴입니다. 고위험 작업이나 중요한 의사결정에서 안전장치로 활용됩니다.

​
왜 필요한가?

사용 시나리오이유
이메일/메시지 발송잘못된 내용 방지
데이터베이스 수정데이터 손실 방지
결제/금융 처리금전적 위험 방지
외부 API 호출부작용 방지

​
interrupt() + Command(resume=…)

LangGraph에서 HITL을 구현하는 핵심 메커니즘입니다. 
from typing import TypedDict, Annotated, Sequence
from langchain.chat_models import init_chat_model
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import MemorySaver

llm = init_chat_model("gpt-4o-mini", temperature=0)

class State(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]
    action: str
    approved: bool

def plan_action(state: State) -> State:
    """Agent가 실행할 작업을 계획"""
    response = llm.invoke([
        SystemMessage(content="사용자 요청을 분석하여 실행할 작업을 계획하세요."),
    ] + state["messages"])
    return {"action": response.content}

def human_review(state: State) -> State:
    """사람의 승인을 요청"""
    # interrupt()로 실행을 일시 정지
    decision = interrupt({
        "action": state["action"],
        "question": "이 작업을 승인하시겠습니까? (yes/no)",
    })
    return {"approved": decision == "yes"}

def execute_action(state: State) -> State:
    """승인된 작업을 실행"""
    if state["approved"]:
        return {"messages": [("assistant", f"작업 실행 완료: {state['action']}")]}
    return {"messages": [("assistant", "작업이 거부되었습니다.")]}

# 그래프 구성
graph = StateGraph(State)
graph.add_node("plan", plan_action)
graph.add_node("review", human_review)
graph.add_node("execute", execute_action)

graph.add_edge(START, "plan")
graph.add_edge("plan", "review")
graph.add_edge("review", "execute")
graph.add_edge("execute", END)

# 체크포인터 필수 (일시 정지 상태 저장)
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

​
실행 흐름

config = {"configurable": {"thread_id": "review-1"}}

# 1. 실행 시작 → human_review에서 일시 정지
result = app.invoke(
    {"messages": [HumanMessage(content="고객에게 프로모션 이메일을 발송해줘")]},
    config=config,
)
# → interrupt()에서 정지, 사람에게 승인 요청

# 2. 사람이 승인 → 실행 재개
result = app.invoke(
    Command(resume="yes"),
    config=config,
)
# → "작업 실행 완료: ..."

​
도구 실행 전 확인

도구 호출 에이전트에서 위험한 도구 실행 전에 사람 확인을 추가합니다. 
from langgraph.prebuilt import ToolNode, tools_condition

def agent(state: State) -> State:
    messages = [SystemMessage(content=SYSTEM_PROMPT)] + state["messages"]
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}

def human_approval(state: State) -> State:
    """도구 호출 전 사람 확인"""
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        tool_names = [tc["name"] for tc in last_message.tool_calls]
        decision = interrupt({
            "tools": tool_names,
            "question": f"다음 도구를 실행할까요? {tool_names}",
        })
        if decision != "yes":
            return {"messages": [("assistant", "도구 실행이 거부되었습니다.")]}
    return state

graph = StateGraph(State)
graph.add_node("agent", agent)
graph.add_node("human_check", human_approval)
graph.add_node("tools", ToolNode(tools))

graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", tools_condition, {
    "tools": "human_check",  # 도구 호출 전 사람 확인
    "__end__": END,
})
graph.add_edge("human_check", "tools")
graph.add_edge("tools", "agent")

​
체크포인터 요구사항

HITL은 체크포인터가 필수입니다. interrupt() 시점의 상태를 저장하고, Command(resume=...) 시 복원합니다.
체크포인터용도설치
MemorySaver개발/테스트langgraph 내장
PostgresSaver프로덕션pip install langgraph-checkpoint-postgres
SqliteSaver경량 프로덕션pip install langgraph-checkpoint-sqlite
MemorySaver는 프로세스 종료 시 상태가 사라집니다. 프로덕션에서는 PostgresSaver나 SqliteSaver를 사용하세요.
사용 시나리오 판단: 모든 도구 호출에 HITL을 적용하면 사용성이 떨어집니다. 고위험 작업(외부 API, 데이터 수정, 메시지 발송)에만 선택적으로 적용하세요.