feat(cli): add adk eval --num_runs N with per-eval-case aggregation

src/google/adk/cli/cli_tools_click.py

@@ -1142,6 +1142,13 @@ def _resolve_eval_config_file_path(
 )
 @click.argument("eval_set_file_path_or_id", nargs=-1)
 @click.option("--config_file_path", help="Optional. The path to config file.")
+@click.option(
+    "--num_runs",
+    type=click.IntRange(min=1),
+    default=1,
+    show_default=True,
+    help="Optional. Number of times each eval set should be run.",
+)
 @click.option(
     "--print_detailed_results",
     is_flag=True,
@@ -1157,6 +1164,7 @@ def cli_eval(
     print_detailed_results: bool,
     eval_storage_uri: str | None = None,
     log_level: str = "INFO",
+    num_runs: int = 1,
 ):
   """Evaluates an agent given the eval sets.
 
@@ -1210,6 +1218,8 @@ def cli_eval(
 
   CONFIG_FILE_PATH: The path to config file.
 
+  NUM_RUNS: Number of times each eval set should be run.
+
   PRINT_DETAILED_RESULTS: Prints detailed results on the console.
   """
   envs.load_dotenv_for_agent(agent_module_file_path, ".")
@@ -1218,6 +1228,8 @@ def cli_eval(
   try:
     import importlib  # noqa: F401
 
+    from ..evaluation._eval_case_result_aggregator import aggregate_eval_case_results
+    from ..evaluation.base_eval_service import EvaluateConfig
     from ..evaluation.base_eval_service import InferenceConfig
     from ..evaluation.base_eval_service import InferenceRequest
     from ..evaluation.eval_config import get_eval_metrics_from_config
@@ -1273,9 +1285,10 @@ def cli_eval(
     inference_config = InferenceConfig(
         use_live=True,
         live_timeout_seconds=eval_config.live_model_config.timeout_seconds,
+        num_runs=num_runs,
     )
   else:
-    inference_config = InferenceConfig(use_live=False)
+    inference_config = InferenceConfig(use_live=False, num_runs=num_runs)
 
   # Check if the first entry is a file that exists, if it does then we assume
   # rest of the entries are also files. We enforce this assumption in the if
@@ -1350,18 +1363,26 @@ def cli_eval(
         app=app,
     )
 
+    # `num_runs` is carried on the InferenceConfig, so the eval service repeats
+    # each eval case through its own parallelism pool.
     inference_results = asyncio.run(
         _collect_inferences(
-            inference_requests=inference_requests, eval_service=eval_service
+            inference_requests=inference_requests,
+            eval_service=eval_service,
         )
     )
+    evaluate_config = EvaluateConfig(eval_metrics=eval_metrics)
     eval_results = asyncio.run(
         _collect_eval_results(
             inference_results=inference_results,
             eval_service=eval_service,
             eval_metrics=eval_metrics,
         )
     )
+    aggregate_eval_results = aggregate_eval_case_results(
+        eval_results,
+        aggregation_strategy=evaluate_config.aggregation_strategy,
+    )
   except ModuleNotFoundError as mnf:
     raise click.ClickException(_missing_eval_dependencies_message()) from mnf
 
@@ -1370,7 +1391,7 @@ def cli_eval(
   )
   eval_run_summary = {}
 
-  for eval_result in eval_results:
+  for eval_result in aggregate_eval_results:
     if eval_result.eval_set_id not in eval_run_summary:
       eval_run_summary[eval_result.eval_set_id] = [0, 0]
 

src/google/adk/evaluation/_eval_case_result_aggregator.py

@@ -0,0 +1,157 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Aggregation of per-run EvalCaseResults into a single result per eval case."""
+
+from __future__ import annotations
+
+import statistics
+
+from .base_eval_service import AggregationStrategy
+from .eval_metrics import _get_metric_threshold
+from .eval_metrics import EvalMetricResult
+from .eval_result import EvalCaseResult
+from .evaluator import EvalStatus
+
+
+def _generate_final_eval_status(
+    overall_eval_metric_results: list[EvalMetricResult],
+) -> EvalStatus:
+  """Returns the final eval status for a case from its overall metric results."""
+  final_eval_status = EvalStatus.NOT_EVALUATED
+  for overall_eval_metric_result in overall_eval_metric_results:
+    overall_eval_status = overall_eval_metric_result.eval_status
+    if overall_eval_status == EvalStatus.PASSED:
+      final_eval_status = EvalStatus.PASSED
+    elif overall_eval_status == EvalStatus.NOT_EVALUATED:
+      continue
+    elif overall_eval_status == EvalStatus.FAILED:
+      return EvalStatus.FAILED
+    else:
+      raise ValueError(f"Unknown eval status: {overall_eval_status}.")
+  return final_eval_status
+
+
+def _has_hard_failed_run(per_case_results: list[EvalCaseResult]) -> bool:
+  """Returns True if any run failed without producing metric results.
+
+  A run whose inferencing raised is recorded as FAILED with no per-invocation
+  metric results. Such a run contributes no scores to the mean, so it must be
+  honored explicitly or a crashed run would be silently dropped from the
+  aggregate verdict. Mirrors
+  `AgentEvaluator._get_failures_from_final_eval_status`.
+  """
+  for result in per_case_results:
+    if result.final_eval_status != EvalStatus.FAILED:
+      continue
+    if not any(
+        invocation.eval_metric_results
+        for invocation in result.eval_metric_result_per_invocation
+    ):
+      return True
+  return False
+
+
+def _mean_over_invocations(
+    per_case_results: list[EvalCaseResult],
+) -> list[EvalMetricResult]:
+  """Returns overall metric results pooled over every invocation of every run.
+
+  Every per-invocation metric result across all runs is pooled by metric name
+  and its scores are averaged into a single overall score for that metric. This
+  matches `AgentEvaluator`, whereas averaging each run's already-overall score
+  (mean-of-means) would weight runs equally regardless of invocation count.
+  """
+  # Pool per-invocation metric results across all runs, keyed by metric name,
+  # preserving first-seen order (which is metric evaluation order).
+  results_by_metric: dict[str, list[EvalMetricResult]] = {}
+  for result in per_case_results:
+    for invocation in result.eval_metric_result_per_invocation:
+      for metric_result in invocation.eval_metric_results:
+        results_by_metric.setdefault(metric_result.metric_name, []).append(
+            metric_result
+        )
+
+  overall_eval_metric_results: list[EvalMetricResult] = []
+  for metric_results in results_by_metric.values():
+    aggregate_metric_result = metric_results[0].model_copy(deep=True)
+    scores = [m.score for m in metric_results if m.score is not None]
+    if scores:
+      aggregate_metric_result.score = statistics.mean(scores)
+      aggregate_metric_result.eval_status = (
+          EvalStatus.PASSED
+          if aggregate_metric_result.score
+          >= _get_metric_threshold(aggregate_metric_result)
+          else EvalStatus.FAILED
+      )
+    else:
+      aggregate_metric_result.score = None
+      aggregate_metric_result.eval_status = EvalStatus.NOT_EVALUATED
+    overall_eval_metric_results.append(aggregate_metric_result)
+
+  return overall_eval_metric_results
+
+
+def aggregate_eval_case_results(
+    eval_case_results: list[EvalCaseResult],
+    aggregation_strategy: AggregationStrategy = (
+        AggregationStrategy.MEAN_OVER_INVOCATIONS
+    ),
+) -> list[EvalCaseResult]:
+  """Aggregates per-run EvalCaseResults into one result per eval case.
+
+  Results are grouped by (eval_set_id, eval_id); each group holds the runs for a
+  single eval case. The returned list has one EvalCaseResult per case, sorted by
+  (eval_set_id, eval_id). A case run once is returned unchanged apart from
+  grouping.
+
+  Args:
+    eval_case_results: The per-run results to aggregate.
+    aggregation_strategy: How to combine per-run results. Only
+      `MEAN_OVER_INVOCATIONS` is currently supported.
+  """
+  if aggregation_strategy != AggregationStrategy.MEAN_OVER_INVOCATIONS:
+    raise ValueError(
+        f"Unsupported aggregation strategy: {aggregation_strategy}."
+    )
+
+  results_by_case: dict[tuple[str, str], list[EvalCaseResult]] = {}
+  for result in eval_case_results:
+    key = (result.eval_set_id, result.eval_id)
+    results_by_case.setdefault(key, []).append(result)
+
+  aggregate_results: list[EvalCaseResult] = []
+  for per_case_results in results_by_case.values():
+    overall_eval_metric_results = _mean_over_invocations(per_case_results)
+    final_eval_status = _generate_final_eval_status(overall_eval_metric_results)
+
+    # A run that crashed before producing any metric results contributes no
+    # scores to the mean above, so honor its failure explicitly.
+    if final_eval_status != EvalStatus.FAILED and _has_hard_failed_run(
+        per_case_results
+    ):
+      final_eval_status = EvalStatus.FAILED
+
+    aggregate_result = per_case_results[0].model_copy(deep=True)
+    aggregate_result.overall_eval_metric_results = overall_eval_metric_results
+    aggregate_result.final_eval_status = final_eval_status
+    # Retain every run's invocations so detailed inspection still has them.
+    aggregate_result.eval_metric_result_per_invocation = [
+        invocation
+        for result in per_case_results
+        for invocation in result.eval_metric_result_per_invocation
+    ]
+    aggregate_results.append(aggregate_result)
+
+  return sorted(aggregate_results, key=lambda x: (x.eval_set_id, x.eval_id))

src/google/adk/evaluation/base_eval_service.py

@@ -31,6 +31,17 @@
 from .eval_result import EvalCaseResult
 
 
+class AggregationStrategy(Enum):
+  """Strategy for aggregating EvalCaseResults of the same case across runs."""
+
+  MEAN_OVER_INVOCATIONS = "mean_over_invocations"
+  """Pools per-invocation metric scores across all runs and takes their mean.
+
+  This matches the aggregation performed by `AgentEvaluator` (the pytest eval
+  entrypoint), keeping multi-run summaries consistent across entrypoints.
+  """
+
+
 class EvaluateConfig(BaseModel):
   """Contains configurations needed to run evaluations."""
 
@@ -54,6 +65,12 @@ class EvaluateConfig(BaseModel):
 """,
   )
 
+  aggregation_strategy: AggregationStrategy = Field(
+      default=AggregationStrategy.MEAN_OVER_INVOCATIONS,
+      description="""Strategy used to aggregate the per-run EvalCaseResults of an
+eval case when it is run more than once (see `InferenceConfig.num_runs`).""",
+  )
+
 
 class InferenceConfig(BaseModel):
   """Contains configurations need to run inferences."""
@@ -69,6 +86,14 @@ class InferenceConfig(BaseModel):
 charges.""",
   )
 
+  num_runs: int = Field(
+      default=1,
+      ge=1,
+      description="""Number of times each eval case should be run. Values greater
+than 1 reduce nondeterminism: the eval case is inferenced `num_runs` times
+(through the same parallelism pool) and the per-run results are aggregated.""",
+  )
+
   parallelism: int = Field(
       default=4,
       description="""Number of parallel inferences to run during an Eval. Few

src/google/adk/evaluation/local_eval_service.py

@@ -198,7 +198,15 @@ async def run_inference(eval_case: EvalCase) -> InferenceResult:
             live_timeout_seconds=inference_request.inference_config.live_timeout_seconds,
         )
 
-    inference_results = [run_inference(eval_case) for eval_case in eval_cases]
+    # Each eval case is inferenced `num_runs` times. Running the repeats here
+    # (rather than in the caller) lets the parallelism semaphore above cover the
+    # repeated runs as well.
+    num_runs = inference_request.inference_config.num_runs
+    inference_results = [
+        run_inference(eval_case)
+        for eval_case in eval_cases
+        for _ in range(num_runs)
+    ]
     for inference_result in asyncio.as_completed(inference_results):
       yield await inference_result