【Langgraph】interrupt機能を使ってChainlitにHuman-in-the-loopを組み込む方法

## はじめに
この記事では、Chainlit上でLangGraphの `interrupt` 関数を利用したHuman-in-the-loopの実装方法についてまとめます。

LLMがツールを呼んだあと、それを実行するか、引数を修正して実行するか、実行を拒否するかを選択できるような機構を作成しました。ここでは、LLMにある都市の天気を返すような `weather_search` というツールを持たせ、Human-in-the-loop により、ツールの実行の許可や拒否が可能なアプリケーションを実装しました。

以下はツールの実行を許可する例です。

![chainlit_hitl_01.gif](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2122187/9ec3354c-ddb4-4883-9afa-e7c40288f3c7.gif)

以下はツールの実行を拒否し、追加の質問文を入力する例です。

![chainlit_hitl_03.gif](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2122187/5f26f742-5b38-4c2c-bf66-d740b48480c8.gif)

下記リポジトリでchainlit上の動作を確認できます。

https://github.com/kossyprg/llm-my-tutorial/tree/main/my-app/human-in-the-loop

## 対象読者
- :white_check_mark: LangGraphの `interrupt()` を用いたHuman-in-the-loopの実装方法について知りたい方
- :white_check_mark: Human-in-the-loopを導入したAIチャットボットをChainlitで試作したい方

この記事では、以下の内容については説明を省略します。

- :x: LangChain、LangGraph、Chainlitの基礎的な内容
- :x: LangChainにおける tool-calling の仕組み

## 環境
この記事の執筆時点(2025/03/01)では、以下のバージョンで実装しています。

- langgraph: 0.3.0
- chainlit: 2.2.1
- Python: 3.11.9

他のライブラリについてはリポジトリの `requirements.lock` を参照してください。

https://github.com/kossyprg/llm-my-tutorial/blob/main/my-app/human-in-the-loop/requirements.lock

## そもそも Human-in-the-loop とは

**Human-in-the-loop（HITL）** とは、端的に言えば「AIの判断に人間がかかわる仕組み」のことです。この記事で扱うチャットボットでは、LLMが実行しようとしたツールについて、その実行を許可するか否かを人間が判定します。

理想的には、AIがすべてのプロセスを処理できればよいのですが、現実的にはAIの精度を100%にすることは困難です。そこで、重要なプロセスには人の判断を挟むことで、より信頼度の高い出力にしようというのがHITLというアイデアです。また、モデルそのものの推論精度が十分に高くなくても、HITLで精度を補填することで早期に運用を始められるというメリットも考えられます[^1]。

## 今回扱うグラフ
HITLが導入されたグラフは、一般的なReActエージェント[^2]のグラフと比較するとわかりやすいので、始めにそちらについて言及します。

