LangGraphで AIエージェントをさらに学んでいく - ブレイクポイントでエージェントを中断する -

#AI

森亮介

2025.02.10

AIエージェントで何らかの作業を自動化する際、人間による判断や承認を挟む場合もあるかと思います。

LangCainではHuman-in-the-Loop（ヒューマン・イン・ザ・ループ）と定義されています。

LangGraphは、以下のようなフローをサポートするために、エージェントの状態を取得・更新するいくつかの方法が提供されています。

承認 - エージェントを中断し、ユーザに状態を提示し、ユーザにアクションを承認させることができる。
デバッグ - グラフを巻き戻して、問題を再現したり回避したりすることができる。
編集 - 状態を修正することができる。

今回は特定のステップでグラフを止めることができるブレイクポイントについて試していきたいと思います。

参考：langchain-academyのブレイクポイントのモジュール

人間の承認のためのブレークポイント

以前も動作確認で使っていたツールを利用します。

from langchain_openai import ChatOpenAI

def multiply(a: int, b: int) -> int:
    """Multiply a and b.

    Args:
        a: first int
        b: second int
    """
    return a * b

# This will be a tool
def add(a: int, b: int) -> int:
    """Adds a and b.

    Args:
        a: first int
        b: second int
    """
    return a + b

def divide(a: int, b: int) -> float:
    """Divide a by b.

    Args:
        a: first int
        b: second int
    """
    return a / b

tools = [add, multiply, divide]
llm = ChatOpenAI(model="gpt-4o")
llm_with_tools = llm.bind_tools(tools)

ブレイクポイントを使うには、コンパイル時にinterrupt_beforeを使用します。

How to add breakpoints¶

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

特定のステップにのみつけることも可能。

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

上記のサンプルのツールで試してみると、

from IPython.display import Image, display

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import MessagesState
from langgraph.graph import START, StateGraph
from langgraph.prebuilt import tools_condition, ToolNode

from langchain_core.messages import AIMessage, HumanMessage, SystemMessage

# System message
sys_msg = SystemMessage(content="You are a helpful assistant tasked with performing arithmetic on a set of inputs.")

# Node
def assistant(state: MessagesState):
   return {"messages": [llm_with_tools.invoke([sys_msg] + state["messages"])]}

# Graph
builder = StateGraph(MessagesState)

# Define nodes: these do the work
builder.add_node("assistant", assistant)
builder.add_node("tools", ToolNode(tools))

# Define edges: these determine the control flow
builder.add_edge(START, "assistant")
builder.add_conditional_edges(
    "assistant",
    # If the latest message (result) from assistant is a tool call -> tools_condition routes to tools
    # If the latest message (result) from assistant is a not a tool call -> tools_condition routes to END
    tools_condition,
)
builder.add_edge("tools", "assistant")

memory = MemorySaver()
graph = builder.compile(interrupt_before=["tools"], checkpointer=memory)

# Show
display(Image(graph.get_graph(xray=True).draw_mermaid_png()))

スクリーンショット 2025-02-10 10.58.50

__interrupt=before が表示されています。ツール実行前に割り込みが挟まっていることを示しています。

ではこのグラフを実行してみましょう。

# Input
initial_input = {"messages": HumanMessage(content="2と3を掛け合わせる")}

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

# Run the graph until the first interruption
for event in graph.stream(initial_input, thread, stream_mode="values"):
    event['messages'][-1].pretty_print()

スクリーンショット 2025-02-10 11.05.56

ツールコールは返ってきますが、実際にツールは実行されませんでした。

続きを実行するには、input引数にNoneを指定すると
LangGraphはツール・コールを持つAIMessageを含む現在のステートを再実行します。そして、ノードから始まるグラフの次のステップを実行します。

for event in graph.stream(None, thread, stream_mode="values"):
    event['messages'][-1].pretty_print()

スクリーンショット 2025-02-10 11.45.25

ユーザーの判断を入れたコードは例えば以下のようになります。

# Input
initial_input = {"messages": HumanMessage(content="2と3を掛け合わせる")}

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

# Run the graph until the first interruption
for event in graph.stream(initial_input, thread, stream_mode="values"):
    event['messages'][-1].pretty_print()

# Get user feedback
user_approval = input("Do you want to call the tool? (yes/no): ")

# Check approval
if user_approval.lower() == "yes":

    # If approved, continue the graph execution
    for event in graph.stream(None, thread, stream_mode="values"):
        event['messages'][-1].pretty_print()

else:
    print("Operation cancelled by user.")