[번역][langgraph Contextual Guides] Human in the loop

langgraph의 공식문서를 번역해 놓은 자료입니다. 문제가 되면 삭제하겠습니다.

https://langchain-ai.github.io/langgraph/concepts/human_in_the_loop/

Human-in-the-loop(또는 "on-the-loop")는 여러 일반적인 사용자 상호작용 패턴을 통해 에이전트의 능력을 향상시킨다.

일반적인 상호작용 패턴에는 다음이 포함된다.

(1) 승인(Approval) - 에이전트의 작업을 중단하고 현재 상태를 사용자에게 표시하여 사용자가 작업을 승인하도록 할 수 있다.

(2) 수정(Editing) - 에이전트의 작업을 중단하고 현재 상태를 사용자에게 표시하여 사용자가 에이전트 상태를 수정할 수 있게 한다.

(3) 입력(Input) - 사용자 입력을 수집하기 위한 그래프 노드를 명시적으로 생성하고 해당 입력을 에이전트 상태에 직접 전달할 수 있다.

이러한 상호작용 패턴의 사용 사례는 다음과 같다.

(1) 도구 호출 검토 - 에이전트를 중단하여 도구 호출의 결과를 검토하고 수정할 수 있다.

(2) 시간 여행(Time Travel) - 에이전트의 과거 작업을 수동으로 다시 실행하거나 분기할 수 있다.

Persistence

이 모든 상호작용 패턴은 LangGraph의 내장된 영속성 계층에 의해 활성화된다. 이 계층은 각 단계에서 그래프 상태의 체크포인트를 기록한다. 영속성을 통해 그래프가 중단되어 사용자가 그래프의 현재 상태를 검토하거나 수정한 후, 해당 입력을 반영하여 다시 실행할 수 있다.

Breakpoints

그래프 흐름의 특정 위치에 브레이크포인트를 추가하는 것은 human-in-the-loop(사람이 개입하는 방식)를 구현하는 한 가지 방법이다. 이 경우, 개발자는 워크플로에서 사람의 입력이 필요한 위치를 알고 있으며, 그 특정 그래프 노드 전에 또는 그 후에 브레이크포인트를 배치하기만 하면 된다.

여기에서 우리는 그래프를 컴파일할 때, checkpointer와 함께 중단하고자 하는 노드 step_for_human_in_the_loop에 브레이크포인트를 추가한다. 그런 다음 위에서 설명한 상호작용 패턴 중 하나를 수행하며, 사람이 그래프 상태를 수정하면 새로운 체크포인트가 생성된다. 이 새로운 체크포인트는 스레드에 저장되며, 입력으로 None을 전달하여 그래프 실행을 다시 시작할 수 있다.

# Compile our graph with a checkpoitner and a breakpoint before "step_for_human_in_the_loop"
graph = builder.compile(checkpointer=checkpoitner, interrupt_before=["step_for_human_in_the_loop"])

# Run the graph up to the breakpoint
thread_config = {"configurable": {"thread_id": "1"}}
for event in graph.stream(inputs, thread_config, stream_mode="values"):
    print(event)

# Perform some action that requires human in the loop

# Continue the graph execution from the current checkpoint 
for event in graph.stream(None, thread_config, stream_mode="values"):
    print(event)

Dynamic Breakpoints

또한, 개발자는 브레이크포인트가 트리거되기 위한 조건을 정의할 수도 있다. 이러한 동적 브레이크포인트 개념은 특정 조건에서 그래프를 중단하고 싶을 때 유용하다. 이는 NodeInterrupt라는 특별한 유형의 예외를 사용하며, 해당 예외는 노드 내부에서 특정 조건에 따라 발생할 수 있다. 예를 들어, 입력이 5자보다 길 경우에 트리거되는 동적 브레이크포인트를 정의할 수 있다.

def my_node(state: State) -> State:
    if len(state['input']) > 5:
        raise NodeInterrupt(f"Received input that is longer than 5 characters: {state['input']}")
    return state

그래프를 동적 브레이크포인트를 트리거하는 입력으로 실행한 후, 입력에 None을 전달하여 그래프 실행을 다시 시작하려고 시도한다고 가정해보자.

# Attempt to continue the graph execution with no change to state after we hit the dynamic breakpoint 
for event in graph.stream(None, thread_config, stream_mode="values"):
    print(event)

그래프는 동일한 그래프 상태로 이 노드가 다시 실행되기 때문에 다시 중단된다. 따라서 동적 브레이크포인트를 트리거하는 조건이 더 이상 충족되지 않도록 그래프 상태를 변경해야 한다. 그래서 우리는 그래프 상태를 5자 미만의 조건을 충족하는 입력으로 간단히 수정하고 노드를 다시 실행할 수 있다.

# Update the state to pass the dynamic breakpoint
graph.update_state(config=thread_config, values={"input": "foo"})
for event in graph.stream(None, thread_config, stream_mode="values"):
    print(event)

