こんにちは！
KDDIアイレットの取り組みとして、8月17日から8月28日まで開催中の「Google Cloud 夏休み自由研究ブログリレー」、本日は8日目の投稿です。
今回は「ADK2.0のHITL（人間参加型）」をテーマに、実際に検証してみた内容をご紹介します！

これまでの記事はこちらです。
記事一覧はこちら

はじめに

今回はADK2.0のアップデートで登場したグラフベースのエージェントワークフロー機能の一部であるHITL（Human In The Loop：人間参加型）について解説していきます。またHITLを組み込んだ簡単な承認フローを検証したのでそちらも合わせて紹介します。
エージェントは手間のかかる作業を自動で行ってくれるのでとても便利です。その一方で、「全ての作業をAIに任せるのは怖い」と思う方も多いのではないでしょうか。そのような思いもHITLを使用すれば解決できるかもしれません。

HITL（Human In The Loop）とは

HITL = Human In The Loop。直訳すると「人間を（処理の）輪の中に入れる」という意味です。
簡単に説明すると、エージェントの自動処理を一時中断して、人間に入力、判断、承認を求め、返事が来たら処理を再開させる仕組みです。
AIエージェントは便利ですが、全て自動で処理させてしまうと実務において危険な場面があります。

• お客様へのメールの返信を自動送信する前に人間が確認したい
• 高額な決済や返金を実行する前に人間が確認したい
• AIによるレビューを確定する前に人間が確認したい

AIに任せると危険な判断を、「人間が責任を持って判断する」というのをワークフローに組み込むことができる機能がHITLです。
生成AIはそれらしい出力をすることがあり、100%正しいとは限りません。だからこそ、間違えると影響が大きい判断（送信・決済・削除など）をする直前に人間をフローに置くことがAIを活用していく上での最適解となります。

また、HITLノードはAIモデルを使わずに動く点も重要です。
単純にユーザーからの入力と承認を待つだけの確実な処理のため、予測可能で信頼できます。
一方で、後述する文章を作成するノード（draft_agent）は裏でGemini（AI）が動くので出力が毎回異なります。
つまりHITLは入力や承認を人間が確実に行うことができるので安全ということです。

HITLの中心：RequestInput クラス

ADKでHITLを実装するにはRequestInputクラスを使用します。
from google.adk.events import RequestInput
設定できるパラメータは以下の3つです。

※messageとpayloadの違いで混乱したのでGeminiを活用してわかりやすく整理しました。
同じ情報でも、message は「人間が読むための文章」、payload は「アプリが画面を組み立てるための生データ」です。役割を分けて渡すことで、ユーザーに見やすいUIを提示しながら判断を仰げます。
つまり、messageは人間向けの文章（データ）で形は文字列だけ。payloadはコンピューター向けのデータで形は構造化データ（オブジェクトやリスト）。

実行環境・準備

本記事のコードは以下の環境で動作を確認しています。

インストールとAPIキーの設定は以下の通りです。

python3 -m venv .venv
source .venv/bin/activate
pip install "google-adk>=2.0.0" # 本記事の検証時は 2.3.0

Geminiを利用するため、プロジェクト直下の .env にAPIキーを記載します（Google AI Studio で取得できます）。

# .env
GOOGLE_API_KEY=（取得したAPIキー）

ファイル構成は以下の通りです。ADKは フォルダ名をエージェント名として認識 し、ワークフローの起点となる Workflow オブジェクトを root_agent という変数名で探します。

ADK/
├── .env # GOOGLE_API_KEY を記載
└── hitl_agent/ # ← このフォルダ名が adk web に表示される
├── __init__.py # 空ファイル（ADKのパッケージ認識用）
└── agent.py # 後述のコードをすべてここに書く（root_agent を定義）

コード内の Workflow(name="approval_flow") の approval_flow はワークフロー内部の名前です。adk web の画面に一覧表示されるのはフォルダ名（hitl_agent）なので、両者が違っても問題ありません。

カスタマー返信の承認フロー作成をしてみた

今回AIがカスタマー返信を下書き → 人間が承認・却下 → 承認なら送信するフローを作成しました。
概要は以下の通りで、ポイントはAIに送信まで自動で処理させないことです。

ユーザーが問い合わせを入力
↓
[store_inquiry] 問い合わせを保存
↓
[draft_agent] 🤖 AIが返信案を「下書き」
↓
[request_approval] ⏸️ 人間に「送っていい?」と聞いて一時停止 ← ★ここがHITL
↓ （人間が approve / reject を入力）
[route_decision] 👤 返事を見て分岐
↓
approve → [send_reply] ✅ 送信
reject → [cancel] 🛑 中止