下図は[一般的なReActエージェントのグラフ](https://langchain-ai.github.io/langgraph/how-tos/create-react-agent/)です。グラフが実行されると最初に `agent` ノードへ到達します。このノードでは、LLMがユーザの質問文と渡されたツールの情報を参照して、ツールの実行が必要かどうかを判定します。もし必要ならば `tools` ノードへ遷移し、ツールが実行されます。必要なければ `__end__` ノードへ到達し、最終的な応答が生成されます。このグラフでは、ツールの必要性をLLMが判定し、人間はその判断に介入できません。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2122187/ed6fd3bf-cba9-405b-98bc-b7397b0be0d0.png)

次にHITLの仕組みが導入されたグラフを下図に示します。グラフは [LangGraphの公式チュートリアル](https://langchain-ai.github.io/langgraph/how-tos/human_in_the_loop/review-tool-calls/) にあるものを使用します（後述の通り、`human_review_node` ノードの内部の実装については独自に修正を加えます）。グラフが実行されると最初に `call_llm` ノードへ到達します。このノードは前述の `agent` ノードと同様、LLMがユーザの質問文と渡されたツールの情報を参照して、ツールの実行が必要かどうかを判定します。もし必要ならば `human_review_node` ノードへ遷移します。先ほどのグラフではすぐにツールが実行されましたが、このグラフでは文字通り、人間によるレビューが入ります。

![graph.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2122187/3552936a-a853-4486-9427-510ce963271c.png)

今回は3通りのレビュー方法を実装します。

- ツールの実行を許可する（`run_tool` へ遷移）
- ツールの引数を修正した上で実行を許可する（`run_tool` へ遷移）
- ツールの実行を拒否する（`call_llm` へ遷移）

1つ目は、LLMの判断に問題がないので、そのままツールを実行するパターンです。2つ目は、ツールを実行するという判断そのものは正しいものの、実行の仕方に問題があるというパターンです。この場合は、ユーザから追加の入力を受け付けて修正したのちにツールを実行します。3つ目は、ツールの実行を完全に拒否するパターンです。LLMのツールを実行するという判断そのものに問題がある場合に相当します。

## 実装（コマンドラインベース）

Chainlitを活用したアプリケーションの実装方法に移る前に、コマンドラインベースでHITLの動作を確認します。実装は [LangGraphの公式チュートリアル](https://langchain-ai.github.io/langgraph/how-tos/human_in_the_loop/review-tool-calls/) を流用しています。LLMに渡すツールとして、都市の名前 `city` を受け取り、単に `Sunny!` とだけ返す `weather_search` を用意します。

詳細は、Google Colab上で動作する ipynb ファイルを参照してください。

https://github.com/kossyprg/llm-my-tutorial/blob/main/my-app/human-in-the-loop/20250301_HITL.ipynb

<details><summary>グラフを作成するコード全体</summary>

```python:create_graph()
from typing_extensions import Literal
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command, interrupt
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage
from langchain.schema.runnable.config import RunnableConfig
from langchain_core.messages import RemoveMessage

def create_graph():
    @tool
    def weather_search(city: str):
        """Search for the weather"""
        print("----")
        print(f"Searching for: {city}")
        print("----")
        return "Sunny!"

    model = ChatOpenAI(model_name="gpt-4o").bind_tools(
        [weather_search]
    )

    class State(MessagesState):
        """Simple state."""

    def call_llm(state):
        return {"messages": [model.invoke(state["messages"])]}

    def human_review_node(state) -> Command[Literal["call_llm", "run_tool"]]:
        last_message = state["messages"][-1]
        tool_call = last_message.tool_calls[-1]

        # this is the value we'll be providing via Command(resume=<human_review>)
        human_review = interrupt(
            {
                "question": "Is this correct?",
                # Surface tool calls for review
                "tool_call": tool_call,
            }
        )

        review_action = human_review["action"]
        review_data = human_review.get("data")

        # if approved, call the tool
        if review_action == "continue":
            return Command(goto="run_tool")

        # update the AI message AND call tools
        elif review_action == "update":
            updated_message = {
                "role": "ai",
                "content": last_message.content,
                "tool_calls": [
                    {
                        "id": tool_call["id"],
                        "name": tool_call["name"],
                        # This the update provided by the human
                        "args": review_data,
                    }
                ],
                # This is important - this needs to be the same as the message you replacing!
                # Otherwise, it will show up as a separate message
                "id": last_message.id,
            }
            return Command(goto="run_tool", update={"messages": [updated_message]})

        # ツールの実行を拒否するように実装を変更
        elif review_action == "cancel":
            # Ref https://langchain-ai.github.io/langgraph/how-tos/memory/delete-messages/
            return Command(goto="call_llm", update={"messages": [RemoveMessage(id=last_message.id), HumanMessage(content=review_data)]})

    def run_tool(state):
        new_messages = []
        tools = {"weather_search": weather_search}
        tool_calls = state["messages"][-1].tool_calls
        for tool_call in tool_calls:
            tool = tools[tool_call["name"]]
            result = tool.invoke(tool_call["args"])
            new_messages.append(
                {
                    "role": "tool",
                    "name": tool_call["name"],
                    "content": result,
                    "tool_call_id": tool_call["id"],
                }
            )
        return {"messages": new_messages}


    def route_after_llm(state) -> Literal[END, "human_review_node"]:
        if len(state["messages"][-1].tool_calls) == 0:
            return END
        else:
            return "human_review_node"


    builder = StateGraph(State)
    builder.add_node(call_llm)
    builder.add_node(run_tool)
    builder.add_node(human_review_node)
    builder.add_edge(START, "call_llm")
    builder.add_conditional_edges("call_llm", route_after_llm)
    builder.add_edge("run_tool", "call_llm")

    # Set up memory
    memory = MemorySaver()

    # Add
    graph = builder.compile(checkpointer=memory)
    return graph
```
</details>

### interrupt 関数を使ったhuman-in-the-loopの実装

`interrupt` 関数は、LangGraphにおいて特定のノードでグラフの実行を一時停止し、人間による操作（承認や編集、追加情報の入力など）を受け取った後に再開するための関数です。この関数を使用することで、途中でユーザーや管理者が確認作業を行ったり、追加の指示を与えたりといった、人間を介在させたワークフロー、まさにHITLを実現することができます。

`interrupt` 関数は LangGraph 0.2.57 からHITLの実装での利用が推奨されています。

> As of LangGraph 0.2.57, the recommended way to set breakpoints is using the interrupt function as it simplifies human-in-the-loop patterns.

今回は `human_review_node` で用います。このノードに遷移すると `interrupt()` が実行され、グラフの実行が一時停止します。その後、`Command` の `resume` に渡されたレビュー結果をもとにグラフを再開します。ここでは `action` キーに3通りの値を期待しています。

| action   | 遷移先ノード   | 処理内容                                 | 
| -------- | -------- | ---------------------------------------- | 
| `continue` | `run_tool` | ツールの実行を許可する                   | 
| `update`   | `run_tool` | ツールの引数を修正した上で実行を許可する | 
| `cancel`   | `call_llm` | ツールの実行を拒否する                   | 


```python:human_review_node
def human_review_node(state) -> Command[Literal["call_llm", "run_tool"]]:
    last_message = state["messages"][-1]
    tool_call = last_message.tool_calls[-1]

    # this is the value we'll be providing via Command(resume=<human_review>)
    human_review = interrupt(
        {
            "question": "Is this correct?",
            # Surface tool calls for review
            "tool_call": tool_call,
        }
    )

    review_action = human_review["action"]
    review_data = human_review.get("data")

    # if approved, call the tool
    if review_action == "continue":
        return Command(goto="run_tool")

    # update the AI message AND call tools
    elif review_action == "update":
        updated_message = {
            "role": "ai",
            "content": last_message.content,
            "tool_calls": [
                {
                    "id": tool_call["id"],
                    "name": tool_call["name"],
                    # This the update provided by the human
                    "args": review_data,
                }
            ],
            # This is important - this needs to be the same as the message you replacing!
            # Otherwise, it will show up as a separate message
            "id": last_message.id,
        }
        return Command(goto="run_tool", update={"messages": [updated_message]})

    # ツールの実行を拒否するように実装を変更
    elif review_action == "cancel":
        # Ref https://langchain-ai.github.io/langgraph/how-tos/memory/delete-messages/
        return Command(goto="call_llm", update={"messages": [RemoveMessage(id=last_message.id), HumanMessage(content=review_data)]})
```

### パターン1: ツールの実行を許可する

まずは、単にツールの実行を許可するパターンを見てみましょう。

```python
# Input
initial_input = {"messages": [{"role": "user", "content": "what's the weather in Tokyo?"}]}

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

# Run the graph until the first interruption
for event in graph.stream(initial_input, thread, stream_mode="updates"):
    print(event)
    print("\n")
```

LLMは `weather_search` が必要と判断し、ツールを呼びました。`tool_calls` が空でなかったため、グラフは `human_review_node` に遷移し、グラフの実行が一時停止しました。

```
{'call_llm': 
    {'messages': 
        [
            AIMessage(content='', 
                     (省略), 
                     tool_calls=[
                         {'name': 'weather_search', 
                          'args': {'city': 'Tokyo'}, 
                          'id': 'call_V3zn2Fd4BBk7u8y7VZtL80nA', 
                          'type': 'tool_call'}
                    ]
            )
        ]
    }
}


{'__interrupt__': (Interrupt(
    value={'question': 'Is this correct?', 
           'tool_call': {'name': 'weather_search', 
                         'args': {'city': 'Tokyo'}, 
                         'id': 'call_V3zn2Fd4BBk7u8y7VZtL80nA', 
                         'type': 'tool_call'}
            }, 
            resumable=True, 
            ns=['human_review_node:16829b7b-a50c-0744-7486-7063849de1d9'], 
            when='during'
        ),
    )
}
```

グラフの実行が一時停止していることは、グラフの状態を `get_state()` メソッドで取得し、`next` 属性を参照することで明らかになります。

```python
print(graph.get_state(thread).next)
```

```
('human_review_node',)
```

続行する場合は `action` として `continue` を渡します。

```python
for event in graph.stream(
    Command(resume={"action": "continue"}),
    dict(**thread, run_name="action"),
    stream_mode="updates",
):
    print(event)
    print("\n")
```

ツールが実行され、最終応答 `The weather in Tokyo is sunny!` が得られました。

```
{'human_review_node': None}


----
Searching for: Tokyo
----
{'run_tool': 
    {'messages': 
        [{'role': 'tool', 
          'name': 'weather_search', 
          'content': 'Sunny!', 
          'tool_call_id': 'call_Uj51DFZK8EKjJrerbqQSGhBO'}]
    }
}


{'call_llm': {'messages': [AIMessage(content='The weather in Tokyo is sunny!', (省略))]}}

```

Langsmithのログを確認してみましょう。`interrupt()` が実行された箇所は `GraphInterrupt` というエラーが発生した扱いになっています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2122187/8b54b092-75ab-446e-ba23-5a810038c294.png)

再開した後のトレースは別のトレースとして記録されます。`run_tool` ノードでツールを実行し、その結果を元に最終応答を生成しています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2122187/6164f201-ba4f-4c9e-bd85-8c14077e9749.png)


### パターン2: ツールの引数を修正した上で実行を許可する

次に、ツールの引数を修正した上で実行を許可するパターンを見てみましょう。

```python
# Input
initial_input = {"messages": [{"role": "user", "content": "what's the weather in Osaka?"}]}

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

# Run the graph until the first interruption
for event in graph.stream(initial_input, thread, stream_mode="updates"):
    print(event)
    print("\n")
```

```
{'call_llm': 
    {'messages': 
        [
            AIMessage(content='', 
                     (省略),
                     tool_calls=[
                         {'name': 'weather_search', 
                         'args': {'city': 'Osaka'}, 
                         'id': 'call_JhPSSmaeXgyr0yxfAHJMWo8Y', 
                         'type': 'tool_call'}
                    ], 
            )
        ]
    }
}


{'__interrupt__': (Interrupt(
    value={'question': 'Is this correct?', 
           'tool_call': {'name': 'weather_search', 
                         'args': {'city': 'Osaka'}, 
                         'id': 'call_JhPSSmaeXgyr0yxfAHJMWo8Y', 
                         'type': 'tool_call'}
          }, 
          resumable=True, 
          ns=['human_review_node:ad7424ea-4c28-4139-baea-351057100632'], 
          when='during'
        ),
    )
}
```

今度はツールの引数を `Osaka` から `Osaka city` に変更した上で、ツールを実行するようにしてみましょう。`action` として `update` を、`data` に新しい引数を渡してグラフを再開します。

```python
# Let's now continue executing from here
for event in graph.stream(
    Command(resume={"action": "update", "data": {"city": "Osaka city"}}),
    dict(**thread, run_name="update"),
    stream_mode="updates",
):
    print(event)
    print("\n")
```

確かにツールの引数が `Osaka city` に変更されていることが分かります。

```
{'human_review_node': 
    {'messages': 
        [{'role': 'ai', 
        　'content': '', 
        　'tool_calls': [
             {'id': 'call_JhPSSmaeXgyr0yxfAHJMWo8Y', 
             'name': 'weather_search', 
             'args': {'city': 'Osaka city'}}
          ], 
          'id': 'run-b68495b3-3403-4e30-b09a-9237ab0003f0-0'
        }]
    }
}


----
Searching for: Osaka city
----
{'run_tool': 
    {'messages': 
        [{'role': 'tool', 
          'name': 'weather_search', 
          'content': 'Sunny!', 
          'tool_call_id': 'call_JhPSSmaeXgyr0yxfAHJMWo8Y'}]
    }
}

{'call_llm': {'messages': [AIMessage(content='The weather in Osaka is currently sunny!', (省略)]}}
```

グラフを再開した後のLangsmith上のトレースを確認してみます。`human_review_node` の出力の `Update` で新しい引数の情報が渡されています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2122187/5f677d3c-3b6d-4802-a67c-d90d0b7aa7a0.png)

