icon

sambaiz-net
github
places
board game arena
aws golang machinelearning kubernetes python etl gcp linux spark web ...

LangGraph のスレッドを永続化する checkpoint と HumanInTheLoopMiddleware によるユーザーへの確認の挙動を確認する

llm

LangGraph にはスレッドを永続化する checkpoint の仕組みがあって、SQLite や PostgreSQL に会話履歴などのステートを保存しておき、スレッドIDを渡すことで続きから再開できる。なお、2025 年 10 月に LangGraph v1.0 がリリースされ API に変更があった。HumanInTheLoopMiddleware をはじめとする middleware もその一つ。

今回は以前のような手動で StateGraph を構築する方法ではなく create_agent() を使う。

LangChain のエコシステム LangGraph で cycle を含むグラフから LLM のエージェントを作る - sambaiz-net

# agent_common.py
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain_core.tools import tool

@tool
def multiply(left: int, right: int) -> int:
    """Multiplies two numbers together."""
    return left * right

def build_agent(model, checkpointer):
    return create_agent(
        model=model,
        tools=[multiply],
        middleware=[
            HumanInTheLoopMiddleware(
                interrupt_on={"multiply": {"allowed_decisions": ["approve", "reject"]}},
                description_prefix="May I run this tool?",
            ),
        ],
        checkpointer=checkpointer,
    )

agent.invoke() すると指定ツールが使われるときに interrupt する。

$ cat session1_start.py
from langchain.chat_models import init_chat_model
from langgraph.checkpoint.sqlite import SqliteSaver
from agent_common import build_agent

DB_PATH = "agent_checkpoints.db"
THREAD_ID = "demo-agent-1"

model = init_chat_model(
    "global.anthropic.claude-sonnet-4-6",
    model_provider="bedrock_converse",
    temperature=0,
)

with SqliteSaver.from_conn_string(DB_PATH) as checkpointer:
    agent = build_agent(model, checkpointer)
    config = {"configurable": {"thread_id": THREAD_ID}}

    result = agent.invoke(
        {"messages": [{"role": "user", "content": "Calculate 9 x 3"}]},
        config=config,
    )
    print("Waiting for approval:", result["__interrupt__"])

$ ./venv/bin/python session1_start.py
Waiting for approval: [Interrupt(value={'action_requests': [{'name': 'multiply', 'args': {'left': 9, 'right': 3}, 'description': "May I run this tool?\n\nTool: multiply\nArgs: {'left': 9, 'right': 3}"}], 'review_configs': [{'action_name': 'multiply', 'allowed_decisions': ['approve', 'reject']}]}, id='7252cf64586dc33c79f722d758b7b335')]

スレッドIDを渡すと続きから再開するので、interrupt の action_requests に対して approve する decisions を resume で渡してやるとツールが実行される。

$ cat session2_resume.py
from langchain.chat_models import init_chat_model
from langgraph.types import Command
from langgraph.checkpoint.sqlite import SqliteSaver

from agent_common import build_agent

DB_PATH = "agent_checkpoints.db"
THREAD_ID = "demo-agent-1"

model = init_chat_model(
    "global.anthropic.claude-sonnet-4-6",
    model_provider="bedrock_converse",
    temperature=0,
)

with SqliteSaver.from_conn_string(DB_PATH) as checkpointer:
    agent = build_agent(model, checkpointer)
    config = {"configurable": {"thread_id": THREAD_ID}}

    snapshot = agent.get_state(config)
    print("pending tasks:", [t.name for t in snapshot.tasks])

    pending_interrupt = snapshot.tasks[0].interrupts[0]
    action_requests = pending_interrupt.value["action_requests"]
    print("Action requests:", action_requests)

    decisions = [{"type": "approve"} for _ in action_requests]
    result = agent.invoke(
        Command(resume={"decisions": decisions}),
        config=config,
    )
    print("Result:", result["messages"][-1].content)

$ ./venv/bin/python session2_resume.py
pending tasks: ['HumanInTheLoopMiddleware.after_model']
Action requests: [{'name': 'multiply', 'args': {'left': 9, 'right': 3}, 'description': "May I run this tool?\n\nTool: multiply\nArgs: {'left': 9, 'right': 3}"}]
Result: 9 × 3 = **27**

並列ノードの一部だけ完了した状態からも再開できるか、fast と slow の2ノードを並列実行して slow 側だけ interrupt するグラフで確認する。

$ cat parallel_graph.py
import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph
from langgraph.types import interrupt


class State(TypedDict):
    results: Annotated[list, operator.add]


def fast_node(state: State) -> dict:
    print(">>> fast_node started")
    return {"results": ["fast-done"]}


def slow_node(state: State) -> dict:
    print(">>> slow_node started")
    decision = interrupt({"question": "May I continue slow_mode?"})
    print(">>> slow_node resumed")
    return {"results": [f"slow-done:{decision}"]}


def combine_node(state: State) -> dict:
    print(">>> combine_node started:", state["results"])
    return {"results": ["combined"]}


def build_graph(checkpointer):
    builder = StateGraph(State)
    builder.add_node("fast", fast_node)
    builder.add_node("slow", slow_node)
    builder.add_node("combine", combine_node)

    builder.add_edge(START, "fast")
    builder.add_edge(START, "slow")
    builder.add_edge("fast", "combine")
    builder.add_edge("slow", "combine")
    builder.add_edge("combine", END)

    return builder.compile(checkpointer=checkpointer)

$ cat parallel_session1.py
from langgraph.checkpoint.sqlite import SqliteSaver
from parallel_graph import build_graph

DB_PATH = "parallel_checkpoints.db"
THREAD_ID = "parallel-1"

with SqliteSaver.from_conn_string(DB_PATH) as checkpointer:
    graph = build_graph(checkpointer)
    config = {"configurable": {"thread_id": THREAD_ID}}

    result = graph.invoke({"results": []}, config=config)
    print("Result:", result)

    snapshot = graph.get_state(config)
    for t in snapshot.tasks:
        print(f"task name={t.name} result={t.result} interrupts={t.interrupts}")

$ ./venv/bin/python parallel_session1.py
>>> fast_node started
>>> slow_node started
Result: {'results': ['fast-done'], '__interrupt__': [Interrupt(value={'question': 'May I continue slow_mode?'},
id='c89ec89a0c3dcc9775c0c2cad0f8beb6')]}
task name=fast result={'results': ['fast-done']} interrupts=()
task name=slow result=None interrupts=(Interrupt(value={'question': 'May I continue slow_mode?'},
id='c89ec89a0c3dcc9775c0c2cad0f8beb6'),)

slow を resume すると前の実行での fast の結果と共に combined に進んだ。

$ cat parallel_session2.py
from langgraph.types import Command
from langgraph.checkpoint.sqlite import SqliteSaver
from parallel_graph import build_graph

DB_PATH = "parallel_checkpoints.db"
THREAD_ID = "parallel-1"

with SqliteSaver.from_conn_string(DB_PATH) as checkpointer:
    graph = build_graph(checkpointer)
    config = {"configurable": {"thread_id": THREAD_ID}}

    result = graph.invoke(Command(resume="approved"), config=config)
    print("Result:", result)

$ ./venv/bin/python parallel_session2.py
>>> slow_node started
>>> slow_node resumed
>>> combine_node started: ['fast-done', 'slow-done:approved']
Result: {'results': ['fast-done', 'slow-done:approved', 'combined']}