【入門】ADK Web で GCS の CSV を読み、AI エージェントに回答させる

はじめに

本記事では ADK Web の最小サンプル（weather_time_agent） をベースに、Google Cloud Storage（GCS）に置いた cities.csv を読み取り、チャットで都市の天気と現在時刻に答えられるように拡張します。

参考（初期サンプルの出典）
Google ADK Quickstart の agent.py（公式）
https://google.github.io/adk-docs/get-started/quickstart/#agentpy
※ 本記事のコードは上記をベースに、説明に不要な部分を一部省略し、GCS 読み込みと alias 対応を追加しています。

---

完成像

• adk web を起動 → ブラウザのチャットで
  • 「Tokyo の天気は？」 → cities.csv の内容から回答
  • 「大阪の現在時刻は？」 → alias の「大阪」が Osaka にマッチして時刻回答
  • 「New York の現在時刻は？」 → tz=America/New_York を使って時刻回答
• コードは 1ファイル（agent.py）。GCS から CSV を取り出し、標準の csv で読みます。

---

この記事でできること / できないこと

• できる：GCS の cities.csv を読み、都市の 天気 と 現在時刻 を返す（日本語の別表記は alias で吸収）
• できない：実天気API連携、自動更新・スケジューリング（必要なら後で拡張）

---

前提

• GCP プロジェクト（課金有効）／Cloud Storage API 有効化
• バケットに cities.csv を配置済み
• 実行サービスアカウントに roles/storage.objectViewer
• ローカルの認証（ADC）例：

# PowerShell
$env:GOOGLE_APPLICATION_CREDENTIALS="C:\path\to\sa.json"
# bash
# export GOOGLE_APPLICATION_CREDENTIALS=/path/to/sa.json

※ 本番（Cloud Run 等）は Workload Identity を推奨（JSON鍵の配布は避ける）

---

ADK Web 最小サンプルを起動

プロジェクトルートで以下を実行して、ブラウザが開くことを確認します。

adk web

---

デモ用 cities.csv を作って GCS にアップ

UTF-8 で保存して GCS に置きます。想定パス：gs://<YOUR_BUCKET>/<YOUR_PREFIX>/cities.csv
alias 列に別名を |（パイプ）で並べます。

city,alias,weather,temp_c,tz
Tokyo,"東京|とうきょう|tokyo",cloudy,28,Asia/Tokyo
Osaka,"大阪|おおさか|osaka",sunny,30,Asia/Tokyo
New York,"ニューヨーク|ny|nyc|new york",sunny,25,America/New_York
London,"ロンドン|london",rainy,18,Europe/London
Berlin,"ベルリン|berlin",clear,22,Europe/Berlin

アップロード例：

gcloud storage cp ./cities.csv gs://<YOUR_BUCKET>/<YOUR_PREFIX>/cities.csv

---

実装

この章で 実装の要点を把握します。全文は「付録A」に掲載。

1) GCS から CSV を読み込む

• google-cloud-storage の storage.Client() で GCS に接続
• download_as_text() で文字列取得 → csv.DictReader で 行＝辞書 に

from google.cloud import storage
import csv, io

def _download_csv_rows(bucket: str, path: str, encoding: str = "utf-8"):
    text = storage.Client().bucket(bucket).blob(path).download_as_text(encoding=encoding)
    return list(csv.DictReader(io.StringIO(text)))

2) 都市名で行を探す（alias 列 + 正規化（NFKC））

• alias を | 区切りで展開
• 前後空白・大小文字・全角/半角の差を NFKC 正規化で吸収

import unicodedata

def _norm(s: str) -> str:
    return unicodedata.normalize("NFKC", (s or "")).strip().lower()

def _split_alias(s: str):
    if not s:
        return []
    return [p.strip() for p in s.split("|") if p.strip()]

def _find_city(rows, city: str):
    target = _norm(city)
    for r in rows:
        names = [r.get("city") or ""]
        names += _split_alias(r.get("alias"))  # alias が無ければ空配列
        if any(_norm(n) == target for n in names):
            return r
    return None

3) ツール関数：get_weather / get_current_time

• ADK の tools に登録すると、チャットから関数呼び出しされます
• 返り値は {"status": "...", "report": "..."} に統一（LLM が扱いやすい）

import datetime
from zoneinfo import ZoneInfo

def get_weather(city: str) -> dict:
    rows = _download_csv_rows(GCS_BUCKET, GCS_PATH)
    row  = _find_city(rows, city)
    if not row:
        return {"status":"error","error_message": f"'{city}' not found in gs://{GCS_BUCKET}/{GCS_PATH}."}
    name    = (row.get("city") or city).strip()
    weather = (row.get("weather") or "").strip()
    temp_c  = (row.get("temp_c") or "").strip()
    msg = f"The weather in {name} is {weather}" + (f" with a temperature of {temp_c}°C." if temp_c else ".")
    return {"status":"success","report": msg}

def get_current_time(city: str) -> dict:
    rows = _download_csv_rows(GCS_BUCKET, GCS_PATH)
    row  = _find_city(rows, city)
    if not row:
        return {"status":"error","error_message": f"'{city}' not found in gs://{GCS_BUCKET}/{GCS_PATH}."}
    name = (row.get("city") or city).strip()
    tz   = (row.get("tz") or "").strip()
    if not tz:
        return {"status":"error","error_message": f"No timezone (tz) for '{name}' in cities.csv."}
    now = datetime.datetime.now(ZoneInfo(tz))
    return {"status":"success","report": f'The current time in {name} is {now.strftime("%Y-%m-%d %H:%M:%S %Z%z")}'}

4) エージェントにツールを登録

from google.adk.agents import Agent

