Evaluate Google ADK Agents: The 6-Step 2026 Production Loop

TL;DR

Steps 1 to 3 are copy-paste runnable. Steps 4 to 6 are integration patterns that adapt to your agent module, dataset, and mapper. The code targets ADK 2.x, and --otel_to_cloud needs ADK 1.17.0 or later. Run Steps 2, 4 and 6, plus any Simulate SDK code, in a separate agent-learning-kit environment.

Why ADK’s Built-in Evaluation Falls Short After Deployment and What This Guide Covers

To evaluate a Google ADK agent, write .test.json or evalset files, set thresholds in test_config.json, and run them with adk eval, pytest’s AgentEvaluator, or the adk web UI. Google’s Agent Development Kit (ADK) scores tool trajectories, final responses, hallucinations, safety and multi-turn conversations.

Things get thin after deployment. ADK’s eval tools are designed for the development loop: you define test cases, run them through pytest or the CLI, and check scores before merging code. That workflow does not extend into production.

Once your agents start handling real user requests, you lose visibility into quality drift, cost attribution per agent, latency bottlenecks across workflow steps, and continuous quality scoring on live traffic. To close that gap between development eval and production Google ADK observability, you instrument ADK agents with traceAI, score each workflow pattern ADK supports, and track cost, latency and quality on live traffic in Observe.

ADK Python 2.0 went GA on May 19, 2026, ADK 2.10 added informational efficiency criteria, and --otel_to_cloud exports traces to Google Cloud. The open layer is continuous, span-level quality scoring on live agent traces, which the rest of this guide builds.

A Concrete Failure Your Tests Won’t Catch

The agent passes every .test.json case in CI, then in production you see a trace like this:

root_agent.run("Find me a vegetarian restaurant in Berlin and book a 7pm table")
├── search_agent.run(...)   tool="google_search"   ok
├── search_agent.run(...)   tool="google_search"   ok   (same query)
├── search_agent.run(...)   tool="google_search"   ok   (third identical call)
└── booking_agent.run(...)  tool="reserve_table"   ok
final_response = "I've booked you a table at a vegetarian restaurant in Berlin for 7pm."

tool_trajectory_avg_score passes with ANY_ORDER. response_match_score passes on keyword overlap. safety_v1 passes. But the agent looped on google_search three times, the final answer omits the restaurant name and address, and the user lands at a phantom reservation.

The eval suite missed it because it graded only the final response and trajectory, with no check on per-step output, loop count or grounding in the search results. Production eval has to catch this class of failure before your users do. The fix is the ADK Production Eval Loop: six steps that go from a first hello-trace to persona-driven simulation and prompt optimization.

Prerequisites

- Python 3.11 or 3.12
- google-adk 2.x (this install resolves 2.1.0, as traceai-google-adk caps google-genai below 2.0)
- A Google API key for Gemini (GOOGLE_API_KEY)
- A Future AGI project + API key + secret (FI_API_KEY, FI_SECRET_KEY). Sign up at https://futureagi.com
- An OpenAI API key for Step 6 (OPENAI_API_KEY)

Install the traced ADK service:

pip install traceai-google-adk "google-adk[eval]"

Two things to know about the packages:

• traceai-google-adk pulls in fi-instrumentation-otel transitively. That’s the package that gives you from fi_instrumentation import register.
• agent-learning-kit gives you from fi.evals import evaluate plus the fi.simulate and fi.opt SDKs for Steps 5 and 6. Install it with pip install agent-learning-kit in a separate virtual environment.

Set the four environment variables:

export FI_API_KEY="..."
export FI_SECRET_KEY="..."
export GOOGLE_API_KEY="..."
export OPENAI_API_KEY="..."

Step 1. Install and Verify the Trace Pipeline

The fastest way to confirm everything works is the canonical hello-trace: a single-agent app that answers a weather question with a tool call. Save this as hello-trace.py:

import asyncio

from google.adk.agents import Agent
from google.adk.runners import InMemoryRunner
from google.genai import types

from traceai_google_adk import GoogleADKInstrumentor
from fi_instrumentation import register
from fi_instrumentation.fi_types import ProjectType

tracer_provider = register(
    project_name="adk_demo",
    project_type=ProjectType.OBSERVE,
)
GoogleADKInstrumentor().instrument(tracer_provider=tracer_provider)

def get_weather(city: str) -> dict:
    if city.lower() == "new york":
        return {"status": "success",
                "report": "New York is sunny, 25°C (77°F)."}
    return {"status": "error", "error_message": f"No data for {city}."}

