Agent system 을 만들다보면 자동으로 결과까지 완벽하게 돌아가주면 좋겠지만 지금의 정확도로는 힘들다. 만약 전체 과정을 Agent가 생각하여 진행하도록 한다면, 중간에 한번의 실수만 있어도 의도와는 매우 다른 결과를 얻게 될 것이다.
그리고 Agent에게 권한을 전적으로 위임하기엔 무서운 오퍼레이션들도 있다. API콜이나, DB 수정, Bash 스크립팅등... 이 있다.

그렇기에 지금(어쩌면 꽤 오랫동안)은 Agent의 수행과정 중 인간의 간섭은 필연적이다.

Langgraph의 Human-in-the-loop

인간이 간섭하는 5가지의 상황을 지원한다.

Approval

다음 실행을 할 것인지 판단을 인간에게 맡김.

Editing

현재 상태를 유저한테 보여주고 수정할 수 있도록 함.

Input

User Input을 받는 단계를 명시하고 실제로 유저한테 값을 받는 것.

Reviewing tool calls

유저가 Tool의 결과 값을 보고 수정할 수 있도록 함.

Time travel

이전 상태(노드 수행전)로 돌아가던가, 돌아가서 다른 진행을 할 수 있도록 하는 것. Like multi-verse.

Persistant Layer

이런 인간의 간섭이 가능하게 하는 건 Langgraph의 persistant layer 덕이다. 인간의 간섭이 필요한 상태를 저장해놨다가 유저의 승인, 수정이 끝나면 다시 그 지점부터 시작이 가능하다. 게임에서 체크포인트 같은 개념이다.

Breakpoint

디버깅 툴에서 Breakpoint 설정하는 것과 같이 breakpoint는 그 지점까지만 신나게 수행하고 잠시 멈추라는 표시이다.

langgraph에서는 graph를 컴파일할때 break point를 명시할 수 있다.

graph = builder.compile(checkpointer=checkpointer, interrupt_before=["step_for_human_in_the_loop"])

Dynamic Breakpoint

컴파일 할 때 선언하는 것은 정적이다. 실행 중에 변경되는 state에 따라서 멈추는 것이 어렵다. Dynamic Breakpoint는 state에 따라서 설정할 수 있는 breakpoint이다. NodeInterrupt라는 특별한 exception이 발생하면 그래프 수행을 멈추고 유저 간섭을 기다린다.

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

Patterns

그림을 보면 좀 더 쉽다.

Approval

# 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)

node_2 전에 그래프 수행이 끝나서 첫번째 for 구문을 탈출하게 되고, 그 후에 다시 graph.stream을 호출해서 이어서 수행한다.

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)

역시 node_2 앞에서 멈추고, 이번에는 아무것도 안하고 다시 graph.stream을 호출하는 대신, 현재 그래프의 state를 변경한다. 그 후에 변경된 state에서 다시 그래프를 수행한다.

이외

이외의 동작들은 아래 랭그래프 공식 도큐먼트에서 확인하자

Reference

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