dipテックブログPublicationへの投稿
📬

【ADK】自分のGメールついて質問できるマルチAIエージェントを作る

Taro
に公開
生成 AI
AIエージェント
マルチエージェント
Agent Development Kit
Agent Development Kit
tech

これは何か

ADKを使って自分のGメールについて質問できるマルチAIエージェントを構築してみました。
とりあえず動かしてみる系の記事です。最小限のコードで動作確認できればゴールなのであしからず。

記載のコードは同じ順番でColabで実行できます。

対象読者

  • ADKを用いて開発を始めたい人
  • とりあえず以下の一歩目を踏み出したい人
    • AIエージェント開発
    • マルチエージェント開発
  • AIエージェントにツールを通してWeb APIを使わせたい人

環境・技術スタック

  • 実行環境 : Google Colaboratory
  • 言語 : Python
  • フレームワーク : ADK（Agent Development Kit）
  • LLMモデル : Gemini2.5 Flash

ADK

ADK（Agent Development Kit）はGoogleが開発しているAIエージェントの開発とデプロイのための柔軟でモジュール式のフレームワークです。

https://google.github.io/adk-docs/

2025年10月現在、Python版とJava版があります。
今回はPythonを使います。

https://github.com/google/adk-python

マルチAIエージェントの実装

マルチエージェントとは、1つのAIエージェントではなく複数のエージェントで役割を分担して自律的にタスクを遂行してくれる構成です。

ルートエージェントがユーザーとの対話と他のエージェントへのタスクの割り振りを行い、サブエージェントが割り振られたタスクを専門的に処理します。

今回は、ADK公式のチュートリアルを参考に作業を進めます。

https://google.github.io/adk-docs/tutorials/agent-team/

ツールの作成

今回は関数としてツールを実装します。多分これが一番簡単な方法です。

Google APIの準備

ライブラリのインストール
GoogleのAPIを呼び出すためのクライアントライブラリと認証用のライブラリをインストールします。

!pip install google-api-python-client google-auth-httplib2 google-auth-oauthlib -q

認証情報の取得
GoogleのAPIを呼び出すにはOAuth 2.0による認証が必要です。
あらかじめ取得したACCESS_TOKENを使って認証オブジェクトを生成します。
ACCESS_TOKENを取得するまでの流れは、前回こちらの記事に書きました。

https://zenn.dev/taro000/articles/6fac047499f87b

from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build

ACCESS_TOKEN = "XXX"

creds = Credentials(token=ACCESS_TOKEN)

メール検索ツール

Gメールを検索できるツールです。
Gmail API使ってメールの検索をしたいので、クライアントライブラリのmessages().list()を使います。
query(検索クエリ)とmaxResults(最大検索結果数)を引数に渡せるので、このツールでも引数として設定します。
戻り値はメッセージのIDとスレッドのIDの辞書のリストです。

user_idには自分のGメールアドレスを入れて下さい。

from typing import Optional, List, Dict
from logging import error

def search_messages(query: str, maxResults: Optional[int] = None) -> List[Dict[str, str]]:
  """
    与えられたクエリに一致するメッセージを検索する関数。
    maxResultsが与えらない場合、適切な件数を設定する。
    
    Arguments:
      query: 検索クエリ
      maxResults: 最大検索結果数

    Returns:
      messages: 検索結果 (list)
        dict:
          id: メッセージのID
          threadId: スレッドのID
  """
  try:
    creds = Credentials(token=ACCESS_TOKEN)
    service = build("gmail", "v1", credentials=creds)

    user_id = "hogehoge@gmail.com"

    # クエリ検索結果を指定された件数取得
    matched_messages = service.users().messages().list(userId=user_id, q=query, maxResults=maxResults).execute()
    messages = matched_messages.get("messages", [])

    if not messages:
      return None
    
    return messages

  except Exception as e:
    error(e)
    return None

メール本文取得ツール

特定のGメールの本文を取得できるツールです。
クライアントライブラリのmessages().list()を使います。
id（メッセージID）が引数として必要なので、このツールでも引数として設定します。
戻り値はメッセージの本文です。

クライアントライブラリで返却される本文データがBase64 URLエンコードされた文字列なので、UTF-8文字列にデコードするユーティリティ関数decode_base64url_dataも実装しました。

import base64

def get_message(message_id: str) -> str:
  """
  メッセージIDに一致するメール本文を返却する関数。

  Arguments:
    message_id: メッセージID

  Returns:
    message_body: メール本文
  """
  try:
    creds = Credentials(token=ACCESS_TOKEN)
    service = build("gmail", "v1", credentials=creds)

    user_id = "hogehoge@gmail.com"

    message = service.users().messages().get(userId=user_id, id=message_id).execute()

    if "data" in message["payload"]["body"]:
      message_body = message["payload"]["body"]["data"]
    else:
      parts = message['payload']['parts']
      parts = [part for part in parts if part['mimeType'] == 'text/plain']
      message_body = parts[0]['body']['data']
    
    if message_body is None:
      return None

    message_body = decode_base64url_data(message_body)
  
    return message_body
  except Exception as e:
    error(e)
    return None

