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

はじめに

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

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

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

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

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

対象読者

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

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

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

環境

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

• langgraph: 0.3.0
• chainlit: 2.2.1
• Python: 3.11.9

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

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

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

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

今回扱うグラフ

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

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

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

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

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

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

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

Chainlitを活用したアプリケーションの実装方法に移る前に、コマンドラインベースでHITLの動作を確認します。実装は LangGraphの公式チュートリアル を流用しています。LLMに渡すツールとして、都市の名前 city を受け取り、単に Sunny! とだけ返す weather_search を用意します。

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

グラフを作成するコード全体

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

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	ツールの実行を拒否する

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: ツールの実行を許可する

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

# 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 属性を参照することで明らかになります。

print(graph.get_state(thread).next)

('human_review_node',)

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

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 というエラーが発生した扱いになっています。

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

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

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

# 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 に新しい引数を渡してグラフを再開します。

# 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 で新しい引数の情報が渡されています。

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

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

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

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

(再掲) 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? という質問文を投げ、その後にツールの呼び出しを取り消すことを考えます。

# 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 にユーザのフィードバックを与えてグラフを再開します。

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上のトレースを確認してみます。ツールの実行はなされておらず、最初のクエリの後にユーザのフィードバックが挿入されていることが分かります。

実装（GUIベース）

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

ユーザのフィードバックを受け取るのに有用なコンポーネント

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

AskActionMessage

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

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上に選択肢が表示されます。

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

AskUserMessage

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

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の表示は下図のようになります。

実装の詳細

セッションが開始したらグラフを作成します。

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

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

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

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 を渡すことでグラフの実行を再開できます。

    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 を押下した場合にはそのままツールが実行されます。

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

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

まとめ

この記事では、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関連

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

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

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

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

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

Chainlit関連

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

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

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

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

APIリファレンス

• interrupt 関数
• get_state メソッド
• 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 ↩