🈸

[ADK] 1.14.0で追加された require_confirmation を利用したHuman-in-the-Loop

に公開

こんにちは、サントリーこと大橋です。

先日(2025/9/11) ADK 1.14.0がリリースされました。前回の記事では、AgentとPluginをカプセル化する新しい概念「App」について紹介しました。

https://zenn.dev/soundtricker/articles/f51b9fb4f788d8

今回は、同じく1.14.0で追加されたもう一つの重要な機能、「Tool Confirmation Flow」について深掘りしていきます。この機能は、AIエージェント開発における重要なパターンであるHuman-in-the-Loop(HIL)を、より安全かつ明確に実装するための専用フローを提供します。

対象読者

  • ADKユーザー
  • ADKをこれから使いたいPythonユーザー
  • ADKをこれから使いたいAI エージェント開発者

※本文中ではADKのAgentを「Agent」とアルファベット表記、ADKだけではなく一般的なAIエージェントを「AIエージェント」とカタカナ表記しています。

課題: 従来のHIL実装とLongRunningFunctionToolの限界

AIエージェントが実世界のシステムに影響を与える処理(経費承認、DB更新など)を行う際、人間の承認を挟むHILは不可欠です。

これまでのADKでは、このHILをLongRunningFunctionToolで実装することが一般的でした。

https://github.com/google/adk-python/tree/9862b7b1e2a4b232663ab5053fc9712865abd70a/contributing/samples/human_in_loop
LongRunningFunctionToolを利用したHILの例

しかし、LongRunningFunctionToolは本来、完了までに時間のかかる非同期タスクを扱うための汎用的な機能であり、承認フロー専用ではありませんでした。そのため、以下のような課題がありました。

  • 意図が不明確: ツールが中断された際、それが「承認待ち」なのか、単なる「処理中」なのかをRunner/クライアント側が明確に判断できませんでした。
  • 構造化された情報を渡せない: 「何を」「どのような選択肢で」承認してほしいのか、といった構造化された情報をAgentからRunner/クライアントに伝える標準的な方法がありませんでした。
  • 条件分岐の難しさ: 「特定の条件下でのみ承認を求める」といったロジックをツール内に記述するのが困難でした。

結果として、Runnerやクライアント側が多くの解釈と複雑な状態管理を強いられることになり、堅牢なHILの実装が難しい状況でした。

新機能: 承認フロー専用の「Tool Confirmation Flow」

この課題を解決するため、ADK 1.14.0では承認フロー専用の「Tool Confirmation Flow」が導入されました。これは、HILのパターンをADKのネイティブ機能として正式にサポートするものです。

adk web を利用している場合、この機能は以下のようなダイアログをユーザーに表示し、承認を促します。

ツール確認UIのスクリーンショット
図1. 高度な確認フローにおける承認要求ダイアログの例

このフローの核心は、ツールが「承認が必要」という明確な意図(RequestConfirmationイベント)をフレームワークに伝え、フレームワークがその後の承認待ちと結果の受け渡しを仲介する点にあります。

Runner/クライアントサイドでの承認処理

重要な点として、この新しいフローを使っても、承認要求を解釈し、ユーザーに提示して、その結果をAgentに送り返す処理は、引き続きRunnerまたはそのクライアントサイドで実装する必要があります。

新機能の価値は、そのやり取りが標準化され、意図が明確になった点にあります。以下は、require_confirmationが設定されたツールを呼び出した際のRunner側の処理のサンプルです。

(※コードは筆者がADKの公式サンプルを基に、確認フロー用に修正したものです)

articles/hil-long-running-tool/main.py
# Copyright 2025 Google LLC
# (中略)

async def main():
  # (中略)
  runner = Runner(
      agent=agent.root_agent,
      app_name=APP_NAME,
      session_service=session_service,
  )

  async def call_agent(query: str):
    content = types.Content(role="user", parts=[types.Part(text=query)])
    print(f'>>> ユーザーのクエリ: "{query}"\n')
    print("--- エージェントの初期ターンを実行中 ---")

    events_async = runner.run_async(
        session_id=session.id, user_id=USER_ID, new_message=content
    )

    request_confirm_function_call: Union[types.FunctionCall, None] = None
    tool_confirmation: Union[ToolConfirmation, None] = None

    async for event in events_async:
      if event.content and event.content.parts:
        for part in event.content.parts:
          # (中略)
          # "adk_request_confirmation" という名前のFunctionCallを捕捉する
          if not request_confirm_function_call and part.function_call and part.function_call.name == "adk_request_confirmation":
              request_confirm_function_call = part.function_call
              print(
                  f"      (承認要求としてキャプチャ: '{part.function_call.name}')"
              )

    print("--- エージェントの初期ターンの終了 ---")

    # 承認要求があった場合
    if request_confirm_function_call:
      print(f"--- 外部承認をシミュレート ---")

      # 本来はユーザーの入力を受け付けるが、ここでは自動的に承認する
      updated_tool_output_data = {
          "confirmed": True
      }

      updated_function_response_part = types.Part(
          function_response=types.FunctionResponse(
              id=request_confirm_function_call.id,
              name=request_confirm_function_call.name, # adk_request_confirmation
              response=updated_tool_output_data,
          )
      )

      print(
          f"--- 承認結果をエージェントに送信 呼び出しID: "
          f"{request_confirm_function_call.id}: {updated_tool_output_data} ---")
      print("--- 承認結果を受け取った後のエージェントのターンを実行中 ---")

      # 再度Agentを実行して承認結果を渡す
      async for event in runner.run_async(
          session_id=session.id,
          user_id=USER_ID,
          new_message=types.Content(
              parts=[updated_function_response_part], role="user"
          ),
      ):
        # (中略)

