Patrick's 데이터 세상
Human-in-the-loop(HIL) 본문
Human-in-the-Loop, 인간 개입 시스템
에이전트나 워크플로우에서 도구 호출하는 것을 검토, 편집 및 승인하려면 LangGraph의 인간 개입 기능을 활용하여 워크플로우의 어느 단계에서든 사람의 개입을 가능하게 합니다. 모델 출력에 대해 검증, 수정 또는 추가적인 맥락이 필요한 경우 대규모 언어 모델(LLM) 기반 애플리케이션에서 특히 유용합니다.
👉🏻 주요 기능
지속적인 실행 상태: 인터럽트는 그래프 상태를 저장하는 LangGraph의 지속성 계층을 사용하여, 사용자가 재개할 때까지 그래프 실행을 무기한 일시 중지할 수 있습니다. LangGraph가 각 단계 후 그래프 상태를 체크포인트하기 때문에 가능하며, 이를 통해 시스템은 실행 컨텍스트를 유지하고 나중에 워크플로를 재개하여 중단된 지점부터 계속할 수 있습니다. 시간 제약 없이 비동기적인 인간 검토 또는 입력을 지원합니다.
Dynamic interrupts, 동적 중단 : 그래프의 현재 상태를 기반으로 특정 노드 내부에서 그래프를 일시 중지하기 위해 중단을 사용합니다.
Static interrupts, 정적 중단 : interrupt_before 및 interrupt_after를 사용하여 노드 실행 전후의 미리 정의된 지점에서 그래프를 일시 중지합니다.
Flexible integration points, 유연한 통합 지점 : 인간 개입 논리는 워크플로우의 어느 지점에서든 도입될 수 있습니다. 이를 통해 API 호출 승인, 출력 수정 또는 대화 유도 등 특정 목적에 따른 인간 개입이 가능합니다.
동적 중단 예제
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command
# 1) State 타입 정의
class State(TypedDict):
some_text: str
# 2) 사람 입력을 기다리는 노드
def human_node(state: State) -> State:
value = interrupt({"text_to_revise": state["some_text"]})
return {"some_text": value}
# 3) 그래프 구성
graph_builder = StateGraph(State)
graph_builder.add_node("human_node", human_node)
graph_builder.add_edge(START, "human_node")
checkpointer = InMemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)
# 4) 실행 → 인터럽트 발생
config = {"configurable": {"thread_id": "some_id"}}
result = graph.invoke({"some_text": "original text"}, config=config)
print(result.get("__interrupt__", None))
# [Interrupt(value={'text_to_revise': 'original text'}, resumable=True, ns=['human_node:f34c685b-1070-df0c-b889-83338ddfec0a'])]
# 5) 인터럽트 재개
resumed = graph.invoke(Command(resume="Edited text"), config=config)
print(resumed)
# {'some_text': 'Edited text'}
- 실행 일시중지
interrupt(payload)를 노드 안에서 호출하면 그 자리에서 그래프가 멈추고, {"text_to_revise": "original text"}와 메타데이터를 담은 Interrupt 객체가 반환됩니다. 이때 클라이언트는 이 값을 사용자에게 보여주고, 사람의 응답을 기다립니다. - 재개(Resume)
사용자의 응답을 Command(resume=<값>)로 넘겨 같은 thread_id 로 다시 호출하면, 인터럽트를 호출했던 “그 노드의 시작부터” 재실행되며 이번에는 interrupt(...)가 멈추지 않고 resume에 넣어준 값(예: "Edited text")을 반환합니다. 그래서 value가 "Edited text"가 되고, 노드의 반환은 {"some_text": "Edited text"}가 됩니다. - 체크포인터 필수
중단/재개는 상태를 저장해야 하므로 compile(checkpointer=...)가 필수입니다. 또한 실행 컨텍스트는 config={"configurable":{"thread_id": ...}} 로 구분합니다.
정적 중단 예제
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
# ---- (예시용) 상태/노드 정의 ----
class State(TypedDict):
x: int
def node_a(state: State) -> State:
return {"x": state["x"] + 1}
def node_b(state: State) -> State:
return {"x": state["x"] * 2}
def node_c(state: State) -> State:
return {"x": state["x"] - 3}
# ---- 그래프 구성 ----
builder = StateGraph(State)
builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)
builder.add_node("node_c", node_c)
builder.add_edge(START, "node_a")
builder.add_edge("node_a", "node_b")
builder.add_edge("node_b", "node_c")
builder.add_edge("node_c", END)
checkpointer = InMemorySaver()
graph = builder.compile(
checkpointer=checkpointer,
# static interrupt: node_a 실행 "전"에 멈추고,
interrupt_before=["node_a"],
# node_b, node_c 실행 "후"에 멈춤
interrupt_after=["node_b", "node_c"],
)
# ---- 실행 & 재개 ----
config = {"configurable": {"thread_id": "some_thread"}}
inputs = {"x": 10}
# 1) 최초 실행: node_a "전"에서 멈춤
res = graph.invoke(inputs, config=config)
print(res.get("__interrupt__", [])) # 인터럽트 메타 확인
# 2) 재개: node_a 실행으로 진행
res = graph.invoke(Command(resume=True), config=config)
print(res.get("__interrupt__", [])) # node_b "후"에서 다시 멈춤
# 3) 재개: node_c로 진행
res = graph.invoke(Command(resume=True), config=config)
print(res.get("__interrupt__", [])) # node_c "후"에서 다시 멈춤
# 4) 마지막 재개: END 도달
final = graph.invoke(Command(resume=True), config=config)
print(final) # {'x': 최종값}
- compile : interrupt_before=["node_a"]: node_a 실행 전에 일시 중지
interrupt_after=["node_b", "node_c"]: node_b, node_c 실행 이후에 각각 중지
node_a 실행 전에 멈추며, __interrupt__ 필드를 통해 인터럽트 메타데이터를 반환받습니다.
- 재개 흐름
- resume=True 호출로 실행이 재개됩니다.
- node_a가 실행된 후 node_b 직후에서 다시 멈춤(interrupt_after)
- __interrupt__에 재차 인터럽트 메타가 포함됩니다.
- 이후의 연속 재개 호출로 node_c 실행 후, 최종 상태(END)에 도달할 때까지 반복됩니다.
👉🏻 패턴
인터럽트와 명령어를 사용하여 구현할 수 있는 네 가지 대표적인 디자인 패턴이 있습니다:
Approve or reject, 승인 또는 거부 : API 호출과 같은 중요한 단계 전에 그래프를 일시 중지하여 작업을 검토하고 승인합니다. 작업이 거부되면 그래프가 해당 단계를 실행하지 못하도록 막고, 대체 조치를 취할 수 있습니다. 이 패턴은 종종 사용자의 입력에 따라 그래프를 라우팅하는 것을 포함합니다.
Edit graph state, 그래프 상태 편집 : 그래프를 일시 중지하여 그래프 상태를 검토하고 편집합니다. 실수를 수정하거나 추가 정보로 상태를 업데이트하는 데 유용합니다. 이 패턴은 종종 사람의 입력으로 상태를 업데이트하는 것을 포함합니다.
Review tool calls, 도구 호출 검토 : 도구 실행 전에 LLM이 요청한 도구 호출을 검토하고 편집하기 위해 그래프를 일시 중지합니다.
Validate human input, 사람 입력 검증 : 다음 단계로 진행하기 전에 사람의 입력을 검증하기 위해 그래프를 일시 중지합니다.
Review tool calls 예제
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command
# ---- 상태 정의 ----
class EmailState(TypedDict):
recipient: str
subject: str
body: str
action: Literal["approve", "edit", "feedback"] | None
updated_body: str | None
# ---- 노드 정의 ----
def email_planner_node(state: EmailState) -> EmailState:
# LLM 역할로 이메일 내용을 작성했다고 가정
return {
**state,
"recipient": "john@example.com",
"subject": "Meeting Reminder",
"body": "Don’t forget our meeting tomorrow at 2 PM.",
"action": None,
"updated_body": None,
}
def human_review_tool_call(state: EmailState) -> Command:
# 사람이 검토할 내용과 질문사항 전달
review = interrupt({
"question": "Review this email before sending",
"email": {
"to": state["recipient"],
"subject": state["subject"],
"body": state["body"],
}
})
action = review.get("action")
data = review.get("data")
# 사람 선택에 따라 다른 경로로 진행
if action == "approve":
return Command(goto="send_email", update={"action": "approve"})
if action == "edit":
return Command(
goto="send_email",
update={"action": "edit", "updated_body": data.get("body", state["body"])},
)
if action == "feedback":
return Command(goto="end", update={"action": "feedback"})
raise ValueError("Invalid review action")
def send_email_node(state: EmailState) -> EmailState:
if state["action"] == "approve":
final_body = state["body"]
elif state["action"] == "edit" and state["updated_body"]:
final_body = state["updated_body"]
else:
final_body = state["body"] # fallback
# 실제 이메일 발송 로직을 여기에 넣을 수 있습니다.
print(f"Sending email to {state['recipient']}")
print(f"Subject: {state['subject']}")
print(f"Body: {final_body}")
return {**state, "body": final_body}
# ---- 그래프 구성 ----
builder = StateGraph(EmailState)
builder.add_node("plan", email_planner_node)
builder.add_node("review", human_review_tool_call)
builder.add_node("send_email", send_email_node)
builder.add_edge(START, "plan")
builder.add_edge("plan", "review")
builder.add_edge("review", "send_email")
builder.add_edge("send_email", END)
graph = builder.compile(checkpointer=InMemorySaver())
# ---- 실행 흐름 ----
config = {"configurable": {"thread_id": "thread1"}}
initial = {"recipient": "", "subject": "", "body": "", "action": None, "updated_body": None}
# (1) 실행 → 인간 검토 포인트에서 중단
res = graph.invoke(initial, config=config)
print(res.get("__interrupt__", []))
# (2) 사람의 응답 (예: approve/edit/feedback 선택)
# 예: 수정 요청 시
resume_payload = {"action": "edit", "data": {"body": "Updated meeting time: 3 PM."}}
res2 = graph.invoke(Command(resume=resume_payload), config=config)
# (3) 이메일 전송까지 이어 실행
res3 = graph.invoke(Command(resume=True), config=config)
print("Final State:", res3)
HIL 패턴 중 가장 유용하다고 판단되는 Reveiw tool calls의 예제에 대해 알아보겠습니다.
이메일 내용을 자동으로 생성하는 그래프라고 가정할 때, 사람 리뷰로 interrupt()를 호출해 사람에게 이메일 내용을 보여주고, 승인 / 수정 / 피드백 입력을 받습니다.
조건 분기는 사람 선택에 따라 “발송”, “수정된 내용 적용 후 발송”, “피드백 후 종료” 등의 흐름을 지정하고 최종 이메일 콘텐츠로 도구 호출 실행합니다.
Command(resume={...}) 또는 단순 Command(resume=True)로 이어서 실행 (체크포인터를 통한 상태 유지) 방식으로 재개합니다.
마치며
오늘 포스팅에서는 LangGraph의 Human-in-the-Loop(HIL)에 대해 알아봤습니다. Human-in-the-Loop는 단순히 자동화에 인간 개입을 끼워 넣는 방식이 아니라, AI의 효율성과 인간의 판단력을 조화롭게 결합하는 지속 가능한 전략입니다.
'Deep Learning > LangGraph' 카테고리의 다른 글
| LangGraph - Persistence 영속성 (2) | 2025.08.31 |
|---|---|
| LangGraph - 개요 (1) | 2025.08.11 |
