DEV Community

Jaeyoun Nam
Jaeyoun Nam

Posted on

1

Langgraph - Human In the Loop

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"])
Enter fullscreen mode Exit fullscreen mode

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

Enter fullscreen mode Exit fullscreen mode

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)
Enter fullscreen mode Exit fullscreen mode

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

Image description

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)
Enter fullscreen mode Exit fullscreen mode

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

Image description

이외

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

Reference

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

Image of Timescale

🚀 pgai Vectorizer: SQLAlchemy and LiteLLM Make Vector Search Simple

We built pgai Vectorizer to simplify embedding management for AI applications—without needing a separate database or complex infrastructure. Since launch, developers have created over 3,000 vectorizers on Timescale Cloud, with many more self-hosted.

Read more →

Top comments (0)

Image of Docusign

🛠️ Bring your solution into Docusign. Reach over 1.6M customers.

Docusign is now extensible. Overcome challenges with disconnected products and inaccessible data by bringing your solutions into Docusign and publishing to 1.6M customers in the App Center.

Learn more