def decode_base64url_data(data):
  """
  データをBase64 -> UTF-8にデコードする関数

  Arguments:
    data: デコードするデータ

  Returns:
    decoded_message: デコードされたデータ
  """
  decoded_bytes = base64.urlsafe_b64decode(data)
  decoded_message = decoded_bytes.decode("UTF-8")
  return decoded_message

サブエージェントの作成

ADKのインストール

!pip install google-adk -q

LLMモデルの準備
LLMを呼び出すための準備も必要です。今回はGeminiを使います。（もちろんOpenAIやAnthropicのモデルも使えます）
まずGoogle AI StudioでAPIキーを発行して下さい。

https://ai.google.dev/gemini-api/docs/api-key?hl=ja

GOOGLE_API_KEYという環境変数を設定しておくことで、ADKがよしなにAPIキーを読み込んでくれます。

import os

os.environ["GOOGLE_API_KEY"] = "XXX"

MODEL_GEMINI_2_5_FLASH = "gemini-2.5-flash"

Gメール検索エージェント

Gメール検索を専門とするサブエージェントです。
先ほど作成したsearch_messagesをツールとして設定します。

from google.adk.agents import Agent

search_gmail_agent = None

search_gmail_agent = Agent(
    model = MODEL_GEMINI_2_5_FLASH,
    name="search_gmail_agent",
    instruction="あなたはメール検索エージェントです。あなたの仕事は検索ワードにマッチするメールを返却することです。"
    "メールの検索には'search_messages'を使って下さい。"
    "もしユーザーが最大検索結果数を指定した場合は、その値をツールに渡して下さい。"
    "他の会話や作業を行わないでください。",
    description="'search_messages' ツールを使ってメールを検索する",
    tools=[search_messages],
)

Gメール本文取得エージェント

Gメールの本文取得を専門とするサブエージェントです。
先ほど作成したget_messageをツールとして設定します。

get_gmail_agent = None

get_gmail_agent = Agent(
    model = MODEL_GEMINI_2_5_FLASH,
    name="get_gmail_agent",
    instruction="あなたはメール本文取得エージェントです。あなたの仕事はIDにマッチするメールの本文を返却することです。"
    "メール本文の取得には'get_message'を使って下さい。"
    "他の会話や作業を行わないでください。",
    description="'get_message' ツールを使ってメール本文を取得する",
    tools=[get_message],
)

ルートエージェントの作成

サブエージェントを束ねて、ユーザからの質問を直接受け付けるルートエージェントです。
先ほど実装したsearch_gmail_agentとget_gmail_agentをサブエージェントに設定し、それぞれの役割を説明しておきます。

これでルートエージェントを中心とするマルチエージェントのチームが構築できました。

gmail_agent_team = None

gmail_agent_team = Agent(
    name="gmail_agent",
    model=MODEL_GEMINI_2_5_FLASH,
    description="メインの司令塔エージェント。リクエストをsearch_gmail_agentやget_gmail_agentに捌き、レスポンスをユーザーに返す。",
    instruction="あなたはGメールエージェントチームの司令塔です。ユーザーとの会話を通して、Gメールに関する業務を指揮してユーザーの役に立ちます。"
    "あなたは特定の仕事に特化したsub-agentsを利用できます。 sub-agents: "
    "1. 'search_gmail_agent': メール検索エージェントです。検索ワードにマッチするメール配列を返却します。検索ワードと最大検索結果数(Optional)を渡せます。"
    "2. 'get_gmail_agent': メール本文取得エージェントです。IDにマッチするメールの本文を返却します。"
    "ユーザーのリクエストを解析し、Gメールの検索が必要であれば'search_gmail_agent'に依頼し、Gメールの本文が必要であれば'get_gmail_agent'に依頼します。"
    "それ以外の事柄については、適切に対応するか、対応できない旨を伝えてください。",
    sub_agents=[search_gmail_agent, get_gmail_agent]
)

セッションとランナーの作成

from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner

session_service = InMemorySessionService()

APP_NAME = "gmail_team_app"
USER_ID = "user_1"
SESSION_ID = "session_001"

session = await session_service.create_session(
    app_name=APP_NAME,
    user_id=USER_ID,
    session_id=SESSION_ID
)

runner = Runner(
    agent=gmail_agent_team,
    app_name=APP_NAME,
    session_service=session_service
)

質問してみる

質問用の関数

Colab上でエージェントと会話するためのテスト関数です。

from google.genai import types