agent = Agent(
    name="weather_agent",
    model="gemini-3.8-flash",
    description="Answer weather questions using the weather tool.",
    instruction="You must use the available tools to find an answer.",
    tools=[get_weather],
)

async def main():
    app_name, user_id, session_id = "weather_app", "demo_user", "demo_session"
    runner = InMemoryRunner(agent=agent, app_name=app_name)
    await runner.session_service.create_session(
        app_name=app_name, user_id=user_id, session_id=session_id,
    )
    async for event in runner.run_async(
        user_id=user_id,
        session_id=session_id,
        new_message=types.Content(
            role="user", parts=[types.Part(text="What is the weather in New York?")]
        ),
    ):
        if event.is_final_response():
            print(event.content.parts[0].text.strip())

if __name__ == "__main__":
    asyncio.run(main())

Run it:

python hello-trace.py
# Example output (Gemini may rephrase): "New York is sunny, 25°C (77°F)."

The Runner.run_async call became one OTel span. The Gemini call inside the agent became a child span. The get_weather tool call became another child span. Open Observe in your Future AGI project, where you’ll see the trace tree under the adk_demo project with the agent invocation, the LLM call, and the tool call as separate spans.

If the trace doesn’t show up, the most common cause is missing env vars. register() reads FI_API_KEY and FI_SECRET_KEY from the environment. You now have a working trace pipeline. Step 2 attaches eval scores to those spans.

Step 2. Score Outputs with the Unified evaluate() API

The unified evaluate() API is the simplest way to score an output against any Future AGI evaluator. It returns an EvalResult with score, passed, reason, and latency_ms. You skip client setup and nested result objects. Pass model="turing_flash" for a Turing judge, or any LiteLLM model with a custom prompt.

from fi.evals import evaluate

# Re-use Step 1's hello-trace output to keep the example concrete:
user_question = "What is the weather in New York?"
context = "New York is sunny, 25°C (77°F)."
response = "It's sunny in New York with a temperature of 25°C."

# Local heuristic. Runs offline, sub-second, free.
r = evaluate("contains", output="hello world", keyword="hello")
print(r.score, r.passed, r.reason)
# 1.0 True Keyword 'hello' found

# Cloud template via Future AGI Turing models. LLM-as-a-judge.
r = evaluate("groundedness", output=response, context=context, model="turing_flash")
print(r.score, r.passed, r.reason)
# 0 to 1 score, pass or fail, judge's reason

# Custom prompt against any LiteLLM-supported model.
r = evaluate(
    prompt="Rate clarity 0-1 of: {output}",
    output="ML is a subset of AI.",
    engine="llm",
    model="gemini/gemini-3.8-flash",
)

You can run multiple evaluators in one call:

from fi.evals import evaluate

# `response`, `context`, `user_question` are the same variables defined above.
results = evaluate(
    ["groundedness", "factual_accuracy", "toxicity"],
    output=response, context=context, input=user_question,
    model="turing_flash",
)
for r in results:
    print(r.eval_name, r.score, r.passed)

Auto-attaching scores to the active span

In your eval job, call Step 1’s register() with set_global_tracer_provider=True so spans reach Observe, then call enable_auto_enrichment() once at startup. After that, every evaluate() result inside an active span context attaches its score, reason, and latency to that span automatically.

from fi.evals import evaluate
from fi.evals.otel import enable_auto_enrichment, get_tracer

enable_auto_enrichment()
tracer = get_tracer()

# `ans` is the agent's answer for the current step; `ctx` is the grounding source.
ans = response                  # from the variables defined above
ctx = context

with tracer.start_as_current_span("rag-pipeline"):
    with tracer.start_as_current_span("generate-answer"):
        r = evaluate("groundedness", output=ans, context=ctx, model="turing_flash")
        # No span_id plumbing required. The score is on this span.

If you want explicit control over which span gets enriched (for sub-agent boundaries, for example), use enrich_span_with_evaluation:

from fi.evals.otel import enrich_span_with_evaluation

with tracer.start_as_current_span("quality-gate") as span:
    r = evaluate("faithfulness", output=ans, context=ctx)
    enrich_span_with_evaluation(
        metric_name="faithfulness",
        score=r.score,
        reason=r.reason[:200],
        latency_ms=r.latency_ms,
        span=span,
    )

Open Observe and you’ll see the eval score as a span attribute next to the trace, so you never stitch logs together to find it.

Score live ADK spans without touching agent code

In Observe, create an eval task on the ADK project, filter it to agent or LLM spans, and attach evals such as Task Completion or Groundedness. A Continuous run scores new spans as they arrive, with a sampling rate and span limit to control cost.

Step 3. Gate Releases on .test.json with AgentEvaluator

