From 33eec34577dc41f98deba09712e399522c99af7d Mon Sep 17 00:00:00 2001 From: Ankur Sharma Date: Fri, 11 Jul 2025 19:14:57 -0700 Subject: [PATCH] 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 | 35 ---- .../adk/evaluation/base_eval_service.py | 11 + .../adk/evaluation/local_eval_service.py | 191 +++++++++++++++++- .../evaluation/test_local_eval_service.py | 188 +++++++++++++++++ 4 files changed, 389 insertions(+), 36 deletions(-) delete mode 100644 src/google/adk/evaluation/agent_creator.py diff --git a/src/google/adk/evaluation/agent_creator.py b/src/google/adk/evaluation/agent_creator.py deleted file mode 100644 index 7299675c..00000000 --- a/src/google/adk/evaluation/agent_creator.py +++ /dev/null @@ -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() diff --git a/src/google/adk/evaluation/base_eval_service.py b/src/google/adk/evaluation/base_eval_service.py index bfbb0f3c..3d576ab2 100644 --- a/src/google/adk/evaluation/base_eval_service.py +++ b/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.""" diff --git a/src/google/adk/evaluation/local_eval_service.py b/src/google/adk/evaluation/local_eval_service.py index 14214ac6..bb83b72f 100644 --- a/src/google/adk/evaluation/local_eval_service.py +++ b/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, diff --git a/tests/unittests/evaluation/test_local_eval_service.py b/tests/unittests/evaluation/test_local_eval_service.py index ca19a1bb..60702fc5 100644 --- a/tests/unittests/evaluation/test_local_eval_service.py +++ b/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])