iteLLMとLangGraphのHuman-in-the-Loopでエージェントに人間の判断を挟んでみる

Python 3.13
litellm 1.83.14
langgraph 1.1.10
langchain-litellm 0.6.4
langchain-core 1.3.2

"""interrupt() の最小サンプル: 途中で人間の入力を待つ。"""

from typing import TypedDict

import litellm
from langchain_litellm import ChatLiteLLM
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt

litellm.modify_params = True

# プラン文の品質を担保するため plan_node はClaudeを採用
llm = ChatLiteLLM(model="anthropic/claude-sonnet-4-6")

class State(TypedDict):
    destination: str
    style: str
    plan: str

def ask_style_node(state: State) -> dict:
    """ユーザーに「どのスタイルで旅行するか」を聞く。"""
    style = interrupt(
        {
            "question": f"行き先『{state['destination']}』の旅行プランを立てます。どのスタイルにしますか？",
            "options": ["観光重視", "グルメ重視", "温泉重視"],
        }
    )
    return {"style": style}

def plan_node(state: State) -> dict:
    response = llm.invoke(
        [
            {
                "role": "user",
                "content": f"行き先『{state['destination']}』の{state['style']}の1日旅行プランを200字で書いて",
            }
        ]
    )
    return {"plan": str(response.content)}

graph = StateGraph(State)
graph.add_node("ask_style", ask_style_node)
graph.add_node("plan", plan_node)
graph.add_edge(START, "ask_style")
graph.add_edge("ask_style", "plan")
graph.add_edge("plan", END)
app = graph.compile(checkpointer=InMemorySaver())

config = {"configurable": {"thread_id": "demo-1"}}

# 行き先をユーザーから受け取る
destination = input("行き先を入力してください（例: 京都）: ").strip() or "京都"

# 初回実行: interrupt で停止する
result = app.invoke(
    {"destination": destination, "style": "", "plan": ""}, config
)
interrupt_data = result["__interrupt__"][0].value
print("\n中断: 人間に判断を仰ぎます")
print(f"  質問: {interrupt_data['question']}")
print(f"  選択肢: {interrupt_data['options']}")

# 人間がオプションから選ぶ
options = interrupt_data["options"]
style = ""
while style not in options:
    style = input(f"どれにしますか？ {options}: ").strip()

# Command(resume=...) で再開
result = app.invoke(Command(resume=style), config)
print("\n=== 完成プラン ===")
print(result["plan"])

$ python interrupt_basic.py
行き先を入力してください（例: 京都）: 京都

中断: 人間に判断を仰ぎます
  質問: 行き先『京都』の旅行プランを立てます。どのスタイルにしますか？
  選択肢: ['観光重視', 'グルメ重視', '温泉重視']
どれにしますか？ ['観光重視', 'グルメ重視', '温泉重視']: 観光重視

=== 完成プラン ===
# 京都1日観光プラン

**朝**：嵐山の竹林を散策し、天龍寺の庭園を鑑賞。

**昼**：錦市場で湯葉や漬物など京グルメを食べ歩き。

**午後**：金閣寺で黄金の舎利殿を見学後、祇園の石畳を散策。運が良ければ舞妓さんに出会えるかも。

**夕方**：清水寺で夕暮れの京都の街並みを一望して締めくくる。

"""ツール実行の前に人間の承認を求めるパターン。

危険なツール（送信、削除、決済など）を呼び出す前に interrupt して
人間に「許可するか／パラメータを修正するか／拒否するか」を選んでもらう。
"""

from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage
from langchain_core.tools import tool
from langchain_litellm import ChatLiteLLM
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.types import Command, interrupt

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """指定された宛先にメールを送信します（実運用では実際にAPIを呼ぶ）。"""
    return f"送信完了: {to} / {subject}"

@tool
def get_emails() -> str:
    """受信トレイのメールを取得します。"""
    return "新着メール: 鈴木さんから案件の確認依頼が1通"

tools = [send_email, get_emails]
tool_map = {t.name: t for t in tools}
# tool_use の互換性が高いGPTで動作確認
llm = ChatLiteLLM(model="openai/gpt-5-mini").bind_tools(tools)

# 承認が必要な「危険なツール」のリスト
DANGEROUS_TOOLS = {"send_email"}

def agent_node(state: MessagesState) -> dict:
    sys = SystemMessage(
        content="メールアシスタントです。ツールを使って対応してください。"
    )
    response = llm.invoke([sys, *state["messages"]])
    return {"messages": [response]}