ADK’s built-in evaluation is the right tool for CI gates. Drop a .test.json fixture in your tests directory, define thresholds in a test_config.json, and let AgentEvaluator.evaluate() run them in pytest. Minimal tests/fixtures/weather_basic.test.json (the schema follows the official ADK evaluation docs):

{
  "eval_set_id": "weather_basic_set",
  "name": "weather basic queries",
  "description": "Single-turn weather agent smoke tests.",
  "eval_cases": [
    {
      "eval_id": "ny_weather",
      "conversation": [
        {
          "invocation_id": "1",
          "user_content": {"parts": [{"text": "What is the weather in New York?"}], "role": "user"},
          "final_response": {"parts": [{"text": "New York is sunny, 25°C (77°F)."}], "role": "model"},
          "intermediate_data": {
            "tool_uses": [{"args": {"city": "New York"}, "name": "get_weather"}],
            "intermediate_responses": []
          }
        }
      ],
      "session_input": {"app_name": "weather_app", "user_id": "test_user", "state": {}}
    }
  ]
}

Thresholds in tests/fixtures/test_config.json:

{
  "criteria": {
    "tool_trajectory_avg_score": 1.0,
    "response_match_score": 0.8
  }
}

AgentEvaluator.evaluate() discovers the agent through an importable Python module path. Save Step 1’s agent object as weather_app/agent.py with the special root_agent name ADK looks for, and put test_config.json in the same folder as the .test.json fixtures so ADK auto-loads it:

weather_app/
  __init__.py         # contains: from . import agent
  agent.py            # exports root_agent (the Agent from Step 1)
tests/
  fixtures/
    test_config.json
    weather_basic.test.json
  test_weather.py

weather_app/agent.py:

# Re-use the Agent definition from Step 1's hello-trace.
from google.adk.agents import Agent

def get_weather(city: str) -> dict:
    if city.lower() == "new york":
        return {"status": "success",
                "report": "New York is sunny, 25°C (77°F)."}
    return {"status": "error", "error_message": f"No data for {city}."}

agent = Agent(
    name="weather_agent",
    model="gemini-3.8-flash",
    description="Answer weather questions using the weather tool.",
    instruction="You must use the available tools to find an answer.",
    tools=[get_weather],
)

root_agent = agent   # AgentEvaluator looks up `root_agent` from the module

Run it from pytest. Pass the folder containing the test config and fixtures:

import pytest
from google.adk.evaluation.agent_evaluator import AgentEvaluator

@pytest.mark.asyncio
async def test_weather_basic():
    await AgentEvaluator.evaluate(
        agent_module="weather_app",
        eval_dataset_file_path_or_dir="tests/fixtures",
    )

ADK’s criteria reference lists every criterion, marks the ones that need a reference answer or a rubric, and flags the efficiency criteria that take no threshold. Start with tool_trajectory_avg_score (1.0) and response_match_score (0.8), ADK’s defaults when no criteria are set, and add hallucinations_v1 once you have grounding context. The efficiency criteria need ADK 2.10 or later, so try them in a plain pip install "google-adk[eval]" environment, not the traced service or the agent-learning-kit environment.

Run the same eval set from the adk CLI

Outside pytest, adk eval runs the same fixture against the agent folder and prints pass or fail per metric. adk eval_set create weather_app weather_regression writes an empty weather_app/weather_regression.evalset.json, and the adk web UI can save real sessions into it. Step 4 covers what to score per workflow pattern.

adk eval weather_app tests/fixtures/weather_basic.test.json \
  --config_file_path=tests/fixtures/test_config.json --print_detailed_results

Step 4. Apply the Right Eval Recipe per ADK Workflow Pattern

ADK gives you four orchestration patterns. SequentialAgent, ParallelAgent, LoopAgent, and dynamic-routing agents that transfer_to_agent. Each fails differently in production (ADK 2.0 adds graph-based and dynamic workflows), so run these recipes in your eval job and use Observe eval tasks on the live service.

SequentialAgent: Score each step’s output to prevent error compounding

from fi.evals import evaluate
from fi.evals.otel import enable_auto_enrichment

enable_auto_enrichment()

# Inside each sub-agent in the chain:
def _planning_step(input_text: str) -> str:
    plan = planning_agent.run(input_text)
    evaluate("prompt_instruction_adherence", instructions=input_text, output=plan, model="turing_flash")
    return plan

Each step’s eval auto-attaches to that step’s span. When the final response is wrong, you can trace which step compounded the error.

ParallelAgent: Score branch quality + merged-output consistency

