Geminiを使ってみる｜Agent Development Kit エージェントを開発する その12(Human in the loopを実装する)
はじめに
この記事では、Google Agent Development Kit（ADK）の Human-in-the-Loop（HITL） を、いちばん小さい例で追います。
やりたいことは次の2つだけです。
ワークフローの途中で処理を止め、人間が文字を入力するまで待つ
その入力を、次のノードで受ける。
人間の入力ごとの分岐はこの記事では扱いません。まず「止まる」と「再開する」と「状態を渡す」が分かれば、あとのパターンは同じ部品の組み合わせです。
人間の入力ノードの実装は公式を参考にすれば簡単に実装できます。
公式のコード例を動かしてみる
RequestInput クラスとユーザーへのテキストプロンプトを使用することで、グラフに人間入力ノードを実装できます。
全体の流れ
起動メッセージ
↓
step01 … ここで「入力:」と聞いて止まる
↓ 人が文字を打つと、同じ場所から再開する
prepare_request … 打った文字を共有 state に保存する
↓
simple_agent … その文字を質問として答えるagent/root_agent/agent.py
from google.adk.events import RequestInput
from google.adk import Workflow
def step1(): # Human input step
yield RequestInput(message="Enter a number:")
def step2(node_input):
return node_input * 2
root_agent = Workflow(
name="root_agent",
edges=[('START', step1, step2)],
)main.py
import asyncio
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.apps import App
from google.genai import types
from agent.root_agent.agent import root_agent
APP_NAME = "simple_app"
USER_ID = "1234"
SESSION_ID = "session1234"
REQUEST_INPUT = "adk_request_input"
app = App(name=APP_NAME, root_agent=root_agent)
def find_request_input(event):
"""イベントから RequestInput（adk_request_input）の function_call を探す"""
for fc in event.get_function_calls():
if fc.name == REQUEST_INPUT and fc.id in (event.long_running_tool_ids or ()):
return fc
return None
async def main():
session_service = InMemorySessionService()
# セッションの作成
await session_service.create_session(
app_name=APP_NAME,
user_id=USER_ID,
session_id=SESSION_ID)
# 最初のメッセージ
content = types.Content(
role='user',
parts=[types.Part(text="こんにちは")])
invocation_id = None
async with Runner(app=app, session_service=session_service) as runner:
while True:
pending = None
async for event in runner.run_async(
user_id=USER_ID,
session_id=SESSION_ID,
invocation_id=invocation_id,
new_message=content,
):
fc = find_request_input(event)
if fc:
print(f"fc: {fc}")
# ワークフローが人間の入力待ちで一時停止した
pending = (fc, event.invocation_id)
elif event.output is not None:
print(f"[{event.node_info.path if event.node_info else ''}] output: {event.output}")
elif event.content and event.content.parts and event.content.parts[0].text:
print(f"[{event.author}] {event.content.parts[0].text}")
if pending is None:
break # 入力待ちが無ければワークフロー完了
fc, invocation_id = pending
user_input = input(f"\n{fc.args.get('message')} ")
try:
value = int(user_input)
except ValueError:
value = user_input
# interruptId に対する FunctionResponse を返して、同じ invocation を再開する
content = types.Content(
role='user',
parts=[types.Part(function_response=types.FunctionResponse(
id=fc.id,
name=REQUEST_INPUT,
response={'result': value},
))])
if __name__ == "__main__":
asyncio.run(main())main.pyが実行コードになります。このサンプルの最初のメッセージは `"こんにちは"` です。ただし `step01` はその引数を受け取っていません。起動の合図には使いますが、エージェントへの質問にはなりません。質問になるのは、止まっているあいだにプロンプトへ打った文字です。
LLMを使ったAgentの作成
main.pyはそのままにして、ワークフローへLLMを使ったAgentを配置します。
agent/root_agent/agent.py
from google.adk import Context, Event, Workflow
from google.adk.events import RequestInput
from google.adk.agents import Agent
from pydantic import BaseModel, Field
simple_agent = Agent(
name="simple_agent",
model="gemini-3.7-flash",
instruction="あなたはAIアシスタントです。ユーザーからの質問に対して、適切な回答をしてください。",
output_key="response",
)
def step1(ctx: Context):
# コンテクストの確認
print(f"Context_session: {ctx.session}")
print(f"Context_id: {ctx.run_id}")
print(f"Context_state: {ctx.state.to_dict()}")
print(f"Context_resume_inputs: {ctx.resume_inputs}")
print(f"Context_node_path: {ctx.node_path}")
yield RequestInput(message="Enter a number:")
"""
def step2(node_input, ctx: Context):
print(f"Context_session_step02: {ctx.session}")
print(f"Context_id_step02: {ctx.run_id}")
print(f"Context_state_step02: {ctx.state.to_dict()}")
print(f"Context_resume_inputs_step02: {ctx.resume_inputs}")
print(f"Context_node_path_step02: {ctx.node_path}")
return node_input * 2
"""
step2 = Agent(
name="cat_agent",
model="gemini-3.7-flash",
instruction="""
あなたはAIアシスタントです。語尾に「ニャー」をつけて回答してください。
これまでの返答: {response}
""",
#output_key="response_step2",
)
root_agent = Workflow(
name="root_agent",
edges=[('START', simple_agent, step1, step2)],
)このワークフローでは、最初のメッセージをsimple_agentで受け取りGeminiが回答を作成します。
次に、step1で次の入力待ちで停止します。
人間からの入力を受けて、step2のcat_agentが応答してワークフローが終了します。
main.py
import asyncio
import os
from dotenv import load_dotenv
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.apps import App
from google.genai import types
from agent.root_agent.agent import root_agent
APP_NAME = "simple_app"
USER_ID = "1234"
SESSION_ID = "session1234"
REQUEST_INPUT = "adk_request_input"
app = App(name=APP_NAME, root_agent=root_agent)
def configure_environment() -> None:
""".env を読み込み、Vertex AI 用の既定値を設定する。"""
load_dotenv()
os.environ.setdefault("GOOGLE_CLOUD_LOCATION", "global")
os.environ.setdefault("GOOGLE_GENAI_USE_ENTERPRISE", "true")
def find_request_input(event):
"""イベントから RequestInput（adk_request_input）の function_call を探す"""
for fc in event.get_function_calls():
if fc.name == REQUEST_INPUT and fc.id in (event.long_running_tool_ids or ()):
return fc
return None
async def main():
configure_environment()
session_service = InMemorySessionService()
# セッションの作成
await session_service.create_session(
app_name=APP_NAME,
user_id=USER_ID,
session_id=SESSION_ID)
# 最初のメッセージ
content = types.Content(
role='user',
parts=[types.Part(text="こんにちは。私の名前はポン吉です")])
invocation_id = None
async with Runner(app=app, session_service=session_service) as runner:
while True:
pending = None
async for event in runner.run_async(
user_id=USER_ID,
session_id=SESSION_ID,
invocation_id=invocation_id,
new_message=content,
):
fc = find_request_input(event)
if fc:
print(f"fc: {fc}")
# ワークフローが人間の入力待ちで一時停止した
pending = (fc, event.invocation_id)
elif event.output is not None:
print(f"[{event.node_info.path if event.node_info else ''}] output: {event.output}")
elif event.content and event.content.parts and event.content.parts[0].text:
print(f"[{event.author}] {event.content.parts[0].text}")
if pending is None:
break # 入力待ちが無ければワークフロー完了
fc, invocation_id = pending
user_input = input(f"\n{fc.args.get('message')} ")
try:
value = int(user_input)
except ValueError:
value = user_input
# interruptId に対する FunctionResponse を返して、同じ invocation を再開する
content = types.Content(
role='user',
parts=[types.Part(function_response=types.FunctionResponse(
id=fc.id,
name=REQUEST_INPUT,
response={'result': value},
))])
if __name__ == "__main__":
asyncio.run(main())output
[simple_agent] こんにちは、ポン吉さん！よろしくお願いします。
～略～
Enter a number: 私の名前を呼んでください
[cat_agent] ポン吉さんニャー！
公式では人間の入力待ちノードでRequestInputクラスを以下の設定オプションで使用でるとありまがそれはまた今度。
message:人間による入力要求を説明するためにユーザーに提供されるテキスト。
payload:人間による入力要求の一部として使用される構造化データ。
response_schema:人間の応答が準拠しなければならないデータ構造。