### 使い方1: シンプルなYes/No承認 (Boolean Confirmation)

ツール側で承認をリクエストする最も簡単な方法は、`FunctionTool`でツールをラップする際に`require_confirmation`パラメータを使用することです。

`require_confirmation=True`と設定すると、そのツールは呼び出されるたびに必ずユーザーの確認を求めるようになります。また、ブール値を返す関数を渡すことで、「特定の条件を満たした場合にのみ確認を求める」という動的な制御も可能です。

(※コードはADKの公式サンプルより引用し、コメントは筆者が日本語化しています。)

```python:articles/hil-function-tool/agent.py
# ... (imports) 

async def confirmation_threshold(amount: int, tool_context: ToolContext) -> bool:
  """金額が1000より大きい場合にtrueを返します。"""
  return amount > 1000

root_agent = Agent(
    # ... (agent config)
    tools=[
        # require_confirmation に呼び出し可能オブジェクトを設定し、
        # 金額が1000を超える場合のみユーザーの確認を要求します。
        FunctionTool(
            reimburse,
            require_confirmation=confirmation_threshold,
        ),
    ],
)

使い方2: 構造化データを扱う高度な承認 (Advanced Confirmation)

承認者が単なるYes/Noだけでなく、具体的な数値(例:承認する日数)を入力できるようにしたい場合は、ツールの中から tool_context.request_confirmation() を呼び出します。

このメソッドは、ユーザーへの指示(hint)と、期待する応答データの構造(payload)を渡すことができます。これにより、Agentと承認者の間で明確な「契約」を結ぶことができます。

(※コードはADKの公式サンプルより引用し、コメントは筆者が日本語化しています。)

articles/hil-function-tool/agent.py
# ... (imports) 

def request_time_off(days: int, tool_context: ToolContext):
  """従業員の休暇を申請します。"""
  # (中略)
  # 承認済みかどうかをtool_confirmationで確認
  tool_confirmation = tool_context.tool_confirmation
  if not tool_confirmation:
    # 承認がない場合は、承認をリクエストする
    tool_context.request_confirmation(
        hint=('休暇申請を承認または拒否してください。'),
        payload={'approved_days': 0,},
    )
    return {'status': 'マネージャーの承認が必要です。'}

  # 承認済みの場合は、そのペイロードを利用して処理を続行
  approved_days = tool_confirmation.payload['approved_days']
  # (中略)

制約事項

まとめ

今回はADK 1.14.0で追加された「Tool Confirmation Flow」について、その真価に焦点を当てて解説しました。

この機能の最大のメリットは、Human-in-the-Loopという重要なパターンに対して、意図が明確で、構造化されたやり取りを可能にする「専用の仕組み」を提供した点にあります。

LongRunningFunctionToolという汎用的な機能で工夫して実装していた時代から、request_confirmationという承認専用の機能を手に入れたことで、Agentの設計がより堅牢になり、何をすべきかが明確になりました。これは業務フローなどを作るうえでは非常に大きな一歩と言えるでしょう。


お知らせ/宣伝

先日、ADKユーザー向けのイベント「ADK UG #0」が開催され、ADK開発者が集う日本語のDiscordコミュニティが誕生しました。ADKに関する情報交換や議論に興味がある方は、ぜひご参加ください!

https://discord.gg/BKpGRzjtqZ

また、ADKの最新のコミットログやリリースノートを分かりやすく解説するPodcastを、月・水・金に配信しています。ADKの動向を追いかけたい方は、ぜひ聴いてみてください。

https://www.youtube.com/playlist?list=PL0Zc2RFDZsM_MkHOzWNJpaT4EH5fQxA8n

Discussion