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

2026-09-10 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="この計算を実行してよいか確認してください",
            ),
        ],
        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": "9かける3を計算して"}]},
        config=config,
    )
    print("承認待ちで停止:", result["__interrupt__"])

$ ./venv/bin/python session1_start.py
承認待ちで停止: [Interrupt(value={'action_requests': [{'name': 'multiply', 'args': {'left': 9, 'right': 3}, 'description': "この計算を実行してよいか確認してください\n\nTool: multiply\nArgs: {'left': 9, 'right': 3}"}], 'review_configs': [{'action_name': 'multiply', 'allowed_decisions': ['approve', 'reject']}]}, id='7252cf64586dc33c79f722d758b7b335')]

この 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)

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

$ ./venv/bin/python session2_resume.py
pending tasks: ['HumanInTheLoopMiddleware.after_model']
承認待ちのツール呼び出し: [{'name': 'multiply', 'args': {'left': 9, 'right': 3}, 'description': "この計算を実行してよいか確認してください\n\nTool: multiply\nArgs: {'left': 9, 'right': 3}"}]
最終回答: 9 × 3 = **27** です！