async def call_agent_async(query: str, runner, user_id, session_id):
  """
  エージェントにユーザーの質問を送り、返答をプリントする関数。
  """
  print(f"\n>>> User Query: {query}")

  # ユーザーの質問をADKのフォーマットにする
  content = types.Content(role='user', parts=[types.Part(text=query)])

  final_response_text = "エージェントが回答に失敗しました。" # デフォルトメッセージ

  # 最終返答が返却されるまで、エージェントを繰り返し実行
  async for event in runner.run_async(user_id=user_id, session_id=session_id, new_message=content):
      # is_final_response()で最終返答が返却されたかを確認
      if event.is_final_response():
          if event.content and event.content.parts: # 最終返答の取り出し
             final_response_text = event.content.parts[0].text
          elif event.actions and event.actions.escalate: # エラー対応
             final_response_text = f"Agent escalated: {event.error_message or 'No specific message.'}"
          break # 最終返答が見つかったらプロセスを出る

  print(f"<<< Agent Response: {final_response_text}")

Let's 質問

質問1

質問1
await call_agent_async(
    "こんにちは。",
    runner=runner,
    user_id=USER_ID,
    session_id=SESSION_ID
)
返答1
>>> User Query: こんにちは。
<<< Agent Response: こんにちは！Gメールに関するご質問やお手伝いできることがあれば、お気軽にお申し付けください。

質問2

質問2
await call_agent_async(
    "Findyに関するメールをいくつか要約して下さい。",
    runner=runner,
    user_id=USER_ID,
    session_id=SESSION_ID
)
返答2
>>> User Query: Findyに関するメールをいくつか要約して下さい。
<<< Agent Response: Findyに関するメールが多数見つかりました。これらのメールを要約するために、まずは最新の数件のメールの本文を取得する必要があります。

何件のメールを要約しますか？

質問3

質問3
await call_agent_async(
    "5件お願いします。",
    runner=runner,
    user_id=USER_ID,
    session_id=SESSION_ID
)
返答3
>>> User Query: 5件お願いします。
<<< Agent Response: Findyに関する直近5件のメールの要約を以下に示します。

*   **メール1: SBギフト株式会社の求人**
    SBギフト株式会社（生成AI活用企業）がWebエンジニアを募集しています。リモートワーク、フルフレックスが可能で、月給は249,000円〜348,000円です。TypeScript, Ruby, PHP, Java, Ruby on Rails, React, Next.js, Angular, AWSなどのモダンな技術を利用し、少数精鋭で裁量権を持って働ける環境です。電子ギフト市場の成長も魅力の一つです。

*   **メール2: 株式会社ハイブリッドテクノロジーズの求人**
    株式会社ハイブリッドテクノロジーズ（生成AI活用企業、東証グロース市場上場）がフルスタックエンジニアを募集しています。プライム案件90%以上で、月給は470,800円〜665,900円です。Python, TypeScript, Swift, PHP, Kotlin, Java, Laravel, Angular, Next.js, Ruby on Rails, Spring Boot, AWSなどを利用し、アジャイル開発やスクラムチームでのBtoB案件が中心です。社長や役員との距離が近く、自身のアイデアを経営層に提案できる環境で、AI, NFT, メタバースなどの最先端技術にも携われます。

*   **メール3: 株式会社unerryの求人**
    株式会社unerry（生成AI活用企業）がフロントエンドエンジニアを募集しています。月間840億件超のログと4.2億IDのリアル人流データを活用し、年俸は4,000,000円〜9,000,000円です。一部リモート可能で、自社サービス開発に携わります。Vue, HTML, CSS, Reactなどを利用し、日本最大級の位置情報ビッグデータを活用したユニークな案件やUI/UX領域に大きな裁量を持って関わることができます。

*   **メール4: Findy Toolsのイベント告知**
    Findy Toolsが「3社のリアルな意思決定を深掘り！技術選定の裏側に迫る」と題したオンラインイベントを10月24日(金)19:00〜20:15に開催します。株式会社ミラティブ、株式会社マイベスト、KDDIアジャイル開発センター株式会社が登壇し、データ基盤のアーキテクチャ見極め方や各社の意思決定、トレードオフ、工夫について語られます。データエンジニア、アナリティクスエンジニア/BIエンジニア、開発組織や技術をリードしている方におすすめです。

*   **メール5: 株式会社APPLILABの求人**
    株式会社APPLILAB（生成AI活用企業）が飲食店向けサービス「K1くん」を開発するフルスタックエンジニアを募集しています。週1回リモート勤務が可能で、月給は290,000円〜です。PHP, JavaScript, HTML, CSS, Node.js, Laravel, CakePHP, AWSなどを利用し、アジャイル開発で自社サービス開発を行います。顧客視点の開発を重視し、ユーザーの意見を直接聞く機会が多く、AIを利用した開発業務の効率化も検討されており、キャリアアップも目指せる環境です。

これらのメールは主にFindyからの求人情報とイベント告知で構成されています。

ご覧の通り、会話のラリーを通してメールを5件要約して貰うことができました！
ユーザーの発言に対して、自律的にサブエージェントに役割を分担し、ツールを用いて正しい情報を取得できていることが分かります。

とても簡単にマルチAIエージェントを構築することができました。
今回の例ではサブエージェントに役割分担するほどの処理内容ではなかったですが、より複雑な処理や専門性の高い処理などはサブエージェントの役割が期待されます。

dipテックブログ
dipテックブログPublication

ディップ株式会社のテックブログです。

Discussion