LangGraphの(LLMなし)Human-in-the-loopを試してみた

Human-in-the-loopとは

ワークフロー内に人間の承認や判断を組み込んで、重要な段階での決定、検証、または修正を可能にします。
エラー許容度の低いシナリオ(コンプライアンス、意思決定、etc)で、人間の関与により修正が可能になり、信頼性が担保されます。

方法

インストール

LangGraphをインストールします。

pip install langgraph

Graphの定義

まずはシンプルなGraphを定義します。

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from IPython.display import Image, display


class State(TypedDict):
    input: str


def node_1(state):
    print("---Node 1---")
    pass


def node_2(state):
    print("---Node 2---")
    pass


builder = StateGraph(State)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
builder.add_edge("node_2", END)

# Set up memory
memory = MemorySaver()

# Add
graph = builder.compile(checkpointer=memory)

# View
display(Image(graph.get_graph().draw_mermaid_png()))

Human-in-the-loopの導入

Node内でinterrupt()を使用することで、グラフの実行を一時停止し、値をクライアントに表示することで、人間が関与するワークフローを可能となります。

node_1とnode_2の間にhuman_feedbackNodeを追加します。

from langgraph.types import interrupt

class State(TypedDict):
    input: str
    user_feedback: str

def human_feedback(state):
    print("---human_feedback---")
    feedback = interrupt("承認しますか？:")
    return {"user_feedback": feedback}

builder.add_node("human_feedback", human_feedback)

builder.add_edge("human_feedback", "node_2")

実行

stream()の引数configで、スレッドIDを定義して実行します。
同一のスレッドIDでないと、human_feedbackNodeが実行されなくなります。

input = {"input": "Hello World"}
thread_config = {"configurable": {"thread_id": "1"}}

for event in graph.stream(input, config=thread, stream_mode="updates"):
    print(event)
    print("\n")

出力

---Node 1---
{'node_1': None}


---human_feedback---
{'__interrupt__': (Interrupt(value='承認しますか？:', resumable=True, ns=['human_feedback:69678c31-8166-d3e8-fb54-630d050e6311'], when='during'),)}

human_feedbackNodeのinterruptのところでグラフが一時中断していることが分かります。

次に人間の入力を行います。
その際は、Commandを使用します。
これはNodeにメッセージを送信するためのものです。
Commandのパラメータresumeに入力内容を記載して、グラフを実行します。

from langgraph.types import Command

for event in graph.stream(
    Command(resume="承認します"), config=thread, stream_mode="updates"
):
    print(event)
    print("\n")

出力

---human_feedback---
{'human_feedback': {'user_feedback': '承認します'}}


---Node 2---
{'node_2': None}

グラフの最初からではなく、続きのnode_2が実行されました。

全体のコード

グラフ定義

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command, interrupt
from IPython.display import Image, display


class State(TypedDict):
    input: str
    user_feedback: str


def node_1(state):
    print("---Node 1---")
    pass

def human_feedback(state):
    print("---human_feedback---")
    feedback = interrupt("承認しますか？:")
    return {"user_feedback": feedback}

def node_2(state):
    print("---Node 2---")
    pass


builder = StateGraph(State)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("human_feedback", human_feedback)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "human_feedback")
builder.add_edge("human_feedback", "node_2")
builder.add_edge("node_2", END)

memory = MemorySaver()

graph = builder.compile(checkpointer=memory)

display(Image(graph.get_graph().draw_mermaid_png()))

human_feedback入力前まで

input = {"input": "Hello World"}

thread = {"configurable": {"thread_id": "1"}}

for event in graph.stream(input, config=thread, stream_mode="updates"):
    print(event)
    print("\n")

human_feedback入力

for event in graph.stream(
    Command(resume="承認します"), thread, stream_mode="updates"
):
    print(event)
    print("\n")

スレッドIDが異なる場合

thread_idを"1"として実行します。

thread = {"configurable": {"thread_id": "1"}}

その後、thread_idを"2"として続きから実行しようとしてみます。

thread = {"configurable": {"thread_id": "2"}}

for event in graph.stream(
    Command(resume="承認します"), config=thread, stream_mode="updates"
):
    print(event)
    print("\n")

出力

InvalidUpdateError: Must write to at least one of ['input', 'user_feedback']

定義したStateの値を入力してくださいとエラーになりました。
スレッドIDを同一にしないと、続きから実行されることはないようです。

おわりに

LangGraphのHuman-in-the-loopを試してみました。
Human-in-the-loopは、人間のチェックや承認が必要な確実性を高めたいAgentシステムには重要な機能だと思いました。
今後はLLMを導入したAgentシステム内に、Human-in-the-loopを組み込んでみて使用感を確かめてみたい。