コンテンツにスキップ

Google ADK 完全ガイド 2026:コードでインテリジェントな AI エージェントを構築

Google ADK とは?

Google ADK (Agent Development Kit) は、AI エージェントを実際のアプリケーションに統合したい開発者のために Google が提供するオープンソースの AI エージェント開発フレームワークです。LangChain や LlamaIndex などのフレームワークとは異なり、ADK は「コードファースト」(code-first)の設計思想を採用しており、エージェントの動作を従来のソフトウェアのように予測可能かつテスト可能なものにします。

コア機能

  • コードファースト:純粋な Python でエージェントの動作を定義。複雑な YAML 設定は不要
  • 型安全:Python の型ヒントを最大限に活用し、IDE の自動補完とエラーチェックを提供
  • テスト容易性:エージェントのロジックは通常の関数と同様にユニットテストが可能
  • プロダクションレディ:ログ、モニタリング、エラーハンドリングなど、本番環境に必要な機能を内蔵
  • マルチモデル対応:Gemini、GPT-4、Claude など主要な大規模言語モデルをサポート

適用シナリオ

  • カスタマーサービスチャットボット
  • 自動化ワークフローアシスタント
  • データ分析とレポート生成
  • コードレビューとプログラミング支援
  • 企業内部のナレッジ QA システム

インストールと設定

環境要件

  • Python 3.10 以降
  • pip パッケージマネージャー
  • Google Cloud アカウント(Gemini モデル使用時)

インストール手順

# 仮想環境の作成(推奨)
python3 -m venv adk-env
source adk-env/bin/activate  # Linux/macOS
# adk-env\Scripts\activate  # Windows

# Google ADK のインストール
pip install google-adk

# 追加依存関係のインストール(必要に応じて)
pip install google-adk[cli]  # CLI ツールを含む
pip install google-adk[testing]  # テストツールを含む

API キーの設定

# 方法 1:環境変数を使用(開発用途に推奨)
export GEMINI_API_KEY="your-api-key-here"

# 方法 2:Google Cloud 認証情報ファイルを使用
export GOOGLE_APPLICATION_CREDENTIALS="/path/to/service-account.json"

# 方法 3:コード内で直接設定(本番環境では非推奨)

コアコンセプト

1. Agent(エージェント)

エージェントは ADK の中核コンポーネントで、ユーザーの入力を受け取り、ツールを呼び出し、レスポンスを生成する役割を担います。

from google.adk import Agent
from google.adk.models import Gemini

# 基本エージェントの作成
agent = Agent(
    name="customer_support",
    model=Gemini(model="gemini-2.0-flash"),
    instruction="あなたは製品の問題解決を支援するプロフェッショナルなカスタマーサポートアシスタントです。",
)

2. Tools(ツール)

ツールは、具体的なタスクを実行するためにエージェントが呼び出せる外部関数です。

from google.adk import Tool
from google.adk.tools import FunctionTool

# ツール関数の定義
def search_knowledge_base(query: str) -> str:
    """ナレッジベースから関連情報を検索する"""
    # 実際の実装ではデータベースや API を呼び出す
    return f"'{query}' の検索結果..."

# ツールの作成
search_tool = FunctionTool(
    func=search_knowledge_base,
    name="search_kb",
    description="製品ナレッジベースを検索する",
)

# ツールをエージェントに追加
agent.tools = [search_tool]

3. Sessions(セッション)

セッションはユーザーとエージェントの対話履歴を管理し、複数ターンの会話をサポートします。

from google.adk import Session

# セッションの作成
session = Session(
    agent=agent,
    user_id="user_123",
)

# メッセージの送信
response = await session.send("製品が起動しません。どうすればいいですか?")
print(response.text)

4. Events(イベント)

イベントシステムは非同期メッセージ処理とストリーミングレスポンスをサポートします。

from google.adk import Event

async def handle_event(event: Event):
    if event.type == "user_message":
        response = await session.send(event.content)
        await event.reply(response.text)

実践例

例 1:シンプルな QA ボット

from google.adk import Agent, Session, Runner
from google.adk.models import Gemini

# エージェントの作成
agent = Agent(
    name="faq_bot",
    model=Gemini(model="gemini-2.0-flash"),
    instruction="""あなたは製品に関する質問に答える FAQ ボットです。
    - 回答は簡潔明瞭に保つ
    - 答えが分からない場合は正直に伝える
    - 関連するヘルプアドバイスを提供する""",
)