def approval_node(state: MessagesState) -> dict:
    """危険なツール呼び出しがあれば承認を求める。"""
    last = state["messages"][-1]
    if not isinstance(last, AIMessage) or not last.tool_calls:
        return {}

    new_messages = []
    for call in last.tool_calls:
        if call["name"] in DANGEROUS_TOOLS:
            decision = interrupt(
                {
                    "type": "tool_approval",
                    "tool": call["name"],
                    "args": call["args"],
                    "options": ["approve", "edit", "reject"],
                }
            )
            if decision["action"] == "approve":
                result = tool_map[call["name"]].invoke(call["args"])
            elif decision["action"] == "edit":
                # 引数を編集して実行
                result = tool_map[call["name"]].invoke(decision["args"])
            else:  # reject
                result = "ユーザーにより拒否されました"
        else:
            result = tool_map[call["name"]].invoke(call["args"])

        new_messages.append(ToolMessage(content=str(result), tool_call_id=call["id"]))
    return {"messages": new_messages}

def has_tool_calls(state: MessagesState) -> str:
    last = state["messages"][-1]
    if isinstance(last, AIMessage) and last.tool_calls:
        return "approval"
    return "end"

graph = StateGraph(MessagesState)
graph.add_node("agent", agent_node)
graph.add_node("approval", approval_node)

graph.add_edge(START, "agent")
graph.add_conditional_edges(
    "agent", has_tool_calls, {"approval": "approval", "end": END}
)
graph.add_edge("approval", "agent")
app = graph.compile(checkpointer=InMemorySaver())

config = {"configurable": {"thread_id": "approval-1"}}

# 1. メール送信を依頼
print("=== ユーザーがメール送信を依頼 ===")
result = app.invoke(
    {"messages": [HumanMessage(content="鈴木さんに『了解しました』と返信して")]},
    config,
)
interrupt_data = result.get("__interrupt__")
print(f"承認待ち: {interrupt_data}")

# 2. 人間が approve を選ぶ
print("\n=== 人間が approve ===")
result = app.invoke(Command(resume={"action": "approve"}), config)
final = result["messages"][-1]
print(f"AI: {final.content}")

$ python tool_approval.py
=== ユーザーがメール送信を依頼 ===
承認待ち: [Interrupt(value={'type': 'tool_approval', 'tool': 'send_email', 'args': {'to': '鈴木さん', 'subject': 'Re: 案件の確認依頼', 'body': '了解しました'}, 'options': ['approve', 'edit', 'reject']}, id='3a716d055879548d99c617de2c91b50e')]

=== 人間が approve ===
AI: 送信しました: 宛先 鈴木さん、件名「Re: 案件の確認依頼」、本文「了解しました」

"""interrupt 中にステートを編集する: タイムトラベルとも組み合わせる。

ユーザーが LLM の出力を直接編集してから次のノードに進む例。
"""

from typing import TypedDict

import litellm
from langchain_litellm import ChatLiteLLM
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt

litellm.modify_params = True

# ドラフト品質を担保するためClaudeを採用
llm = ChatLiteLLM(model="anthropic/claude-sonnet-4-6")

class State(TypedDict):
    request: str
    draft: str
    final: str

def draft_node(state: State) -> dict:
    response = llm.invoke(
        [{"role": "user", "content": f"次の依頼に150字で答えて: {state['request']}"}]
    )
    return {"draft": str(response.content)}

def review_node(state: State) -> dict:
    """人間にドラフトを見せ、編集または承認させる。"""
    edited = interrupt(
        {
            "type": "review",
            "draft": state["draft"],
            "instruction": "編集後のテキストを返すか、空文字なら承認とみなします。",
        }
    )
    if edited.strip():
        return {"final": edited}
    return {"final": state["draft"]}

graph = StateGraph(State)
graph.add_node("draft", draft_node)
graph.add_node("review", review_node)

graph.add_edge(START, "draft")
graph.add_edge("draft", "review")
graph.add_edge("review", END)
app = graph.compile(checkpointer=InMemorySaver())

config = {"configurable": {"thread_id": "edit-1"}}

# 初回実行
result = app.invoke(
    {"request": "AIエージェントとは何かを説明する1文", "draft": "", "final": ""},
    config,
)
draft = result["__interrupt__"][0].value["draft"]
print(f"AIのドラフト: {draft}")

# 人間が編集して resume
edited_text = "AIエージェントとは、LLMを核に自律的に思考し、ツールを使って目的を達成するシステムです。"
result = app.invoke(Command(resume=edited_text), config)
print(f"\n=== 最終出力 ===\n{result['final']}")

$ python state_edit.py
AIのドラフト: AIエージェントとは、与えられた目標を達成するために環境を認識し、自律的に判断・計画を立て、ツールの使用や外部サービスとの連携などの行動を繰り返しながら、人間の継続的な指示なしにタスクを遂行できるAIシステムのことです。

=== 最終出力 ===
AIエージェントとは、LLMを核に自律的に思考し、ツールを使って目的を達成するシステムです。