from fi.evals import evaluate
from fi.evals.otel import enrich_span_with_evaluation

def _evaluate_parallel_outputs(branches: list[str], merged: str) -> None:
    for i, branch in enumerate(branches):
        r = evaluate("conversation_coherence", conversation=branch, model="turing_flash")
        # Rename the span attribute so each branch lands distinctly in Observe.
        enrich_span_with_evaluation(
            metric_name=f"branch_{i}_quality",
            score=r.score, reason=r.reason, latency_ms=r.latency_ms,
        )
    evaluate("factual_accuracy", output=merged, context="\n".join(branches),
             model="turing_flash")

Per-branch coherence catches a single bad branch dragging the merged output. Whole-output factual accuracy catches the merged result diverging from any individual branch.

LoopAgent: Score iteration count and per-iteration output

from fi.evals import evaluate
from fi.evals.otel import enrich_span_with_evaluation

def _evaluate_loop(iterations: list[str], goal: str, max_iters: int) -> None:
    if len(iterations) >= max_iters:
        evaluate("contains", output="MAX_ITERATIONS_HIT", keyword="MAX_ITERATIONS_HIT")
    for i, iter_output in enumerate(iterations):
        r = evaluate("prompt_instruction_adherence", instructions=goal, output=iter_output, model="turing_flash")
        enrich_span_with_evaluation(
            metric_name=f"loop_iter_{i}",
            score=r.score, reason=r.reason, latency_ms=r.latency_ms,
        )

Loop count is the runaway-cost signal. Per-iteration adherence catches the loop drifting away from the original goal.

Dynamic Routing: Score routing accuracy as classification

from fi.evals import evaluate
from fi.evals.otel import enrich_span_with_evaluation

def _evaluate_routing(query: str, chosen_agent: str, expected_agent: str) -> None:
    r = evaluate("contains", output=chosen_agent, keyword=expected_agent)
    enrich_span_with_evaluation(
        metric_name="routing_accuracy",
        score=r.score, reason=r.reason, latency_ms=r.latency_ms,
    )

Routing is a classification task. Treat it that way and score it on a known-routing dataset before you ship the routing rules.

ADK Built-in vs Future AGI for ADK at a glance:

Step 5. Run Pre-Production Scenarios with a Multi-Turn Simulator

Per-step eval and CI gates catch most failures before production. The remaining class is cross-turn behavior, where the agent looks fine on each turn but fails the conversation as a whole. That class needs scenario testing. Two tools cover this for ADK:

ADK’s user simulation fits chat ADK agents: a model plays the user from a ConversationScenario (starting prompt, conversation plan, optional persona such as NOVICE). Score those runs with the multi-turn criteria; reference-based ones such as response_match_score do not support simulated users.

Future AGI Simulate generates chat and voice test scenarios from personas you define, such as an impatient or confused customer, and scores each conversation with built-in evals such as loop detection and context retention. Fix My Agent then analyzes the simulation traces and recommends prompt changes.

For either path, the loop is the same: drive the persona, capture the conversation, score the transcript with ADK’s multi-turn criteria or Future AGI’s built-in conversation evals. The output you carry forward to Step 6 is the set of failing transcripts.

Step 6. Close the Loop with Prompt Optimization

If your scenarios surface a failure mode the agent reproduces (an instruction it ignores, a tool it picks wrong, a hallucinated detail), the next step is to optimize the prompt that drives the misbehavior. ADK’s adk optimize rewrites agent instructions from eval results; fi.opt in agent-learning-kit searches prompt variants against any evaluator, as below.

from fi.opt.optimizers import BayesianSearchOptimizer
from fi.opt.base.evaluator import Evaluator as OptEvaluator
from fi.opt.datamappers import BasicDataMapper
from fi.evals.metrics import BLEUScore

# `failing_trace_dataset` is whatever you load from your Step 2 sampled failures.
# Each row needs `context` and `question` fields plus a `reference` answer.
failing_trace_dataset = [
    {"context": "...", "question": "...", "reference": "..."},
    # ...
]

# `mapper` feeds the generated answer and the reference to BLEUScore.
mapper = BasicDataMapper(
    key_map={"response": "generated_output", "expected_response": "reference"},
)

optimizer = BayesianSearchOptimizer(
    inference_model_name="gpt-6-luna",
    teacher_model_name="gpt-6.1-sol",
    n_trials=10,
)

result = optimizer.optimize(
    evaluator=OptEvaluator(BLEUScore()),
    data_mapper=mapper,
    dataset=failing_trace_dataset,
    initial_prompts=["Given context: {context}, answer: {question}"],
)