# インタラクティブセッションの実行
async def main():
    runner = Runner(agent=agent)
    await runner.run_interactive()

# python3 faq_bot.py

例 2:ツール付きデータ分析アシスタント

from google.adk import Agent, Tool
from google.adk.tools import FunctionTool
from google.adk.models import Gemini
import pandas as pd

# データ分析ツールの定義
def load_csv(file_path: str) -> dict:
    """CSV ファイルを読み込み、基本情報を返す"""
    df = pd.read_csv(file_path)
    return {
        "rows": len(df),
        "columns": list(df.columns),
        "dtypes": df.dtypes.astype(str).to_dict(),
    }

def calculate_stats(file_path: str, column: str) -> dict:
    """指定された列の統計情報を計算する"""
    df = pd.read_csv(file_path)
    numeric_col = pd.to_numeric(df[column], errors="coerce")
    return {
        "mean": float(numeric_col.mean()),
        "median": float(numeric_col.median()),
        "std": float(numeric_col.std()),
        "min": float(numeric_col.min()),
        "max": float(numeric_col.max()),
    }

# ツールの作成
tools = [
    FunctionTool(func=load_csv, name="load_csv", description="CSV ファイルを読み込む"),
    FunctionTool(func=calculate_stats, name="calc_stats", description="統計情報を計算する"),
]

# エージェントの作成
agent = Agent(
    name="data_analyst",
    model=Gemini(model="gemini-2.0-flash"),
    instruction="あなたはデータ分析アシスタントで、CSV データファイルの分析を支援します。",
    tools=tools,
)

例 3:マルチエージェント協調システム

from google.adk import Agent, Runner
from google.adk.models import Gemini

# 複数の専門エージェントを作成
researcher = Agent(
    name="researcher",
    model=Gemini(model="gemini-2.0-flash"),
    instruction="あなたは情報の収集と研究を担当し、詳細な事実とデータを提供します。",
)

writer = Agent(
    name="writer",
    model=Gemini(model="gemini-2.0-flash"),
    instruction="あなたは研究結果を構造化された読みやすいレポートにまとめる担当です。",
)

reviewer = Agent(
    name="reviewer",
    model=Gemini(model="gemini-2.0-flash"),
    instruction="あなたはレポートの品質を審査し、正確性と完全性を確認する担当です。",
)

# ワークフローで複数のエージェントをオーケストレーション
async def collaborative_workflow(user_query: str):
    # 研究フェーズ
    research_result = await researcher.run(user_query)

    # 執筆フェーズ
    draft = await writer.run(f"以下の研究結果に基づいてレポートを作成してください:{research_result}")

    # 審査フェーズ
    final_report = await reviewer.run(f"以下のレポートを審査して改善してください:{draft}")

    return final_report

高度な機能

ストリーミングレスポンス

from google.adk import Agent, Session
from google.adk.models import Gemini

agent = Agent(
    name="streaming_bot",
    model=Gemini(model="gemini-2.0-flash"),
)

session = Session(agent=agent)

# ストリーミングレスポンスの処理
async for chunk in session.send_stream("AI についての物語を教えてください"):
    print(chunk.text, end="", flush=True)

カスタムミドルウェア

from google.adk import Middleware

class LoggingMiddleware(Middleware):
    """すべてのリクエストとレスポンスを記録するログミドルウェア"""

    async def on_request(self, request):
        print(f"[REQUEST] {request.user_id}: {request.content}")
        return request

    async def on_response(self, response):
        print(f"[RESPONSE] {response.text[:100]}...")
        return response

# ミドルウェアの追加
agent.middleware = [LoggingMiddleware()]

エラーハンドリングとリトライ

from google.adk import Agent
from google.adk.models import Gemini
from google.adk.retry import RetryConfig

agent = Agent(
    name="robust_bot",
    model=Gemini(model="gemini-2.0-flash"),
    retry_config=RetryConfig(
        max_retries=3,
        retry_delay=1.0,
        retry_on=[TimeoutError, ConnectionError],
    ),
)

ベストプラクティス

1. プロンプトエンジニアリング

