【ADK入門】Developer Knowledge API で「ハルシネーションしない」資格学習エージェントを作る

はじめに

ビジネスアクセラレーション事業部の森田です。
皆さん、資格の勉強をするときに生成 AI を使いますか？
私は使います。Google Cloud の資格をいくつか取ってきたのですが、分からないことが出てくるとつい聞いてしまいます。
ただ、生成 AI はそれっぽい嘘を、堂々と教えてくることがあるんですよね。制限値が微妙に違っていたり、既に存在しないサービス名が出てきたり。
そんなときに Developer Knowledge API というものを見つけました。Google の公式ドキュメントを検索・取得できる API です。
これを使えば、ちゃんと公式ドキュメントを読んでから答えてくれるやつが作れるのでは？と思い、前から気になっていた ADK（Agent Development Kit） の練習も兼ねて、作ってみることにしました。
手順もひととおり書いておくので、よければ一緒に作ってみてください。
ちなみに、今回作成する資格学習エージェントの全体像はこんな感じです。

ADK 入門

まずは ADK について、私が理解した範囲で書いておきます。

ADK とは

Agent Development Kit の略で、Google が出している AI エージェント開発用のフレームワークです。Python と Java に対応しています。
Gemini 向けに最適化されてはいるものの、モデルもデプロイ先も固定されてはいません。ローカルで動かすこともできますし、Cloud Run や Vertex AI Agent Engine に載せることもできるようです。今回はローカル（というか Cloud Shell）で動かすところまでをやります。
Google 製ということもあって Gemini との相性が良く、公式ドキュメントも一通り揃っているので、入門するにはちょうどいいかなと思いました。

エージェントとは何をするものか

そもそもエージェントとは何なのか、という話をしておきます。
普通に Gemini API を叩くと、質問して回答が返ってくる、それで終わりです。
エージェントにすると、LLM が自分で判断してツールを呼びながら進むループになります。「これは検索が必要だな」→ 検索ツールを実行 →「まだ足りないから全文を取ろう」→ 取得ツールを実行 → 回答、といった具合です。
つまり、LLM に「道具」と「使っていいですよ」という許可を与えたものがエージェント、というのが私の理解です。何をどの順で使うかは LLM が決めます。
このループを自前で書くとまあまあ面倒らしいのですが、今回は ADK を使うことで簡単に実装することが出来ました。

出てくる用語

ADK を触っていると独特の言葉がいくつか出てきます。この記事でも使うので、先にまとめておきます。

Agent

エージェントそのもの。モデル・指示・ツールをまとめたもの。

instruction

エージェントへの指示文。いわゆるシステムプロンプト。

ツール

LLM が呼び出せる関数。ADK では普通の Python 関数を渡すだけ。

Runner

エージェントの実行を回す人。モデル呼び出しとツール実行を仕切る。

Event

実行中に起きたことの記録。ツールを呼んだ、結果が返った、など。

SessionService

Event を貯めておく場所。会話の履歴になる。

session state

会話をまたいで値を持ち回れる箱。

これらがどう繋がっているかは、図にするとこうなります。

ポイントは、1 ターンの中でモデル呼び出しとツール実行が何度も往復しているところです。ユーザーから見ると 1 回のやりとりですが、裏では何度も Gemini に聞きに行っています。
そしてその過程はすべて Event として記録され、次のリクエストに履歴として積まれます。この「全部記録して毎回送る」という仕組みは後半で効いてきます（トークンの話のところで触れます）。

動かすのは adk コマンド

書いたエージェントは adk というコマンドで動かします。いくつかサブコマンドがあるのですが、入門で使うのはこのあたりです。

今回はずっと adk web を使います。ブラウザで話しかけられるだけでなく、エージェントが裏で何をしたかを覗けるのが便利でした。

この記事で「ちゃんとツールを呼んでいるか」を確認しているのは、だいたい Events タブを見ています。
なお SessionService は adk web の既定だとインメモリ保存です。プロセスを落とすと消えます。--session_service_uri で永続化先を変えられるようですが、今回は試していません。

ディレクトリ構成にお作法がある

エージェントをどこに置くかにも決まりがあります。adk web はディレクトリをパッケージとして読み込んで、その中の root_agent という変数を探しにきます。この名前は固定です。

my-project/              ← adk web を実行する場所
└── my_agent/
    ├── __init__.py      ← from . import agent と書く
    ├── agent.py         ← root_agent を定義
    └── .env

.env は ADK が自動で読んでくれます。

docstring がプロンプトになる

ADK を使ってみて一番驚いたのがツールの作り方です。
普通の Python 関数を書いて tools に渡すだけで、ADK が型ヒントと docstring を読んで、Gemini 用の function declaration を勝手に作ってくれます。

def get_weather(city: str) -> dict:
    """指定された都市の天気を取得する。

    Args:
        city: 都市名。例: "Osaka"

    Returns:
        status と天気情報を含む dict。
    """
    return {"status": "success", "weather": "晴れ"}

これを渡すと、ADK が裏でこういう宣言を組み立てて Gemini に送ります。Request タブで実物が見られます。

{
  "name": "get_weather",
  "description": "指定された都市の天気を取得する。\n\nArgs:\n    city: 都市名。例: \"Osaka\"\n\nReturns:\n    status と天気情報を含む dict。",
  "parameters_json_schema": {
    "properties": { "city": { "title": "City", "type": "string" } },
    "required": ["city"],
    "type": "object"
  }
}

見てのとおり、docstring がまるごと description に入っています。関数名も型ヒントもそのまま使われています。
つまり docstring はコメントではなくプロンプトの一部なんですね。ここが雑だとツールがあるのに呼んでくれませんし、逆に「〜する前に必ず呼ぶこと」と書けば従ってくれます。コメントの書き方で挙動が変わるのは新鮮でした。

状態を持たせる

LLM は毎回まっさらな状態でリクエストを受けるので、「さっき大阪の天気を聞いたよね？」という記憶はありません。そこで ADK には session state という箱が用意されています。ツールの引数に tool_context: ToolContext を足すとアクセスできます。

from google.adk.tools.tool_context import ToolContext

def set_city(city: str, tool_context: ToolContext) -> dict:
    """対象の都市を設定する。"""
    tool_context.state["city"] = city
    return {"status": "success"}

ここで「引数が 2 つあるけど、LLM は tool_context に何を入れればいいの？」と思いませんか。私は思いました。
答えは「何も入れなくていい」でした。tool_context は ADK が実行時に勝手に差し込むもので、LLM から見えていません。
Request タブで実際に送られている宣言を確認してみます。

{
  "name": "set_city",
  "parameters_json_schema": {
    "properties": {
      "city": { "title": "City", "type": "string" }
    },
    "required": ["city"],
    "type": "object"
  }
}

city しかありません。tool_context は消えています。
つまり LLM に見せる引数と、実装が必要とする引数を分けられるということですね。LLM には「都市名を教えて」とだけ頼めばよく、state への書き込みは裏側で完結します。
なお tool_context という名前は固定です。別の名前にすると注入されません。
保存した値は instruction から読み出せます。instruction には文字列だけでなく関数も渡せて、こう書くと毎ターン組み立て直してくれます。

def build_instruction(context: ReadonlyContext) -> str:
    city = context.state.get("city", "未設定")
    return f"あなたは天気の案内役です。現在の対象都市: {city}"

root_agent = Agent(instruction=build_instruction, ...)

state に入れた値を、そのままシステムプロンプトに差し込めるわけです。

Developer Knowledge API について

対象が Google の公式ドキュメントに限られた API です。
主なメソッドは 3 つあります。

ぱっと見 answerQuery が一番ラクそうですよね。でも 1 日 50 回しか実行できません。1 セッションで数回叩くとして、10 回ちょっとで終わってしまいます。試験勉強に使うには現実的ではないので、検索と取得を組み合わせる方式にしました。

セットアップ

ここから実際に作っていきます。ローカルに何も入れずに済ませたかったので、コンソールと Cloud Shell だけで進めました。

① API を有効化する

下記のリンクから Developer Knowledge API を有効化します。

https://console.cloud.google.com/start/api?id=developerknowledge.googleapis.com

② API キーを 2 本用意する

1 本目は Developer Knowledge API 用です。
APIとサービス → 認証情報 → + 認証情報を作成 → APIキー で作って、「キーを制限」から Developer Knowledge API だけを選びます。
そして 2 本目が Gemini 用です。Gemini のキーは Google AI Studio で別途取ります。
最初は 1 本にまとめようとしたんですが、「現在選択されている API の制限と組み合わせることはできません」と怒られました。
調べてみると、AI Studio で作るキーは自動的に「認証キー」になり、既定で Gemini API 専用に制限されるそうです。他の API と混ぜられません。

③ 疎通確認する

ハンズオン的に進んだ感が出るので、疎通確認をしておきます。
下記のコマンドを Cloud Shell を開いて実行します。

export DEVELOPERKNOWLEDGE_API_KEY="作成したキー"

curl -s -G "https://developerknowledge.googleapis.com/v1/documents:searchDocumentChunks" \
  --data-urlencode "query=Cloud Run minimum instances cold start" \
  --data-urlencode 'filter=data_source="docs.cloud.google.com"' \
  --data-urlencode "pageSize=3" \
  --data-urlencode "fields=results(document(title,uri))" \
  --data-urlencode "key=$DEVELOPERKNOWLEDGE_API_KEY" | jq

docs.cloud.google.com 配下の URI が返ってくれば大丈夫です。403 が返る場合は、有効化が反映されるまでもう少し待ってみてください。
ちなみに、上記例では英語のクエリで実行していますが、日本語のクエリで実行可能です。日本語の応答が返ってきます。

④ Cloud Shell に環境を作る

Cloud Shell には Python も gcloud も最初から入っているので、実行環境がすぐに手に入ります。

mkdir -p ~/adk-cert-agent/cert_study_agent && cd ~/adk-cert-agent
python3 -m venv .venv && source .venv/bin/activate
pip install google-adk httpx

.env はこんな感じです。

GOOGLE_GENAI_USE_VERTEXAI=FALSE
GOOGLE_API_KEY=AI Studio で取ったキー
DEVELOPERKNOWLEDGE_API_KEY=コンソールで作ったキー

ちなみに .env は隠しファイルなので、エディタのツリーに出てきません。cloudshell edit ~/adk-cert-agent/cert_study_agent/.env で直接開けます。

実装する

ファイルは 3 つに分けました。API を叩く dk_tools.py、state を扱う exam_tools.py、エージェント本体の agent.py です。全文は記事の最後に貼っておくので、ここでは工夫したところだけ書きます。

検索まわり（dk_tools.py）

searchDocumentChunks は filter と fields を効かせるとだいぶ扱いやすくなります。

params = {
    "query": query,
    "pageSize": max(1, min(max_results, 10)),
    "filter": f'data_source="{data_source}"',
    "fields": "results(content,document(name,title,uri))",
    "key": API_KEY,
}

filter は data_source / uri / update_time を条件にできます。技術ドキュメントに絞りたかったので docs.cloud.google.com を指定したところ、ノイズが減って精度が上がりました。fields はフィールドマスクで、ドキュメントの name / title / uri のみを返すように制限しました。
pageSize を制限しているのは、LLM が平気で 50 とか入れてくるからです。ツールの引数は信用しないほうがいいです。
そしてレスポンスなのですが、ここが引っかかりました。返ってくるのは「チャンク」の配列で、名前のとおりドキュメント単位ではありません。試しに Cloud Storage について検索したら、5 件中 4 件が同じページの別部分でした。
その場合、title と uri が重複してトークンを食います。なので document.name をキーにしてまとめました。

grouped = {}
for ch in chunks:
    doc = ch.get("document", {})
    key = doc.get("name", "")
    if key not in grouped:
        grouped[key] = {"title": doc.get("title"), "uri": doc.get("uri"),
                        "doc_name": key, "excerpts": []}
    grouped[key]["excerpts"].append((ch.get("content") or "")[:MAX_CHUNK_CHARS])

ハルシネーション対策：ツール側

「モデルの記憶ではなく、いま取ってきた一次情報に基づいて答えさせる」ことをグラウンディングと言うそうです。
ここで大事なのは、公式ドキュメントを引いても解釈を間違える可能性があるということです。ただ出典が付いていれば、生成 AI から与えられた情報が本当に正しいかどうか人の手で確かめることが出来ます。
上記の通り、検索ツールを繋いだだけでは全然ダメでした。LLM は検索結果と自分の記憶をしれっと混ぜてきます。工夫はツールの外側にも必要です。
まずツール側でやったことから。

① エラーは例外ではなく dict で返す

    except Exception as e:
        return {"status": "error", "error_message": f"検索に失敗しました: {e}"}

例外を投げるとエージェントごと止まりますが、dict で返せば LLM が読んで判断できます。status は success / error / empty の 3 種類にしました。「見つからなかった」と「失敗した」は意味が違うからです。

② 切ったことをちゃんと伝える

    return {
        "status": "success",
        "truncated": len(content) > MAX_DOC_CHARS,
        "content": content[:MAX_DOC_CHARS],
    }

数万文字のページがあるのでクライアント側で切るのですが、切ったことを伝えないとまずいです。LLM は途切れた文章を情報の全てであると解釈して、「そんな記述はありません」と自信満々に断定してくることがあるからです。

③ 引数を検証する

    if not doc_name.startswith("documents/"):
        return {"status": "error", "error_message": "doc_name の形式が不正です。"}

LLM が渡してきた文字列をそのまま URL にくっつけるので、一応チェックを入れています。変な値が混ざったときの事故防止です。

ハルシネーション対策：instruction 側

instruction には、できることだけでなくできないことも書きました。これが結構効きました。

参照できない:
  - 公式試験ガイド・出題範囲・配点（コーパスに含まれていない）
  - 試験の合格ライン、出題数などの試験運用情報
  - 他クラウド（AWS など）のドキュメント

参照できない範囲について聞かれた場合は、検索を試みず、
「この情報は参照先に含まれていないため回答できません」と伝えること。

これがないと、コーパスに入っていない話題でも一応検索して、関係ないページを拾って作文してしまいます。何を持っていないかも教えてあげないといけないんですね。
もうひとつ、公式試験ガイドは PDF で配られていて、この API のコーパスには入っていません。なので PDF をチャットに添付してもらう方式にしました。
読み取った内容は save_exam_scope という自作ツールで state に入れています。入門のところで書いた tool_context と build_instruction の組み合わせがここで効いてきて、一度保存すれば以降のターンでは instruction に自動で差し込まれます。
もちろんここでも「ユーザーが出していない出題範囲を、記憶や推測で作ってはいけない」と制限をかけています。

起動して動かしてみる

Cloud Shell から起動します。

cd ~/adk-cert-agent
adk web --allow_origins "regex:https://.*\.cloudshell\.dev" --reload_agents

--allow_origins は Cloud Shell のプロキシ経由で CORS に弾かれるのを防ぐためです。--reload_agents を付けるとコードの変更が即反映されるので、instruction をいじりながら試すのに便利でした。
ターミナルに出てくる http://127.0.0.1:8000 をクリックすると新しいタブで Dev UI が開きます。

まずは断ってもらう

試験範囲を PDF で渡す前に「ハルシネーションをしない」設計が動作しているか確認します。

Q: PCA の出題範囲を教えて
A: この情報は参照先に含まれていないため回答できません。Google Cloud の認定資格の出題範囲については、公式サイトで最新の試験ガイドをご確認ください。もし公式試験ガイドの該当箇所を貼り付けていただければ、その内容を保存して今後の学習支援に役立てることができます。

Developer Knowledge API では取得できない & 現時点では渡していない情報に関して質問をしたので、ちゃんと断ってくれました！
しかもよく見ると、検索すらしていません。質問された時点で「これは範囲外だな」と判断しています。

出題範囲を渡す

先ほどは Developer Knowledge API では取得できない & 現時点では渡していない情報に関して質問をしてしまったので、答えてくれませんでした。そこで、公式試験ガイドの PDF を渡してあげます。
公式試験の出題範囲について質問してみると以下の回答が得られました。ちゃんと読み込めていることがわかります。

普通の質問とクイズ

技術的な質問も大丈夫でした。「Cloud Run の最小インスタンス数を設定するとコールドスタートはどうなる？」に対して検索してから答えて、末尾に参照 URL を 3 件。

クイズを作らせたら、ドキュメントを引いてから 4 択を出して、選択肢ごとに正解・不正解の理由まで説明してくれました。試験対策としてはこれが一番使えそうです。

おわりに

ということで、公式ドキュメントを読んでから答えてくれる資格学習エージェントを作ってみました。
やってみて一番の学びは、検索ツールを付けただけでは AI は嘘をつくのをやめてくれないということです。情報源を繋ぐだけじゃなくて、答える前に必ず調べさせて、分からないなら黙らせる。そこまでやってようやくまともになりました。
次に ADK で記事を書くとしたら、回答の応答の質を計測してみたいです。ADK には adk eval があるので、ツールをちゃんと呼んだ割合とか、出した URL が実在する割合とか、情報がないときに正しく「分かりません」と言えた割合とか、そのあたりを測れるようにしたいなと。
ここまで読んでくださりありがとうございました！

サンプルコード全文

最後に、今回作ったものを全部貼っておきます。ディレクトリ構成はこうなっています。

adk-cert-agent/
├── .venv/
└── cert_study_agent/
    ├── __init__.py
    ├── .env
    ├── dk_tools.py
    ├── exam_tools.py
    ├── agent.py
    └── verify.py

cert_study_agent/__init__.py

ADK にパッケージとして認識させるためのファイルです。これがないとエージェントが見つかりません。

from . import agent

cert_study_agent/.env

キーは 2 本必要です。GOOGLE_API_KEY は AI Studio、DEVELOPERKNOWLEDGE_API_KEY はコンソールで取得したものを入れてください。

GOOGLE_GENAI_USE_VERTEXAI=FALSE
GOOGLE_API_KEY=AI Studio で取得したキー
DEVELOPERKNOWLEDGE_API_KEY=コンソールで作成したキー

cert_study_agent/dk_tools.py

Developer Knowledge API を叩く部分です。チャンクのグルーピングや文字数のガードはここに入っています。

"""Developer Knowledge API を ADK の FunctionTool として使うためのラッパー。

参考:
  https://developers.google.com/knowledge/howto
  https://developers.google.com/knowledge/reference/rest/v1/documents/searchDocumentChunks
"""

import os
from typing import Any

import httpx

BASE_URL = "https://developerknowledge.googleapis.com/v1"
API_KEY = os.environ.get("DEVELOPERKNOWLEDGE_API_KEY", "")

# コーパスでは Google Cloud 関連が 2 つの data_source に分かれている。
#   docs.cloud.google.com … 技術ドキュメント本体（設定手順、制限値など）。主戦場。
#   cloud.google.com      … 製品概要・ソリューションページ。
#
# 【検証済み】公式試験ガイド（cloud.google.com/learn/certification/...）は
# コーパスに含まれていない。uri の完全一致フィルタで検索しても結果ゼロだった。
# ドメインが対象でも全パスが索引化されているわけではない点に注意。
DOCS_SOURCE = "docs.cloud.google.com"
MARKETING_SOURCE = "cloud.google.com"
DEFAULT_DATA_SOURCE = DOCS_SOURCE

# 1 チャンクあたりの最大文字数。コンテキスト爆発を防ぐためのガード。
MAX_CHUNK_CHARS = 1500
MAX_DOC_CHARS = 12000


def _client() -> httpx.Client:
    if not API_KEY:
        raise RuntimeError("環境変数 DEVELOPERKNOWLEDGE_API_KEY が未設定です")
    return httpx.Client(timeout=30.0)


def search_google_docs(
    query: str,
    max_results: int = 5,
    data_source: str = DEFAULT_DATA_SOURCE,
) -> dict[str, Any]:
    """Google 公式ドキュメントを検索し、関連する抜粋（チャンク）を返す。

    Google Cloud の仕様・制限値・ベストプラクティスなど、正確さが求められる
    質問に答える前に必ず呼び出すこと。

    クエリは日本語・英語どちらでも実用的にヒットする（検証済み）。
    ユーザーの質問の語をそのまま使ってよい。ヒットが弱い場合は、
    サービス名や設定項目名を英語に置き換えて再検索すると改善することがある。

    Args:
        query: 検索クエリ。日本語可（例: "Cloud Run 最小インスタンス"）。
        max_results: 取得する**チャンク**の最大件数。1〜10 程度。
            同一ドキュメントの複数箇所がヒットすることがあるため、
            返るドキュメント数はこれより少なくなる場合がある。
            幅広く探したいときは 8〜10 を指定する。
        data_source: 検索対象ドメイン。技術仕様・設定手順は
            "docs.cloud.google.com"（既定）。製品概要やソリューションの
            比較は "cloud.google.com"。試験ガイドはどちらにも含まれない。

    Returns:
        status と results を含む dict。results の各要素はドキュメント単位で、
        title / uri / doc_name と、そのドキュメント内の該当箇所を集めた
        excerpts（文字列のリスト）を持つ。
        参照を示す際は uri をドキュメントごとに一度だけ挙げること。
    """
    params = {
        "query": query,
        "pageSize": max(1, min(max_results, 10)),
        "filter": f'data_source="{data_source}"',
        "fields": "results(content,document(name,title,uri))",
        "key": API_KEY,
    }
    try:
        with _client() as c:
            r = c.get(f"{BASE_URL}/documents:searchDocumentChunks", params=params)
            r.raise_for_status()
            data = r.json()
    except Exception as e:  # noqa: BLE001
        return {"status": "error", "error_message": f"検索に失敗しました: {e}"}

    chunks = data.get("results") or data.get("documentChunks") or []

    # searchDocumentChunks は「チャンク単位」で返すため、同一ドキュメントの
    # 別部分が複数ヒットする（例: storage-classes が5件中4件を占める）。
    # title / uri の重複でトークンを浪費し、LLM が別々の情報源だと
    # 誤認する原因にもなるため、ドキュメント単位にまとめる。
    grouped: dict[str, dict[str, Any]] = {}
    for ch in chunks:
        doc = ch.get("document", {})
        key = doc.get("name", "")
        if key not in grouped:
            grouped[key] = {
                "title": doc.get("title", ""),
                "uri": doc.get("uri", ""),
                "doc_name": key,
                "excerpts": [],
            }
        content = (ch.get("content") or "")[:MAX_CHUNK_CHARS]
        if content:
            grouped[key]["excerpts"].append(content)

    results = list(grouped.values())

    if not results:
        return {"status": "empty", "message": "該当するドキュメントが見つかりませんでした。"}
    return {"status": "success", "results": results}


def get_google_doc(doc_name: str) -> dict[str, Any]:
    """検索結果の doc_name を指定して、ドキュメント全文（Markdown）を取得する。

    抜粋だけでは判断できない場合（手順の全体像、表形式の制限値の確認など）
    にのみ使うこと。1 回の応答で 2 件までに抑える。

    Args:
        doc_name: search_google_docs が返した doc_name（例: "documents/cloud.google.com/run/docs/..."）。

    Returns:
        status と title / uri / content を含む dict。
    """
    if not doc_name.startswith("documents/"):
        return {"status": "error", "error_message": "doc_name は 'documents/' で始まる必要があります。"}

    try:
        with _client() as c:
            r = c.get(f"{BASE_URL}/{doc_name}", params={"key": API_KEY})
            r.raise_for_status()
            doc = r.json()
    except Exception as e:  # noqa: BLE001
        return {"status": "error", "error_message": f"取得に失敗しました: {e}"}

    content = doc.get("content", "")
    return {
        "status": "success",
        "title": doc.get("title", ""),
        "uri": doc.get("uri", ""),
        "truncated": len(content) > MAX_DOC_CHARS,
        "content": content[:MAX_DOC_CHARS],
    }

cert_study_agent/exam_tools.py

出題範囲と苦手トピックを session state に保存するツールです。

"""出題範囲まわりのツール。

【設計変更の経緯】
当初は Developer Knowledge API から公式試験ガイドを取得する設計だった。
しかし実際に叩いて確認したところ、cloud.google.com/learn/certification/ 配下は
コーパスに含まれていなかった（uri フィルタで完全一致検索しても結果ゼロ）。

コーパスリファレンスに cloud.google.com がドメインとして載っていたため
「試験ガイドも入っている」と判断したが、ドメインが対象でも全パスが
索引化されているわけではなかった。

そのため出題範囲は「ユーザーが提示したものだけを保存する」方式に変更した。
エージェントが推測で出題範囲を作らないよう、instruction 側でも縛っている。
"""

from typing import Any

from google.adk.tools.tool_context import ToolContext

STATE_KEY_SCOPE = "exam_scope"
STATE_KEY_WEAK = "weak_sections"


def save_exam_scope(
    exam_name: str,
    sections: list[str],
    tool_context: ToolContext,
    guide_uri: str = "",
) -> dict[str, Any]:
    """ユーザーが提示した出題範囲をセッションに保存する。

    ユーザーが試験ガイドを添付（PDF など）または貼り付けた場合に、
    その場で必ず呼ぶこと。読み取るだけで保存しないのは不可。

    保存しないと以下の問題が起きる:
      - 新しいセッションで内容が失われる
      - PDF の全文が毎ターン再送信され、コンテキストとコストを圧迫する

    重要: ユーザーが提示していない出題範囲を、記憶や推測で作ってはいけない。
    出題範囲が不明な場合は、公式試験ガイドを添付するようユーザーに依頼すること。

    Args:
        exam_name: 資格名。ユーザーが述べた、または資料に記載されたとおりに記録する。
        sections: 出題範囲の大分類。資料に記載されたものをそのまま記録する。
            配点が示されていれば含める。小項目は含めない。
            例: ["セクション1: クラウドソリューションアーキテクチャの設計と計画 (約24%)"]
        guide_uri: 試験ガイドの URL。判明している場合のみ設定する。

    Returns:
        保存結果。
    """
    if not sections:
        return {
            "status": "error",
            "error_message": "sections が空です。ユーザーが提示した出題範囲が必要です。",
        }

    scope = {
        "exam_name": exam_name,
        "sections": sections,
        "guide_uri": guide_uri,
    }
    tool_context.state[STATE_KEY_SCOPE] = scope
    return {"status": "success", "saved": scope}


def record_weak_section(topic: str, tool_context: ToolContext) -> dict[str, Any]:
    """ユーザーが誤答した／理解が浅かったトピックを記録する。

    クイズで不正解だった時や、ユーザーが苦手だと述べた時に呼ぶ。

    Args:
        topic: 具体的なトピック名（例: "Cloud Run のオートスケーリング"）。

    Returns:
        現在記録されている苦手トピックの一覧。
    """
    weak: list[str] = list(tool_context.state.get(STATE_KEY_WEAK, []))
    if topic not in weak:
        weak.append(topic)
    tool_context.state[STATE_KEY_WEAK] = weak
    return {"status": "success", "weak_topics": weak}

cert_study_agent/agent.py

エージェント本体です。instruction がやたら長いですが、ここが今回の肝だと思っています。
instruction の中で ## を使って見出しをつけています。LLM は Markdown の構造を読んでくれるので、こうやって区切っておくと指示が通りやすい印象でした。

"""資格学習支援エージェント。

Developer Knowledge API で公式ドキュメントを参照し、根拠つきで回答する。

対応範囲:
  ○ 技術仕様・設定手順・制限値（docs.cloud.google.com から取得）
  × 試験ガイド・出題範囲（コーパスに含まれないため、ユーザー提示のみ）
"""

from google.adk.agents import Agent
from google.adk.agents.readonly_context import ReadonlyContext

