こちらの記事の続編です。
ここでは、Checkpointer と Human-in-the-loop でより実用的なエージェントにしていきます。
-
Checkpointer（永続性）: 会話の履歴をデータベースに保存し、チャットボットのように過去の文脈を記憶させる
-
Human-in-the-loop: ツールを実行する前に「本当に実行していいですか」と人間に承認を求めるステップを挟む
ディレクトリ構成
langgraph-search-agent/
├── requirements.txt
├── .env.example
├── main.py
├── chat.py # ←新たに追加
└── src/
├── __init__.py
├── state.py
├── nodes.py
├── edges.py
└── graph.py # ←修正を加える
state.py / nodes.py / edges.py の中身は前編・後編で作ったものと同じです。今回変更が入るのは graph.py と、新規に追加する chat.py だけです。
事前準備
Checkpointerの永続化先としてSQLiteを使うため、専用パッケージを追加でインストールします。
pip install -U langgraph-checkpoint-sqlite
1. Checkpointerとは
これまでの app.stream(inputs) は呼び出すたびに全く新しい会話として扱われ、一つ前のやり取りを覚えていませんでした。Checkpointer を使うと、グラフの実行結果（State）がスナップショットとして保存され、thread_id という会話 ID ごとに続きから再開できるようになります。
LangGraph には複数の Checkpointer 実装がありますが、今回はファイルベースで永続化される SqliteSaver を使います。
from langgraph.checkpoint.sqlite import SqliteSaver
with SqliteSaver.from_conn_string("checkpoints.sqlite") as checkpointer:
app = workflow.compile(checkpointer=checkpointer)
checkpointer を渡してコンパイルしたグラフを実行する際は、config に thread_id を指定します。
config = {"configurable": {"thread_id": "user-1"}}
app.stream(inputs, config)
同じ thread_id を指定して再度呼び出せば、checkpoints.sqlite に保存された過去の会話履歴（State）を引き継いだ状態から再開されます。
2. Human-in-the-loopとは
Agent が自律的にツールを呼び出せるのは便利ですが、実行前に人間が内容を確認したい場面もあります（取り返しのつかない操作は特に）。
LangGraph では、グラフをコンパイルするときに interrupt_before へノード名を渡すだけで、そのノードの直前で実行を一時停止できます。
app = workflow.compile(checkpointer=checkpointer, interrupt_before=["tools"])
停止中は app.get_state(config) で現在のStateと「次に実行される予定のノード」を確認できます。
snapshot = app.get_state(config)
snapshot.next # 例: ("tools",) -> toolsノードの実行待ちで停止中
再開する場合は、app.stream(None, config) のように入力を None にして呼び出します。人間が実行を拒否した場合は、app.update_state() でツール呼び出しに対する却下メッセージ（ToolMessage）を直接Stateへ書き込み、Agentに「ツールを使わずに答えて」と伝えます。
3. graph.py の編集：Checkpointer / Human-in-the-loop対応にする
src/graph.py の build_graph を、checkpointer と interrupt_before_tools を受け取れるように拡張します。
（どちらも指定しなければ以前と全く同じ挙動になります。）
from typing import Optional
from langgraph.checkpoint.base import BaseCheckpointSaver
from langgraph.graph import END, START, StateGraph
from langgraph.graph.state import CompiledStateGraph
from .edges import should_continue
from .nodes import call_model, tool_node
from .state import State
def build_graph(
checkpointer: Optional[BaseCheckpointSaver] = None,
interrupt_before_tools: bool = False,
) -> CompiledStateGraph:
"""グラフを組み立ててコンパイルする。
Args:
checkpointer: 指定すると会話履歴（State）が永続化され、
同じ thread_id で再度実行した際に続きから会話できるようになる（Checkpointer）。
interrupt_before_tools: True にすると "tools" ノードの直前で実行を
一旦中断し、人間の承認を待つようになる（Human-in-the-loop）。
"""
workflow = StateGraph(State)
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)
workflow.add_edge(START, "agent")
workflow.add_conditional_edges(
"agent",
should_continue,
{"tools": "tools", END: END},
)
workflow.add_edge("tools", "agent")
compile_kwargs = {}
if checkpointer is not None:
compile_kwargs["checkpointer"] = checkpointer
if interrupt_before_tools:
compile_kwargs["interrupt_before"] = ["tools"]
return workflow.compile(**compile_kwargs)
4. chat.py の追加：対話型 CLI を作る
これまでの main.py は「1回質問して終わり」のデモでしたが、chat.py では複数ターンの対話を続けられるCLIチャットボットとして実装します。
import truststore
truststore.inject_into_ssl()
import uuid
from dotenv import load_dotenv
load_dotenv()
from langchain_core.messages import ToolMessage
from langgraph.checkpoint.sqlite import SqliteSaver
from src.graph import build_graph
DB_PATH = "checkpoints.sqlite"
def print_latest_message(state_values: dict) -> None:
"""State内の直近のメッセージを表示する。"""
last_message = state_values["messages"][-1]
role = last_message.__class__.__name__
if getattr(last_message, "content", None):
print(f"[{role}] {last_message.content}")
else:
print(f"[{role}] (ツール呼び出しのみ・本文なし)")
def handle_pending_tool_calls(app, config: dict) -> None:
"""interrupt_before=['tools'] で止まっている場合、承認を取って再開する。
ユーザーが拒否した場合は、ToolMessageで「却下された」ことをStateに
書き込み、Agentに再考させる。
"""
snapshot = app.get_state(config)
while snapshot.next: # 次に実行されるノードが残っている = 中断中
last_message = snapshot.values["messages"][-1]
for call in last_message.tool_calls:
print(f"\n[承認待ち] ツール `{call['name']}` を次の引数で実行しようとしています:")
print(f" {call['args']}")
answer = input("実行してよいですか？ (y/n): ").strip().lower()
if answer == "y":
for event in app.stream(None, config, stream_mode="values"):
print_latest_message(event)
else:
print("-> ツールの実行をキャンセルしました。")
cancel_messages = [
ToolMessage(
tool_call_id=call["id"],
content="ユーザーがこのツールの実行を却下しました。ツールを使わずに答えてください。",
)
for call in last_message.tool_calls
]
# ToolNodeを経由せず、直接Stateにキャンセル結果を書き込む
app.update_state(config, {"messages": cancel_messages})
for event in app.stream(None, config, stream_mode="values"):
print_latest_message(event)
snapshot = app.get_state(config)
def main() -> None:
with SqliteSaver.from_conn_string(DB_PATH) as checkpointer:
app = build_graph(checkpointer=checkpointer, interrupt_before_tools=True)
thread_id = str(uuid.uuid4())
config = {"configurable": {"thread_id": thread_id}}
print(f"会話ID: {thread_id}")
print(f"({DB_PATH} に会話が保存されます。'exit' で終了)")
while True:
user_input = input("\nYou: ").strip()
if user_input.lower() in {"exit", "quit"}:
break
if not user_input:
continue
inputs = {"messages": [("user", user_input)]}
for event in app.stream(inputs, config, stream_mode="values"):
print_latest_message(event)
# ツール呼び出し待ちで中断していないか確認し、必要なら承認フローへ
handle_pending_tool_calls(app, config)
if __name__ == "__main__":
main()
コードのポイント
-
app.get_state(config).nextが空でない間はツール呼び出し待ちの状態なので、ループで承認を求め続けます - 承認（
y）した場合はapp.stream(None, config, ...)で中断したところから再開し、拒否した場合はToolMessageで却下理由を直接Stateに書き込んでから Agent に再考させます
実行してみる
python chat.py
雑談だけなら、Human-in-the-loopの確認は挟まれず今まで通りすぐに返答が返ってきます。一方、検索が必要な質問をすると、ツール実行前に確認が入るようになります。
実際の出力：
会話ID: 3f1a3e2e-xxxx-xxxx-xxxx-xxxxxxxxxxxx
(checkpoints.sqlite に会話が保存されます。'exit' で終了)
[AIMessage] こんにちは！何かお手伝いできることがありますか？
You: 2027年のAIのトレンド予測を教えて
[HumanMessage] 2027年のAIのトレンド予測を教えて
[AIMessage] (ツール呼び出しのみ・本文なし)
[承認待ち] ツール `tavily_search` を次の引数で実行しようとしています:
{'query': '2027 AI trends forecast', 'time_range': 'year', 'topic': 'general'}
実行してよいですか？ (y/n): y
[AIMessage] (ツール呼び出しのみ・本文なし)
[ToolMessage] No search results found for '2027 AI trends forecast'. Suggestions: Remove time_range argument, Try a more detailed search using 'advanced' search_depth. Try modifying your search parameters with one of these approaches.
[AIMessage] (ツール呼び出しのみ・本文なし)
[承認待ち] ツール `tavily_search` を次の引数で実行しようとしています:
{'query': 'AI trends forecast 2027', 'search_depth': 'advanced'}
実行してよいですか？ (y/n): y
以下のような検索結果が回答されました。
1. **経済への影響**:
- ゴールドマン・サックスの研究によると、2027年にはAIがアメリカのGDPに測定可能な影響を与え始めると予測されています。AIは、先進国の労働タスクの約25%、新興国では10-20%を自動化する可能性があり、これが経済成長を促進する要因となるとされています。具体的には、AIの導入が進むことで、今後数年間で経済成長に寄与することが期待されています。[詳細はこちら](https://www.goldmansachs.com/insights/articles/ai-may-start-to-boost-us-gdp-in-2027)
2. **スーパーヒューマンコーダーの登場**:
- AI 2027の予測によると、2027年には「スーパーヒューマンコーダー」と呼ばれるAIシステムが開発される可能性があります。このシステムは、最も優れたエンジニアと同等のコーディングタスクを迅速かつ安価にこなすことができるとされています。AIの進化により、コーディングタスクの処理能力が飛躍的に向上し、2027年には80%の信頼性で人間のエンジニアが数年かかるタスクを完了できるようになると予測されています。[詳細はこちら](https://ai-2027.com)
これらの予測は、AI技術の進化とその経済的影響に関する重要な洞察を提供しています。今後の動向に注目が集まります。
ここでさらに以下のように過去の文脈が必要になる入力をします。
You: その内容をもっと詳しく教えて
[HumanMessage] その内容をもっと詳しく教えて
[AIMessage] (ツール呼び出しのみ・本文なし)
[承認待ち] ツール `tavily_search` を次の引数で実行しようとしています:
{'query': 'Goldman Sachs AI GDP impact 2027', 'search_depth': 'advanced'}
するとちゃんと文脈が引き継がれていて、Checkpointer による永続化が機能していることがわかります。
[AIMessage] 2027年のAIに関するトレンド予測について、さらに詳しい情報を以下にまとめます。
### 1. AIの経済への影響
- **GDPへの影響**:
- ゴールドマン・サックスの研究によると、2027年にはAIがアメリカのGDPに測定可能な影響を与え始めると予測されています。AIは、先進国の労働タスクの約25%、新興国では10-20%を自動化する可能性があります。
- AIの普及が進むことで、アメリカの生産性成長率は年間1.5ポイント向上する可能性があり、他の主要先進国でも同様の効果が期待されています。新興国では、農業や建設などAIの影響が少ない分野が多いため、0.7-1.3ポイントの影響が見込まれています。
- ただし、これらの経済的利益が実際の数字に現れるまでには数年かかるとされています。AIの導入タイミングやICT投資の変化、規制の影響などが、経済成長にどのように寄与するかを左右します。[詳細はこちら](https://www.goldmansachs.com/insights/articles/ai-may-start-to-boost-us-gdp-in-2027)
### 2. スーパーヒューマンコーダーの登場
- **スーパーヒューマンコーダー（SC）**:
- 2027年3月には、スーパーヒューマンコーダーと呼ばれるAIシステムが開発されると予測されています。このシステムは、最も優れた人間のエンジニアと同等のコーディングタスクを、より迅速かつコスト効率よく実行できる能力を持つとされています。
- SCの登場により、AIの研究開発が加速し、2027年末までに人工超知能（ASI）に至る可能性があるとされています。具体的には、SCはAI研究におけるコーディングタスクを30倍の速度で処理できるとされています。[詳細はこちら](https://www.lesswrong.com/posts/5bZ8xFBBHhuFNXSNY/a-guide-to-ai-2027)
- **今後の展望**:
- SCの開発が進むことで、AIの研究者や開発者の役割が変わる可能性があります。AIが自動化することで、従来の人間の仕事が減少する一方で、新たな職種や役割が生まれることも考えられます。
- ただし、AIの進化には不確実性が伴い、技術の進展が予想通りに進まない可能性もあります。特に、計算能力の向上やアルゴリズムの進化がどのように進むかが重要な要素となります。[詳細はこちら](https://www.lesswrong.com/posts/QdaMzqaBJi6kupKtD/superhuman-coding-in-ai-2027-not-so-fast)
また、n と答えた場合はツールが実行されず、Agent がツールなしで回答します。
一度スクリプトを終了しても、同じ thread_id を config に指定して再実行すれば、checkpoints.sqlite から続きの会話を再開できます。
まとめ
どちらも StateGraph のコンパイルオプションを少し変えるだけで実現できており、LangGraph がグラフ構造をベースにしていることの強みが体験できました。
ちなみに、今回は実行時のオプションをつくっただけなので、graph.png に描かれるフローは変わりません。