print("final_score:", result.final_score)
print("best_prompt:", result.best_generator.get_prompt_template())

The dataset is your collection of failing traces from production (sampled by Step 2’s eval scores). The evaluator can be a public metric like BLEUScore or any custom evaluator you’ve built. The optimizer searches prompt variants and returns the one that lifts your evaluator score the most. Promote that prompt through your CI gate, redeploy, watch the failure rate drop in Observe.

Production Hardening

Sample production traffic for eval. Running every production trace through evaluate() is too expensive, so choose a sampling rate, for example 5 to 10 percent, and tune it to your eval budget. The unified evaluate() call is synchronous; for async production scoring, run it in a background worker or job queue you control so the agent’s user-facing response path is never blocked on evaluation latency.

Roll trace cost up to per-user-request budgets. An Observe dashboard puts error rate, latency, cost and eval scores in one view. OTel GenAI semantic conventions (such as gen_ai.provider.name and gen_ai.request.model) give you the per-call inputs you need; rolling them up the trace tree is the platform’s job.

Use --otel_to_cloud if you’re already on Google Cloud Observability. ADK 1.17+ exports natively. traceAI sends agent, model and tool spans to a Future AGI Observe project, where eval tasks score them and Error Feed groups repeated failures into issues.

How to Know the Eval Is Working

After running through Steps 1 to 6, three checks confirm the pipeline is wired correctly:

1. Trace tree visible in Observe. Open the latest run. The agent invocation, every sub-agent, every LLM call, and every tool call should be separate spans under one root span.
2. Eval score on the span. Click into a span where you called evaluate(). You should see the eval name, score, reason, and latency as span attributes in Observe.
3. CI fails on a deliberately broken case. Add a .test.json case where the agent should fail (wrong city in the weather example, missing tool call). Run pytest. AgentEvaluator.evaluate() should fail. If it passes, your test_config.json thresholds are too loose.

If a check fails, start with the env vars and the register() settings from Step 1. As Step 1 notes, missing env vars are the most common reason a trace never shows up.

Pitfalls

Five mistakes to check first:

1. pip install futureagi followed by from fi.evals import Evaluator fails. The futureagi package excludes fi/evals. fi.evals ships in agent-learning-kit; install it in your eval environment.
2. runner.run_async(new_message="...") raises AttributeError: 'str' object has no attribute 'role'. ADK’s run_async expects new_message=types.Content(role="user", parts=[types.Part(text="...")]). Build a Content object; don’t pass a string.
3. runner.run_async raises Session not found. You skipped await runner.session_service.create_session(app_name=..., user_id=..., session_id=...). The session must exist before the first run_async call.
4. Eval scores don’t show up on spans. Your eval job’s register() lacks set_global_tracer_provider=True, or you called evaluate() outside an active span context. Wrap the call in with tracer.start_as_current_span(...).
5. AgentEvaluator.evaluate() passes everything in CI. Your test_config.json thresholds are too low or absent. Pick the tight defaults: tool_trajectory_avg_score: 1.0, response_match_score: 0.8, then raise response_match_score as your references stabilize.

Why Use Future AGI for Google ADK Evaluation and Observability

ADK’s built-in evaluation covers the development loop: .test.json and evalset files, pytest and adk eval, user simulation and adk optimize. It runs on test data you prepare, so live-traffic scoring and production triage belong to your observability layer.

Future AGI fills that layer. traceAI turns ADK invocations, agent runs, model calls and tool calls into OpenTelemetry spans, Observe eval tasks score a sample continuously, and Error Feed groups traces that failed the same way into one issue. agent-learning-kit brings evaluate(), the Simulate SDK and optimizers such as GEPA to CI.

For the broader 2026 production AI context that motivates the Production Eval Loop, see Generative AI Trends 2026: 8 Shifts Reshaping Builds & Buys. For the per-model comparison and benchmark trust framework, see LLM Benchmarks 2026: GPT-5, Claude Opus 4.7, Gemini 2.5 Pro, Grok 4 Compared. For recent model picks, see Best LLMs of September 2026.

Catching the Phantom Reservation Before Users Do

The Berlin booking that passed CI is the failure this loop exists for. With traceAI on the ADK service, the three identical searches sit in one trace, an Observe eval task scores it, and Error Feed files it with every run that failed the same way.

Ready to evaluate your first ADK agent? Get started with Future AGI and see the evaluation platform and tracing pages.

Sources

• Open In Colab
• evaluation framework
• ADK 1.17+ --otel_to_cloud flag and Vertex AI Agent Engine integration
• ADK evaluation criteria reference
• Google ADK Python source
• OpenTelemetry GenAI semantic conventions