mirror of
https://github.com/encounter/adk-python.git
synced 2026-07-09 18:19:28 -07:00
Ankur SharmaandCopybara-Service
bab3be2cf3
If the EvalRunResultsManager is provided to LocalEvalService, then we want to persist the eval run results using it. PiperOrigin-RevId: 782196848
351 lines
12 KiB
Python
351 lines
12 KiB
Python
# 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 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_set_results_manager import EvalSetResultsManager
|
|
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
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_eval_sets_manager():
|
|
return mock.create_autospec(EvalSetsManager)
|
|
|
|
|
|
@pytest.fixture
|
|
def dummy_agent():
|
|
llm = LLMRegistry.new_llm("gemini-pro")
|
|
return LlmAgent(name="test_agent", model=llm)
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_eval_set_results_manager():
|
|
return mock.create_autospec(EvalSetResultsManager)
|
|
|
|
|
|
@pytest.fixture
|
|
def eval_service(
|
|
dummy_agent, mock_eval_sets_manager, mock_eval_set_results_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,
|
|
eval_set_results_manager=mock_eval_set_results_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,
|
|
):
|
|
eval_set = EvalSet(
|
|
eval_set_id="test_eval_set",
|
|
eval_cases=[
|
|
EvalCase(eval_id="case1", conversation=[], session_input=None),
|
|
EvalCase(eval_id="case2", conversation=[], session_input=None),
|
|
],
|
|
)
|
|
mock_eval_sets_manager.get_eval_set.return_value = eval_set
|
|
|
|
mock_inference_result = mock.MagicMock()
|
|
eval_service._perform_inference_sigle_eval_item = mock.AsyncMock(
|
|
return_value=mock_inference_result
|
|
)
|
|
|
|
inference_request = InferenceRequest(
|
|
app_name="test_app",
|
|
eval_set_id="test_eval_set",
|
|
inference_config=InferenceConfig(parallelism=2),
|
|
)
|
|
|
|
results = []
|
|
async for result in eval_service.perform_inference(inference_request):
|
|
results.append(result)
|
|
|
|
assert len(results) == 2
|
|
assert results[0] == mock_inference_result
|
|
assert results[1] == mock_inference_result
|
|
mock_eval_sets_manager.get_eval_set.assert_called_once_with(
|
|
app_name="test_app", eval_set_id="test_eval_set"
|
|
)
|
|
assert eval_service._perform_inference_sigle_eval_item.call_count == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_perform_inference_with_case_ids(
|
|
eval_service,
|
|
dummy_agent,
|
|
mock_eval_sets_manager,
|
|
):
|
|
eval_set = EvalSet(
|
|
eval_set_id="test_eval_set",
|
|
eval_cases=[
|
|
EvalCase(eval_id="case1", conversation=[], session_input=None),
|
|
EvalCase(eval_id="case2", conversation=[], session_input=None),
|
|
EvalCase(eval_id="case3", conversation=[], session_input=None),
|
|
],
|
|
)
|
|
mock_eval_sets_manager.get_eval_set.return_value = eval_set
|
|
|
|
mock_inference_result = mock.MagicMock()
|
|
eval_service._perform_inference_sigle_eval_item = mock.AsyncMock(
|
|
return_value=mock_inference_result
|
|
)
|
|
|
|
inference_request = InferenceRequest(
|
|
app_name="test_app",
|
|
eval_set_id="test_eval_set",
|
|
eval_case_ids=["case1", "case3"],
|
|
inference_config=InferenceConfig(parallelism=1),
|
|
)
|
|
|
|
results = []
|
|
async for result in eval_service.perform_inference(inference_request):
|
|
results.append(result)
|
|
|
|
assert len(results) == 2
|
|
eval_service._perform_inference_sigle_eval_item.assert_any_call(
|
|
app_name="test_app",
|
|
eval_set_id="test_eval_set",
|
|
eval_case=eval_set.eval_cases[0],
|
|
root_agent=dummy_agent,
|
|
)
|
|
eval_service._perform_inference_sigle_eval_item.assert_any_call(
|
|
app_name="test_app",
|
|
eval_set_id="test_eval_set",
|
|
eval_case=eval_set.eval_cases[2],
|
|
root_agent=dummy_agent,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_perform_inference_eval_set_not_found(
|
|
eval_service,
|
|
mock_eval_sets_manager,
|
|
):
|
|
mock_eval_sets_manager.get_eval_set.return_value = None
|
|
|
|
inference_request = InferenceRequest(
|
|
app_name="test_app",
|
|
eval_set_id="not_found_set",
|
|
inference_config=InferenceConfig(parallelism=1),
|
|
)
|
|
|
|
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, mock_eval_set_results_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
|
|
assert mock_eval_set_results_manager.save_eval_set_result.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, mock_eval_set_results_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])
|