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

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

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

# 方法
## インストール
LangGraphをインストールします。
```bash
pip install langgraph
```


## Graphの定義
まずはシンプルなGraphを定義します。
```python
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()))
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3968617/ee122089-c4aa-9794-5c25-8aa7f25c1209.png)


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

https://langchain-ai.github.io/langgraph/reference/types/#langgraph.types.interrupt

node_1とnode_2の間に`human_feedback`Nodeを追加します。

```python
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")
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3968617/774a162b-0d2d-673f-77d3-bb55bce47fa4.png)


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

```python
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")
```

出力
```bash
---Node 1---
{'node_1': None}


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

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

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

https://langchain-ai.github.io/langgraph/reference/types/#langgraph.types.Command

```python
from langgraph.types import Command

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

出力
```bash
---human_feedback---
{'human_feedback': {'user_feedback': '承認します'}}


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

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

<details><summary>全体のコード</summary>

グラフ定義
```python
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入力前まで
```python
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入力
```python
for event in graph.stream(
    Command(resume="承認します"), thread, stream_mode="updates"
):
    print(event)
    print("\n")
```

</details>

## スレッドIDが異なる場合
`thread_id`を`"1"`として実行します。
```python
thread = {"configurable": {"thread_id": "1"}}
```

その後、`thread_id`を`"2"`として続きから実行しようとしてみます。
```python
thread = {"configurable": {"thread_id": "2"}}

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

出力
```bash
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を組み込んでみて使用感を確かめてみたい。