root_agent = Agent(
    name="weather_time_agent",
    model="gemini-2.0-flash",
    description="Answer weather/time by reading cities.csv on GCS.",
    instruction=("Read weather/time info from a CSV stored in GCS. "
                 "When asked about a city, look it up in the CSV and answer in Japanese."),
    tools=[get_weather, get_current_time],
)

---

起動と動作確認

adk web
# http://127.0.0.1:8000 を開く

チャット例

• 「Tokyo の天気は？」 → cloudy, 28℃ のように回答
• 「大阪の現在時刻は？」 → alias の「大阪」→ Osaka にマッチして回答
• 「New York の現在時刻は？」 → America/New_York の時刻を返す
• 存在しない都市名 → わかりやすいエラーメッセージ

---

まとめ

• GCS から CSV を文字列で取得 → csv.DictReader で行＝辞書に
• 日本語表記などのゆらぎは alias 列 + NFKC 正規化で吸収
• dict 形式の返り値（status / report or error_message）で LLM が扱いやすい応答に
• まずは依存最小・毎回読み直しで“動く”を優先（必要に応じてキャッシュ化や統合ツール化も可能）

---

付録A： agent.py

# agent.py
# 依存：pip install google-adk google-cloud-storage
# 認証（ローカル）：
#   PowerShell: $env:GOOGLE_APPLICATION_CREDENTIALS="C:\path\to\sa.json"
#   bash      : export GOOGLE_APPLICATION_CREDENTIALS=/path/to/sa.json

import csv
import io
import datetime
import unicodedata
from zoneinfo import ZoneInfo
from google.cloud import storage
from google.adk.agents import Agent

# ===== 設定（※環境に合わせて変更） =====
GCS_BUCKET = "<YOUR_BUCKET>"              # 例: my-demo-gcs-2025
GCS_PATH   = "<YOUR_PREFIX>/cities.csv"   # 例: data/demo/cities.csv
ENCODING   = "utf-8"  # 文字化けする場合は "shift_jis" などに変更

def _download_csv_rows(bucket: str, path: str, encoding: str = ENCODING):
    """GCS から CSV をダウンロードして、行＝辞書の配列にして返す。"""
    client = storage.Client()
    text = client.bucket(bucket).blob(path).download_as_text(encoding=encoding)
    return list(csv.DictReader(io.StringIO(text)))

def _norm(s: str) -> str:
    """前後空白除去 → NFKC 正規化（全角/半角差を吸収）→ 小文字化。"""
    return unicodedata.normalize("NFKC", (s or "")).strip().lower()

def _split_alias(s: str):
    """alias 列を '|' 区切りで分割。空要素は落とす。"""
    if not s:
        return []
    return [p.strip() for p in s.split("|") if p.strip()]

def _find_city(rows, city: str):
    """
    city 列と alias 列（任意）を使って都市を特定する。
    - alias は "大阪|Osaka|おおさか" のように '|' 区切りを想定。
    - 大文字/小文字、全角/半角、前後空白の違いは吸収する。
    見つからなければ None を返す。
    """
    target = _norm(city)
    for r in rows:
        names = [r.get("city") or ""]
        names += _split_alias(r.get("alias"))
        if any(_norm(n) == target for n in names):
            return r
    return None

def get_weather(city: str) -> dict:
    """cities.csv から天気と気温を取得して返す。"""
    try:
        rows = _download_csv_rows(GCS_BUCKET, GCS_PATH)
        row = _find_city(rows, city)
        if not row:
            return {"status": "error",
                    "error_message": f"'{city}' not found in gs://{GCS_BUCKET}/{GCS_PATH}."}
        name    = (row.get("city") or city).strip()
        weather = (row.get("weather") or "").strip()
        temp_c  = (row.get("temp_c") or "").strip()
        if temp_c:
            report = f"The weather in {name} is {weather} with a temperature of {temp_c}°C."
        else:
            report = f"The weather in {name} is {weather}."
        return {"status": "success", "report": report}
    except Exception as e:
        return {"status": "error", "error_message": f"get_weather failed: {e}"}

def get_current_time(city: str) -> dict:
    """cities.csv の tz（IANA タイムゾーン）を使って現在時刻を返す。"""
    try:
        rows = _download_csv_rows(GCS_BUCKET, GCS_PATH)
        row = _find_city(rows, city)
        if not row:
            return {"status": "error",
                    "error_message": f"'{city}' not found in gs://{GCS_BUCKET}/{GCS_PATH}."}
        name = (row.get("city") or city).strip()
        tz   = (row.get("tz") or "").strip()
        if not tz:
            return {"status": "error",
                    "error_message": f"No timezone (tz) for '{name}' in cities.csv."}
        now = datetime.datetime.now(ZoneInfo(tz))
        report = f'The current time in {name} is {now.strftime("%Y-%m-%d %H:%M:%S %Z%z")}'
        return {"status": "success", "report": report}
    except Exception as e:
        return {"status": "error", "error_message": f"get_current_time failed: {e}"}

root_agent = Agent(
    name="weather_time_agent",
    model="gemini-2.0-flash",
    description="Answer weather/time by reading cities.csv on GCS.",
    instruction=("Read weather/time info from a CSV stored in GCS. "
                 "When asked about a city, look it up in the CSV and answer in Japanese."),
    tools=[get_weather, get_current_time],
)

---

付録B：cities.csv

city,alias,weather,temp_c,tz
Tokyo,"東京|とうきょう|tokyo",cloudy,28,Asia/Tokyo
Osaka,"大阪|おおさか|osaka",sunny,30,Asia/Tokyo
New York,"ニューヨーク|ny|nyc|new york",sunny,25,America/New_York
London,"ロンドン|london",rainy,18,Europe/London
Berlin,"ベルリン|berlin",clear,22,Europe/Berlin