[langgraph] Human-in-the-loop

원문 출처: https://langchain-ai.github.io/langgraph/agents/human-in-the-loop/

에이전트의 도구 호출을 검토, 수정, 승인하려면 LangGraph의 내장 Human-In-the-Loop(HIL) 기능, 특히 interrupt() 프리미티브를 사용할 수 있습니다.

LangGraph는 실행을 무기한(수분, 수시간, 심지어 수일) 동안 일시정지하고, 인간 입력이 들어올 때까지 대기할 수 있습니다.

이것이 가능한 이유는 에이전트 상태가 데이터베이스에 체크포인트되어, 실행 컨텍스트가 영속적으로 저장되고, 나중에 중단된 지점부터 워크플로우를 재개할 수 있기 때문입니다.

Human-in-the-loop 개념에 대한 더 깊은 설명은 개념 가이드를 참고하세요.

human-in-the-loop

인간은 에이전트의 출력(특히 도구 호출 결과)을 검토 및 편집한 후 다음 단계로 진행할 수 있습니다. 이는 도구 호출이 민감하거나 인간의 감독이 필요한 애플리케이션에서 매우 중요합니다.

도구 호출 검토(Review tool calls)

도구에 인간 승인 단계를 추가하려면:

도구 내에서 interrupt()를 사용해 실행을 일시정지합니다.
인간 입력에 따라 Command(resume=...)로 재개합니다.

API Reference: InMemorySaver | interrupt | create_react_agent

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt
from langgraph.prebuilt import create_react_agent

# 민감한 도구 예시 (인간 검토 필요)
def book_hotel(hotel_name: str):
    """호텔 예약"""
    response = interrupt(  # (1)!
        f"Trying to call `book_hotel` with args {{'hotel_name': {hotel_name}}}. "
        "Please approve or suggest edits."
    )
    if response["type"] == "accept":
        pass
    elif response["type"] == "edit":
        hotel_name = response["args"]["hotel_name"]
    else:
        raise ValueError(f"Unknown response type: {response['type']}")
    return f"Successfully booked a stay at {hotel_name}."

checkpointer = InMemorySaver() # (2)!

agent = create_react_agent(
    model="anthropic:claude-3-5-sonnet-latest",
    tools=[book_hotel],
    checkpointer=checkpointer, # (3)!
)

interrupt()는 에이전트 그래프를 특정 노드에서 일시정지합니다. 이 예시에서는 도구 함수 시작 부분에서 호출하여, 도구 실행 노드에서 그래프가 멈춥니다. interrupt()에 전달된 정보(예: 도구 호출)는 인간에게 보여줄 수 있으며, 인간 입력(승인/수정/피드백)으로 그래프를 재개할 수 있습니다.
InMemorySaver는 각 단계의 에이전트 상태를 메모리에 저장합니다. 실제 서비스에서는 데이터베이스에 저장합니다.
에이전트 생성 시 checkpointer를 전달합니다.

에이전트는 stream() 메서드로 실행하며, config에 thread ID를 지정합니다. 이렇게 하면 이후에도 같은 대화를 이어갈 수 있습니다.

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

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "book a stay at McKittrick hotel"}]},
    config
):
    print(chunk)
    print("\n")

위 코드를 실행하면, 에이전트가 interrupt()에 도달할 때까지 실행되고, 그 시점에서 인간 입력을 기다리며 일시정지합니다.

인간 입력에 따라 Command(resume=...)로 그래프를 재개할 수 있습니다.

API Reference: Command

from langgraph.types import Command

for chunk in agent.stream(
    Command(resume={"type": "accept"}),  # (1)!
    # Command(resume={"type": "edit", "args": {"hotel_name": "McKittrick Hotel"}}),
    config
):
    print(chunk)
    print("\n")

interrupt()는 Command 객체와 함께 사용되어, 인간이 제공한 값으로 그래프를 재개합니다.

Agent Inbox와 함께 사용하기(Using with Agent Inbox)

어떤 도구에도 interrupt를 쉽게 추가할 수 있는 래퍼를 만들 수 있습니다.

아래 예시는 Agent Inbox UI 및 Agent Chat UI와 호환되는 참조 구현입니다.

Human-in-the-loop 래퍼 예시

from typing import Callable
from langchain_core.tools import BaseTool, tool as create_tool
from langchain_core.runnables import RunnableConfig
from langgraph.types import interrupt 
from langgraph.prebuilt.interrupt import HumanInterruptConfig, HumanInterrupt

def add_human_in_the_loop(
    tool: Callable | BaseTool,
    *,
    interrupt_config: HumanInterruptConfig = None,
) -> BaseTool:
    """도구에 human-in-the-loop 검토를 추가하는 래퍼.""" 
    if not isinstance(tool, BaseTool):
        tool = create_tool(tool)

    if interrupt_config is None:
        interrupt_config = {
            "allow_accept": True,
            "allow_edit": True,
            "allow_respond": True,
        }

    @create_tool(
        tool.name,
        description=tool.description,
        args_schema=tool.args_schema
    )
    def call_tool_with_interrupt(config: RunnableConfig, **tool_input):
        request: HumanInterrupt = {
            "action_request": {
                "action": tool.name,
                "args": tool_input
            },
            "config": interrupt_config,
            "description": "Please review the tool call"
        }
        response = interrupt([request])[0]
        # 승인
        if response["type"] == "accept":
            tool_response = tool.invoke(tool_input, config)
        # 인자 수정
        elif response["type"] == "edit":
            tool_input = response["args"]["args"]
            tool_response = tool.invoke(tool_input, config)
        # 사용자 피드백 전달
        elif response["type"] == "response":
            user_feedback = response["args"]
            tool_response = user_feedback
        else:
            raise ValueError(f"Unsupported interrupt response type: {response['type']}")

        return tool_response

    return call_tool_with_interrupt

이 래퍼는 도구 실행 이전에 interrupt()를 호출하는 새 도구를 만듭니다.
interrupt()는 Agent Inbox UI에서 기대하는 특수 입력/출력 포맷을 사용합니다.

add_human_in_the_loop로 어떤 도구에도 interrupt()를 쉽게 추가할 수 있습니다.

API Reference: InMemorySaver | create_react_agent

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.prebuilt import create_react_agent

checkpointer = InMemorySaver()

def book_hotel(hotel_name: str):
   """호텔 예약"""
   return f"Successfully booked a stay at {hotel_name}."

agent = create_react_agent(
    model="anthropic:claude-3-5-sonnet-latest",
    tools=[
        add_human_in_the_loop(book_hotel), # (1)!
    ],
    checkpointer=checkpointer,
)

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

# 에이전트 실행
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "book a stay at McKittrick hotel"}]},
    config
):
    print(chunk)
    print("\n")

add_human_in_the_loop 래퍼를 사용해 도구에 interrupt()를 추가합니다. 이로써 도구 호출 전 실행이 일시정지되고, 인간 입력을 기다립니다.

위 코드를 실행하면, 에이전트가 interrupt()에 도달할 때까지 실행되고, 그 시점에서 인간 입력을 기다리며 일시정지합니다.

인간 입력에 따라 Command(resume=...)로 그래프를 재개할 수 있습니다.

API Reference: Command

from langgraph.types import Command 

for chunk in agent.stream(
    Command(resume=[{"type": "accept"}]),
    # Command(resume=[{"type": "edit", "args": {"args": {"hotel_name": "McKittrick Hotel"}}}]),
    config
):
    print(chunk)
    print("\n")