# 良い例:明確で具体的な指示
agent = Agent(
    instruction="""あなたはプロフェッショナルなテクニカルサポートアシスタントです。

    回答ルール:
    1. まずユーザーの問題を理解していることを確認する
    2. ステップバイステップの解決策を提供する
    3. 問題が複雑な場合は人工サポートへの連絡を提案する
    4. 友好的でプロフェッショナルなトーンを保つ""",
)

# 避けるべき例:曖昧で一般的な指示
agent = Agent(
    instruction="ユーザーの問題を解決する",  # 曖昧すぎる
)

2. ツール設計

# 良い例:単一責任、明確なドキュメント
def get_order_status(order_id: str) -> dict:
    """
    注文ステータスの照会

    Args:
        order_id: 注文番号(形式:ORD-XXXXXX)

    Returns:
        status、estimated_delivery などの情報を含む辞書
    """
    pass

# 避けるべき例:多機能、パラメーター過多
def handle_order(action: str, order_id: str, user_id: str, 
                 items: list, address: dict, payment: dict) -> dict:
    # 複雑すぎてメンテナンスが困難
    pass

3. テスト戦略

import pytest
from google.adk import Agent, Session

@pytest.fixture
def test_agent():
    return Agent(
        name="test_bot",
        model=Gemini(model="gemini-2.0-flash"),
        instruction="あなたはテストアシスタントです",
    )

@pytest.mark.asyncio
async def test_agent_response(test_agent):
    session = Session(agent=test_agent)
    response = await session.send("こんにちは")
    assert response.text is not None
    assert len(response.text) > 0

4. パフォーマンス最適化

# コネクションプールの使用
from google.adk.models import Gemini

model = Gemini(
    model="gemini-2.0-flash",
    max_concurrent_requests=10,
    timeout=30.0,
)

# 頻繁なレスポンスのキャッシュ
from functools import lru_cache

@lru_cache(maxsize=100)
def get_faq_answer(question_hash: str) -> str:
    # キャッシュロジック
    pass

デプロイメントガイド

ローカル開発サーバー

from google.adk import Agent, Runner
from google.adk.models import Gemini

agent = Agent(
    name="web_bot",
    model=Gemini(model="gemini-2.0-flash"),
)

# Web サーバーの起動
runner = Runner(agent=agent)
runner.serve(host="0.0.0.0", port=8000)

# アクセス http://localhost:8000

Docker デプロイメント

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["python", "main.py"]

Google Cloud Run

# cloud-run.yaml
apiVersion: serving.knative.dev/v1
kind: Service
metadata:
  name: adk-agent
spec:
  template:
    spec:
      containers:
      - image: gcr.io/your-project/adk-agent
        env:
        - name: GEMINI_API_KEY
          valueFrom:
            secretKeyRef:
              name: api-keys
              key: gemini

よくある質問

Q1: ADK と LangChain の違いは何ですか?

A: ADK はコードファーストと型安全を重視し、プログラミング経験のある開発者に適しています。LangChain は豊富なプリビルドコンポーネントと YAML 設定を提供し、迅速なプロトタイピングに適しています。

Q2: どの大規模言語モデルがサポートされていますか?

A: Google Gemini シリーズモデルをネイティブにサポートしています。カスタム Provider を介して、OpenAI GPT、Anthropic Claude、ローカルモデルなども統合できます。

Q3: エージェントの動作をデバッグするにはどうすればよいですか?

A: ADK 内蔵のデバッグモードを使用します:

from google.adk import Runner

runner = Runner(agent=agent, debug=True)
# 詳細なログがコンソールに出力されます

Q4: 本番環境では何に注意すべきですか?

A: - 環境変数を使用して API キーを管理する - 適切なレート制限を設定する - ログ記録とモニタリングを実装する - エラーハンドリングとリトライメカニズムを追加する - 十分な負荷テストを行う

まとめ

Google ADK は、開発者がプロダクションレベルの AI エージェントアプリケーションを構築するための強力で柔軟なツールを提供します。コードファーストの設計思想により、エージェント開発は従来のソフトウェアエンジニアリングの実践に近づき、テスト、メンテナンス、拡張が容易になります。

次のステップ

  1. 公式ドキュメントを読む
  2. GitHub サンプルリポジトリを参照する
  3. 開発者コミュニティに参加する
  4. 最初のエージェントプロジェクトの構築に挑戦する

関連リソース: