The built-in ADK metrics compare tool calls and answer text, or ask a judge model. None of them checks the rule the docs assistant lives by: end every answer with the page it came from, and only cite a page whose text a tool returned. That rule is implemented in Python, and once it is a metric, ADK can do more than report it. adk optimize runs the agent, scores it with your metric and lets an optimizer model rewrite the instruction until the score goes up. We start the docs assistant from a one-line instruction with no citation rule, write the metric, and let GEPA improve the instruction on four train questions while three other questions check the result. The score on the validation set before and after this recorded optimization run:
When you need this. A docs assistant whose answers link to their source page breaks the link when it forgets the citation or cites the wrong page. You need a metric for a rule like that as soon as you change the prompt or the model, and you can compare candidate instructions against that metric instead of changing the prompt without measuring it.
{
"before": {
"local-model": 0,
"ollama-unreachable": 0,
"drop-index": 0
},
"after": {
"drop-index": 1,
"ollama-unreachable": 1,
"local-model": 1
}
}Lab setup. Download the complete lab ZIP, extract it and open a terminal in adk-custom-metric-optimize/. Keep the bundled code, tests and fixtures together. Create the pinned environment before running the commands below:
python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements-lock.txt1. Start from an instruction that says nothing about citations
adk optimize needs the [eval] extra, which brings gepa. The agent’s instruction is one sentence. DOCS_INSTRUCTION_FILE lets us evaluate an optimized instruction later without editing the file.
"""The docs assistant before optimization: its instruction says nothing about citations."""
import os
from pathlib import Path
from google.adk import Agent
from google.adk.models.lite_llm import LiteLlm
from google.genai import types
# Gemini through ADK's native client; set DOCS_MODEL=ollama_chat/gemma4:e2b to run Gemma locally.
MODEL = os.environ.get("DOCS_MODEL", "gemini-3.8-flash")
DOCS = Path(__file__).resolve().parent / "docs"
INSTRUCTION = "You answer questions about the docs-assistant project. Use the tools to look things up."
# To evaluate an optimized instruction without editing this file: DOCS_INSTRUCTION_FILE=optimized-instruction.txt
if os.environ.get("DOCS_INSTRUCTION_FILE"):
INSTRUCTION = Path(os.environ["DOCS_INSTRUCTION_FILE"]).read_text(encoding="utf-8").strip()
def list_pages() -> dict:
"""List the docs pages by path."""
return {"pages": sorted(page.name for page in DOCS.glob("*.md"))}
def read_page(path: str) -> dict:
"""Return the Markdown of one docs page, by the path list_pages returned."""
page = DOCS / path
if page.parent != DOCS or not page.is_file():
return {"error": f"No page {path!r}. Call list_pages for valid paths."}
return {"path": path, "content": page.read_text(encoding="utf-8")}
root_agent = Agent(
name="docs_assistant",
model=LiteLlm(model=MODEL) if MODEL.startswith("ollama_chat/") else MODEL,
instruction=INSTRUCTION,
tools=[list_pages, read_page],
generate_content_config=types.GenerateContentConfig(
temperature=0.2, thinking_config=types.ThinkingConfig(thinking_level="low")),
)The eval sets are ordinary ADK eval sets in the agent folder: train.evalset.json with four questions, validation.evalset.json with three different ones. Each expected answer ends with the page that answers it.
{
"eval_set_id": "train",
"name": "train",
"eval_cases": [
{
"eval_id": "rebuild-index",
"conversation": [
{
"invocation_id": "inv1",
"user_content": {
"role": "user",
"parts": [
{
"text": "How do I rebuild the search index?"
}
]
},
"final_response": {
"role": "model",
"parts": [
{
"text": "Run `docs-assistant index --rebuild` (cli.md)."
}
]
}
}
],
"session_input": {
"app_name": "docs_assistant",
"user_id": "dev"
}
},
{
"eval_id": "python-version",
"conversation": [
{
"invocation_id": "inv1",
"user_content": {
"role": "user",
"parts": [
{
"text": "Which Python version do I need?"
}
]
},
"final_response": {
"role": "model",
"parts": [
{
"text": "Python 3.12 or later (install.md)."
}
]
}
}
],
"session_input": {
"app_name": "docs_assistant",
"user_id": "dev"
}
},
{
"eval_id": "wrong-citation",
"conversation": [
{
"invocation_id": "inv1",
"user_content": {
"role": "user",
"parts": [
{
"text": "The assistant cites the wrong page. What should I do?"
}
]
},
"final_response": {
"role": "model",
"parts": [
{
"text": "Rebuild the search index with `docs-assistant index --rebuild` (troubleshooting.md)."
}
]
}
}
],
"session_input": {
"app_name": "docs_assistant",
"user_id": "dev"
}
},
{
"eval_id": "ask-output",
"conversation": [
{
"invocation_id": "inv1",
"user_content": {
"role": "user",
"parts": [
{
"text": "What does the ask command print?"
}
]
},
"final_response": {
"role": "model",
"parts": [
{
"text": "`docs-assistant ask QUESTION` prints an answer with citation links (cli.md)."
}
]
}
}
],
"session_input": {
"app_name": "docs_assistant",
"user_id": "dev"
}
}
]
}2. Write the metric
"""A custom ADK eval metric: the answer cites the expected page, and the agent read that page first."""
import re
import statistics
from typing import Optional
from google.adk.evaluation.conversation_scenarios import ConversationScenario
from google.adk.evaluation.eval_case import Invocation, get_all_tool_calls_with_responses
from google.adk.evaluation.eval_metrics import EvalMetric, EvalStatus
from google.adk.evaluation.evaluator import EvaluationResult, PerInvocationResult
CITATION = re.compile(r"\(([\w-]+\.md)\)\W*$")
def cited_page(invocation: Optional[Invocation]) -> Optional[str]:
"""The page path in parentheses at the end of the final response, if any."""
if invocation is None or invocation.final_response is None:
return None
text = "".join(part.text or "" for part in invocation.final_response.parts or [])
match = CITATION.search(text.strip())
return match.group(1) if match else None
def pages_read(invocation: Invocation) -> set[str]:
# A proposed call is not a completed read. Require a matched result with page text.
pages = set()
for call, response in get_all_tool_calls_with_responses(invocation.intermediate_data):
if call.name != "read_page" or not call.id or response is None or response.name != "read_page":
continue
body, path = response.response, (call.args or {}).get("path")
if (isinstance(path, str) and isinstance(body, dict) and body.get("path") == path
and isinstance(body.get("content"), str) and not {"error", "error_code"} & body.keys()):
pages.add(path)
return pages
def score_invocation(actual: Invocation, expected: Optional[Invocation]) -> float:
page = cited_page(actual)
if page is None or page not in pages_read(actual):
return 0.0 # no citation, or a citation of a page the agent never read
return 1.0 if expected is None or page == cited_page(expected) else 0.0
def cited_page_was_read(
eval_metric: EvalMetric,
actual_invocations: list[Invocation],
expected_invocations: Optional[list[Invocation]],
conversation_scenario: Optional[ConversationScenario],
) -> EvaluationResult:
threshold = eval_metric.criterion.threshold # EvalMetric.threshold is deprecated in favor of criterion
if expected_invocations is not None and len(expected_invocations) != len(actual_invocations):
raise ValueError("Actual and expected turn counts must match; refusing a partial score.")
expected_invocations = ([None] * len(actual_invocations)
if expected_invocations is None else expected_invocations)
results = []
for actual, expected in zip(actual_invocations, expected_invocations):
score = score_invocation(actual, expected)
results.append(PerInvocationResult(
actual_invocation=actual, expected_invocation=expected, score=score,
eval_status=EvalStatus.PASSED if score >= threshold else EvalStatus.FAILED))
if not results:
return EvaluationResult(overall_eval_status=EvalStatus.NOT_EVALUATED)
overall = statistics.mean(r.score for r in results)
return EvaluationResult(
overall_score=overall, per_invocation_results=results,
overall_eval_status=EvalStatus.PASSED if overall >= threshold else EvalStatus.FAILED)ADK calls a custom metric with the EvalMetric, the agent’s actual Invocations, the expected ones from the eval set, and the ConversationScenario when the case has one. Each Invocation is one user turn: the user’s message, the final response, and intermediate_data with every event of the turn. get_all_tool_calls_with_responses matches calls to results by ID. pages_read requires a read_page call with a matching named result, the same path, page content and no error field. A proposed call or a failed read does not count.
score_invocation gives 1.0 only when the answer ends with a page in parentheses, the agent read that page in the same turn, and it is the page the expected answer cites. A citation of a page the agent never read scores 0, which is the case the outage lesson showed the model can produce. The function returns an EvaluationResult with a PerInvocationResult per turn and the mean as overall_score. Actual and expected turn counts must match; otherwise scoring raises instead of silently ignoring turns. When scenario cases have no expected turns, it checks successful reads without choosing an expected page. The pass threshold comes from eval_metric.criterion.threshold, because EvalMetric.threshold is deprecated.
Why not a judge model? A judge could be asked the same question, but it costs a model call per turn and can disagree with itself between runs. A rule that code can decide exactly should be code; save judges for qualities such as tone or completeness.
A citation of a page that was never read. A text-only check would pass an answer that ends with (cli.md) after a failed read. Require a matching successful read_page result with content; a call alone does not show a read succeeded.
Mismatched turn counts. Scoring fewer expected turns than actual turns could omit the extra responses. Reject unequal turn counts before computing a mean.
3. Register the metric
{
"criteria": {
"cited_page_was_read": 1.0
},
"custom_metrics": {
"cited_page_was_read": {
"code_config": {
"name": "citation_metric.cited_page_was_read"
},
"metric_info": {
"metric_name": "cited_page_was_read",
"description": "The answer cites the expected page, and the agent read that page.",
"metric_value_info": {
"interval": {
"min_value": 0.0,
"max_value": 1.0
}
}
}
}
}
}The saved model runs were rescored with the corrected call-and-result check; their scores stayed the same. These scores check the citation rule, not whether the answer follows from the page text.
The metric needs two entries with the same name: one under criteria with its threshold, and one under custom_metrics with code_config.name, the import path of the function. metric_info is optional, but if you give it, it must be complete: metric_name and metric_value_info are required, as the recorded validation error shows.
Partial metric_info. adk eval stops with a ValidationError for EvalConfig: metricName and metricValueInfo are required. Give metric_name and metric_value_info.interval, or leave metric_info out.
# Run partial-metric-info
$ adk eval docs_assistant train --config_file_path config-partial-metric-info.json
pydantic_core._pydantic_core.ValidationError: 2 validation errors for EvalConfigPYTHONPATH=. ADK_DISABLE_LOAD_DOTENV=1 adk eval docs_assistant validation --config_file_path eval_config.json --print_detailed_resultsPYTHONPATH=. is needed because adk imports the metric by name, and a console script does not put the current folder on the import path. The baseline failed every validation question. The answers were right, but none of them named a page:
# Run baseline-validation
$ PYTHONPATH=. adk eval docs_assistant validation --config_file_path eval_config.json --print_detailed_results
Tests passed: 0
Tests failed: 3
prompt: How do I point the assistant at a local model?
actual_response: To point the assistant at a local model, set the `OLLAMA_API_BASE` environment
variable to your local model endpoint (such as your local Ollama server address)
instead of using Gemini.
read_page(
path='cli.md')
read_page(
path='install.md')
cited_page_was_read: FAILED (0.0)
prompt: The assistant can't reach Ollama. What do I check?
(troubleshooting.md).
actual_response: Check that the `OLLAMA_API_BASE` environment variable points to a running Ollama
server.
read_page(
path='troubleshooting.md')
cited_page_was_read: FAILED (0.0)
prompt: What command drops the search index?
(cli.md).
actual_response: The command that drops the search index (and rebuilds it from the docs folder)
is:
```bash
docs-assistant index --rebuild
```
read_page(
path='cli.md')
read_page(
path='troubleshooting.md')
cited_page_was_read: FAILED (0.0)Metric module not importable. Every case is NOT_EVALUATED with Could not import custom metric function. The agent ran and cost tokens; only scoring failed. Run adk with the metric’s folder on PYTHONPATH, or ship the metric in an installed package.
# Run metric-not-on-path
$ adk eval docs_assistant train --config_file_path eval_config.json
Eval Run Summary
train:
Tests passed: 0
Tests failed: 4
2026-10-21 21:04:00,126 - ERROR - local_eval_service.py:436 - Metric evaluation failed for metric `cited_page_was_read` for eval case id 'wrong-citation' with following error `Could not import custom metric function from ...
ModuleNotFoundError: No module named 'citation_metric'
ImportError: Could not import custom metric function from citation_metric.cited_page_was_read4. Optimize the instruction
{
"eval_config": {
"criteria": {
"cited_page_was_read": 1.0
},
"custom_metrics": {
"cited_page_was_read": {
"code_config": {
"name": "citation_metric.cited_page_was_read"
},
"metric_info": {
"metric_name": "cited_page_was_read",
"description": "The answer cites the expected page, and the agent read that page.",
"metric_value_info": {
"interval": {
"min_value": 0.0,
"max_value": 1.0
}
}
}
}
}
},
"app_name": "docs_assistant",
"train_eval_set": "train",
"validation_eval_set": "validation"
}{
"optimizer_model": "gemini-3.8-flash",
"max_metric_calls": 40,
"reflection_minibatch_size": 2
}The sampler config tells LocalEvalSampler which eval config to score with and which eval sets to use. The optimizer config sets GEPA’s reflection model, a budget of 40 metric calls (the default is 100) and two train cases per reflection step.
PYTHONPATH=. ADK_DISABLE_LOAD_DOTENV=1 adk optimize docs_assistant --sampler_config_file_path sampler_config.json --optimizer_config_file_path optimizer_config.json --print_detailed_results$ PYTHONPATH=. adk optimize docs_assistant --sampler_config_file_path sampler_config.json --optimizer_config_file_path optimizer_config.json --print_detailed_results
Iteration 0: Base program full valset score: 0.0 over 3 / 3 examples
Iteration 1: Accepted candidate (subsample score 0.0 -> 2.0); running full eval.
Iteration 1: Valset score for new program: 1.0 (coverage 3 / 3)
Iteration 2: Accepted candidate (subsample score 1.0 -> 2.0); running full eval.
Iteration 2: Valset score for new program: 1.0 (coverage 3 / 3)
Iteration 3: Reflective mutation did not propose a new candidate
Iteration 4: Accepted candidate (subsample score 1.0 -> 2.0); running full eval.
Iteration 4: Valset score for new program: 1.0 (coverage 3 / 3)
Iteration 5: Reflective mutation did not propose a new candidate
Iteration 6: Reflective mutation did not propose a new candidate
Iteration 7: Reflective mutation did not propose a new candidate
Iteration 8: Reflective mutation did not propose a new candidate
Iteration 9: Reflective mutation did not propose a new candidate
Iteration 10: Reflective mutation did not propose a new candidate
Iteration 11: Reflective mutation did not propose a new candidateThe first proposed instruction already scored 1.0 on all three validation questions. Later iterations proposed variants that tied, and once every sampled train score was perfect, GEPA had no failures to reflect on and proposed nothing. This is the instruction adk optimize printed as the best:
You are an assistant that answers questions about the docs-assistant project by looking up information in the project documentation.
### Workflow & Tool Usage
1. Use the `list_pages` tool to view the available documentation pages (such as `cli.md`, `install.md`, `troubleshooting.md`).
2. Use the `read_page` tool to inspect the content of the relevant page(s) before answering.
3. Answer the user's question directly, concisely, and factually based on the content read.
### Citation Requirement
- Every answer MUST cite the source page from which the information was retrieved.
- Format the citation by appending the filename in parentheses at the end of the statement or answer, for example: `(<filename.md>)` (e.g., `(troubleshooting.md)` or `(install.md)`).
- Only cite pages that were actively retrieved and read using `read_page`.
### Known Project Reference Information
- **Installation & Requirements (`install.md`)**:
- Python requirement: Python 3.12 or later.
- Local model configuration: Set `OLLAMA_API_BASE` to use a local model instead of Gemini.
- **Troubleshooting (`troubleshooting.md`)**:
- If the assistant cites the wrong page: Rebuild the search index with `docs-assistant index --rebuild`.
- If the assistant cannot reach Ollama: Check that `OLLAMA_API_BASE` points to a running Ollama server.
- **CLI Commands (`cli.md`)**: CLI usage and commands.5. Review the optimized instruction before you ship it
The citation rules are what we wanted. The “Known Project Reference Information” section is not. GEPA’s reflection model saw the pages the agent read during the train runs and copied their facts into the instruction: the Python version, the OLLAMA_API_BASE setting, the rebuild command. Two things follow, and both are in the recording.
First, every model call now carries those facts. The first model call of each validation question used these prompt tokens before and after:
{
"local-model": {
"before": 134,
"after": 457
},
"ollama-unreachable": {
"before": 137,
"after": 460
},
"drop-index": {
"before": 130,
"after": 453
}
}Second, the facts go stale when the docs change. record.py edited install.md to say Python 3.13 and asked the optimized agent:
# Run optimized-after-docs-change (install.md now says Python 3.13)
> Which Python version do I need?
call list_pages {}
call read_page {'path': 'install.md'}
answer You need Python 3.13 or later (install.md).This time the agent read the page and answered from it, because the instruction also says to cite only pages it read. The instruction still states 3.12; a later model or question may favor that stale text instead of the page. Delete the copied facts before you commit the instruction, then score it again; a metric that rewards reading the page is what made the facts redundant.
Docs facts copied into the instruction. The optimized instruction repeats page content, costs prompt tokens on every call and goes stale when the docs change. Review the diff, delete copied facts, and score the edited instruction again.
How GEPA selects root-agent instructions
adk optimize loads the root agent, wraps the sampler config in a LocalEvalSampler and runs GEPARootAgentPromptOptimizer, which changes only the root agent’s instruction (not sub-agents, tools or skills). GEPA keeps a pool of candidate instructions. Each iteration picks a candidate, runs it on a minibatch of train cases through the sampler, and gives the reflection model the instruction plus the scored runs, including the agent’s tool calls and answers. The reflection model writes a new instruction. If the new one scores better on the same minibatch, GEPA scores it on the full validation set and adds it to the pool; candidates are kept on a Pareto front per validation case, so an instruction that is best on one case survives even if its average is lower.
The optimizer uses validation scores to choose candidates, so this is a selection set, not an independent final test. Its reflection prompts use the sampled train runs. A perfect selection score covers these three questions under this metric; reserve another untouched set before estimating how a chosen instruction behaves on new questions.
Test the metric offline
test_citation_metric.py re-scores the saved baseline and optimized runs with the metric, checks both config files with ADK’s EvalConfig, and imports the metric the way the eval process does.
python -m unittest -v test_citation_metric.pytest_a_citation_of_an_unread_page_scores_zero (test_citation_metric.CitationMetric.test_a_citation_of_an_unread_page_scores_zero) ... ok
test_a_failed_read_does_not_count_as_reading_the_page (test_citation_metric.CitationMetric.test_a_failed_read_does_not_count_as_reading_the_page) ... ok
test_metric_info_without_its_required_fields_is_rejected (test_citation_metric.CitationMetric.test_metric_info_without_its_required_fields_is_rejected) ... ok
test_mismatched_turn_counts_are_rejected_instead_of_truncated (test_citation_metric.CitationMetric.test_mismatched_turn_counts_are_rejected_instead_of_truncated) ... ok
test_missing_or_mismatched_read_results_do_not_count (test_citation_metric.CitationMetric.test_missing_or_mismatched_read_results_do_not_count) ... ok
test_the_baseline_agent_never_cites_a_page (test_citation_metric.CitationMetric.test_the_baseline_agent_never_cites_a_page) ... ok
test_the_metric_module_must_be_importable_by_the_eval_process (test_citation_metric.CitationMetric.test_the_metric_module_must_be_importable_by_the_eval_process) ... ok
test_the_optimized_instruction_passes_the_validation_set (test_citation_metric.CitationMetric.test_the_optimized_instruction_passes_the_validation_set) ... ok
test_the_optimizer_copied_docs_facts_into_the_instruction (test_citation_metric.CitationMetric.test_the_optimizer_copied_docs_facts_into_the_instruction) ... ok
----------------------------------------------------------------------
Ran 9 tests in 1.076s
OKAll 9 tests pass.
Read the test and recording scripts
"""Re-score the recorded runs with the custom metric offline, and check the configs and the optimized instruction.
Run: python -m unittest -v test_citation_metric.py
"""
import json
import subprocess
import sys
import unittest
import warnings
from pathlib import Path
warnings.filterwarnings("ignore")
from google.adk.evaluation.eval_config import EvalConfig, get_eval_metrics_from_config # noqa: E402
from google.adk.evaluation.eval_result import EvalSetResult # noqa: E402
from pydantic import ValidationError # noqa: E402
import citation_metric # noqa: E402
HERE = Path(__file__).resolve().parent
CONFIG = EvalConfig.model_validate_json((HERE / "eval_config.json").read_text())
METRIC = next(m for m in get_eval_metrics_from_config(CONFIG) if m.metric_name == "cited_page_was_read")
RESULTS = {p.stem: EvalSetResult.model_validate_json(p.read_text()) for p in (HERE / "results").glob("*.json")}
def invocations(run):
return [(i.actual_invocation, i.expected_invocation)
for case in RESULTS[run].eval_case_results for i in case.eval_metric_result_per_invocation]
def rescore(run):
pairs = invocations(run)
return citation_metric.cited_page_was_read(METRIC, [a for a, _ in pairs], [e for _, e in pairs], None)
def first_prompt_tokens(run):
return sorted(next(e.usage_metadata.prompt_token_count for e in a.intermediate_data.invocation_events if e.usage_metadata)
for a, _ in invocations(run))
def setUpModule():
warnings.simplefilter("ignore")
class CitationMetric(unittest.TestCase):
def test_the_baseline_agent_never_cites_a_page(self):
for run in ("baseline-train", "baseline-validation"):
result = rescore(run)
self.assertEqual(result.overall_score, 0.0)
self.assertTrue(all(citation_metric.cited_page(a) is None for a, _ in invocations(run)))
def test_the_optimized_instruction_passes_the_validation_set(self):
self.assertEqual(rescore("optimized-validation").overall_score, 1.0)
recorded = [m.score for c in RESULTS["optimized-validation"].eval_case_results
for m in c.overall_eval_metric_results if m.metric_name == "cited_page_was_read"]
self.assertEqual(recorded, [1.0] * 3)
def test_a_citation_of_an_unread_page_scores_zero(self):
actual, expected = invocations("optimized-validation")[0]
unread = actual.model_copy(deep=True)
unread.intermediate_data.invocation_events = [
e for e in unread.intermediate_data.invocation_events
if not (e.content and any(p.function_call and p.function_call.name == "read_page" for p in e.content.parts))]
self.assertEqual(citation_metric.score_invocation(actual, expected), 1.0)
self.assertEqual(citation_metric.score_invocation(unread, expected), 0.0)
def test_a_failed_read_does_not_count_as_reading_the_page(self):
actual, expected = invocations("optimized-validation")[0]
failed = actual.model_copy(deep=True)
for event in failed.intermediate_data.invocation_events:
for part in event.content.parts if event.content else []:
if part.function_response and part.function_response.name == "read_page":
part.function_response.response = {"error_code": 503, "error_message": "Unavailable"}
self.assertEqual(citation_metric.score_invocation(failed, expected), 0.0)
def test_missing_or_mismatched_read_results_do_not_count(self):
actual, expected = invocations("optimized-validation")[0]
for mode in ("missing", "wrong-id", "wrong-path", "no-content"):
changed = actual.model_copy(deep=True)
for event in changed.intermediate_data.invocation_events:
if not event.content:
continue
if mode == "missing":
event.content.parts = [p for p in event.content.parts if not p.function_response]
for part in event.content.parts:
response = part.function_response
if response and response.name == "read_page":
if mode == "wrong-id":
response.id = "unrelated-call"
elif mode == "wrong-path":
response.response["path"] = "unrelated.md"
else:
response.response.pop("content", None)
with self.subTest(mode=mode):
self.assertEqual(citation_metric.score_invocation(changed, expected), 0.0)
def test_mismatched_turn_counts_are_rejected_instead_of_truncated(self):
actual, expected = invocations("optimized-validation")[0]
for actuals, expecteds in (([actual, actual], [expected]), ([actual], [expected, expected]),
([actual], []), ([], [expected])):
with self.subTest(actual=len(actuals), expected=len(expecteds)):
with self.assertRaisesRegex(ValueError, "turn counts"):
citation_metric.cited_page_was_read(METRIC, actuals, expecteds, None)
def test_metric_info_without_its_required_fields_is_rejected(self):
with self.assertRaises(ValidationError):
EvalConfig.model_validate_json((HERE / "config-partial-metric-info.json").read_text())
self.assertEqual(CONFIG.custom_metrics["cited_page_was_read"].metric_info.metric_value_info.interval.max_value, 1.0)
def test_the_metric_module_must_be_importable_by_the_eval_process(self):
code = ("from google.adk.evaluation.custom_metric_evaluator import _get_metric_function as get; "
"get('citation_metric.cited_page_was_read')")
isolated = subprocess.run([sys.executable, "-I", "-c", code], cwd=HERE, capture_output=True, text=True)
self.assertIn("Could not import custom metric function", isolated.stderr)
on_path = subprocess.run([sys.executable, "-c", code], cwd=HERE, capture_output=True, text=True,
env={"PYTHONPATH": str(HERE), "PATH": ""})
self.assertEqual(on_path.returncode, 0, on_path.stderr)
self.assertIn("No module named 'citation_metric'", (HERE / "run-output.txt").read_text())
def test_the_optimizer_copied_docs_facts_into_the_instruction(self):
instruction = (HERE / "optimized-instruction.txt").read_text()
self.assertIn("Python 3.12 or later", instruction)
self.assertIn("Python 3.12 or later", (HERE / "docs_assistant/docs/install.md").read_text())
baseline, optimized = first_prompt_tokens("baseline-validation"), first_prompt_tokens("optimized-validation")
self.assertTrue(all(o > b for o, b in zip(optimized, baseline)))
self.assertIn("answer You need Python 3.13 or later (install.md).", (HERE / "run-output.txt").read_text())
if __name__ == "__main__":
unittest.main()"""Score the docs assistant with the custom citation metric, optimize its instruction with GEPA,
and score the optimized instruction on the held-out validation set.
Run from this folder with GOOGLE_API_KEY set, in a venv with google-adk[eval]: python record.py
Writes run-output.txt, optimize-output.txt, optimized-instruction.txt and results/<run>.json.
"""
import json
import os
import re
import shutil
import subprocess
import sys
from pathlib import Path
HERE = Path(__file__).resolve().parent
ADK = str(Path(sys.executable).with_name("adk"))
HISTORY = HERE / "docs_assistant" / ".adk" / "eval_history"
ENV = {**os.environ, "ADK_DISABLE_LOAD_DOTENV": "1", "PYTHONWARNINGS": "ignore"}
WITH_PATH = {"PYTHONPATH": str(HERE)} # so `adk` can import citation_metric
EVAL = ["eval", "docs_assistant"]
RUNS = {
"partial-metric-info": (EVAL + ["train", "--config_file_path", "config-partial-metric-info.json"], {}),
"metric-not-on-path": (EVAL + ["train", "--config_file_path", "eval_config.json"], {}),
"baseline-train": (EVAL + ["train", "--config_file_path", "eval_config.json", "--print_detailed_results"], WITH_PATH),
"baseline-validation": (EVAL + ["validation", "--config_file_path", "eval_config.json", "--print_detailed_results"], WITH_PATH),
}
def clean(line: str) -> str:
line = re.sub(r"^.*?(\d{4}-\d\d-\d\d \d\d:)", r"\1", line)
return line if len(line) <= 220 else line[:220] + " ..."
def compact(value):
if isinstance(value, dict):
return {k: compact(v) for k, v in value.items() if v is not None and k != "thought_signature"}
if isinstance(value, list):
return [compact(v) for v in value]
return value
def adk(name: str, args: list, extra: dict) -> str:
shutil.rmtree(HISTORY, ignore_errors=True)
result = subprocess.run([ADK, *args], cwd=HERE, env={**ENV, **extra}, capture_output=True, text=True, timeout=3600)
errors = [clean(l) for l in result.stderr.splitlines() if re.search(r"(Error|Exception):|ERROR -", l)]
saved = sorted(HISTORY.glob("*.json"))
if saved:
(HERE / "results" / f"{name}.json").write_text(json.dumps(compact(json.loads(saved[-1].read_text())), indent=1))
env = "".join(f"{k}={Path(v).relative_to(HERE) or '.'} " for k, v in extra.items())
return f"# Run {name}\n$ {env}adk {' '.join(args)}\n" + result.stdout.strip() + (
"\n" + "\n".join(dict.fromkeys(errors)) if errors else "") + "\n"
PROBE = """
import asyncio, sys
from google.adk.runners import InMemoryRunner
from google.genai import types
from docs_assistant.agent import root_agent
async def main():
runner = InMemoryRunner(agent=root_agent, app_name="docs_assistant")
session = await runner.session_service.create_session(app_name="docs_assistant", user_id="dev")
message = types.Content(role="user", parts=[types.Part(text=sys.argv[1])])
async for event in runner.run_async(user_id="dev", session_id=session.id, new_message=message):
for part in event.content.parts if event.content else []:
if part.function_call:
print("call ", part.function_call.name, dict(part.function_call.args or {}))
elif part.text and not part.thought and event.is_final_response():
print("answer", part.text.strip())
asyncio.run(main())
"""
def changed_docs_probe() -> str:
"""Edit install.md, then ask the optimized agent what the page now says."""
page = HERE / "docs_assistant" / "docs" / "install.md"
original = page.read_text()
page.write_text(original.replace("Python 3.12 or later.", "Python 3.13 or later."))
try:
question = "Which Python version do I need?"
result = subprocess.run([sys.executable, "-c", PROBE, question], cwd=HERE, capture_output=True, text=True,
env={**ENV, "DOCS_INSTRUCTION_FILE": str(HERE / "optimized-instruction.txt")}, timeout=300)
finally:
page.write_text(original)
return (f"# Run optimized-after-docs-change (install.md now says Python 3.13)\n> {question}\n"
+ result.stdout.strip() + "\n")
def main() -> None:
(HERE / "results").mkdir(exist_ok=True)
output = [adk(name, args, extra) for name, (args, extra) in RUNS.items()]
optimize = adk("optimize", ["optimize", "docs_assistant", "--sampler_config_file_path", "sampler_config.json",
"--optimizer_config_file_path", "optimizer_config.json", "--print_detailed_results"], WITH_PATH)
(HERE / "optimize-output.txt").write_text(optimize)
found = re.search(r"Optimized root agent instructions:\n-+\n(.*?)\n=+", optimize, re.S)
if found:
(HERE / "optimized-instruction.txt").write_text(found.group(1).strip() + "\n")
output.append(adk("optimized-validation", EVAL + ["validation", "--config_file_path", "eval_config.json",
"--print_detailed_results"],
{**WITH_PATH, "DOCS_INSTRUCTION_FILE": str(HERE / "optimized-instruction.txt")}))
output.append(changed_docs_probe())
shutil.rmtree(HERE / "docs_assistant" / ".adk", ignore_errors=True)
(HERE / "run-output.txt").write_text("\n".join(output))
if __name__ == "__main__":
main()You rerun adk optimize with the same configs after adding two pages to the docs. The new best instruction scores 1.0 on validation again and lists facts from all five pages.
Read the explanation
An optimizer maximizes the metric you give it; anything the metric doesn't check, including copied facts, is yours to review.
- Apply safety and security guardrails to agent tools: control which tools an agent may call, for whom, and with whose approval.
- Test an agent against simulated users: use the same custom metric on multi-turn scenario cases, which pass a
ConversationScenarioinstead of expected turns.
References
Official sources checked October 21, 2026, against google-adk 2.12.0.