AI（人工知能）が私たちの周りでますます活躍するようになりました。その中でも「AIエージェント」は、特定の目的を持って自律的に考え、行動するプログラムです。まるで、賢いデジタルアシスタントが、私たちのために情報を集めたり、タスクをこなしたりしてくれるようなものです。
Agent Development Kit (ADK) は、Googleが提供する、このようなAIエージェントを構築するためのオープンソースの道具箱（フレームワーク）です。ADKを使えば、アイデアを形にし、インテリジェントなエージェントを比較的簡単に開発・実行できます。
ADKには、特にAI開発初心者やGoogleの技術に関心がある方にとって、いくつかの利点があります。
このガイドは、ADKを初めて使う方を対象としています。Pythonの基本的な知識（変数、関数など）があれば十分です。特別なAIの専門知識は必要ありません。
読み終える頃には、あなたは以下のことができるようになっているはずです。
python --version (または python3 --version) で確認し、なければ python.org からインストールしてください。準備ができたら、さっそく環境を整えましょう。
快適な開発のために、まずは作業スペースを準備します。
mkdir adk_beginner_project
cd adk_beginner_project
source env/bin/activate
.\env\Scripts\activate(env) と表示されれば成功です。仮想環境が有効な状態で、pipを使ってADKをインストールします。
Geminiモデルを使うために、取得したAPIキーを環境変数 GOOGLE_API_KEY に設定します。
export GOOGLE_API_KEY='あなたのAPIキー'
set GOOGLE_API_KEY=あなたのAPIキー
$env:GOOGLE_API_KEY = "あなたのAPIキー"
注意: APIキーは機密情報です。コード内に直接書き込まず、環境変数などを使って安全に扱ってください。
これで開発の準備が整いました。
最も基本的なエージェントは、大規模言語モデル（LLM）の能力をそのまま利用するものです。ADKの LlmAgent を使って、簡単な応答を生成させてみましょう。
プロジェクトフォルダ内に simple_responder.py というファイルを作成します。
simple_responder.py に以下のコードを記述します。
# simple_responder.py
import os
import asyncio
from adk.agents import LlmAgent
from adk.sessions import Session
# APIキーが設定されているか確認 (デバッグ用)
# print(f"APIキー設定済み: {os.getenv('GOOGLE_API_KEY') is not None}")
async def main():
"""シンプルな LlmAgent を使って応答を生成する"""
print("エージェントを準備中...")
# LlmAgentインスタンスを作成。名前を付けておく。
# GOOGLE_API_KEYが設定されていれば、デフォルトでGeminiモデルを使用。
agent = LlmAgent(name="SimpleResponder")
# 対話のコンテキストを管理するセッションを作成
session = Session()
# エージェントに与える指示（プロンプト）
# 以前とは違う、少しクリエイティブなタスクを依頼
prompt = "「夏の終わり」をテーマにした短い詩を一句作ってください。"
print(f"\nエージェントへの指示: '{prompt}'")
# agent.invoke を使ってエージェントを実行
# await で非同期処理の完了を待つ
response_data = await agent.invoke(prompt, session=session)
# 応答データからテキスト出力を取得して表示
print("\nエージェントからの応答:")
# response_data は辞書形式。'output'キーに応答テキストが入っていることが多い。
print(response_data.get('output', '応答がありませんでした。'))
# 参考：セッションに記録されたイベント（処理の流れ）を表示
print("\n--- 処理の記録 ---")
for event in session.get_events():
print(event)
print("------------------")
if __name__ == "__main__":
# 非同期関数 main を実行
asyncio.run(main())
async def main() / await agent.invoke() / asyncio.run(main()): ADKは非同期処理を多用するため、async/await 構文と asyncio ライブラリが使われます。LlmAgent(name="..."): LLM（ここではGemini）の能力を直接利用するエージェントを作成します。Session(): エージェントとの一回のやり取り（対話）を管理するオブジェクトです。入力、出力、途中の処理ステップなどが記録されます。prompt = "...": エージェントに何をしてもらいたいかを伝える指示文です。今回は詩の作成を依頼しました。response_data = await agent.invoke(prompt, session=session): エージェントを起動し、指示を与えます。session を渡すことで、処理の記録が可能になります。response_data.get('output', ...): invoke の結果（辞書）から、キー 'output' に対応する値（LLMが生成したテキスト）を取り出します。.get() を使うとキーが存在しなくてもエラーになりません。session.get_events(): セッションに記録された処理のステップ（イベント）を取得します。デバッグに役立ちます。ターミナルでスクリプトを実行します（仮想環境が有効であること、APIキーが設定されていることを確認してください）。
python simple_responder.py
実行すると、エージェントが「夏の終わり」に関する詩（一句）を生成し、表示するはずです。応答内容は実行するたびに変わる可能性があります。また、処理の記録として AgentStart, LlmStart, LlmEnd, AgentEnd といったイベントが表示されます。
エージェントを準備中...
エージェントへの指示: '「夏の終わり」をテーマにした短い詩を一句作ってください。'
エージェントからの応答:
ひぐらしの声遠くなり 風涼し
--- 処理の記録 ---
AgentStart(agent_name="SimpleResponder", input={'input': '「夏の終わり」をテーマにした短い詩を一句作ってください。'})
LlmStart(...)
LlmEnd(...)
AgentEnd(agent_name="SimpleResponder", output={'output': 'ひぐらしの声遠くなり 風涼し'})
------------------
これで、ADKを使ってLLMに簡単なタスクを実行させる最初のステップが完了しました。
より高度なエージェントを作るために、ADKの主要な構成要素をもう少し詳しく見てみましょう。
LlmAgent: 大規模言語モデル（LLM）の推論能力を主に利用します。対話、文章生成、ツール使用の判断などを行います。SequentialAgent, ParallelAgent, etc.): 複数のエージェントやツールを、決められた順序や条件で実行します。処理の流れを制御します。LlmAgent の「頭脳」にあたる部分で、通常はGeminiのようなLLMを指します。指示を理解し、応答を生成し、ツールを使うべきか判断します。LlmAgent 作成時に、使用するモデル（例：gemini-1.5-pro）を指定できます。指定しない場合は、ADKが適切なデフォルトモデル（APIキーに基づき）を選びます。invoke呼び出し）全体を管理する箱のようなものです。入力、最終出力、途中で発生したイベント（LLM呼び出し、ツール実行など）を保持します。デバッグや処理追跡に不可欠です。invoke呼び出し）にわたって情報を記憶するための仕組みです。これにより、エージェントは過去のやり取りを踏まえた応答（「前の質問で言った都市のことだね」など）が可能になります。Memory オブジェクトを作成し、それを Session に関連付けます。対話の履歴は Store と呼ばれる場所に保存されます（単純なメモリ内ストアや、永続化可能なストアなどがあります）。これらの要素を組み合わせることで、様々な能力を持つエージェントを構築できます。
LLMだけでは現実世界の情報を取得したり、特定の計算を正確に行ったりするのは苦手です。そこで「ツール」の出番です。簡単な「天気予報ツール（シミュレーション）」をエージェントに追加してみましょう。
まず、ツールとして使いたい機能を持つPython関数を定義します。今回は、指定された都市の天気を返す（ふりをする）関数を作ります。
weather_agent.py というファイルを作成し、以下のコードを追加します。
# weather_agent.py (開始点)
import os
import asyncio
from adk.agents import LlmAgent
from adk.sessions import Session
from adk.tools import define_function_tool # ツール定義用ヘルパー
# --- 天気予報ツール関数 ---
def get_current_weather(city: str) -> str:
"""
指定された都市の現在の天気を取得します（シミュレーション）。
Args:
city: 天気を知りたい都市名 (例: "東京", "大阪").
Returns:
その都市の天気情報を含む文字列。不明な都市の場合はその旨を返します。
"""
print(f"--- ツール実行: get_current_weather(city='{city}') ---")
city = city.lower() # 都市名を小文字に統一
if "東京" in city:
return "東京の現在の天気は晴れ、気温は28度です。"
elif "大阪" in city:
return "大阪の現在の天気は曇り、気温は26度です。"
elif "札幌" in city:
return "札幌の現在の天気は雨、気温は18度です。"
else:
return f"すみません、「{city}」の天気情報は現在取得できません。"
# --- エージェント設定と実行部分 (後で追加) ---
async def main():
pass # ここにエージェント処理を書く
if __name__ == "__main__":
asyncio.run(main())
get_current_weather 関数は、都市名を受け取り、事前に定義された天気情報を返します。docstringと型ヒント（city: str, -> str）が重要です。
define_function_tool を使って、この関数をADKが認識できるツールに変換します。
weather_agent.py の main 関数を以下のように更新します。
# weather_agent.py (続き)
import os
import asyncio
from adk.agents import LlmAgent
from adk.sessions import Session
from adk.tools import define_function_tool
from adk.models import Model # 必要に応じてモデル指定用
# --- get_current_weather 関数 (上記で定義) ---
def get_current_weather(city: str) -> str:
# ... (関数実装は上と同じ) ...
print(f"--- ツール実行: get_current_weather(city='{city}') ---")
city = city.lower()
if "東京" in city: return "東京の現在の天気は晴れ、気温は28度です。"
elif "大阪" in city: return "大阪の現在の天気は曇り、気温は26度です。"
elif "札幌" in city: return "札幌の現在の天気は雨、気温は18度です。"
else: return f"すみません、「{city}」の天気情報は現在取得できません。"
async def main():
"""天気予報ツールを持つエージェントを作成し、実行する"""
print("天気予報ツールを定義中...")
# Python関数をADKツールに変換
weather_tool = define_function_tool(get_current_weather)
print(f"定義されたツール: {weather_tool.name}") # 'get_current_weather'
print("\nツールを持つエージェントを準備中...")
# LlmAgent作成時に、toolsリストに作成したツールを渡す
agent = LlmAgent(
name="WeatherReporterAgent",
tools=[weather_tool]
# 必要であれば model=Model(...) でモデルを指定
)
session = Session()
prompt = "現在の大阪の天気を教えてください。"
print(f"\nエージェントへの指示: '{prompt}'")
response_data = await agent.invoke(prompt, session=session)
print("\nエージェントからの最終応答:")
print(response_data.get('output', '応答がありませんでした。'))
print("\n--- 処理の記録 ---")
# イベントの中に ToolStart, ToolEnd が含まれることを確認
for event in session.get_events():
print(event)
print("------------------")
if __name__ == "__main__":
asyncio.run(main())
define_function_tool(get_current_weather): Python関数 get_current_weather をADKツールオブジェクトに変換します。関数の名前、docstring（説明）、引数（型ヒント含む）が自動的に解析され、LLMがツールを理解・利用するために使われます。LlmAgent(..., tools=[weather_tool]): エージェントを作成する際に tools 引数にツールのリストを渡すことで、エージェントがそのツールを利用できるようになります。get_current_weather）が必要になるような指示を与えます。実行結果を見ると、エージェントが get_current_weather ツールを呼び出し、その結果を使って応答を生成していることがわかります。
天気予報ツールを定義中...
定義されたツール: get_current_weather
ツールを持つエージェントを準備中...
エージェントへの指示: '現在の大阪の天気を教えてください。'
--- ツール実行: get_current_weather(city='大阪') ---
エージェントからの最終応答:
はい、大阪の現在の天気は曇りで、気温は26度です。
--- 処理の記録 ---
AgentStart(agent_name="WeatherReporterAgent", input={'input': '現在の大阪の天気を教えてください。'})
LlmStart(...)
ToolStart(tool_name="get_current_weather", tool_input={'city': '大阪'})
ToolEnd(tool_name="get_current_weather", tool_output="大阪の現在の天気は曇り、気温は26度です。")
LlmStart(...)
LlmEnd(...)
AgentEnd(agent_name="WeatherReporterAgent", output={'output': 'はい、大阪の現在の天気は曇りで、気温は26度です。'})
------------------
イベントログに ToolStart と ToolEnd が記録されており、エージェントがプロンプトに応じてツールを正しく呼び出したことが確認できます。LLMはツールの説明（docstring）を見て、天気に関する質問には get_current_weather ツールが適していると判断し、必要な引数（city='大阪'）を抽出してツール実行を指示したわけです。
一問一答だけでなく、前の会話の内容を踏まえたやり取りができると、エージェントはより自然で賢く見えます。ADKの「メモリ」機能を使って、会話履歴を記憶させてみましょう。天気予報エージェントを拡張します。
ConversationMemory と、メモリの保存場所となる SimpleStore をインポートします。
weather_agent.py の先頭に以下を追加します（または memory_weather_agent.py を作成）。
# memory_weather_agent.py (または weather_agent.py を更新)
# ... 他のインポート ...
from adk.memory import ConversationMemory
from adk.store import SimpleStore # シンプルなインメモリ（RAM上）ストア
会話履歴を保存するための SimpleStore と、それを管理する ConversationMemory を作成します。
main 関数を以下のように変更します。
# ... (get_current_weather 関数 と define_function_tool は同じ) ...
async def main():
"""会話メモリを持つ天気予報エージェントを実行する"""
weather_tool = define_function_tool(get_current_weather)
print(f"定義されたツール: {weather_tool.name}")
# --- メモリ設定 ---
print("\n会話メモリを準備中...")
# 1. 履歴の保存場所（ストア）を作成
memory_store = SimpleStore()
# 2. ストアを使ってメモリ管理オブジェクトを作成
conversation_memory = ConversationMemory(
store=memory_store,
session_id="weather_chat_01", # 会話ごとのID
input_key="input", # ユーザー入力のキー名
output_key="output" # エージェント応答のキー名
)
# --------------------
print("\nツールとメモリを持つエージェントを準備中...")
agent = LlmAgent(
name="MemoryWeatherAgent",
tools=[weather_tool]
)
# --- 連続した対話 ---
# セッションを作成し、memory引数に準備したメモリ オブジェクトを渡す
session = Session(memory=conversation_memory)
prompts = [
"今日の札幌の天気は？",
"ありがとう。じゃあ、その都市の明日の天気はわかりますか？" # 「その都市」が札幌を指すことを期待
]
for turn, prompt_text in enumerate(prompts):
print(f"\n--- 対話ターン {turn + 1} ---")
print(f"ユーザー: '{prompt_text}'")
# 毎回同じセッション（メモリ付き）を使ってエージェントを呼び出す
response_data = await agent.invoke(prompt_text, session=session)
print("\nエージェント:")
print(response_data.get('output', '応答がありませんでした。'))
print("\n--- 最終ターン後のメモリ内容 ---")
# メモリに保存された全履歴を確認
history = await conversation_memory.load()
print(history)
print("-----------------------------")
if __name__ == "__main__":
asyncio.run(main())
SimpleStore() / ConversationMemory(...): 会話履歴をRAM上に保持する設定です。session_id は複数の会話を区別する場合に使います。input_key と output_key は、メモリがユーザー入力とエージェント応答を正しく識別・保存するために必要です。session = Session(memory=conversation_memory): セッション作成時に memory 引数を指定するのが重要です。これにより、セッションが対話履歴を管理できるようになります。session オブジェクトを使い回して agent.invoke を複数回呼び出します。ADKは内部で、LLMを呼び出す前に conversation_memory から過去の履歴を読み込み、プロンプトに含めます。応答後には、最新のやり取りをメモリに保存します。get_current_weather ツールは明日の天気を返せないので、その旨の応答になるでしょう）。python memory_weather_agent.py # または更新したファイル名
memory.load() で取得した履歴が表示され、2つのターンのやり取りが記録されていることが確認できます。... (ターン1の出力) ...
--- 対話ターン 2 ---
ユーザー: 'ありがとう。じゃあ、その都市の明日の天気はわかりますか？'
--- ツール実行: get_current_weather(city='札幌') ---
エージェント:
申し訳ありませんが、現在の天気しかお伝えできません。get_current_weatherツールでは、札幌の明日の天気予報を取得することはできません。
--- 最終ターン後のメモリ内容 ---
{'history': "Human: 今日の札幌の天気は？\nAI: 札幌の現在の天気は雨、気温は18度です。\nHuman: ありがとう。じゃあ、その都市の明日の天気はわかりますか？\nAI: 申し訳ありませんが、現在の天気しかお伝えできません。get_current_weatherツールでは、札幌の明日の天気予報を取得することはできません。"}
-----------------------------
これで、エージェントが会話の流れを記憶し、文脈に応じた応答をするようになりました。
エージェントに自由な判断をさせるだけでなく、特定のステップを順番に実行させたい場合があります。SequentialAgent は、複数のエージェントやツールを定義された順序で実行するためのワークフローエージェントです。
例として、「トピックを与えると、それに関する簡単な質問を生成し、その質問に自分で答える」という2段階のワークフローを作ってみましょう。
# sequential_qa_agent.py
# ... 他のインポート ...
from adk.agents import SequentialAgent
今回は2つの LlmAgent を使います。一つは質問生成用、もう一つは回答生成用です。
# sequential_qa_agent.py (完全な例)
import os
import asyncio
from adk.agents import LlmAgent, SequentialAgent
from adk.sessions import Session
from adk.models import Model
async def main():
"""質問生成と回答生成を順番に行う SequentialAgent を実行する"""
print("ワークフローのステップを定義中...")
# --- ステップ1: 質問生成エージェント ---
question_generator = LlmAgent(
name="QuestionGenerator",
system_prompt=(
"与えられたトピックに関する簡単な質問を一つだけ生成してください。"
"出力は質問文のみとします。"
)
# このエージェントにツールは不要
)
# --- ステップ2: 回答生成エージェント ---
answer_generator = LlmAgent(
name="AnswerGenerator",
system_prompt=(
"与えられた質問に対して、簡潔に答えてください。"
"出力は回答文のみとします。"
)
# このエージェントにもツールは不要
)
# --- Sequential Agent の定義 ---
print("\nSequential Agent を作成中...")
# children リストに、実行したいエージェントやツールを順番に入れる
# ステップ1 (question_generator) の出力が、ステップ2 (answer_generator) の入力となる
qa_workflow = SequentialAgent(
name="SimpleQAWorkflow",
children=[
question_generator, # 最初に実行
answer_generator # 次に実行
]
)
# --- ワークフローの実行 ---
session = Session()
initial_topic = "日本の宇宙開発"
print(f"\nワークフローを開始します。最初のトピック: '{initial_topic}'")
# invoke に最初の入力を渡す
response_data = await qa_workflow.invoke(initial_topic, session=session)
print("\nワークフローの最終応答 (回答生成エージェントの出力):")
print(response_data.get('output', '応答がありませんでした。'))
print("\n--- ワークフロー全体の処理記録 ---")
for event in session.get_events():
print(event)
print("---------------------------------")
if __name__ == "__main__":
asyncio.run(main())
question_generator: 最初のLLMエージェント。トピック（例：「日本の宇宙開発」）を受け取り、それに関する質問（例：「日本の主なロケットは何ですか？」）を生成するように指示されています（system_prompt）。answer_generator: 2番目のLLMエージェント。前のステップで生成された質問を受け取り、それに簡潔に答えるように指示されています。SequentialAgent(children=[...]): ワークフローを定義します。children リスト内の要素が順番に実行されます。SequentialAgent は、前のステップ（question_generator）の出力（生成された質問文）を、自動的に次のステップ（answer_generator）の入力として渡します。answer_generator）の出力になります。python sequential_qa_agent.py
イベントログを見ると、SimpleQAWorkflow が開始され、まず QuestionGenerator が実行されて質問を生成し、次にその出力が AnswerGenerator に渡されて回答が生成され、それが最終的な出力となっている流れが確認できます。
ワークフローのステップを定義中...
Sequential Agent を作成中...
ワークフローを開始します。最初のトピック: '日本の宇宙開発'
ワークフローの最終応答 (回答生成エージェントの出力):
日本の主力ロケットにはH-IIAやH3ロケットがあります。
--- ワークフロー全体の処理記録 ---
AgentStart(agent_name="SimpleQAWorkflow", input={'input': '日本の宇宙開発'})
AgentStart(agent_name="QuestionGenerator", input={'input': '日本の宇宙開発'})
LlmStart(...)
LlmEnd(..., response="日本の主力ロケットにはどのようなものがありますか？")
AgentEnd(agent_name="QuestionGenerator", output={'output': '日本の主力ロケットにはどのようなものがありますか？'})
AgentStart(agent_name="AnswerGenerator", input={'input': '日本の主力ロケットにはどのようなものがありますか？'})
LlmStart(...)
LlmEnd(..., response="日本の主力ロケットにはH-IIAやH3ロケットがあります。")
AgentEnd(agent_name="AnswerGenerator", output={'output': '日本の主力ロケットにはH-IIAやH3ロケットがあります。'})
AgentEnd(agent_name="SimpleQAWorkflow", output={'output': '日本の主力ロケットにはH-IIAやH3ロケットがあります。'})
---------------------------------
このように SequentialAgent を使うことで、複数の処理ステップを順番に実行する、より制御されたワークフローを構築できます。
これでADKの基本的な使い方を学びました。シンプルな応答生成から、ツール連携、会話記憶、そして簡単なワークフロー制御まで、AIエージェント開発の第一歩を踏み出せたはずです。
ADKには、さらに多くの可能性があります。
ParallelAgent): 複数のタスクを同時に実行して効率化する。LoopAgent): 条件に応じて処理を繰り返す。これらの詳細については、ぜひ公式ドキュメントやサンプルコードを参照してください。
Google Agent Development Kit (ADK) は、アイデアをインテリジェントなAIエージェントとして形にするための強力なツールキットです。Geminiモデルとの連携を活かし、ツールやメモリ、ワークフロー制御を組み合わせることで、様々なタスクを実行できるエージェントを開発できます。
このガイドを通じて、ADKの基本を掴むことができたなら幸いです。AIエージェント開発の世界は広大で、常に進化しています。ぜひ、ここで学んだことを基礎として、さらに探求し、あなただけのエージェントを作り上げてください。
Be the first to comment