そのあとはパターン1と同様に、ツールの実行結果を元に最終出力が生成されています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2122187/3b15704c-ae2d-4aa1-9d7d-70de366e3bbf.png)



### パターン3: ツールの実行を拒否する

最後にツールの実行そのものを拒否することを検討します。[LangGraphの公式チュートリアル](https://langchain-ai.github.io/langgraph/how-tos/human_in_the_loop/review-tool-calls/) では、ユーザからのフィードバックをツールの実行結果に埋め込む方法が紹介されています。この方法は、ツールの実行履歴そのものがなくなる訳ではありません。個人的には、ツールの実行そのものを取りやめる手段に興味があったので、実装方法を一部変更しています。

具体的には、`review_action` として `cancel` を受け取った時、ツール呼びが含まれる `AIMessage` を削除し、ユーザから受け取ったメッセージを `HumanMessage` として追加するという実装を行っています。メッセージの削除は `RemoveMessage` で実現できます。

```python:(再掲) human_review_node
def human_review_node(state) -> Command[Literal["call_llm", "run_tool"]]:
    last_message = state["messages"][-1]
    tool_call = last_message.tool_calls[-1]

    # this is the value we'll be providing via Command(resume=<human_review>)
    human_review = interrupt(
        {
            "question": "Is this correct?",
            # Surface tool calls for review
            "tool_call": tool_call,
        }
    )

    review_action = human_review["action"]
    review_data = human_review.get("data")

    # if approved, call the tool
    if review_action == "continue":
        return Command(goto="run_tool")
        
    # 中略...
    
    # ツールの実行を拒否するように実装を変更
    elif review_action == "cancel":
        # Ref https://langchain-ai.github.io/langgraph/how-tos/memory/delete-messages/
        return Command(goto="call_llm", 
                       update={"messages": [RemoveMessage(id=last_message.id),
                                            HumanMessage(content=review_data)]})
```

ここでは `what's the weather in Qiita?` という質問文を投げ、その後にツールの呼び出しを取り消すことを考えます。

```python
# Input
initial_input = {"messages": [{"role": "user", "content": "what's the weather in Qiita?"}]}

# Thread
thread = {"configurable": {"thread_id": "3"}}

# Run the graph until the first interruption
for event in graph.stream(initial_input, thread, stream_mode="updates"):
    print(event)
    print("\n")
```

```
{'call_llm': 
    {'messages': 
        [
            AIMessage(content='', 
                     (省略),
                     tool_calls=[
                         {'name': 'weather_search', 
                         'args': {'city': 'Qiita'}, 
                         'id': 'call_IJLDTs411rqVGbh3cPly12ZB', 
                         'type': 'tool_call'}
                    ],
            )
        ]
    }
}


{'__interrupt__': (Interrupt(
    value={'question': 'Is this correct?', 
           'tool_call': {'name': 'weather_search', 
                         'args': {'city': 'Qiita'}, 
                         'id': 'call_IJLDTs411rqVGbh3cPly12ZB', 
                         'type': 'tool_call'}
           }, 
           resumable=True, 
           ns=['human_review_node:be502500-a2a4-d4d3-0074-ec2365814f17'], 
           when='during'
        ),
    )
}
```

LLMは Qiita という都市の天気を調べようとしています。もちろんこれはユーザ側の冗談なので、陳謝した上で Qiita について知っていることを答えてもらいましょう。`action` に `cancel` を、`data` にユーザのフィードバックを与えてグラフを再開します。

```python
for event in graph.stream(
    Command(resume={"action": "cancel", "data": "Sorry, just kidding. There's no city called Qiita. By the way, do you know anything about Qiita?"}),
    dict(**thread, run_name="cancel"),
    stream_mode="updates",
):
    print(event)
    print("\n")
```

```
{'human_review_node': 
    {'messages': 
        [RemoveMessage(content='', 
                       additional_kwargs={}, 
                       response_metadata={}, 
                       id='run-c1acdeb9-e8d6-4d45-a54c-3a85b2fb47e7-0'),
         HumanMessage(content="Sorry, just kidding. There's no city called Qiita. By the way, do you know anything about Qiita?", 
                      additional_kwargs={}, 
                      response_metadata={}, 
                      id='4fd3eaff-6ed1-4ff2-baaf-7f83daf211ea')]
    }
}


{'call_llm': 
    {'messages': 
        [
            AIMessage(content='Yes, Qiita is a popular online platform primarily for
                               developers and engineers to share knowledge, tips, 
                               and code. It operates as a community-driven site 
                               where users can write and post articles related to 
                               programming and software development. These articles 
                               can cover a wide range of topics such as tutorials, 
                               coding techniques, insights into new technologies, 
                               and solutions to specific technical problems. Qiita 
                               is especially popular in Japan and mainly features 
                               content in Japanese, making it a valuable resource 
                               for Japanese-speaking developers.', )]}}
```

グラフを再開した後のLangsmith上のトレースを確認してみます。ツールの実行はなされておらず、最初のクエリの後にユーザのフィードバックが挿入されていることが分かります。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2122187/9363923f-ed7f-4f6a-b513-16acb8e1470d.png)

