【マルチモーダル・マルチエージェント開発の総仕上げ】Agent Development Kit でイメージデザイン・アシスタントを作成する
はじめに
次の 3 つの記事では、ADK で会話型エージェントを作る基本的な方法、そして、マルチエージェントを含めたアーキテクチャーについて学びました。
今回の記事では、これらの知識をまとめた総仕上げとして、次の図のアーキテクチャーを持つ、「イメージデザイン・アシスタント」のアプリケーションを作成します。このアプリケーションでは、ユーザーは、画像生成エージェントで画像を生成して、さらに、生成した画像を画像分析エージェントで分析できます。この後の使用例で見る様に、新しいマグカップのデザインを生成して、これを SNS で宣伝するためのキャチコピーを考えるといった使い方ができます。
イメージデザイン・アシスタントのアーキテクチャー
これまでにない新しい要素として、セッション中に作成したデータを一時保存する「アーティファクト」やエージェントの出力を後処理する「コールバック」の使い方も説明します。
この後の説明では、前述の 3 つの記事で説明した用語が登場するので、まずはこれらの記事に目を通しておくことをお勧めします。
アーティファクトの仕組み
先ほどのアーキテクチャー図には、ルートエージェントにひもづいたサブエージェントとして、「画像生成エージェント」と「画像分析エージェント」があります。画像生成エージェントは、Gemini API の画像生成機能（Imagen 3）を用いて、ユーザーが希望する画像を生成します。この際、エージェントが生成した画像は、まずは一時保存領域に保存しておき、フロントエンドのアプリケーションで画面表示します。その後、ユーザーが気にいったらファイル名を指定してストレージバケットに永続保存します。
ここに、ADK の「アーティファクト」を利用します。これは、セッション中に生成したデータを一時保存する機能を提供するのもので、今回は、生成した画像をアーティファクトとして一時保存します。具体的なコードを見た方がわかりやすいので、次の例で説明します。
async def generate_image_tool(
description: str, purpose: str,
tool_context: ToolContext
) -> dict:
...(中略）...
image_bytes = generate_image(image_prompt) # PNG 画像のバイナリーデータ
image_id = uuid.uuid4().hex
image_artifact = Part( # バイナリーデータを含む Part オブジェクトを作成
inline_data=types.Blob(
mime_type='image/png',
data=image_bytes
)
)
# アーティファクトとして保存
await tool_context.save_artifact(filename=image_id, artifact=image_artifact)
...(以下省略)...
関数 generate_image_tool() は、画像生成エージェントが画像生成ツールとして利用するものです。一般に、ADK のツールとして利用する関数は、引数に tool_context: ToolContext を含めると、tool_context オブジェクトを利用して、ADK の独自機能が利用できます。ここでは、save_artifact() メソッドを用いて、画像データをアーティファクトとして保存しています。
まず、関数 generate_image() は Imagen 3 で画像を生成して、生成された PNG 画像のバイナリーデータを返却します。これを Part オブジェクトに変換したものを save_artifact() メソッドで、ファイル名を指定して保存します。それぞれのアーティファクトは、このファイル名で識別されます。このコードからわかるように、ADK の個々のアーティファクトは、Part オブジェクトの形式になります。
Part オブジェクトは Gemini API で LLM とデータをやり取りする際に使用するデータ形式で、LLM からの応答には、一般に複数の Part オブジェクト（Part オブジェクトのリスト）が含まれます。ADK のエージェントからの応答にも同様に、Part オブジェクトのリストが含まれます。
コールバックの仕組み
これで、画像生成ツールが生成した画像データをアーティファクトとして保存できましたが、このデータをフロントエンドにどうやって受け渡せばよいのでしょうか? 今回は、画像生成エージェントからの応答データに、アーティファクトから抽出した画像データを含める方法を用います。
ただし、ADK のエージェント自身には、保存済みのアーティファクトを取り出したり、これを自身の応答データに追加する機能はありません。どうすれば、そんなことができるのでしょうか? ——— ここで登場するのがコールバックの機能です。これを利用すると、エージェントが応答データを生成した後に、コールバック関数を呼び出して、この中で応答データを書き換えることができます。
今回は、エージェントにはアーティファクトの画像を表示したい部分に、<artifact>_ImageID_</artifact>（_ImageID_ はアーティファクトのファイル名）というテキストを書き込むように指示します。コールバック関数の中で、このテキストを見つけて、該当するアーティファクトから取り出した画像データを応答データに追加します。
実際に使用するものとは異なりますが、簡単化したコールバック関数は次のようになります。
async def callback_load_artifact(
callback_context: CallbackContext,
llm_response: LlmResponse
) -> LlmResponse:
parts_new = []
# 応答データ llm_response に含まれる Part オブジェクトをチェックしながら
# parts_new にコピーしていく
for part in copy.deepcopy(llm_response.content.parts):
parts_new.append(part)
pattern = '<artifact>(.+?)</artifact>'
inline_images = re.findall(pattern, part.text)
for filename in inline_images:
# <artifact>_ImageID_</artifact> のパターンがある場合は
# 該当のアーティファクトの画像データを parts_new に追加する
image_artifact = await callback_context.load_artifact(filename=filename)
image_bytes = image_artifact.inline_data.data
mime_string = to_mime_string(image_bytes) # 'data:image/jpeg;base64,...'
parts_new.append(Part.from_text(text=mime_string))
# parts_new に内容を置き換えた応答データを返却する
llm_response_new = copy.deepcopy(llm_response)
llm_response_new.content.parts = parts_new
return llm_response_new
フロントエンドのアプリケーションは、この後処理が行われた応答データを受け取ります。この応答データには Part オブジェクトのリストが含まれていますが、ある Part オブジェクトのテキストに <artifact>_ImageID_</artifact> というパターンがあれば、その直後の Part オブジェクトのテキストは、対応する画像データ（Base64 エンコードされた文字列）になるので、これを画面表示します。
なお、ツール関数の中でも、保存済みのアーティファクトを取得できます。上記の関数では、引数 callback_context: CallbackContext で受け取った callback_context オブジェクトの load_artifact() メソッドでアーティファクトを取得していますが、ツール関数の中では、tool_context オブジェクトの同じメソッドを使用します。
それでは、さっそく、「イメージデザイン・アシスタント」を実装していきましょう。まずは、サブエージェントの 1 つである、画像生成エージェントから作成します。
画像生成エージェントの作成
環境準備
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)」の新規ノートブックを作成します。
初期設定
ここからは、ノートブックのセルでコードを実行していきます。
はじめに、Agent Development Kit (ADK) のパッケージをインストールします。
%pip install --upgrade --user \
google-adk==0.5.0 \
google-cloud-aiplatform==1.93.0
インストールしたパッケージを利用可能にするために、次のコマンドでカーネルを再起動します。
import IPython
app = IPython.Application.instance()
_ = app.kernel.do_shutdown(True)
再起動を確認するポップアップが表示されるので [Ok] をクリックします。
次に、必要なモジュールをインポートして、実行環境の初期設定を行います。
import base64, copy, json, os, re, time, uuid
from io import BytesIO
import matplotlib.pyplot as plt
from PIL import Image
import vertexai
from google.cloud import storage
from google import genai
from google.genai import types
from google.genai.types import (
HttpOptions, GenerateContentConfig, GenerateImagesConfig,
Part, UserContent, ModelContent,
)
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
from google.adk.tools import ToolContext
from google.adk.agents.callback_context import CallbackContext
from google.adk.models import LlmRequest, LlmResponse
[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'
BUCKET_NAME = f'{PROJECT_ID}-adk-design'
最後の変数 BUCKET_NAME は、画像ファイルを保存するストレージバケットを指定します。必要に応じて変更しても構いません。次のコマンドで、バケットを作成します。
!gsutil mb -b on -l {LOCATION} gs://{BUCKET_NAME}
補助関数の定義
ツール、および、アプリケーションが内部で利用する補助関数を用意します。
次の generate_response() は、Gemini API で、単体の LLM として Gemini のモデルを使用する関数です。
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]
)
次の generate_image() は、Imagen 3 を用いて画像を生成します。image_prompt で記述された画像を生成して、PNG 画像のバイナリーデータを返します。
def generate_image(image_prompt, model='imagen-3.0-generate-002'):
client = genai.Client(vertexai=True,
project=PROJECT_ID, location=LOCATION,
http_options=HttpOptions(api_version='v1'))
response = client.models.generate_images(
model=model,
prompt=image_prompt,
config=GenerateImagesConfig(
number_of_images=1,
)
)
pil_image = response.generated_images[0].image._pil_image
output = BytesIO()
pil_image.save(output, format='PNG')
image_bytes = output.getvalue()
return image_bytes
次の upload_image() は、PNG 画像のバイナリーデータをファイルとしてストレージバケットに保存します。
def upload_image(image_bytes, blog_name, bucket_name=BUCKET_NAME):
client = storage.Client()
blob = client.bucket(bucket_name).blob(blog_name)
blob.upload_from_string(image_bytes, content_type='image/png')
次の 2 つの関数は、画像をノートブックの画面上に表示します。display_image_bytes() は、PNG、もしくは、JPEG 画像のバイナリーデータを受け取り、display_image() は、ストレージバケット内のファイルパスを受け取ります。
def display_image_bytes(image_bytes, title):
img = Image.open(BytesIO(image_bytes))
fig, ax = plt.subplots(figsize=(4, 4))
ax.imshow(img)
ax.set_title(title, fontsize=10)
ax.axis('off')
def display_image(blob_name, title, bucket_name=BUCKET_NAME):
client = storage.Client()
bucket = client.bucket(bucket_name)
image_bytes = bucket.blob(blob_name).download_as_bytes()
display_image_bytes(image_bytes, title)
ツール関数の定義
続いて、ツールとして使用する関数を定義します。
次の generate_image_tool() は、ユーザーの指示に従って画像を生成します。この時、ユーザーは、Imagen 3 に与えるプロンプトを直接に指定する必要はありません。生成したい画像の簡単な説明 description と画像の使用目的 purpose を与えると、これらを元にして、Gemini のモデル（LLM）が、Imagen 3 に与える適切なプロンプトを生成します。これを用いて Imagen 3 で画像を生成して、得られた結果をアーティファクトとして保存した上で、そのファイル名を返却します。
ファイル名は、ランダムなハッシュ値で生成されます。この後、画像生成エージェントに与えるインストラクションでは、このファイル名をイメージ ID（ImageID）と呼んでいます。
async def generate_image_tool(
description: str, purpose: str,
tool_context: ToolContext
) -> dict:
"""
Generate an image based on description and its purpose.
Args:
description (str): short description of an image that the user wants to create.
purpose (str): the purpose of an image that the user wants to use the image for.
Returns:
dict: A dictionary with the following keys:
image_id: unique id of the image
error: error message in case of exception
"""
try:
system_instruction = '''
Generate a text string that will be used as a prompt to generate an image with Imagen.
Based on the following user input, add detailed information as much as possibe.
[description]: Description of an image.
[purpose]: The purpose of an image that the user wants to use the image for.
'''
contents = f'''
[description]
{description}
[purpose]
{purpose}
'''
response_schema = {
"type": "object",
"properties": {
"prompt": {"type": "string"},
},
"required": ["prompt"],
}
response = json.loads(
generate_response(system_instruction, contents, response_schema)
)
image_bytes = generate_image(response['prompt'])
image_artifact = Part(
inline_data=types.Blob(
mime_type='image/png',
data=image_bytes
)
)
# Save image as artifact
image_id = uuid.uuid4().hex
await tool_context.save_artifact(filename=image_id, artifact=image_artifact)
return {'image_id': image_id}
except Exception as e:
return {'error': f'Error creating image: {e}'}
次の save_image_tool() はアーティファクトを指定のファイル名でストレージバケットにアップロードします。この際、指定されたファイル名の頭に、ユーザー ID（user_id）のフォルダ名を追加します。これにより、ユーザーごとにフォルダを分けて保存します。
async def save_image_tool(
image_id: str, filename: str,
tool_context: ToolContext
) -> dict:
"""
Tool to save an image.
Args:
image_id (str): unique id of the image
filename (str): destination filename
Returns:
dict: A dictionary containing the plan with the following keys:
result: a text message about the result
"""
user_id = tool_context._invocation_context.user_id
filename = f'{user_id}/{filename}'
try:
assert(filename.endswith('png'))
image_artifact = await tool_context.load_artifact(filename=image_id)
image_bytes = image_artifact.inline_data.data
upload_image(image_bytes, filename)
return {'result': 'Succeeded.'}
except Exception as e:
return {'result': f'Error saving image: {e}'}
さらにここで、冒頭で説明したコールバック関数 callback_load_artifact() を用意します。これは、エージェントの応答テキストに含まれる <artifact>_ImageID_</artifact> というパターンを検出して、対応するアーティファクトの画像データを応答データに追加します。引数 llm_response でエージェントの応答データを受け取って、これを修正したもの返却します。エージェントの利用者には、ここで修正された応答データが受け渡されます。
async def callback_load_artifact(
callback_context: CallbackContext,
llm_response: LlmResponse
) -> LlmResponse:
try:
if not (llm_response.content and llm_response.content.parts):
return llm_response
parts_new = []
for part in copy.deepcopy(llm_response.content.parts):
parts_new.append(part)
if not part.text:
continue
# load images from artifacts
pattern = '<artifact>(.+?)</artifact>'
inline_images = re.findall(pattern, part.text)
for filename in inline_images:
image_artifact = await callback_context.load_artifact(filename=filename)
if not image_artifact:
part.text = part.text.replace(
f'<artifact>{filename}</artifact>',
f'<artifact>_none_{filename}</artifact>'
)
parts_new[-1] = part
continue
# Convert PNG to JPEG to reduce data size
image_bytes = image_artifact.inline_data.data
img = Image.open(BytesIO(image_bytes)).convert('RGB')
img = img.resize((500, int(img.height * (500 / img.width))))
jpg_buffer = BytesIO()
img.save(jpg_buffer, 'JPEG', quality=70)
jpg_binary = jpg_buffer.getvalue()
base64_encoded = base64.b64encode(jpg_binary).decode('utf-8')
mime_string = f'data:image/jpeg;base64,{base64_encoded}'
parts_new.append(Part.from_text(text=mime_string))
llm_response_new = copy.deepcopy(llm_response)
llm_response_new.content.parts = parts_new
return llm_response_new
except Exception: # fall back to the original response
return llm_response
ADK による画像生成エージェントの実装
これでようやく準備が整いました。先に用意したツール関数とコールバック関数を用いて、画像生成エージェントを次の様に定義します。
image_creator_agent = LlmAgent(
name='image_creator_agent',
model='gemini-2.0-flash',
instruction='''
You are an agent that creates images for the user.
Output in Japanese.
[conditions]
* Avoid mentioning tool names in your reply.
* Show underscore in filenames as "_" instead of "\_".
- Good example: "my_file.png"
- Bad example: "my\_file.png"
[interaction flow]
1. Collect information.
1-1 Collect the following two items from the user:
- description: short description of an image.
- purpose: the purpose of an image that they want to use for.
If possible, infer these items from the previous conversation.
2. Summarize and get an approval.
2-1 Present the summary of the collected information in a bullet list.
2-2 Ask the user if they want to create an image based on it or update the information.
2-3 If you get an approval, go to step 3. If not, ask the user to update the information.
3. Create an image.
3-1 Use generate_image_tool() to create an image using the information you collected.
The tool returns an unique ImageID. Make sure to use the tool. Don't fake the result.
- If ImageID is empty, some error has occured. Retry to create an image.
3-2 Display the image by inserting a line "<artifact>ImageID</artifact>" in your message where
replacing ImageID with the real ImageID you received.
The user will see the real image in that part.
3-3 Ask the user 'Do you want to save the image with specifying a filename ending with .png,
or update information to create another image?'
You need to mention the fact that the user needs to specify a filename.
- If they want to save, go to step 4.
- If they want to update information, go back to step 2 with keeping the current information.
4. Save the image.
4-1 Use save_image_tool() to save the image using the ImageID and filename.
Make sure to use the tool. Don't fake the result.
- You don't need to say anything before saving an image.
- If the filename lacks file extension, or the extension is different from '.png',
you should change the extension to '.png' (case sensitive.)
You don't need to get a permission from the user for this modification.
- Once you saved the image, report the filename that you used to the user.
4-2 Ask the user if they want to update information to create another image.
- If they want to update information, go back to step 2 with keeping the current information.
''',
description='''
An agent that creates and saves an image by collecting necessary information.
''',
tools = [
generate_image_tool,
save_image_tool,
],
after_model_callback=callback_load_artifact
)
最後のオプション after_model_callback は、エージェントが応答データを生成した後に呼び出すコールバック関数を指定します。
エージェントに与えるインストラクション instruction が長く感じるかもしれませんが、ここでは、画像を生成・保存する際の次のフローを記述しています。
- ユーザーから
descriptionとpurposeの情報を収集する。
↓ - 収集した情報を提示して、これで画像を作成するか確認する。
↓ - 生成した画像を表示して、ファイル名を指定して保存するか確認する。
↓ - 指定のファイル名でストレージバケットに保存する。
生成された画像が不満な場合は、保存せずに、生成する画像の情報を修正するなどのやり取りができます。
ここで、簡易的なローカルアプリケーションのクラス LocalApp を定義して、実際にエージェントを使ってみましょう。クラス LocalApp を次の様に定義します。
class LocalApp:
def __init__(self, agent, app_name='local_app', user_id='default_user'):
self._agent = agent
self._user_id = user_id
self._app_name = app_name
self._runner = Runner(
app_name=self._app_name,
agent=self._agent,
artifact_service=InMemoryArtifactService(),
session_service=InMemorySessionService(),
memory_service=InMemoryMemoryService(),
)
self._session = self._runner.session_service.create_session(
app_name=self._app_name,
user_id=self._user_id,
state={},
session_id=uuid.uuid4().hex,
)
def _process_parts(self, author, parts, num_images):
def replace(match):
nonlocal num_images
num_images += 1
return f'## Displayed image [{num_images}] ##'
response = ''
while parts:
part = parts.pop(0)
if not part.text:
continue
# Display images in artifacts
pattern = '<artifact>(.+?)</artifact>'
inline_images = re.findall(pattern, part.text)
for c, filename in enumerate(inline_images):
if filename.startswith('_none_'):
continue
base64_data = parts.pop(0).text.split(',')[1]
image_bytes = base64.b64decode(base64_data)
display_image_bytes(image_bytes, f'[{num_images+c+1}]')
part.text = re.sub(pattern, replace, part.text)
# Display images in the storage bucket
pattern = '<image>(.+?)</image>'
bucket_images = re.findall(pattern, part.text)
for c, filename in enumerate(bucket_images):
title = f'[{num_images+c+1}] {filename}'
display_image(f'{self._user_id}/{filename}', title)
part.text = re.sub(pattern, replace, part.text)
response += f'[{author}]\n\n{part.text}\n'
return response, num_images
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 = []
num_images = 0
async for event in async_events:
if DEBUG:
print(f'----\n{event}\n----')
if not (event.content and event.content.parts):
continue
parts = copy.deepcopy(event.content.parts)
response, num_images = self._process_parts(
event.author, parts, num_images
)
if response:
print(response)
result.append(response)
return result
プライベートメソッド _process_parts() では、エージェントからの応答データに含まれる Part オブジェクトを順番に調べて、応答テキストに <artifact>_ImageID_</artifact> のパターンがあった場合は、その直後の Part オブジェクトから画像データを取り出して表示する処理を行っています。また、この後で実装する画像分析エージェントでは、ストレージバケットの画像データを表示する必要があり、そのための処理も行っています。
それでは、実際に利用してみましょう。
client = LocalApp(image_creator_agent, user_id='alice')
DEBUG = False
query = '''
何ができますか？
'''
_ = await client.stream(query)
[出力結果]
[image_creator_agent]
画像を作成できます。画像に関する説明と、その画像の用途を教えてください。
指示どおりに、用途と説明を伝えます。
query = '''
お腹が空いたので、ピザの画像を見て我慢します。
できるだけ美味しそうで、満腹になりそうなピザの画像を作成して。
'''
_ = await client.stream(query)
[出力結果]
[image_creator_agent]
ピザの画像ですね。かしこまりました。
画像を生成するために、以下の情報でよろしいでしょうか？
* 説明: 美味しそうで、満腹になりそうなピザ
* 用途: お腹が空いたときに見て我慢するため
上記の情報で画像を作成しますか？それとも、情報を更新しますか？
問題なさそうです。作成を依頼します。
query = '''
作成して。
'''
_ = await client.stream(query)
[出力結果]
[image_creator_agent]
ピザの画像を生成します。少々お待ちください。
[image_creator_agent]
ピザの画像が生成できました。
## Displayed image [1] ##
画像をファイル名指定で保存しますか？それとも、情報を更新して別の画像を作成しますか？
ファイル名を指定する場合は、末尾に.pngをつけてください。
美味しそうで、満腹になりそうなピザができました! 保存しておきましょう。
query = '''
pizza で保存
'''
_ = await client.stream(query)
[出力結果]
[image_creator_agent]
画像は pizza.png という名前で保存されました。
情報を更新して別の画像を作成しますか？
興奮のあまりファイル名に .png をつけ忘れましたが、エージェントが気をきかせて補完してくれました。次の様に、ユーザー名のフォルダーにファイルが保存されています。
!gsutil ls gs://{BUCKET_NAME}/alice/
[出力結果]
gs://[BUKET_NAME]/alice/pizza.png
画像生成エージェントの動作確認はここまでにします。ストレージバケットに保存されたファイルは、いったん削除しておきます。
!gsutil rm gs://{BUCKET_NAME}/alice/*
画像分析エージェントの作成
もう一つのサブエージェントである「画像分析エージェント」を作成します。このエージェントの仕組みは、下記の記事で作成したものとほぼ同じですが、ユーザー ID のフォルダー内の画像データを分析対象とするように修正しておきます。
まず、フォルダー内のファイルの一覧を取得するツール関数 list_files_tool() を定義します。
async def list_files_tool(tool_context: ToolContext) -> dict:
"""Tool to list available image files for the user."""
user_id = tool_context._invocation_context.user_id
try:
storage_client = storage.Client()
bucket = storage_client.bucket(BUCKET_NAME)
blobs = bucket.list_blobs()
files = []
for blob in blobs:
if not blob.name.startswith(f'{user_id}/'):
continue
if blob.name.endswith('/'):
continue
filename = blob.name.lstrip(f'{user_id}/')
files.append(filename)
return {'files': files}
except Exception as e:
return {'error': e}
続いて、指定の画像ファイルを分析するツール関数 analyze_image_tool() を定義します。
async def analyze_image_tool(instruction: str, filename: str, tool_context: ToolContext):
"""
Tool to analyze an image using LLM.
Args:
instruction (str): Instruction to LLM on how to analyze an image
filename (str): Filename of an image
Returns:
dict: A dictionary containing the plan with the following keys:
filename: Filename of an image
result: Result of the analysis
"""
user_id = tool_context._invocation_context.user_id
filename = f'{user_id}/{filename}'
try:
storage_client = storage.Client()
bucket = storage_client.bucket(BUCKET_NAME)
blob = bucket.blob(filename)
blob.reload()
mime_type = blob.content_type
image = Part.from_uri(
file_uri=f'gs://{BUCKET_NAME}/{filename}',
mime_type=mime_type
)
parts = []
parts.append(Part.from_text(text='[instruction]\n{instruction}'))
parts.append(Part.from_text(text='[image]'))
parts.append(image)
contents = contents=[UserContent(parts=parts)]
system_instruction = '''
You are a professional image analyst.
Analyze the [image] based on the [instruction].
'''
response_schema = {
"type": "object",
"properties": {
"filename": {"type": "string"},
"result": {"type": "string"},
},
"required": ["filename", "result"],
}
result = generate_response(
system_instruction, contents,
response_schema, model='gemini-2.0-flash-001'
)
return json.loads(result)
except Exception as e:
return {
'filename': filename,
'result': f'An unexpected error occurred: {e}'
}
そして、これらの関数をツールとして利用する、画像分析エージェント image_analyst_agent を次の様に定義します。
image_analyst_agent = LlmAgent(
name='image_analyst_agent',
model='gemini-2.0-flash',
instruction='''
You are an agent that analyze image files.
Output in Japanese.
[conditions]
* Avoid mentioning tool names in your reply.
* Show underscore in filenames as "_" instead of "\_".
- Good example: "my_file.png"
- Bad example: "my\_file.png"
[tasks]
** List image files
* You can get a list of image files with list_files_tool().
Make sure to use the tool. Don't fake the result.
** Display image
* You can insert an image in your message by inserting a line "<image>_filename_</image>".
Replace _filename_ with the filename of the image file.
Then the user will see the real image in that part.
** Analyze image
* If you need to analyze an image, use analyze_image_tool().
Make sure to use the tool. Don't fake the result.
** Answer questions
* You give an answer to questions from the user regarding images.
- Analyze the image with the tool "analyze_image" if necessary.
- Based on the previous conversation, infer the filename of the image
that the user is asking about.
''',
description='''
An agent that manage and handling the user's requests regarding image files.
This agent can access saved images, and analyze them.
''',
tools = [
list_files_tool,
analyze_image_tool,
],
)
イメージデザイン・アシスタントの作成
これで、冒頭のアーキテクチャー図に示した 2 つのエージェント「画像生成エージェント」と「画像分析エージェント」ができたので、これらをサブエージェントとして束ねるルートエージェントとして、イメージデザイン・アシスタント image_design_assistant を定義します。ここでは、global_instruction で、それぞれのエージェントにニックネームをつけています。
image_design_assistant = LlmAgent(
name='image_design_assistant',
model='gemini-2.0-flash',
global_instruction='''
* A nickname of the root agent (image_design_assistant) is 'コンシェルジュ・イマージュ'.
* A nickname of image_creator_agent is 'ジェネシス・ヴィジョン'.
* A nickname of image_analyst_agent as 'アナリティクス・アイ'.
* Each agent should say their name at the first conversation
or when they were transferred from another agent.
* You should transfer the agent without asking the user.
''',
instruction='''
You are an agent that help users create and analyze images.
Output in Japanese.
''',
description='''
An agent that helps users create and analyze images.
''',
sub_agents = [
copy.deepcopy(image_creator_agent),
copy.deepcopy(image_analyst_agent),
]
)
さっそく利用してみましょう。
client = LocalApp(image_design_assistant, user_id='alice')
DEBUG = False
query = '''
何ができますか？
'''
_ = await client.stream(query)
[出力結果]
[image_design_assistant]
コンシェルジュ・イマージュです。私は画像作成と分析のお手伝いをすることができます。画像を作成したい場合はジェネシス・ヴィジョンに、画像を分析したい場合はアナリティクス・アイに転送できます。
なるほど。画像作成をお願いしてみましょう。
query = '''
画像を作成してください。
ショップで販売する新しいマグカップのデザインを考えていて、
パステルカラーで洗練された、全面がカピバラ柄のサンプルが欲しいです。
'''
_ = await client.stream(query)
[出力結果]
[image_creator_agent]
ジェネシス・ヴィジョンです。
画像を生成するために、以下の情報でよろしいでしょうか？
* 説明: パステルカラーで洗練された、全面がカピバラ柄のマグカップ
* 目的: ショップで販売する新しいマグカップのデザインサンプル
上記で画像を作成しますか？ または、情報を更新しますか？
ジェネシス・ヴィジョンが登場して、依頼情報をまとめてくれました。これで作成しましょう。
query = '''
作成して。
'''
_ = await client.stream(query)
[出力結果]
[image_creator_agent]
画像を生成します。少々お待ちください。
[image_creator_agent]
こちらが生成された画像です。
## Displayed image [1] ##
画像をファイル名（末尾は.png）を指定して保存しますか？ または、情報を更新して別の画像を生成しますか？
よさそうですね。ファイル名を指定して保存します。
query = '''
kapi_mug.png で保存して。
'''
_ = await client.stream(query)
[出力結果]
[image_creator_agent]
画像を kapi_mug.png で保存しました。
情報を更新して別の画像を生成しますか？
次は、この商品の販売戦略を考えましょう。
query = '''
いいえ。この画像からSNSで拡散するキャッチコピーを考えて。
'''
_ = await client.stream(query)
[出力結果]
[image_creator_agent]
そのご質問にはアナリティクス・アイが適していますので、転送します。
[image_analyst_agent]
アナリティクス・アイです。
画像ファイル kapi_mug.png からSNSで拡散するキャッチコピーを考えるのですね。承知いたしました。どのようなキャッチコピーをご希望ですか？
例えば、
* 商品のターゲット層（例：20代女性、ファミリー層など）
* アピールしたいポイント（例：かわいさ、機能性、限定感など）
* SNSのプラットフォーム（例：Instagram、Twitterなど）
これらの情報があると、より効果的なキャッチコピーを提案できます。
アナリティクス・アイに依頼が転送されましたが、ファイル名など、必要な情報はきちんと引き継がれています。
ターゲット層をしぼって、メッセージを作成してもらいます。
query = '''
この柄を特に好むターゲット層を考えて、そこにアピールするシンプルで効果的なメッセージをお願い。
'''
_ = await client.stream(query)
[出力結果]
[image_analyst_agent]
画像 kapi_mug.png のターゲット層を分析し、アピールするシンプルで効果的なメッセージを提案します。
[image_analyst_agent]
画像の分析結果と、カピバラ柄という点から、以下のようなターゲット層が考えられます。
* **年齢層:** 20代～40代
* **性別:** 女性
* **興味関心:** 動物好き、可愛いもの好き、癒しを求める人、SNS映えするアイテムに関心がある人
上記を踏まえ、以下のようなキャッチコピーを提案します。
* **「#カピバラ と #ほっこりティータイム 🫖 あなたの日常に、#癒やし をプラスしませんか？」**
* **「#カピバラマグ で #カフェ気分 ☕️ 見るたび #きゅん とする #可愛すぎる デザイン💕」**
* **「#頑張るあなたに #カピバラさん から #エール 📣 休憩のお供に #癒やしのマグカップ はいかが？」**
これらのキャッチコピーと画像を組み合わせてSNSに投稿することで、ターゲット層に効果的にアピールできるかと思います。
この後は、さらに分析を続けることもできますし、新しい画像の作成を依頼することもできます。画像の作成を依頼すると、ジェネシス・ヴィジョンに依頼が引き継がれます。
（参考）Agent Engine にデプロイする場合の考慮点
参考として、今回作成したエージェント image_design_assistant を Agent Engine にデプロイして利用する際のポイントをまとめておきます。
まず、このエージェントは、ストレージバケットにアクセスするので、クラウド上での実行に使用される「Vertex AI Reasoning Engine サービスエージェント」に storage.objectUser ロールを設定します。具体的には、Cloud Shell から次のコマンドを実行します。
PROJECT_ID=$(gcloud config list --format 'value(core.project)')
PROJECT_NUMBER=$(gcloud projects describe $PROJECT_ID --format='value(projectNumber)')
gcloud projects add-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:service-${PROJECT_NUMBER}@gcp-sa-aiplatform-re.iam.gserviceaccount.com" \
--role='roles/storage.objectUser'
ただし、プロジェクトで初回のデプロイを実行するまでは、該当のサービスエージェントが存在しないため、まずは、サンプルのエージェントをデプロイしてサービスエージェントを作成した後で、上記のコマンドを実行します。
また、このエージェントは、画像を扱うライブラリー Pillow を用いるので、Agent Engine にデプロイする際は、requirements に 'pillow' を追加します。
from vertexai import agent_engines
remote_agent = agent_engines.create(
agent_engine=image_design_assistant,
requirements=[
'google-adk==0.5.0',
'google-cloud-aiplatform[agent_engines]==1.93.0',
'pillow',
]
)
デプロイしたリモートエージェントをノートブック上で利用する際は、次の RemoteApp クラスを使用します。これは、LocalApp クラスをリモートエージェント用に修正したものです。
class RemoteApp:
def __init__(self, remote_agent, user_id='default_user'):
self._remote_agent = remote_agent
self._user_id = user_id
self._session = remote_agent.create_session(user_id=self._user_id)
def _process_parts(self, author, parts, num_images):
def replace(match):
nonlocal num_images
num_images += 1
return f'## Displayed image [{num_images}] ##'
response = ''
while parts:
part = parts.pop(0)
if 'text' not in part:
continue
# Display images in artifacts
pattern = '<artifact>(.+?)</artifact>'
inline_images = re.findall(pattern, part['text'])
for c, filename in enumerate(inline_images):
if filename.startswith('_none_'):
continue
base64_data = parts.pop(0)['text'].split(',')[1]
image_bytes = base64.b64decode(base64_data)
display_image_bytes(image_bytes, f'[{num_images+c+1}]')
part['text'] = re.sub(pattern, replace, part['text'])
# Display images in the storage bucket
pattern = '<image>(.+?)</image>'
bucket_images = re.findall(pattern, part['text'])
for c, filename in enumerate(bucket_images):
title = f'[{num_images+c+1}] {filename}'
display_image(f'{self._user_id}/{filename}', title)
part['text'] = re.sub(pattern, replace, part['text'])
response += f'[{author}]\n\n{part["text"]}\n'
return response, num_images
def _stream(self, query):
events = self._remote_agent.stream_query(
user_id=self._user_id,
session_id=self._session['id'],
message=query,
)
result = []
num_images = 0
for event in events:
if DEBUG:
print(f'----\n{event}\n----')
if not ('content' in event and 'parts' in event['content']):
continue
parts = copy.deepcopy(event['content']['parts'])
response, num_images = self._process_parts(
event['author'], parts, num_images
)
if response:
print(response)
result.append(response)
return result
def stream(self, query):
for _ in range(5):
result = self._stream(query)
if result:
return result
if DEBUG:
print('----\nRetrying...\n----')
time.sleep(3)
return 'Error: No response from Agent Engine.'
リモートエージェントのオブジェクトを格納した remote_agent を用いて、次の様に利用します。
remote_client = RemoteApp(remote_agent, user_id='bob')
DEBUG = False
query = '''
何ができますか？
'''
_ = remote_client.stream(query)
[出力結果]
[image_design_assistant]
コンシェルジュ・イマージュです。私は画像を作成したり、分析したりするお手伝いができます。どのようなご要望がありますか？
まとめ
この記事では、ADK によるエージェント作成の総仕上げとして、ツール関数を備えたサブエージェントをルートエージェントでまとめるという、少し複雑なアーキテクチャーに挑戦しました。今回のアーキテクチャーを振り返ると、ユーザーに提供する機能を整理した上で、関連する機能をサブエージェントにまとめて、それぞれのサブエージェントが必要な機能をツール関数として実装するという戦略が見えてきます。
また、今回のアプリケーションでは、ADK のアーティファクトを利用して、生成した画像データを一時保存する仕組みを作りました。ここでは、画像データをフロントエンドに受け渡す際に、JPEG に変換して圧縮するなどの工夫をしましたが、動画ファイルなど、データサイズが大きくなると、この方法にも限界があります。そのような場合は、コールバック関数でアーティファクトをストレージバケットにコピーしておき、フロントエンドのアプリケーションは、応答テキストに含まれるファイル名からストレージバケットのデータを参照するなどの方法を用いるのがよいでしょう。あるいは、GcsArtifactService モジュールを利用して、アーティファクトをストレージバケットに直接保存する方法もあります。
このように、作成したエージェントをアプリケーションとして作り上げるには、アプリケーション開発の技術も大切になります。MCP など、エージェントに特化した技術に加えて、アプリケーション設計・開発の技術を広く学ぶことで、エージェントの応用範囲を大きく広げることができるでしょう。
そしてまた、エージェントの作成を続けていくと、どこかのタイミングで、別々に作成したエージェントを組み合わせて利用したいシーンが生まれるかもしれません。このような場合は、既存のエージェントを A2A で連携させる方法が使えるかもしれません。A2A を利用したアーキテクチャーについては、また別の機会に紹介したいと思います。お楽しみに!
Discussion
とてもわかりやすい記事をありがとうございます！まさに
Agent Engine の環境におけるアーティファクトの利用部分でつまづいていたので早速試してみようと思います！A2A を利用したアーキテクチャーについても楽しみにしております！