from .dk_tools import get_google_doc, search_google_docs
from .exam_tools import (
    STATE_KEY_SCOPE,
    STATE_KEY_WEAK,
    record_weak_section,
    save_exam_scope,
)

BASE_INSTRUCTION = """
あなたは Google Cloud 認定資格の学習を支援するチューターです。
ユーザーは日本語で質問します。回答も必ず日本語で行ってください。

## 最重要ルール: 根拠のない回答をしない
1. 技術的な事実（仕様、制限値、料金モデル、推奨構成、手順）を含む回答をする
   前に、必ず `search_google_docs` を呼び出して公式ドキュメントを確認すること。
   記憶だけで答えてはいけない。
2. 検索クエリは日本語のままでよい（検証済み）。ヒットが弱いと感じたら、
   サービス名や設定項目名を英語に置き換えて再検索すること。
3. 検索はチャンク単位で行われるため、同じドキュメントの複数箇所が
   まとめて返ることがある。結果がドキュメント1件に偏った場合は、
   別の観点のクエリで検索し直して視野を広げること。
4. 抜粋で足りない場合のみ `get_google_doc` で全文取得。1 応答 2 件まで。
5. 裏付けが取れない場合は推測で埋めず「公式ドキュメントで確認できませんでした」
   と明示すること。
6. 回答末尾に必ず「参照」セクションを設け、タイトルと URL を列挙すること。
   **同じ URL を重複して挙げないこと**（1ドキュメント1行）。

## 参照できる範囲と、できない範囲
参照先は Developer Knowledge API のコーパスに限られる。以下を正確に区別すること。

参照できる:
  - Google Cloud の技術ドキュメント（data_source: docs.cloud.google.com）
  - 製品概要・ソリューションページ（data_source: cloud.google.com）

参照できない:
  - **公式試験ガイド・出題範囲・配点**（コーパスに含まれていない）
  - 試験の合格ライン、出題数などの試験運用情報
  - 他クラウド（AWS など）のドキュメント

参照できない範囲について聞かれた場合は、検索を試みず、
「この情報は参照先に含まれていないため回答できません」と伝え、
公式サイトで確認するよう案内すること。憶測で答えてはいけない。

## 出題範囲の扱い
下に「対策中の資格」が表示されていない状態で出題範囲に関わる話題が出たら、
公式試験ガイドを添付するか、該当箇所を貼り付けるようユーザーに依頼すること。
公式試験ガイドは PDF で配布されているため、ファイル添付が最も確実である。

ユーザーが試験ガイド（PDF またはテキスト）を渡してきたら、**その場で**
以下を実行すること。後回しにしてはいけない。
1. セクションの見出しと配点を読み取る
2. `save_exam_scope` を呼んで保存する
3. 保存した内容をユーザーに提示し、認識が合っているか確認する

保存は必須である。ファイルの内容は会話履歴に残るだけでは不十分で、
保存しないと新しいセッションで失われ、毎ターンのコンテキストも圧迫する。

保存する際の注意:
  - 配点の数値は原文のまま正確に写すこと。丸めたり推測したりしない。
  - セクション名は日本語に訳してよいが、意味を変えないこと。
  - 小項目は保存しない（大分類のみ）。詳細が必要になったら再度添付を依頼する。

**ユーザーが提示していない出題範囲を、記憶や推測で作ってはいけない。**
「たしか5つのセクションがあって…」のような曖昧な記憶で埋めるのは禁止。

## 資格対策としての振る舞い
- 出題範囲が保存されている場合のみ、回答冒頭で該当セクションを示すこと。
  保存されていない場合は、無理に対応づけようとしないこと。
- 仕様の羅列で終わらせず、試験で問われる観点（紛らわしい類似サービスとの
  違い、コスト vs 可用性のトレードオフなど）を添える。
- 「問題を出して」と言われたら、まず該当分野を検索して裏を取ってから 4 択を
  作る。ドキュメントに書かれていない内容を問題にしてはいけない。
- ユーザーが誤答したら `record_weak_section` でトピックを記録すること。

## 注意
- 試験の実際の出題内容（本番問題）は扱わない。
"""


def build_instruction(context: ReadonlyContext) -> str:
    """session state を見て、必要な分だけ instruction を組み立てる。

    InstructionProvider を使うと ADK は {} の自動注入を行わないため、
    JSON 例などの波括弧をそのまま書ける利点もある。
    """
    parts = [BASE_INSTRUCTION]

    scope = context.state.get(STATE_KEY_SCOPE)
    if scope:
        sections = "\n".join(f"  - {s}" for s in scope.get("sections", []))
        block = f"## 対策中の資格: {scope['exam_name']}\n出題範囲（ユーザー提示）:\n{sections}\n"
        if scope.get("guide_uri"):
            block += f"出典: {scope['guide_uri']}\n"
        parts.append(block)

    weak = context.state.get(STATE_KEY_WEAK) or []
    if weak:
        parts.append("## 記録済みの苦手トピック\n" + "\n".join(f"- {w}" for w in weak) + "\n")

    return "\n".join(parts)