## 実装（GUIベース）
コマンドラインベースで動作の確認が取れたので、これをChainlitに組み込むことを考えます。

### ユーザのフィードバックを受け取るのに有用なコンポーネント
HITLではLLMの推論結果に応じてユーザからフィードバックを受け取る仕組みを実装する必要があります。その仕組みを実装するのに役立つコンポーネントとして、Chainlitでは `AskActionMessage`、`AskUserMessage` が用意されています。

#### AskActionMessage
`AskActionMessage` はプログラムを続行する前に、次に取るべき行動をユーザに尋ねるためのコンポーネントです。使用例を示します。

```python:AskActionMessage の使用例
action_msg = await cl.AskActionMessage(
    content="Pick an action!",
    actions=[
        cl.Action(name="continue", payload={"action": "continue"}, label="✅ Continue"),
        cl.Action(name="update",   payload={"action": "update"},   label="📝 Edit"),
        cl.Action(name="cancel",   payload={"action": "cancel"},   label="❌ Cancel"),
    ],
    timeout=300, # sec
).send()

action = action_msg.get("payload").get("action")

await cl.Message(content=f"{action} is selected!").send()
```

実行すると、GUI上に選択肢が表示されます。

![AskUserAction_01.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2122187/f801d574-0d7c-4df0-8d85-f4a00c0aff03.png)

