feat: Adding implementation of evaluate method in LocalEvalService

Also, delete agent_creator.py file. We added this file by mistake.

PiperOrigin-RevId: 782193593

src/google/adk/evaluation/agent_creator.py

@@ -1,35 +0,0 @@
-# Copyright 2025 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.
-
-from __future__ import annotations
-
-from typing_extensions import override
-
-from ..agents import BaseAgent
-
-
-class IdentityAgentCreator:
-  """An implementation of the AgentCreator interface that always returns a copy of the root agent."""
-
-  def __init__(self, root_agent: BaseAgent):
-    self._root_agent = root_agent
-
-  @override
-  def get_agent(
-      self,
-  ) -> BaseAgent:
-    """Returns a deep copy of the root agent."""
-    # TODO: Use Agent.clone() when the PR is merged.
-    # return self._root_agent.model_copy(deep=True)
-    return self._root_agent.clone()

src/google/adk/evaluation/base_eval_service.py

@@ -42,6 +42,17 @@ class EvaluateConfig(BaseModel):
       description="""The list of metrics to be used in Eval.""",
   )
 
+  parallelism: int = Field(
+      default=4,
+      description="""Number of parallel evaluations to run during an Eval. Few
+factors to consider while changing this value:
+
+1) Your available quota with the model, especially for those metrics that use
+a model as a judge. Models tend to enforce per-minute or per-second SLAs. Using
+a larger value could result in the eval quickly consuming the quota.
+""",
+  )
+
 
 class InferenceConfig(BaseModel):
   """Contains configurations need to run inferences."""

src/google/adk/evaluation/local_eval_service.py

@@ -15,6 +15,7 @@
 from __future__ import annotations
 
 import asyncio
+import inspect
 import logging
 from typing import AsyncGenerator
 from typing import Callable
@@ -31,15 +32,22 @@ from ..sessions.base_session_service import BaseSessionService
 from ..sessions.in_memory_session_service import InMemorySessionService
 from ..utils.feature_decorator import working_in_progress
 from .base_eval_service import BaseEvalService
+from .base_eval_service import EvaluateConfig
 from .base_eval_service import EvaluateRequest
 from .base_eval_service import InferenceRequest
 from .base_eval_service import InferenceResult
 from .base_eval_service import InferenceStatus
+from .eval_case import Invocation
+from .eval_metrics import EvalMetric
+from .eval_metrics import EvalMetricResult
+from .eval_metrics import EvalMetricResultPerInvocation
 from .eval_result import EvalCaseResult
 from .eval_set import EvalCase
 from .eval_set_results_manager import EvalSetResultsManager
 from .eval_sets_manager import EvalSetsManager
 from .evaluation_generator import EvaluationGenerator
+from .evaluator import EvalStatus
+from .evaluator import EvaluationResult
 from .metric_evaluator_registry import DEFAULT_METRIC_EVALUATOR_REGISTRY
 from .metric_evaluator_registry import MetricEvaluatorRegistry
 
