Langgraph - Human In the Loop

When creating an agent system, it would be nice to automatically return the results perfectly, but it is difficult with the current accuracy. If you let the Agent think through the entire process, even one mistake will result in a result very different from what you intended.
And there are operations that are too scary to fully delegate authority to the Agent. There are API calls, DB modification, Bash scripting, etc.

So now (maybe for quite some time), human intervention during the agent's execution process is inevitable.

Human-in-the-loop by Langgraph

Supports 5 situations where humans interfere.

Approval

Leaves the decision on whether to proceed next to a human.

Editing

Shows the current status to the user and allows modification.

Input

Specifying the steps to receive user input and actually receiving value from the user.

Reviewing tool calls

Allow users to view and modify the result value of the tool.

Time travel

Return to the previous state (before node execution) or go back and proceed again. Like multi-verse.

Persistent Layer

This kind of human intervention is possible thanks to Langgraph’s persistent layer. It is possible to save a state that requires human intervention and start again from that point once the user has approved or modified it. It's like a checkpoint in a game.

Breakpoint

Like setting a breakpoint in a debugging tool, a breakpoint is an indication to proceed only up to that point and stop for a moment.

In langgraph, break points can be specified when compiling the graph.

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

Dynamic Breakpoint

Declarations at compile time are static. It is difficult to stop depending on the state changing during execution. Dynamic Breakpoint is a breakpoint that can be set depending on the state. When a special exception called NodeInterrupt occurs, graph execution stops and waits for user intervention.

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

It's easier if you look at the picture.

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)

The graph execution is completed before node_2, so the first for statement is escaped, and then graph.stream is called again to continue execution.

Editing

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

It also stops in front of node_2, and this time, instead of doing nothing and calling graph.stream again, it changes the state of the current graph. After that, the graph is performed again in the changed state.

Other than

For other operations, check the Langgraph official document below

Reference

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

The above is the detailed content of Langgraph - Human In the Loop. For more information, please follow other related articles on the PHP Chinese website!