① 承認結果の「型」を定義する（response_schema用）

先に人間が返す答えの「型」を決めます。

from typing import Literal
from pydantic import BaseModel, Field

class ApprovalDecision(BaseModel):
"""人間が返す承認結果の形式"""
decision: Literal["approve", "reject"] = Field(
description="承認する場合は approve、却下する場合は reject"
)
comment: str = Field(default="", description="却下理由や修正指示など（任意）")

• decision を approve か reject のどちらかに縛ることで、後の分岐処理が確実に動きます。この型を response_schema としてHITLノードに渡します

② 問い合わせを state（共有メモ）に保存

from google.adk import Event

def store_inquiry(node_input: str):
"""ユーザーが入力した問い合わせを state に保存する"""
return Event(state={"inquiry": node_input})

• node_input … 一つ前のノードから渡ってくるデータ（ここでは最初の入力＝問い合わせ文）
• Event(state={...}) … ワークフロー全体で共有できる「メモ帳」に保存。後で draft_agent が {inquiry} として読み出します

③ AIが返信案を下書きするエージェント

from google.adk import Agent

draft_agent = Agent(
name="draft_agent",
model="gemini-2.5-flash",
instruction="""あなたはカスタマーサポート担当です。
以下のお客様からの問い合わせに対して、丁寧で分かりやすい返信文を日本語で作成してください。

問い合わせ:
{inquiry}
""",
output_key="draft", # 生成した返信案を state["draft"] にも保存
)

• {inquiry} … ②で保存したメモを自動で差し込みます
• output_key="draft" … AIの出力（返信案）を state["draft"] にも保存
• このエージェントの出力は、次のノードに node_input として渡ります

④ ★HITLの中心：人間に承認を求める

from google.adk.events import RequestInput

def request_approval(node_input):
"""AIが作った返信案(node_input)を提示し、人間の承認を待つ"""
yield RequestInput(
message=(
"🔍 AIが返信案を作成しました。この内容で送信してよいですか？\n"
"承認する場合は approve、却下する場合は reject と入力してください。"
),
payload=node_input, # 返信案を判断材料として画面に渡す
response_schema=ApprovalDecision, # 期待する回答形式
)

• HITLではreturnではなくyieldを使う点が重要です。yieldでRequestInputを呼び出すことで処理を一時中断してユーザーの入力を待つことができます。ユーザーからの入力を確認できたら処理を再開することができます
• そして重要なのが、ユーザーが入力した回答は、次のノード（route_decision）の引数 node_input として自動的に渡されるという点です。つまり「request_approval で一時停止 → 人間が approve/reject を入力 → その入力が route_decision に流れ込んで再開」という形でフローがつながります

「本番でも使えるの？」―― 中断中の状態はどこに保存される？

「一時停止して人間を待つ」あいだ、ワークフローの途中状態はセッションに保存されます。再開時は RequestInput が持つ interrupt_id（どの中断への入力かを識別するID）を手がかりに、止まった場所から処理が続きます。今回の adk web での検証はメモリ上に保持する InMemorySessionService（＝プロセスを止めると消える）ですが、ADKには永続化する DatabaseSessionService や VertexAiSessionService も用意されています。承認待ちが数時間〜数日に及ぶような本番運用では、これらの永続セッションに切り替えることで「承認待ちのまま安全に待たせる」ことができます。

⑤ 人間の判断に応じて分岐する

def route_decision(node_input):
"""人間の返答(node_input)を見て、次のノードを決める"""
text = str(node_input).lower()
decision = "approve" if "approve" in text else "reject"
return Event(
route=decision,
message=f"👤 あなたの判断: **{decision}**",
)

• ここでの node_input には、④のHITLノードでユーザーが入力した回答（approve / reject）がそのまま渡ってきます。その値を見て Event(route=...) に渡すことで、route の値をキーに次のノードへと飛びます

⑥ 承認 or 却下の処理

def send_reply(node_input): # approve のとき
return Event(message="✅ 返信を送信しました。対応完了です。")

def cancel(node_input): # reject のとき
return Event(message="🛑 送信を中止しました。返信案を見直してください。")

• ⑤で解説した部分のroute_decisionでapprove, rejectかを判定しています

