[langgraph] 동적 중단 지점을 추가하는 방법

LangGraph 공식문서를 번역한 내용입니다. 필요한 경우 부연 설명을 추가하였고 이해하기 쉽게 예제를 일부 변경하였습니다. 문제가 되면 삭제하겠습니다.

https://langchain-ai.github.io/langgraph/how-tos/human_in_the_loop/dynamic_breakpoints/

사람이 개입하는(HIL) 상호작용은 에이전트 시스템에서 매우 중요하다. 브레이크포인트는 HIL 상호작용의 일반적인 패턴으로, 그래프가 특정 단계에서 멈추고 진행하기 전에 사용자 승인을 요청할 수 있도록 한다(예: 민감한 작업에 대해).

LangGraph에서는 노드가 실행되기 전이나 후에 중단 지점을 추가할 수 있다. 그러나 특정 조건에 따라 주어진 노드 내부에서 동적으로 그래프를 중단시키는 것이 유용할 때도 있다. 이 경우, 해당 중단이 발생한 이유에 대한 정보를 포함하는 것이 도움이 될 수 있다.

이 가이드는 NodeInterrupt를 사용하여 그래프를 동적으로 중단시키는 방법을 보여준다. NodeInterrupt는 노드 내부에서 발생시킬 수 있는 특별한 예외 케이스이다. 이를 실제로 어떻게 사용하는지 살펴보자.

준비

우선, 필요한 패키지를 설치하자.

pip install -U langgraph

그래프 정의

from typing_extensions import TypedDict
from IPython.display import Image, display

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.errors import NodeInterrupt


class State(TypedDict):
    input: str


def step_1(state: State) -> State:
    print("---Step 1---")
    return state


def step_2(state: State) -> State:
    # 입력이 5자보다 긴 경우 NodeInterrupt를 선택적으로 발생시킨다.
    if len(state["input"]) > 5:
        raise NodeInterrupt(
            f"Received input that is longer than 5 characters: {state['input']}"
        )

    print("---Step 2---")
    return state


def step_3(state: State) -> State:
    print("---Step 3---")
    return state


builder = StateGraph(State)
builder.add_node("step_1", step_1)
builder.add_node("step_2", step_2)
builder.add_node("step_3", step_3)

builder.add_edge(START, "step_1")
builder.add_edge("step_1", "step_2")
builder.add_edge("step_2", "step_3")
builder.add_edge("step_3", END)

memory = MemorySaver()

graph = builder.compile(checkpointer=memory)

display(
    Image(
        graph.get_graph().draw_mermaid_png(
            output_file_path="how-to-add-dynamic-breakpoints.png"
        )
    )
)

동적 Interrupt로 그래프 실행

먼저, 5자 이하의 입력값으로 그래프를 실행해 보자. 이는 우리가 정의한 중단 조건을 무시하고, 그래프 실행이 끝난 후 원래 입력값을 안전하게 반환해야 한다.

initial_input = {"input": "hello"}
thread_config = {"configurable": {"thread_id": "1"}}

for event in graph.stream(initial_input, thread_config, stream_mode="values"):
    print(event)

{'input': 'hello'}
---Step 1---
{'input': 'hello'}
---Step 2---
{'input': 'hello'}
---Step 3---
{'input': 'hello'}

이 시점에서 그래프를 확인해 보면 실행할 남은 작업이 없으며, 그래프가 실제로 실행을 완료했음을 알 수 있다.

state = graph.get_state(thread_config)
print(state.next)
print(state.tasks)

()
()

이제 5자보다 긴 입력값으로 그래프를 실행해 보자. 우리가 step_2 노드 내에서 NodeInterrupt 오류를 발생시켜 정의한 동적 인터럽트를 발생할 것이다.

initial_input = {"input": "hello world"}
thread_config = {"configurable": {"thread_id": "2"}}

for event in graph.stream(initial_input, thread_config, stream_mode="values"):
    print(event)

{'input': 'hello world'}
---Step 1---
{'input': 'hello world'}

여기서 그래프가 step_2를 실행하는 동안 중지된 것을 볼 수 있다. 이 시점에서 그래프 상태를 확인하면, 다음에 실행될 노드(step_2)와 인터럽트를 발생시킨 노드(step_2) 및 인터럽트에 대한 추가 정보가 포함된 것을 확인할 수 있다.

state = graph.get_state(thread_config)
print(state.next)
print(state.tasks)

('step_2',)
(PregelTask(id='365d4518-bcff-5abd-8ef5-8a0de7f510b0', name='step_2', error=None, interrupts=(Interrupt(value='Received input that is longer than 5 characters: hello world', when='during'),)),)

그래프를 중단점에서 재개하려고 하면, 입력과 그래프 상태가 변경되지 않았기 때문에 다시 인터럽트가 발생할 것이다.

for event in graph.stream(None, thread_config, stream_mode="values"):
    print(event)

state = graph.get_state(thread_config)
print(state.next)
print(state.tasks)

('step_2',)
(PregelTask(id='365d4518-bcff-5abd-8ef5-8a0de7f510b0', name='step_2', error=None, interrupts=(Interrupt(value='Received input that is longer than 5 characters: hello world', when='during'),)),)

그래프 상태 업데이트

이를 해결하기 위해 여러 가지 방법을 사용할 수 있다.

먼저, 처음 했던 것처럼 다른 스레드에서 그래프를 더 짧은 입력으로 실행할 수 있다. 또는 그래프 실행을 중단점에서 재개하려면, 상태를 업데이트하여 인터럽트 조건인 5자 미만의 입력을 넣을 수 있다.

graph.update_state(config=thread_config, values={"input": "foo"})
for event in graph.stream(None, thread_config, stream_mode="values"):
    print(event)

state = graph.get_state(thread_config)
print(state.next)
print(state.values)

---Step 2---
{'input': 'foo'}
---Step 3---
{'input': 'foo'}
()
{'input': 'foo'}

또한 상태를 업데이트하여 step_2(중단된 노드)를 건너뛰도록 할 수도 있다.

initial_input = {"input": "hello world"}
thread_config = {"configurable": {"thread_id": "3"}}

for event in graph.stream(initial_input, thread_config, stream_mode="values"):
    print(event)

{'input': 'hello world'}
---Step 1---
{'input': 'hello world'}

# NOTE: 여기서 step_2를 건너뛰도록 한다.
graph.update_state(config=thread_config, values=None, as_node="step_2")
for event in graph.stream(None, thread_config, stream_mode="values"):
    print(event)

state = graph.get_state(thread_config)
print(state.next)
print(state.values)

---Step 3---
{'input': 'hello world'}
()
{'input': 'hello world'}