또는 현재 입력을 유지하면서 체크를 수행하는 노드(my_node)를 건너뛰고 싶다면 어떻게 할까? 이를 위해 우리는 as_node="my_node"를 사용하여 그래프 업데이트를 수행하고, 값으로 None을 전달하면 된다. 이렇게 하면 그래프 상태에 대한 업데이트는 이루어지지 않지만, my_node로 업데이트를 실행하여 노드를 건너뛰고 동적 브레이크포인트를 우회할 수 있다.

# This update will skip the node `my_node` altogether
graph.update_state(config=thread_config, values=None, as_node="my_node")
for event in graph.stream(None, thread_config, stream_mode="values"):
    print(event)

자세한 방법은 가이드를 참조하자.

Interaction Patterns

Approval

때때로 우리는 에이전트 실행의 특정 단계를 승인하고 싶다.

우리는 승인하고자 하는 단계 이전의 브레이크포인트에서 에이전트를 중단할 수 있다.

이것은 일반적으로 민감한 작업(예: 외부 API 사용 또는 데이터베이스에 쓰기)에 권장다.

영속성을 통해 현재 에이전트 상태와 다음 단계를 사용자에게 검토 및 승인을 위해 표시할 수 있다.

승인되면 그래프는 마지막으로 저장된 체크포인트에서 실행을 재개하며, 이 체크포인트는 스레드에 저장된다.

# Compile our graph with a checkpoitner and a breakpoint before the step to approve
graph = builder.compile(checkpointer=checkpoitner, interrupt_before=["node_2"])

# Run the graph up to the breakpoint
for event in graph.stream(inputs, thread, stream_mode="values"):
    print(event)

# ... Get human approval ...

# If approved, continue the graph execution from the last saved checkpoint
for event in graph.stream(None, thread, stream_mode="values"):
    print(event)

자세한 방법은 가이드를 참조하자.

Editing

때때로 우리는 에이전트의 상태를 검토하고 수정하고 싶다.

승인과 마찬가지로, 확인하고자 하는 단계 이전의 브레이크포인트에서 에이전트를 중단할 수 있다.

현재 상태를 사용자에게 표시하고 사용자가 에이전트 상태를 수정할 수 있도록 할 수 있다.

예를 들어, 에이전트가 실수를 했을 경우(예: 아래 도구 호출 섹션 참조) 이를 수정하는 데 사용할 수 있다.

현재 체크포인트를 분기하여 그래프 상태를 수정할 수 있으며, 이 체크포인트는 스레드에 저장된다.

그런 다음 이전에 수행한 것처럼 분기된 체크포인트에서 그래프를 계속 진행할 수 있다.

# Compile our graph with a checkpoitner and a breakpoint before the step to review
graph = builder.compile(checkpointer=checkpoitner, interrupt_before=["node_2"])

# Run the graph up to the breakpoint
for event in graph.stream(inputs, thread, stream_mode="values"):
    print(event)

# Review the state, decide to edit it, and create a forked checkpoint with the new state
graph.update_state(thread, {"state": "new state"})

# Continue the graph execution from the forked checkpoint
for event in graph.stream(None, thread, stream_mode="values"):
    print(event)

자세한 방법은 가이드를 참조하자.

Input

때때로 우리는 그래프의 특정 단계에서 명시적으로 인간 입력을 받아야 할 때가 있다.

이를 위해 전용 그래프 노드를 생성할 수 있다(예: 예제 다이어그램의 human_input).

승인 및 수정과 마찬가지로, 이 노드 이전의 브레이크포인트에서 에이전트를 중단할 수 있다.

그런 다음 인간 입력을 포함하는 상태 업데이트를 수행할 수 있으며, 이는 상태 수정 시와 동일하다.

하지만 한 가지를 추가한다.

상태 업데이트와 함께 as_node=human_input을 사용하여 상태 업데이트가 노드로 처리되어야 함을 지정할 수 있다.

이 점은 미묘하지만 중요하다.

수정할 때 사용자는 그래프 상태를 수정할지 여부에 대한 결정을 내린다.

하지만 입력을 받을 때는 인간 입력을 수집하기 위해 그래프에 노드를 명시적으로 정의한다.

그러면 인간 입력이 포함된 상태 업데이트가 이 노드로 실행된다.

# Compile our graph with a checkpoitner and a breakpoint before the step to to collect human input
graph = builder.compile(checkpointer=checkpoitner, interrupt_before=["human_input"])

# Run the graph up to the breakpoint
for event in graph.stream(inputs, thread, stream_mode="values"):
    print(event)

# Update the state with the user input as if it was the human_input node
graph.update_state(thread, {"user_input": user_input}, as_node="human_input")

# Continue the graph execution from the checkpoint created by the human_input node
for event in graph.stream(None, thread, stream_mode="values"):
    print(event)

자세한 방법은 가이드를 참조하자.

Use-cases

Tool-calling 리뷰

일부 사용자 상호작용 패턴은 위의 아이디어를 결합한다.

예를 들어, 많은 에이전트는 결정을 내리기 위해 도구 호출을 사용한다.