⑦ ワークフローグラフとして全部つなぐ

from google.adk import Workflow

root_agent = Workflow(
name="approval_flow",
description="AIが返信案を作成し、人間の承認を得てから送信するHITLワークフロー",
edges=[
# START → 入力保存 → 返信案生成 → 承認要求(HITL) → 判断分岐
("START", store_inquiry, draft_agent, request_approval, route_decision),
# 人間の判断に応じて送信 or 中止へ分岐
(route_decision, {"approve": send_reply, "reject": cancel}),
],
)

• edgesの中身を解説します。
• タプルの中では（A, B, C…）の順番で実行します
• (node, {key: next_node}) → node が返した route 値でジャンプ先を決定しています

動作確認

adk web

上記コマンドで http://127.0.0.1:8000 を開き、左上のプルダウンから フォルダ名 の hitl_agent を選択し、下記の動作確認用の問い合わせ文を送信します。

先日購入した商品がまだ届きません。状況を教えてください。

AIが返信案を下書きしたあと、処理が request_approval で一時停止し、承認フォームが表示されます。

ここで注目すべきポイントが3つあります。
1. response_schema から入力フォームが自動生成された
ApprovalDecision（decision と comment）を渡したことで、adk web が Decision / Comment の入力フォームを自動生成してくれています。人間が形式に沿って答えやすくなります。
2. Form / JSON / Payload / Schema のタブ
Payload タブを開くと、payload で渡した返信案の中身を確認できます。「判断材料（payload）」と「回答形式（schema）」が別々に見られる設計です。
3. route_decision の ⚠️ [NO DEFAULT] 警告
分岐ノードに「approve / reject 以外の値が来たときのデフォルト行き先が無い」という警告が出ます（後述の改善ポイント）。

承認パターン

Decision欄に approve と入力して Submit すると、一時停止が解除され route: approve → send_reply へ進みます。

左のグラフでも send_reply → END の経路が緑にハイライトされ、「✅ 返信を送信しました。対応完了です。」 が表示されました。

却下パターン

New Session で新しく試し、今度は reject を入力すると route: reject → cancel へ分岐します。

「🛑 送信を中止しました。返信案を見直してください。」 が表示され、cancel → END の経路に進みました。人間の判断で処理が変わることが確認できます。

ハマりどころ・改善ポイント

• response_schemaは入力を自動整形しない
  response_schemaはあくまで期待する入力の形式を宣言するものです。入力の内容を自動で整形、バリデーションしてくれるわけではないという点を理解することが正しいフローを処理させる上で重要だと感じました。「approveしないで」と入力するとapproveという文字を拾って承認してしまいます。そこで、本検証ではシンプル版として実装していますが、以下のように route_decision を定義して完全一致の判定を行い、想定外の入力に対しては安全にフォールバックさせる仕組みが推奨されています。

def route_decision(node_input):
text = str(node_input).strip().lower() # 前後の空白も除去

if text == "approve":
decision = "approve"
elif text == "reject":
decision = "reject"
else:
# 想定外の入力（"承認します" "approveしないで" 等）は
# 勝手に送信せず、安全側の reject（保留）に倒す
decision = "reject"

return Event(
route=decision,
message=f"👤 判定: **{decision}**（入力: {text}）",
)

• [NO DEFAULT] 警告 ― 分岐にデフォルトを用意する
  分岐マップに定義した値以外が来たときの行き先がないと警告が出ます。今回の検証ではapproveかrejectの2択しか返さないので安全ですが、より堅牢にするなら、想定外の値を受け止める既定ルートを用意しておくとより安心です。
• HITLノードは yield（ジェネレータ）で書く
  return RequestInput(...) ではなく yield RequestInput(...) です。ここを間違えると一時停止が働かないので注意が必要です。

おわりに

今回ADK2.0のアップデートの一つであるHITLについて学ぶことができました。
生成AIはとても便利で私たちの作業効率を格段に良くしていますが、その一方でハルシネーションを起こしたり、責任の所在が曖昧だったりという課題点も存在しています。今回のHITLは、そうした「AIに任せきりにできない部分」を、ワークフローの中に確実な形で組み込める手段として、とても実用的だと感じました。
一方で、HITLはAIの間違いを人間が食い止める機能ですが、その人間の判断も常に完璧とは限らない、という前提は持っておく必要があると感じました。今後は人間による判断基準の差などに対してどのような対策が登場するか、注目しながらエージェント活用を進めていきたいと思います。