`Continue` を選択すると、`Selected: Continue` と表示されます。なお、`actions` で設定した `payload` を取得して後段の処理に使用することができます。

![AskUserAction_02.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2122187/82b4a41f-dbcc-4013-8789-e37a7316ac9e.png)


https://docs.chainlit.io/api-reference/ask/ask-for-action

#### AskUserMessage
`AskUserMessage` はプログラムを続行する前に、ユーザの入力を受け付けるためのコンポーネントです。使用例を示します。

```python:AskUserMessage の使用例（公式ドキュメントより引用）
res = await cl.AskUserMessage(content="What is your name?", timeout=10).send()
if res:
    await cl.Message(
        content=f"Your name is: {res['output']}",
    ).send()
```

GUIの表示は下図のようになります。

![AskUserMessage.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2122187/bd449743-2810-4938-8785-f9a0d0fd340f.png)

https://docs.chainlit.io/api-reference/ask/ask-for-input

### 実装の詳細
セッションが開始したらグラフを作成します。

```python
@cl.on_chat_start
async def main():
    graph = create_graph()
    cl.user_session.set("graph", graph)
```

次にユーザの質問文を受け付けたときの処理を記述します。通常であれば、ユーザが質問文を一度入力すれば、`handle_on_message` 関数が応答をGUI上に表示して終了になります。一方で、HITLを含む場合はグラフの実行が一時停止された段階で、もう一度ユーザの入力を受け付ける必要があります。その機構は、グラフの状態を [get_state() メソッド](https://langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.state.CompiledStateGraph.get_state)で取得、その [next属性](https://langchain-ai.github.io/langgraph/reference/types/#langgraph.types.StateSnapshot.next) を参照し、それが空でないなら `while` 文でループさせることで実現できます。この方法は[こちら](https://langchain-ai.github.io/langgraph/tutorials/customer-support/customer-support/#example-conversation_2)を参考にしました。

以下はユーザの入力を受け付けた時に実行されるメソッドで、処理の大枠を示しています。グラフの実行が一時停止した後の処理については省略しています。

```python:HITL を含む処理の大枠
@cl.on_message
async def handle_on_message(msg: cl.Message):
    graph = cl.user_session.get("graph")

    # Ref https://docs.chainlit.io/integrations/langchain#with-langgraph
    thread = {"configurable": {"thread_id": cl.context.session.id}}
    cb = cl.LangchainCallbackHandler()
    config = RunnableConfig(callbacks=[cb], **thread)

    # Ref https://langchain-ai.github.io/langgraph/tutorials/customer-support/customer-support/#example-conversation_2
    response = graph.invoke({"messages": [HumanMessage(content=msg.content)]}, config=config)
    snapshot = graph.get_state(config)

    # 一時停止状態であれば続行
    while snapshot.next:
        ##################################
        # ユーザの入力を受け付け、それに応じて様々な処理をおこなう
        ##################################

        # 次の状態を取得
        snapshot = graph.get_state(thread)

    # 最終出力
    msg = cl.Message(content=response["messages"][-1].content)
    await msg.send()
```

最後に、`while` 文の中身を実装します。この部分はユースケースによってさまざまな実装が考えられますが、ここでは実行しようとしているツールの情報をユーザに提示し、`AskActionMessage` で選択させるようにします。繰り返しになりますが、本実装の選択肢は以下の3択です。

| action   | 遷移先ノード   | 処理内容                                 | 
| -------- | -------- | ---------------------------------------- | 
| `continue` | `run_tool` | ツールの実行を許可する                   | 
| `update`   | `run_tool` | ツールの引数を修正した上で実行を許可する | 
| `cancel`   | `call_llm` | ツールの実行を拒否する                   | 

以下は `while` 文の中の実装例です。`action` の `payload` をグラフ内で定義したアクションに合わせておくことで、取得した `action` を `Command` にそのまま利用することができます。なお、コマンドラインベースで実装した際は `graph.stream()` を使いましたが、`graph.invoke()` でも同様に `Command` を渡すことでグラフの実行を再開できます。

```python:
    while snapshot.next:
        tool_calls = response['messages'][-1].tool_calls[0] # dict

        # 実行予定のツールの情報
        tool_calling_info = f"""The AI agent is trying to execute a tool called {tool_calls['name']} with the following arguments.

`json # 記事の表示上の都合でバッククォートの数を3つから1つに減らしています。
{tool_calls['args']}
`
"""

        # Ref https://docs.chainlit.io/api-reference/ask/ask-for-action
        action_msg = await cl.AskActionMessage(
            content=tool_calling_info,
            actions=[
                cl.Action(name="continue", payload={"action": "continue"}, label="✅ Continue"), # ツール実行を許可
                cl.Action(name="update",   payload={"action": "update"},   label="📝 Edit"),     # 引数を修正した上で許可
                cl.Action(name="cancel",   payload={"action": "cancel"},   label="❌ Cancel"),   # ツール実行を拒否
            ],
            timeout=300, # sec
        ).send()

        action = action_msg.get("payload").get("action")

        if action == "continue":
            response = graph.invoke(
                Command(resume={"action": action}),
                config=config,
            )

        elif action == "update":
            # Ref https://docs.chainlit.io/api-reference/ask/ask-for-input
            user_msg = await cl.AskUserMessage(content="Please enter a new city name.", timeout=300).send()
            
            response = graph.invoke(
                Command(resume={"action": action, "data": {"city": user_msg['output']}}),
                config=config,
            )
        
        elif action == "cancel":
            user_msg = await cl.AskUserMessage(content="Please provide additional instructions for the AI.", timeout=300).send()

            response = graph.invoke(
                Command(resume={"action": action, "data": user_msg['output']}),
                config=config,
            )
```

### アプリの動作
完成したアプリの動作について示します。
最初にクエリを入力し、ツール呼びがあった場合には3つの選択肢が表示されます。`Continue` を押下した場合にはそのままツールが実行されます。

![chainlit_hitl_01.gif](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2122187/9ec3354c-ddb4-4883-9afa-e7c40288f3c7.gif)

`Edit` を押下した場合は、引数を編集してからツールが実行されます。ツールへの入力を見ると、ユーザが再入力した文字列に変更されていることが分かります。

![chainlit_hitl_02.gif](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2122187/68562d4b-a9f3-4852-bb8e-13044645d5a6.gif)

`Cancel` を押下した場合はツール呼びがキャンセルされます。追加の指示を与えると、それに従った応答が生成されます。

![chainlit_hitl_03.gif](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2122187/5f26f742-5b38-4c2c-bf66-d740b48480c8.gif)

## まとめ
この記事では、LangGraph の `interrupt()` を用いた Human-in-the-loop の実装と、Chainlit に Human-in-the-loop の仕組みを実装する方法について解説しました。

- LangGraphの0.2.57から実装された `interrupt()` を使うことで、Human-in-the-loop の仕組みを簡単に実装できる
- Chainlit に標準実装されている `AskActionMessage` で次の取るべき行動を受け取り、`AskUserMessage` で追加の指示を受け取る
- グラフが一時停止している際は `StateSnapshot.next` に次のノードの情報が入るので、それを利用してユーザから追加の入力を受け取るような機構を作成する


## 参考リンク
参照日はいずれも 2025/03/01 になります。

### LangGraph関連

以下で記述されているグラフをもとに作成しました。

https://langchain-ai.github.io/langgraph/how-tos/human_in_the_loop/review-tool-calls/#how-to-review-tool-calls

`snapshot.next` を条件にして `while` 文でループさせるヒントは以下から得ました。

https://langchain-ai.github.io/langgraph/tutorials/customer-support/customer-support/#example-conversation_2

LangGraphでメッセージを削除する方法

https://langchain-ai.github.io/langgraph/how-tos/memory/delete-messages/#how-to-delete-messages

Human-in-the-loop の概念的な話

https://langchain-ai.github.io/langgraph/concepts/human_in_the_loop/#human-in-the-loop

通常のReActエージェントについて

https://langchain-ai.github.io/langgraph/how-tos/create-react-agent/#how-to-use-the-pre-built-react-agent

### Chainlit関連
Chainlitのcookbookで紹介されているHITLの例です。ユーザの入力を受け付ける機構をツールとして実装しており、`interrupt()` などは用いていません。

https://github.com/Chainlit/cookbook/tree/main/langchain-ask-human

Chainlit上でサポートされているHITL機能に関するIssueです。`interrupt_before` 引数を用いる方法がいいという結論になっており、`interrupt()` の利用については触れられていません。

https://github.com/Chainlit/chainlit/issues/1129

HITLに使えるChainlit標準搭載のコンポーネント

https://docs.chainlit.io/api-reference/ask/ask-for-action

https://docs.chainlit.io/api-reference/ask/ask-for-input

LangGraphを使うときの Chainlit の実装例。セッションIDやコールバックについてはこちらを参考にしました。HITLは使われていません。

https://docs.chainlit.io/integrations/langchain#with-langgraph

### APIリファレンス
- [`interrupt` 関数](https://langchain-ai.github.io/langgraph/reference/types/#langgraph.types.interrupt)
- [`get_state` メソッド](https://langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.state.CompiledStateGraph.get_state)
- [`StateSnapshot.next` 属性](https://langchain-ai.github.io/langgraph/reference/types/#langgraph.types.StateSnapshot.next)


[^1]: https://xtech.nikkei.com/atcl/nxt/keyword/18/00002/101100209/
[^2]: ReActは「Reasoning and Acting」の略で、どのツールを使用するかを推論する（Reasoning）過程と、実際にツールを使う行動（Acting）過程を交互に繰り返す手法です。原論文：[ReAct: Synergizing Reasoning and Acting in Language Models](https://par.nsf.gov/biblio/10451467)