도구 호출은 두 가지를 올바르게 수행해야 하기 때문에 도전 과제를 제시한다.

(1) 호출할 도구의 이름

(2) 도구에 전달할 인수

도구 호출이 올바르더라도 우리는 여전히 재량을 적용하고 싶을 수 있다.

(3) 도구 호출이 승인해야 하는 민감한 작업일 수 있다.

이러한 점을 염두에 두고, 우리는 위의 아이디어를 결합하여 도구 호출에 대한 human-in-the-loop 검토를 생성할 수 있다.

# Compile our graph with a checkpoitner and a breakpoint before the step to to review the tool call from the LLM 
graph = builder.compile(checkpointer=checkpoitner, interrupt_before=["human_review"])

# Run the graph up to the breakpoint
for event in graph.stream(inputs, thread, stream_mode="values"):
    print(event)

# Review the tool call and update it, if needed, as the human_review node
graph.update_state(thread, {"tool_call": "updated tool call"}, as_node="human_review")

# Otherwise, approve the tool call and proceed with the graph execution with no edits 

# Continue the graph execution from either: 
# (1) the forked checkpoint created by human_review or 
# (2) the checkpoint saved when the tool call was originally made (no edits in human_review)
for event in graph.stream(None, thread, stream_mode="values"):
    print(event)

자세한 방법은 가이드를 참조하자.

Time Travel

에이전트와 작업할 때, 우리는 종종 그들의 의사 결정 과정을 면밀히 검토하고 싶어한다.

(1) 원하는 최종 결과에 도달하더라도, 그 결과에 이르게 한 추론은 종종 중요하게 검토해야 한다.

(2) 에이전트가 실수를 할 경우, 그 이유를 이해하는 것이 종종 유용하다.

(3) 위의 두 경우 모두에서, 대안적인 의사 결정 경로를 수동으로 탐색하는 것이 유용하다.

이러한 개념을 통틀어 우리는 디버깅 개념인 시간 여행(time-travel)이라고 부르며, 이는 재생(replaying)과 분기(forking)로 구성된다.

replaying

때때로 우리는 에이전트의 과거 행동을 단순히 재생하고 싶다.

위에서는 그래프의 현재 상태(또는 체크포인트)에서 에이전트를 실행하는 경우를 보여주었다.

이는 스레드와 함께 입력으로 None을 전달함으로써 수행할 수 있다.

thread = {"configurable": {"thread_id": "1"}}
for event in graph.stream(None, thread, stream_mode="values"):
    print(event)

이제 우리는 체크포인트 ID를 전달하여 특정 체크포인트에서 과거 행동을 재생하도록 이를 수정할 수 있다.

특정 체크포인트 ID를 얻으려면, 스레드에 있는 모든 체크포인트를 쉽게 가져오고 우리가 원하는 체크포인트로 필터링할 수 있다.

all_checkpoints = []
for state in app.get_state_history(thread):
    all_checkpoints.append(state)

각 체크포인트는 고유한 ID를 가지며, 이를 사용하여 특정 체크포인트에서 재생할 수 있다.

체크포인트를 검토한 결과, xxx라는 체크포인트에서 재생하고 싶다고 가정해보자.

그래프를 실행할 때 체크포인트 ID를 전달하기만 하면 된다.

config = {'configurable': {'thread_id': '1', 'checkpoint_id': 'xxx'}}
for event in graph.stream(None, config, stream_mode="values"):
    print(event)

중요하게도, 그래프는 이전에 실행된 체크포인트를 알고 있다.

따라서, 그래프는 다시 실행하는 대신 이전에 실행된 노드를 재생한다.

재생에 대한 관련 컨텍스트는 이 추가 개념 가이드를 참조하자.

시간 여행을 수행하는 방법에 대한 자세한 내용은 이 가이드를 참조하자.

Forking

때때로 우리는 에이전트의 과거 행동을 분기하고 그래프를 통해 다른 경로를 탐색하고 싶다.

위에서 논의한 대로, 수정은 현재 그래프 상태에서 이를 수행하는 방법이다.

하지만 그래프의 과거 상태를 분기하고 싶다면 어떻게 할까?

예를 들어, xxx라는 특정 체크포인트를 수정하고 싶다고 가정해보자.

그래프의 상태를 업데이트할 때 이 checkpoint_id를 전달하면 된다.

config = {"configurable": {"thread_id": "1", "checkpoint_id": "xxx"}}
graph.update_state(config, {"state": "updated state"}, )

이렇게 하면 새로운 분기된 체크포인트인 xxx-fork가 생성되며, 우리는 이 체크포인트에서 그래프를 실행할 수 있다.

config = {'configurable': {'thread_id': '1', 'checkpoint_id': 'xxx-fork'}}
for event in graph.stream(None, config, stream_mode="values"):
    print(event)

분기에 대한 관련 컨텍스트는 이 추가 개념 가이드를 참조하자.

시간 여행을 수행하는 방법에 대한 자세한 내용은 이 가이드를 참조하자.