root_agent = Agent(
    model="gemini-2.5-flash",
    name="cert_study_agent",
    description="Google Cloud 認定資格の学習を、公式ドキュメントに基づいて支援するチューター。",
    instruction=build_instruction,  # 文字列ではなく callable を渡す
    tools=[
        search_google_docs,
        get_google_doc,
        save_exam_scope,
        record_weak_section,
    ],
)

cert_study_agent/verify.py

ADK を挟まずに API の疎通だけを確認するスクリプトです。何かおかしいときはまずこれを流すと切り分けが楽になります。

#!/usr/bin/env python3
"""ADK に組み込む前の疎通確認スクリプト。

確認内容:
  1. API キーが有効で、技術ドキュメントが引けるか
  2. 日本語クエリと英語クエリでヒットが変わるか（instruction の方針判断用）
  3. コーパスに何が入っていないかの確認（試験ガイドは入っていない）

使い方:
    export DEVELOPERKNOWLEDGE_API_KEY="..."
    python verify.py
"""

import os
import sys

import httpx

BASE = "https://developerknowledge.googleapis.com/v1"
KEY = os.environ.get("DEVELOPERKNOWLEDGE_API_KEY", "")


def search(query: str, data_source: str, page_size: int = 5) -> list[dict]:
    params = {
        "query": query,
        "pageSize": page_size,
        "filter": f'data_source="{data_source}"',
        "fields": "results(document(name,title,uri))",
        "key": KEY,
    }
    r = httpx.get(f"{BASE}/documents:searchDocumentChunks", params=params, timeout=30.0)
    r.raise_for_status()
    data = r.json()
    return data.get("results") or data.get("documentChunks") or []


def show(label: str, query: str, source: str) -> list[str]:
    print(f"\n--- {label}")
    print(f"    query={query!r}")
    try:
        results = search(query, source)
    except httpx.HTTPStatusError as e:
        print(f"    NG: HTTP {e.response.status_code} — {e.response.text[:200]}")
        return []
    except Exception as e:  # noqa: BLE001
        print(f"    NG: {e}")
        return []

    uris = []
    for r in results:
        doc = r.get("document", {})
        uri = doc.get("uri", "")
        uris.append(uri)
        print(f"      {doc.get('title', '(no title)')[:55]}")
        print(f"        {uri}")
    if not uris:
        print("      （結果ゼロ件）")
    return uris


def main() -> int:
    if not KEY:
        print("環境変数 DEVELOPERKNOWLEDGE_API_KEY が未設定です")
        return 1

    print("=" * 60)
    print("1. 技術ドキュメントが引けるか")
    print("=" * 60)
    en = show("英語クエリ", "Cloud Run minimum instances cold start", "docs.cloud.google.com")
    if not en:
        print("\n技術ドキュメントが引けていません。キーと有効化を確認してください。")
        return 1

    print("\n" + "=" * 60)
    print("2. 日本語クエリでも同等にヒットするか")
    print("   （同じ結果なら instruction の「英語で検索」ルールは不要）")
    print("=" * 60)
    ja = show("日本語クエリ", "Cloud Run 最小インスタンス コールドスタート", "docs.cloud.google.com")

    if ja and set(ja[:3]) == set(en[:3]):
        print("\n  → 上位3件が一致。日本語クエリでも問題なさそうです。")
    elif ja:
        print("\n  → 結果が異なります。英語クエリのほうが適切か目視で判断してください。")
    else:
        print("\n  → 日本語クエリはヒットしません。instruction で英語クエリを強制すべきです。")

    print("\n" + "=" * 60)
    print("3. 試験ガイドはコーパスに含まれない（確認用）")
    print("=" * 60)
    cert = show(
        "試験ガイド検索",
        "Professional Cloud Architect certification exam guide",
        "cloud.google.com",
    )
    hits = [u for u in cert if "/learn/certification/" in u]
    if hits:
        print("\n  → 試験ガイドが引けました。設計を見直せます。")
    else:
        print("\n  → 想定どおり試験ガイドは含まれていません。")
        print("     出題範囲はユーザーに貼り付けてもらう方式で進めます。")

    print("\n" + "=" * 60)
    print("技術ドキュメントが引けていれば ADK に進めます。")
    return 0


if __name__ == "__main__":
    sys.exit(main())

動かすには pip install google-adk httpx だけで足ります。

参考

• Developer Knowledge API
• Search and retrieve documents
• Corpus reference
• Quotas and limits
• Agent Development Kit
• Gemini Developer API pricing