@@ -136,7 +144,188 @@ class LocalEvalService(BaseEvalService):
       evaluate_request: The request to perform metric evaluations on the
         inferences.
     """
-    raise NotImplementedError()
+    semaphore = asyncio.Semaphore(
+        value=evaluate_request.evaluate_config.parallelism
+    )
+
+    async def run_evaluation(inference_result):
+      async with semaphore:
+        return await self._evaluate_single_inference_result(
+            inference_result=inference_result,
+            evaluate_config=evaluate_request.evaluate_config,
+        )
+
+    evaluation_tasks = [
+        run_evaluation(inference_result)
+        for inference_result in evaluate_request.inference_results
+    ]
+    for evaluation_task in asyncio.as_completed(evaluation_tasks):
+      yield await evaluation_task
+
+  async def _evaluate_single_inference_result(
+      self, inference_result: InferenceResult, evaluate_config: EvaluateConfig
+  ) -> EvalCaseResult:
+    """Returns EvalCaseResult for the given inference result.
+
+    A single inference result can have multiple invocations. For each
+    invocaiton, this method evaluates the metrics present in evaluate config.
+
+    The EvalCaseResult contains scores for each metric per invocation and the
+    overall score.
+    """
+    eval_case = self._eval_sets_manager.get_eval_case(
+        app_name=inference_result.app_name,
+        eval_set_id=inference_result.eval_set_id,
+        eval_case_id=inference_result.eval_case_id,
+    )
+
+    if eval_case is None:
+      raise NotFoundError(
+          f'Eval case with id {inference_result.eval_case_id} not found for'
+          f' app {inference_result.app_name} and eval set'
+          f' {inference_result.eval_set_id}.'
+      )
+
+    # Metric results for each invocation
+    eval_metric_result_per_invocation = []
+
+    # We also keep track of the overall score for a metric, derived from all
+    # invocation. For example, if we were keeping track the metric that compares
+    # how well is the final resposne as compared to a golden answer, then each
+    # invocation will have the value of this metric. We will also have an
+    # overall score using aggregation strategy across all invocations. This
+    # would be the score for the eval case.
+    overall_eval_metric_results = []
+
+    if len(inference_result.inferences) != len(eval_case.conversation):
+      raise ValueError(
+          'Inferences should match conversations in eval case. Found'
+          f'{len(inference_result.inferences)} inferences '
+          f'{len(eval_case.conversation)} conversations in eval cases.'
+      )
+
+    # Pre-creating the EvalMetricResults entries for each invocation.
+    for actual, expected in zip(
+        inference_result.inferences, eval_case.conversation
+    ):
+      eval_metric_result_per_invocation.append(
+          EvalMetricResultPerInvocation(
+              actual_invocation=actual,
+              expected_invocation=expected,
+              # We will fill this as we evaluate each metric per invocation.
+              eval_metric_results=[],
+          )
+      )
+
+    for eval_metric in evaluate_config.eval_metrics:
+      # Perform evaluation of the metric.
+      evaluation_result = await self._evaluate_metric(
+          eval_metric=eval_metric,
+          actual_invocations=inference_result.inferences,
+          expected_invocations=eval_case.conversation,
+      )
+
+      # Track overall scrore across all invocations.
+      overall_eval_metric_results.append(
+          EvalMetricResult(
+              metric_name=eval_metric.metric_name,
+              threshold=eval_metric.threshold,
+              score=evaluation_result.overall_score,
+              eval_status=evaluation_result.overall_eval_status,
+          )
+      )
+
+      if len(evaluation_result.per_invocation_results) != len(
+          eval_metric_result_per_invocation
+      ):
+        raise ValueError(
+            'Eval metric should return results for each invocation. Found '
+            f'{len(evaluation_result.per_invocation_results)} results for '
+            f'{len(eval_metric_result_per_invocation)} invocations.'
+        )
+
+      # Track score across individual invocations.
+      for invocation_result, invocation in zip(
+          evaluation_result.per_invocation_results,
+          eval_metric_result_per_invocation,
+      ):
+        invocation.eval_metric_results.append(
+            EvalMetricResult(
+                metric_name=eval_metric.metric_name,
+                threshold=eval_metric.threshold,
+                score=invocation_result.score,
+                eval_status=invocation_result.eval_status,
+            )
+        )
+
+    final_eval_status = self._generate_final_eval_status(
+        overall_eval_metric_results
+    )
+    user_id = (
+        eval_case.session_input.user_id
+        if eval_case.session_input and eval_case.session_input.user_id
+        else 'test_user_id'
+    )
+
+    return EvalCaseResult(
+        eval_set_file=inference_result.eval_set_id,
+        eval_set_id=inference_result.eval_set_id,
+        eval_id=inference_result.eval_case_id,
+        final_eval_status=final_eval_status,
+        overall_eval_metric_results=overall_eval_metric_results,
+        eval_metric_result_per_invocation=eval_metric_result_per_invocation,
+        session_id=inference_result.session_id,
+        user_id=user_id,
+    )
+
+  async def _evaluate_metric(
+      self,
+      eval_metric: EvalMetric,
+      actual_invocations: list[Invocation],
+      expected_invocations: list[Invocation],
+  ) -> EvaluationResult:
+    """Returns EvaluationResult obtained from evaluating a metric using an Evaluator."""
+
+    # Get the metric evaluator from the registry.
+    metric_evaluator = self._metric_evaluator_registry.get_evaluator(
+        eval_metric=eval_metric
+    )
+
+    if inspect.iscoroutinefunction(metric_evaluator.evaluate_invocations):
+      # Some evaluators could be async, for example those that use llm as a
+      # judge, so we need to make sure that we wait on them.
+      return await metric_evaluator.evaluate_invocations(
+          actual_invocations=actual_invocations,
+          expected_invocations=expected_invocations,
+      )
+    else:
+      # Metrics that perform computation synchronously, mostly these don't
+      # perform any i/o. An example of this would calculation of rouge_1 score.
+      return metric_evaluator.evaluate_invocations(
+          actual_invocations=actual_invocations,
+          expected_invocations=expected_invocations,
+      )
+
+  def _generate_final_eval_status(
+      self, overall_eval_metric_results: list[EvalMetricResult]
+  ) -> EvalStatus:
+    final_eval_status = EvalStatus.NOT_EVALUATED
+    # Go over the all the eval statuses and mark the final eval status as
+    # passed if all of them pass, otherwise mark the final eval status to
+    # failed.
+    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:
+        final_eval_status = EvalStatus.FAILED
+        break
+      else:
+        raise ValueError(f'Unknown eval status: {overall_eval_status}.')
+
+    return final_eval_status
 
   async def _perform_inference_sigle_eval_item(
       self,

tests/unittests/evaluation/test_local_eval_service.py

@@ -16,13 +16,26 @@ from unittest import mock
 
 from google.adk.agents.llm_agent import LlmAgent
 from google.adk.errors.not_found_error import NotFoundError
+from google.adk.evaluation.base_eval_service import EvaluateConfig
+from google.adk.evaluation.base_eval_service import EvaluateRequest
 from google.adk.evaluation.base_eval_service import InferenceConfig
 from google.adk.evaluation.base_eval_service import InferenceRequest
+from google.adk.evaluation.base_eval_service import InferenceResult
+from google.adk.evaluation.eval_case import Invocation
+from google.adk.evaluation.eval_metrics import EvalMetric
+from google.adk.evaluation.eval_metrics import EvalMetricResult
+from google.adk.evaluation.eval_result import EvalCaseResult
 from google.adk.evaluation.eval_set import EvalCase
 from google.adk.evaluation.eval_set import EvalSet
 from google.adk.evaluation.eval_sets_manager import EvalSetsManager
+from google.adk.evaluation.evaluator import EvalStatus
+from google.adk.evaluation.evaluator import EvaluationResult
+from google.adk.evaluation.evaluator import Evaluator
+from google.adk.evaluation.evaluator import PerInvocationResult
 from google.adk.evaluation.local_eval_service import LocalEvalService
+from google.adk.evaluation.metric_evaluator_registry import DEFAULT_METRIC_EVALUATOR_REGISTRY
 from google.adk.models.registry import LLMRegistry
+from google.genai import types as genai_types
 import pytest
 
 
@@ -39,12 +52,42 @@ def dummy_agent():
 
 @pytest.fixture
 def eval_service(dummy_agent, mock_eval_sets_manager):
+  DEFAULT_METRIC_EVALUATOR_REGISTRY.register_evaluator(
+      metric_name="fake_metric", evaluator=FakeEvaluator
+  )
   return LocalEvalService(
       root_agent=dummy_agent,
       eval_sets_manager=mock_eval_sets_manager,
   )
 
 
+class FakeEvaluator(Evaluator):
+
+  def __init__(self, eval_metric: EvalMetric):
+    self._eval_metric = eval_metric
+
+  def evaluate_invocations(
+      self,
+      actual_invocations: list[Invocation],
+      expected_invocations: list[Invocation],
+  ):
+    per_invocation_results = []
+    for actual, expected in zip(actual_invocations, expected_invocations):
+      per_invocation_results.append(
+          PerInvocationResult(
+              actual_invocation=actual,
+              expected_invocation=expected,
+              score=0.9,
+              eval_status=EvalStatus.PASSED,
+          )
+      )
+    return EvaluationResult(
+        overall_score=0.9,
+        overall_eval_status=EvalStatus.PASSED,
+        per_invocation_results=per_invocation_results,
+    )
+
+
 @pytest.mark.asyncio
 async def test_perform_inference_success(
     eval_service, dummy_agent, mock_eval_sets_manager
@@ -142,3 +185,148 @@ async def test_perform_inference_eval_set_not_found(
   with pytest.raises(NotFoundError):
     async for _ in eval_service.perform_inference(inference_request):
       pass
+
+
+@pytest.mark.asyncio
+async def test_evaluate_success(eval_service, mock_eval_sets_manager):
+  inference_results = [
+      InferenceResult(
+          app_name="test_app",
+          eval_set_id="test_eval_set",
+          eval_case_id="case1",
+          inferences=[],
+          session_id="session1",
+      ),
+      InferenceResult(
+          app_name="test_app",
+          eval_set_id="test_eval_set",
+          eval_case_id="case2",
+          inferences=[],
+          session_id="session2",
+      ),
+  ]
+  eval_metric = EvalMetric(metric_name="fake_metric", threshold=0.5)
+  evaluate_request = EvaluateRequest(
+      inference_results=inference_results,
+      evaluate_config=EvaluateConfig(eval_metrics=[eval_metric], parallelism=2),
+  )
+
+  mock_eval_case = mock.MagicMock(spec=EvalCase)
+  mock_eval_case.conversation = []
+  mock_eval_case.session_input = None
+  mock_eval_sets_manager.get_eval_case.return_value = mock_eval_case
+
+  results = []
+  async for result in eval_service.evaluate(evaluate_request):
+    results.append(result)
+
+  assert len(results) == 2
+  assert isinstance(results[0], EvalCaseResult)
+  assert isinstance(results[1], EvalCaseResult)
+  assert mock_eval_sets_manager.get_eval_case.call_count == 2
+
+
+@pytest.mark.asyncio
+async def test_evaluate_eval_case_not_found(
+    eval_service, mock_eval_sets_manager
+):
+  inference_results = [
+      InferenceResult(
+          app_name="test_app",
+          eval_set_id="test_eval_set",
+          eval_case_id="case1",
+          inferences=[],
+          session_id="session1",
+      ),
+  ]
+  eval_metric = EvalMetric(metric_name="fake_metric", threshold=0.5)
+  evaluate_request = EvaluateRequest(
+      inference_results=inference_results,
+      evaluate_config=EvaluateConfig(eval_metrics=[eval_metric], parallelism=1),
+  )
+
+  mock_eval_sets_manager.get_eval_case.return_value = None
+
+  with pytest.raises(NotFoundError):
+    async for _ in eval_service.evaluate(evaluate_request):
+      pass
+
+  mock_eval_sets_manager.get_eval_case.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_evaluate_single_inference_result(
+    eval_service, mock_eval_sets_manager
+):
+  invocation = Invocation(
+      user_content=genai_types.Content(
+          parts=[genai_types.Part(text="test user content.")]
+      ),
+      final_response=genai_types.Content(
+          parts=[genai_types.Part(text="test final response.")]
+      ),
+  )
+  inference_result = InferenceResult(
+      app_name="test_app",
+      eval_set_id="test_eval_set",
+      eval_case_id="case1",
+      inferences=[
+          invocation.model_copy(deep=True),
+          invocation.model_copy(deep=True),
+          invocation.model_copy(deep=True),
+      ],
+      session_id="session1",
+  )
+  eval_metric = EvalMetric(metric_name="fake_metric", threshold=0.5)
+  evaluate_config = EvaluateConfig(eval_metrics=[eval_metric], parallelism=1)
+
+  mock_eval_case = mock.MagicMock(spec=EvalCase)
+  mock_eval_case.conversation = [
+      invocation.model_copy(deep=True),
+      invocation.model_copy(deep=True),
+      invocation.model_copy(deep=True),
+  ]
+  mock_eval_case.session_input = None
+  mock_eval_sets_manager.get_eval_case.return_value = mock_eval_case
+
+  result = await eval_service._evaluate_single_inference_result(
+      inference_result=inference_result, evaluate_config=evaluate_config
+  )
+
+  assert isinstance(result, EvalCaseResult)
+  assert result.eval_id == "case1"
+  assert result.session_id == "session1"
+  assert len(result.overall_eval_metric_results) == 1
+  assert result.overall_eval_metric_results[0].metric_name == "fake_metric"
+  assert result.overall_eval_metric_results[0].score == 0.9
+  mock_eval_sets_manager.get_eval_case.assert_called_once_with(
+      app_name="test_app", eval_set_id="test_eval_set", eval_case_id="case1"
+  )
+
+  assert len(result.eval_metric_result_per_invocation) == 3
+  for i in range(3):
+    invocation_result = result.eval_metric_result_per_invocation[i]
+    assert invocation_result.actual_invocation == inference_result.inferences[i]
+    assert (
+        invocation_result.expected_invocation == mock_eval_case.conversation[i]
+    )
+    assert len(invocation_result.eval_metric_results) == 1
+    metric_result = invocation_result.eval_metric_results[0]
+    assert metric_result.metric_name == "fake_metric"
+    assert metric_result.score == 0.9
+    assert metric_result.eval_status == EvalStatus.PASSED
+
+
+def test_generate_final_eval_status_doesn_t_throw_on(eval_service):
+  # How to fix if this test case fails?
+  # This test case has failed mainly because a new EvalStatus got added. You
+  # mostly need to update _generate_final_eval_status method to handle the new
+  # eval case.
+
+  # We go over all the possible values of EvalStatus one by one and expect
+  # the _generate_final_eval_status to handle it without throwing an exeception.
+  for status in EvalStatus:
+    eval_metric_result = EvalMetricResult(
+        metric_name="metric1", threshold=0.5, eval_status=status
+    )
+    eval_service._generate_final_eval_status([eval_metric_result])