はじめに
Agent Development Kit (ADK) が GitHub のリポジトリで公開されました。これは、会話型の LLM エージェントを作成するためのフレームワークで、複数の外部ツールを使い分けたり、あるいは、複数のエージェントが協調動作するマルチエージェントが作成できます。特に、ADK の特徴として、会話の流れを自然言語で簡単に記述・定義できるという点があるので、ここでは、この特徴を活かした LLM エージェントの構成例を紹介します。
環境準備
Vertex AI workbench のノートブック上で実装しながら説明するために、まずは、ノートブックの実行環境を用意しましょう。新しいプロジェクトを作成したら、Cloud Shell のコマンド端末を開いて、必要な API を有効化します。
gcloud services enable \
aiplatform.googleapis.com \
notebooks.googleapis.com \
cloudresourcemanager.googleapis.com
続いて、Workbench のインスタンスを作成します。
PROJECT_ID=$(gcloud config list --format 'value(core.project)')
gcloud workbench instances create agent-development \
--project=$PROJECT_ID \
--location=us-central1-a \
--machine-type=e2-standard-2
クラウドコンソールのナビゲーションメニューから「Vertex AI」→「Workbench」を選択すると、作成したインスタンス agent-develpment があります。インスタンスの起動が完了するのを待って、「JUPYTERLAB を開く」をクリックしたら、「Python 3(ipykernel)」の新規ノートブックを作成します。
この後は、ノートブックのセルでコードを実行していきます。
まず、Gemini 2.0 の使用に必要な GenAI SDK と、Agent Development Kit (ADK) のパッケージをインストールします。
%pip install --upgrade --user google-genai google-adk
インストール時に表示される ERROR: pip's dependency resolver does not currently take into... というエラーは無視してください。
インストールしたパッケージを利用可能にするために、次のコマンドでカーネルを再起動します。
import IPython
app = IPython.Application.instance()
_ = app.kernel.do_shutdown(True)
再起動を確認するポップアップが表示されるので [Ok] をクリックします。
エージェントのアーキテクチャー
本記事のゴールは、会話型のエージェントを実装することですが、ここでは、イベントの企画立案を支援するエージェントを作成します。新入社員の歓迎イベントの企画などに役立つエージェントで、次の 3 つの機能を持ちます。
- ユーザーの提示したゴールに応じて企画案を生成する
- ユーザーの指示に基づいて企画案を修正する
- 得られた企画案を評価して改善案を提示する
これまでに LLM でエージェントを作成した経験のある方には想像ができると思いますが、これらの機能を作り込む際は、LLM にプロンプトで与える指示（システムインストラクション）を十分に練り込んでいく必要があります。3 つの機能それぞれでインストラクションが異なるので、1 つのエージェントにこれら 3 つの機能を詰め込むのは大変です。
そこでまずは、それぞれの機能を会話機能を持たない「タスク特化型エージェント」として作り込んでおき、その上で、これらの機能を会話形式で提供する「クライアントエージェント」を用意します。今回は、「クライアントエージェント」の部分を ADK で作成します。
タスク特化型エージェントをクライアントエージェントでまとめる構成
タスク特化型エージェントの実装
それではまず、タスク特化型エージェントとして、「企画生成エージェント」「企画修正エージェント」「企画評価エージェント」を実装します。これらは会話機能を持たないので、Gemini API で Gemini 2.0 を呼び出す単純な Python の関数として実装します。
はじめに、Gemini API の利用に使うモジュールをインポートして、実行環境の初期設定を行います。
import json, os, pprint, time, uuid
import vertexai
from google import genai
from google.genai.types import (
HttpOptions, GenerateContentConfig,
Part, UserContent, ModelContent
)
[PROJECT_ID] = !gcloud config list --format 'value(core.project)'
LOCATION = 'us-central1'
vertexai.init(project=PROJECT_ID, location=LOCATION,
staging_bucket=f'gs://{PROJECT_ID}')
os.environ['GOOGLE_CLOUD_PROJECT'] = PROJECT_ID
os.environ['GOOGLE_CLOUD_LOCATION'] = LOCATION
os.environ['GOOGLE_GENAI_USE_VERTEXAI'] = 'True'
最後の 3 つの環境変数は、ADK のエージェントをローカルで実行する際に使用する Gemini API の環境を指定します。
次に、Gemini API で Gemini 2.0 の応答を得る汎用の関数 generate_response() を定義します。ここでは、Controlled Generation の機能を用いて、response_schema で指定したスキーマの JSON で応答を返すようにしています。
def generate_response(system_instruction, contents,
response_schema, model='gemini-2.0-flash-001'):
client = genai.Client(vertexai=True,
project=PROJECT_ID, location=LOCATION,
http_options=HttpOptions(api_version='v1'))
response = client.models.generate_content(
model=model,
contents=contents,
config=GenerateContentConfig(
system_instruction=system_instruction,
temperature=0.4,
response_mime_type='application/json',
response_schema=response_schema,
)
)
return '\n'.join(
[p.text for p in response.candidates[0].content.parts if p.text]
)
使用例は次のようになります。
response_schema = {
"type": "object",
"properties": {
"greeting": {"type": "string"},
},
"required": ["greeting"],
}
system_instruction = '''
ネットショップの仮想店員として、丁寧で、かつ、フレンドリーな雰囲気の挨拶を返してください。
架空の商品名などは含めないこと。
'''
contents = 'こんにちは、中井です。何か、おすすめはありますか？'
print(generate_response(system_instruction, contents, response_schema))
出力結果
{
"greeting": "中井様、こんにちは！いつもありがとうございます。何かお探しですか？最近、特におすすめなのは、新入荷のアイテムです。ぜひチェックしてみてください！もちろん、何かご希望があれば、お気軽にお知らせくださいね。"
}
企画生成エージェントの実装
これを用いて、企画生成エージェントを次のように実装します。
def _generate_plan(goal):
system_instruction = '''
You are a professional event planner. Work on the following tasks.
[task]
A. generate event contents to achieve the given [goal].
[format instruction]
In Japanese. No markdowns. The output has the following three items:
"title": a short title of the event
"summary": three sentence summary of the event
"timeline": timeline of the event such as durations and contents in a bullet list
'''
response_schema = {
"type": "object",
"properties": {
"title": {"type": "string"},
"summary": {"type": "string"},
"timeline": {"type": "string"},
},
"required": ["title", "summary", "timeline"],
}
parts = []
parts.append(Part.from_text(text=f'[goal]\n{goal}'))
contents=[UserContent(parts=parts)]
return generate_response(system_instruction, contents, response_schema)
def generate_plan(goal:str) -> dict:
"""
Create an initial plan to achieve the goal.
Args:
goal (str): The goal of the event.
Returns:
dict: A dictionary containing the plan with the following keys:
title: title of the event
summary: a short summary of the event
timeline: timeline of the event
"""
response = _generate_plan(goal)
return json.loads(response)
ここでは、ユーザーの指示 goal に応じて企画を生成する内部関数 _generate_plan() を定義した上で、そのラッパー generate_plan() を用意しています。ラッパー側は、型アノテーションを加えた上で、入出力に関するドキュメントストリングを追加していますが、これには理由があります。この後、ADK でこの関数を外部ツールとして利用する際、ADK は型アノテーションやドキュメントストリングから、関数の呼び出し方法を自動的に認識します。 なお、内部関数とラッパーを分けることは必須ではありません。ここでは、型アノテーションやドキュメントストリングの必要性を強調するために、このように実装しています。
実行例は、この後であらためて紹介します。
企画修正エージェントの実装
同様に、企画修正エージェントは次のようになります。
def _update_plan(goal, plan, evaluation):
system_instruction = '''
You are a professional event planner. Work on the following tasks.
[task]
A. given [goal] and current [plan] for event contents.
Generate an improved plan based on the given [evaluation].
[format instruction]
In Japanese. No markdowns. The output has the following three items:
"title": a short title of the event
"summary": three sentence summary of the event
"timeline": timeline of the event such as durations and contents in a bullet list
"update: one sentence summary of the update from the previous plan
'''
response_schema = {
"type": "object",
"properties": {
"title": {"type": "string"},
"summary": {"type": "string"},
"timeline": {"type": "string"},
"update": {"type": "string"},
},
"required": ["title", "summary", "timeline", "update"],
}
parts = []
parts.append(Part.from_text(text=f'[goal]\n{goal}'))
parts.append(Part.from_text(text=f'[plan]\n{plan}'))
parts.append(Part.from_text(text=f'[evaluation]\n{evaluation}'))
contents=[UserContent(parts=parts)]
return generate_response(system_instruction, contents, response_schema)
def update_plan(goal:str, plan:str, evaluation:str) -> dict:
"""
Create an updated plan to achieve the goal given the current plan and an evaluation comment.
Args:
goal (str): The goal of the event
plan (str): Current plan
evaluation (str): Evaluation comment in plain text or a JSON string
Returns:
dict: A dictionary containing the plan with the following keys:
title: title of the event
summary: a short summary of the event
timeline: timeline of the event
update: one sentence summary of the update from the previous plan
"""
response = _update_plan(goal, plan, evaluation)
return json.loads(response)
goal にイベントの目的、plan に現在の企画案、そして、evaluation に現在の企画の評価や改善方針を入力すると、修正された新しい企画案が得られます。ドキュメントストリングを見ると、evaluation は、プレインテキスト、もしくは、JSON ストリングになっています。この次に用意する企画評価エージェントは、JSON ストリングで結果を出力するので、それを入力することもできるようにしてあります。
企画評価エージェント
最後に、企画評価エージェントの実装です。
def _evaluate_plan(goal, plan):
system_instruction = '''
You are a professional event planner. Work on the following tasks.
[task]
A. given [goal] and [plan] for event contents, evaluate if the plan is effective to achieve the goal.
B. Also give 3 ideas to improve the plan.
[condition]
A. Event contents should include detailed descriptions.
[format instruction]
In Japanese. No markdowns. The output has the following three items:
"evaluation": three sentence evaluation of the plan.
"improvements": a list of 3 ideas to improve the plan. Each idea is in a single sentence.
'''
response_schema = {
"type": "object",
"properties": {
"evaluation": {"type": "string"},
"improvements": {
"type": "array",
"items": {
"type": "object",
"properties": {
"idea": {"type": "string"},
},
"required": ["idea"],
},
},
},
"required": ["evaluation", "improvements"],
}
parts = []
parts.append(Part.from_text(text=f'[goal]\n{goal}'))
parts.append(Part.from_text(text=f'[plan]\n{plan}'))
contents=[UserContent(parts=parts)]
return generate_response(system_instruction, contents, response_schema)
def evaluate_plan(goal:str, plan:str) -> dict:
"""
Generate an evaluation for the plan against the goal.
Args:
goal (str): The goal of the event
plan (str): Current plan
Returns:
dict: A dictionary containing the evaluation comment with the following keys:
evaluation: evaluation comment
improvements: list of ideas for improvements
"""
response = _evaluate_plan(goal, plan)
return json.loads(response)
goal にイベントの目的、plan に現在の企画案を入力すると、評価コメントと 3 つの改善案が得られます。
ワークフローで連携する例
最終ゴールは、これらのタスク特化型エージェントをまとめた会話型のエージェントを作ることですが、まずは、タスク特化型エージェントの動作確認をかねて、これらをワークフロー的に連携して実行してみましょう。
まずは、企画生成エージェントで、最初の企画を立案します。
goal = 'クラウドネイティブなアプリ開発企業の新入社員歓迎イベントを11:00-14:00の180分の構成で考える'
plan = generate_plan(goal)
pprint.pp(plan)
出力結果
{'summary': 'クラウドネイティブなアプリ開発企業の新入社員歓迎イベントです。新入社員が会社に馴染み、既存社員との交流を深めることを目的とします。ランチを楽しみながら、チームビルディングアクティビティや会社紹介を通じて、親睦を深めます。',
'timeline': '・11:00-11:10 オープニング（10分）：歓迎の挨拶とイベント概要説明\n'
'・11:10-12:00 ランチ交流会（50分）：軽食を取りながら自己紹介と懇親\n'
'・12:00-13:00 チームビルディングアクティビティ（60分）：チーム対抗のゲームや課題解決\n'
'・13:00-13:50 会社紹介プレゼンテーション（50分）：会社の歴史、ビジョン、事業内容の説明\n'
'・13:50-14:00 クロージング（10分）：今後の抱負と記念撮影',
'title': '新入社員歓迎！クラウドネイティブ交流会'}
それっぽい結果が得られましたが、これを企画評価エージェントで評価してみます。
evaluation = evaluate_plan(goal, plan)
pprint.pp(evaluation)
出力結果
{'evaluation': 'この計画は、新入社員が会社に馴染み、既存社員との交流を深めるという目標に対して、時間配分や内容のバランスが取れており、効果的であると考えられます。特に、ランチ交流会とチームビルディングアクティビティは、参加者同士の親睦を深める上で非常に有効でしょう。会社紹介プレゼンテーションも、新入社員が会社の理解を深める上で重要です。',
'improvements': [{'idea': 'チームビルディングアクティビティの内容を、クラウドネイティブな技術に関連するものにすることで、会社の特徴をより強く印象付けることができます。'},
{'idea': 'ランチ交流会では、部署やチームごとにテーブルを分け、新入社員が様々な社員と話せる機会を設けると、より広範な交流が期待できます。'},
{'idea': 'クロージングで、新入社員一人ひとりが簡単に自己紹介と今後の抱負を述べる時間を設けることで、一体感を高めることができます。'}]}
評価は悪くありませんが、さらなる改善のアイデアも得られました。そこで、この評価結果と改善案を企画修正エージェントに入力して、企画をアップデートします。
plan2 = update_plan(goal, plan, json.dumps(evaluation))
pprint.pp(plan2)
出力結果
{'summary': 'クラウドネイティブなアプリ開発企業の新入社員歓迎イベントです。新入社員が会社に馴染み、既存社員との交流を深めることを目的とします。クラウド技術に関連するチームビルディングと部署を超えた交流で、親睦を深めます。',
'timeline': '・11:00-11:10 オープニング（10分）：歓迎の挨拶とイベント概要説明\n'
'・11:10-12:00 ランチ交流会（50分）：軽食を取りながら自己紹介と懇親。部署ごとにテーブルを分け交流を促進。\n'
'・12:00-13:00 チームビルディングアクティビティ（60分）：クラウド技術に関連するチーム対抗のゲームや課題解決\n'
'・13:00-13:40 会社紹介プレゼンテーション（40分）：会社の歴史、ビジョン、事業内容の説明\n'
'・13:40-14:00 クロージング（20分）：新入社員一人ずつの自己紹介と今後の抱負、記念撮影',
'title': '新入社員歓迎！クラウドネイティブ交流会',
'update': 'チームビルディングの内容をクラウド技術に関連付け、ランチ交流会での部署交流、クロージングでの新入社員の自己紹介を追加しました。'}
改善案を取り入れた新しい企画案が得られました。
ここでは、シンプルにそれぞれの関数を順番に実行していきましたが、複数のタスク特化型エージェントをワークフロー的に連携するのであれば、下記の記事で紹介した LangGraph などを用いることもできるでしょう。
一方、これらの機能を会話型のエージェントにまとめることができれば、得られた企画案に応じてユーザー自身が自分のアイデアを提示したり、あるいは、エージェントに改善案を出してもらうなど、より柔軟な使い方ができるようになります。この部分を ADK で実装してみましょう。
ADK による会話型エージェントの実装
はじめに、ADK で使用するモジュールをインポートします。
from google.adk.agents.llm_agent import LlmAgent
from google.adk.artifacts import InMemoryArtifactService
from google.adk.memory.in_memory_memory_service import InMemoryMemoryService
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
そして、ADK による会話型のクライアントエージェント planning_client_agent を次のように作成します。
instruction = """
You are an agent who handles event contents.
Your outputs should be in Japanese without markdown.
**Interaction flow:**
1. Initial plan:
* When you receive a goal of the event, you should first generate an initial plan using generate_plan().
2. Present the plan and ask for evaluation:
* Present the plan to the user, and ask the evaluation and improvement ideas.
- Show in a human readable format.
3. (Optional) Get an evaluation from 主任
* If the user requests to get an evaluation from "主任", get an evaluation using evaluate_plan()
- Present the result to the user in a human readable format, and ask if the user accept it or not.
* If the user accept it, go to step 4.
- When a user say something affirmative, think about if it means to accept 主任's evaluation, or other things.
4. Upadate plan:
* Once you get an evaluation from the user or "主任", generate an updated plan using update_plan().
* Go back to step 2.
"""
planning_client_agent = LlmAgent(
model='gemini-2.0-flash-001',
name='planning_client_agent',
description=(
'This agent creates and updates event contents given the goal of the event.'
),
instruction=instruction,
tools=[
generate_plan,
update_plan,
evaluate_plan,
],
)
このように、LlmAgent クラスのインスタンスとして、会話型のエージェントが作成できます。ここでは、次の 3 つの設定が特に重要な役割を果たします。
-
description: このエージェントの役割を記述 -
instruction: このエージェントの動作内容（処理のフロー）を記述 -
tools: このエージェントが利用する外部ツール（関数）を記述
description は、他のエージェントがこのエージェントを利用する際に、このエージェントに何を依頼できるかを理解するために参照されます。今回は、他のエージェントから呼び出される想定ではないので、この部分は重要ではありません。
tools には、このエージェントが利用する外部ツールを記述します。今回は、先に作成した 3 つのタスク特化型エージェントの関数を指定しています。関数を呼び出す方法は、前述のように、関数の型アノテーションやドキュメントストリングから認識します。
そして、instruction には、このエージェントの動作内容を記述します。これによって、このエージェントがユーザーとどのように会話するかが決まります。ここでは、ユーザーの入力に応じて初期の企画案を生成した後、ユーザーの指示にしたがって企画を修正していきます。特に、ユーザーが「主任の評価が知りたい」と言った場合は、企画評価エージェントによる評価結果を取得して、それをユーザーに提示した後に、この評価を受け入れるかを問い合わせるように指示しています。
それでは、このエージェントと会話してみましょう。まず、ADK のエージェントと会話するための簡易的なアプリのクラス LocalApp を作成します。
class LocalApp:
def __init__(self, agent):
self._agent = agent
self._user_id = 'local_app'
self._runner = Runner(
app_name=self._agent.name,
agent=self._agent,
artifact_service=InMemoryArtifactService(),
session_service=InMemorySessionService(),
memory_service=InMemoryMemoryService(),
)
self._session = self._runner.session_service.create_session(
app_name=self._agent.name,
user_id=self._user_id,
state={},
session_id=uuid.uuid4().hex,
)
async def stream(self, query):
content = UserContent(parts=[Part.from_text(text=query)])
async_events = self._runner.run_async(
user_id=self._user_id,
session_id=self._session.id,
new_message=content,
)
result = []
async for event in async_events:
if DEBUG:
print(f'----\n{event}\n----')
if (event.content and event.content.parts):
response = '\n'.join([p.text for p in event.content.parts if p.text])
if response:
print(response)
result.append(response)
return result
このクラスはセッション管理機能を備えており、インスタンスごとに個別のセッション（会話の履歴）を保持します。次のように、作成したエージェントを指定して、インスタンスを生成します。新しい会話を始める際は、同じコマンドで LocalApp クラスのインスタンスを再生成してください。
client = LocalApp(planning_client_agent)
まずは、初期の企画案を出してもらいます。
DEBUG = False
query = 'クラウドネイティブなアプリ開発企業の新入社員歓迎イベントを11:00-14:00の180分の構成で考えて'
_ = await client.stream(query)
出力結果
新入社員歓迎イベントの計画を作成します。
初期計画は以下の通りです。
タイトル: 新入社員歓迎会：クラウドネイティブの世界へようこそ！
概要: クラウドネイティブな開発を推進する企業の新入社員歓迎イベントです。新入社員同士の親睦を深め、会社への帰属意識を高めることを目的とします。ランチを楽しみながら、チーム対抗のゲームで交流を深めます。
タイムライン:
11:00-11:10 オープニング（10分）：社長挨拶、新入社員紹介
11:10-12:00 チーム対抗ゲーム（50分）：クラウドネイティブに関するクイズ、チームビルディングゲーム
12:00-13:00 ランチ（60分）：立食形式でのランチ、歓談
13:00-13:50 先輩社員からのメッセージ（50分）：部署紹介、キャリアパス紹介
13:50-14:00 クロージング（10分）：記念撮影、今後の抱負
この計画について、改善点や評価はありますか？
この結果について、主任の意見を聞いてみましょう。
query = '主任の意見を聞きたい'
_ = await client.stream(query)
出力結果
承知いたしました。主任にこの計画の評価を依頼します。
主任からの評価は以下の通りです。
評価:
計画は、新入社員同士の親睦を深め、会社への帰属意識を高めるという目標に対して、時間配分やコンテンツ内容から見て概ね効果的であると考えられます。チーム対抗ゲームやランチ、先輩社員からのメッセージなど、多様なプログラムを通じて交流を促進する構成となっています。しかし、クラウドネイティブに関するクイズの内容や難易度によっては、新入社員の知識レベルに差が生じ、一部の社員が疎外感を感じる可能性がある点に注意が必要です。
改善案:
* チーム対抗ゲームの内容を、クラウドネイティブの知識だけでなく、チームワークやコミュニケーション能力を測る要素も取り入れることで、より多くの新入社員が楽しめるようにする。
* ランチの時間に、各テーブルに先輩社員を配置し、新入社員が気軽に質問や相談ができる機会を設けることで、より深い交流を促す。
* 先輩社員からのメッセージに、新入社員が今後のキャリアパスを考える上で役立つ具体的な事例やアドバイスを含めることで、会社への帰属意識を高める。
主任の評価を受け入れ、計画を更新しますか？
instruction で指定したように、評価結果を受け入れるか聞いてきました。ここは、主任の意見を受け入れることにしましょう。
query = 'そうします。'
_ = await client.stream(query)
出力結果
主任の評価を受け入れるとのこと、承知いたしました。評価を反映して計画を更新します。
更新された計画は以下の通りです。
タイトル: 新入社員歓迎会：クラウドネイティブの世界へ飛び込もう！
概要: クラウドネイティブな開発を推進する企業の新入社員歓迎イベントです。新入社員同士の親睦を深め、会社への帰属意識を高めることを目的とします。チーム対抗ゲームで交流を深め、ランチ時には先輩社員との交流を促進します。
タイムライン:
11:00-11:10 オープニング（10分）：社長挨拶、新入社員紹介
11:10-12:00 チーム対抗ゲーム（50分）：クラウドネイティブに関するクイズ、チームビルディングゲーム、コミュニケーションゲーム
12:00-13:00 ランチ（60分）：立食形式でのランチ、各テーブルに先輩社員を配置し歓談
13:00-13:50 先輩社員からのメッセージ（50分）：部署紹介、キャリアパス紹介、具体的な事例やアドバイス
13:50-14:00 クロージング（10分）：記念撮影、今後の抱負
変更点:
チーム対抗ゲームの内容を充実させ、ランチ時の交流を促進し、先輩社員からのメッセージに具体的な事例やアドバイスを追加しました。
この計画について、さらに改善点や評価はありますか？
せっかくなので、自分の意見も盛り込んでみましょう。
query = 'ランチ中は、会話のきっかけになるようなプロジェクト紹介の動画を流して欲しい。先輩のメッセージは30分で十分'
_ = await client.stream(query)
出力結果
承知いたしました。ランチ中にプロジェクト紹介の動画を流し、先輩のメッセージ時間を短縮するよう計画を更新します。
更新された計画は以下の通りです。
タイトル: 新入社員歓迎会：クラウドネイティブの世界へ飛び込もう！
概要: クラウドネイティブな開発を推進する企業の新入社員歓迎イベントです。新入社員同士の親睦を深め、会社への帰属意識を高めることを目的とします。チーム対抗ゲームで交流を深め、ランチ時には会話のきっかけになるような動画を上映し、先輩社員との交流を促進します。
タイムライン:
11:00-11:10 オープニング（10分）：社長挨拶、新入社員紹介
11:10-12:00 チーム対抗ゲーム（50分）：クラウドネイティブに関するクイズ、チームビルディングゲーム、コミュニケーションゲーム
12:00-13:00 ランチ（60分）：立食形式でのランチ、各テーブルに先輩社員を配置し歓談、プロジェクト紹介動画上映
13:00-13:30 先輩社員からのメッセージ（30分）：部署紹介、キャリアパス紹介、具体的な事例やアドバイス
13:30-13:50 チーム対抗ゲーム結果発表、表彰（20分）
13:50-14:00 クロージング（10分）：記念撮影、今後の抱負
変更点:
ランチ中に会話のきっかけになるようなプロジェクト紹介動画を上映し、先輩社員からのメッセージの時間を短縮し、チーム対抗ゲームの結果発表と表彰を追加しました。
この計画でよろしいでしょうか？
これでよいことにしましょう。
query = 'ばっちぐー'
_ = await client.stream(query)
出力結果
承知いたしました。最終計画でイベントの準備を進めます。
エージェントは「イベントの準備を進めます」と言っていますが、もちろん、今回のエージェントにこのようなアクションを実行する機能はありません。instruction を修正して、このような余計な一言は言わないように修正した方がよいかもしれません。もちろん、実際にアクションを実行する関数を用意して、それを実行させるような作り込みも可能です。
Agent Engine へのデプロイ
次は、このエージェントを Agent Engine の環境にデプロイして、リモートエージェントとして利用します。次のコマンドで、先ほど作成した planning_client_agent をデプロイします。
from vertexai import agent_engines
remote_agent = agent_engines.create(
agent_engine=planning_client_agent,
requirements=[
'google-cloud-aiplatform[adk,agent_engines]',
]
)
[出力結果]
Deploying google.adk.agents.Agent as an application.
Identified the following requirements: {'cloudpickle': '3.1.1', 'google-cloud-aiplatform': '1.88.0'}
The following requirements are missing: {'cloudpickle'}
The following requirements are appended: {'cloudpickle==3.1.1'}
The final list of requirements: ['google-cloud-aiplatform[adk,agent_engines]', 'cloudpickle==3.1.1']
...（中略）...
AgentEngine created. Resource name: projects/879055303739/locations/us-central1/reasoningEngines/3036138632382513152
To use this AgentEngine in another session:
agent_engine = vertexai.agent_engines.get('projects/879055303739/locations/us-central1/reasoningEngines/3036138632382513152')
デプロイに成功すると、リモートエージェントのクライアントオブジェクトが remote_agent に保存されます。これを用いてリモートエージェントと会話する、簡易的なアプリのクラス RemoteApp を作成します。
class RemoteApp:
def __init__(self, remote_agent):
self._remote_agent = remote_agent
self._user_id = 'client_agent'
self._session = remote_agent.create_session(user_id=self._user_id)
def _stream(self, query):
events = self._remote_agent.stream_query(
user_id=self._user_id,
session_id=self._session['id'],
message=query,
)
result = []
for event in events:
if DEBUG:
print(f'----\n{event}\n----')
if ('content' in event and 'parts' in event['content']):
response = '\n'.join(
[p['text'] for p in event['content']['parts'] if 'text' in p]
)
if response:
print(response)
result.append(response)
return result
def stream(self, query):
# TODO: avoid infinite loop in case of permanent error
while True:
result = self._stream(query)
if result:
break
if DEBUG:
print('----\nRetrying...\n----')
time.sleep(3)
return result
次のように、先ほどのリモートエージェント remote_agent を指定して、インスタンスを生成します。
client = RemoteApp(remote_agent)
ローカル版と同様に、インスタンスごとに個別のセッション（会話の履歴）を保持するように作ってあるので、新しい会話を始める際は、同じコマンドで RemoteApp クラスのインスタンスを再生成してください。
この後は、ローカル版と同様に stream() メソッドで会話します。ローカル版と異なり、await は不要です。
DEBUG = False
query = 'クラウドネイティブなアプリ開発企業の新入社員歓迎イベントを11:00-14:00の180分の構成で考えて'
_ = client.stream(query)
[出力結果]
新入社員歓迎イベントの計画を作成します。まず、イベントの目標と時間に基づいて初期計画を生成します。
初期計画は以下の通りです。
**タイトル:** 新入社員歓迎会
**概要:** クラウドネイティブな開発を推進する企業の新入社員歓迎イベントです。新入社員が会社の文化に触れ、社員同士の交流を深めることを目的とします。ランチを挟み、リラックスした雰囲気で会社への理解を深めます。
**タイムライン:**
* 11:00-11:10 (10分): オープニング、歓迎の挨拶
* 11:10-11:40 (30分): 会社概要とクラウドネイティブに関する説明
* 11:40-12:00 (20分): 新入社員自己紹介
* 12:00-13:00 (60分): ランチ交流会
* 13:00-13:45 (45分): 先輩社員パネルディスカッション
* 13:45-14:00 (15分): 質疑応答とクロージング
この計画について、何か評価や改善点があれば教えてください。
query = '主任の意見を聞きたい'
_ = client.stream(query)
[出力結果]
主任にこの計画の評価を依頼します。
主任からの評価は以下の通りです。
**評価:**
この計画は、新入社員が会社の文化に触れ、社員同士の交流を深めるという目標に対して、時間配分や内容のバランスが取れており、概ね効果的であると考えられます。特に、ランチ交流会や先輩社員パネルディスカッションは、新入社員が会社の雰囲気を理解し、既存社員との関係を築く上で非常に有効でしょう。しかし、自己紹介の時間が短く、もう少し時間を確保することで、より深い相互理解につながる可能性があります。
**改善案:**
* 新入社員の自己紹介時間を30分に延長し、趣味や特技など、よりパーソナルな情報を共有する機会を設ける。
* ランチ交流会にゲーム要素を取り入れ、チーム対抗のクイズ大会などを実施して、より活発なコミュニケーションを促進する。
* パネルディスカッションのテーマを新入社員が興味を持ちやすい内容にし、事前に質問を募集することで、よりインタラクティブな時間にする。
主任の評価を受け入れ、計画を更新しますか？
今回の実装例はここまでです。Agent Engine にデプロイしたエージェントをまとめて削除する際は、次のコマンドを実行します。
for agent in agent_engines.list():
print(agent.gca_resource.name)
agent.delete(force=True)
まとめ
この記事では、Python の関数として素朴に実装したタスク特化型エージェントを ADK で実装した会話型エージェントを通じて、対話的に利用するという使い方を紹介しました。会話型エージェントの場合、ユーザーがどのような入力をしてくるか予想が付かないので、大雑把な流れを instruction で指定すると、会話の流れに応じて適切な処理を自動的に判断する ADK の機能はとても便利です。その一方で、今回の例のように、システムプロンプトの作り込みが必要な個々のタスクは、独立したエージェントとして実装しておいて、個別にチューニングを進めるのが良さそうです。
なお、個々の「タスク特化型エージェント」を今回のようにシンプルな関数として実装するのか、こちらも ADK を用いた会話型のエージェントとして実装するのかは、個々のユースケースに応じて適切に設計する必要がありそうです。関数として実装したエージェントであれば、ADK のエージェントに対しては、Function calling で連携する外部ツールとして登録することになります。一方、会話型のエージェントとして実装した場合は、別途、「会話型のエージェント同士が相互に通信する仕組み」が必要になります。
——— はい。その一例がまさに Agent 2 Agent (A2A)プロトコルですね。A2A は会話に必要なコンテキストやアーティファクトを共有しながら、会話型エージェントが連携動作するためのプロトコルです。今回のように、ワンショットで実行する関数として実装されたエージェントであれば、わざわざ A2A を使用しなくても、シンプルな外部ツールとしても利用できる点に注意してください。
また、すべての場合で会話型エージェントが適切な実装とは限りません。本文中でも触れましたが、処理の流れが明確なワークフロー型のエージェントであれば、LangGraph などのツールでシンプルに実装することも可能です。これからは、ユースケースに応じて、エージェントの実装方法や連携方法を適切に設計するスキルが重要になりそうです。
Discussion