encounter/adk-python
0
0
Fork 0
mirror of https://github.com/encounter/adk-python.git synced 2026-07-09 18:19:28 -07:00
Code Issues Packages Projects Releases Wiki Activity

feat: Refactored AgentEvaluator and updated it to use LocalEvalService

Browse Source 
With this change we ensure that all three eval entry points, web, cli and pytest use the common LocalEvalService.

Updates to web and cli happened in a previous change.

PiperOrigin-RevId: 786445632
This commit is contained in:
Ankur Sharma
2025-07-23 15:26:41 -07:00
committed by Copybara-Service
parent 927c75f0ee
commit 1355bd643b
2 changed files with 265 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/google/adk/evaluation/agent_evaluator.py
+264 -80
View File
@@ -14,10 +14,12 @@
from __future__ import annotations
import importlib
import json
import logging
import os
from os import path
import statistics
from typing import Any
from typing import Dict
from typing import List
@@ -26,15 +28,21 @@ from typing import Union
import uuid
from google.genai import types as genai_types
from pydantic import BaseModel
from pydantic import ValidationError
from ..agents.base_agent import BaseAgent
from .constants import MISSING_EVAL_DEPENDENCIES_MESSAGE
from .eval_case import IntermediateData
from .eval_case import Invocation
from .eval_metrics import EvalMetric
from .eval_metrics import EvalMetricResult
from .eval_metrics import PrebuiltMetrics
from .eval_result import EvalCaseResult
from .eval_set import EvalSet
from .eval_sets_manager import EvalSetsManager
from .evaluator import EvalStatus
from .evaluator import EvaluationResult
from .evaluator import Evaluator
from .in_memory_eval_sets_manager import InMemoryEvalSetsManager
from .local_eval_sets_manager import convert_eval_set_to_pydanctic_schema
logger = logging.getLogger("google_adk." + __name__)
@@ -42,12 +50,13 @@ logger = logging.getLogger("google_adk." + __name__)
# Constants for default runs and evaluation criteria
NUM_RUNS = 2
TOOL_TRAJECTORY_SCORE_KEY = "tool_trajectory_avg_score"
TOOL_TRAJECTORY_SCORE_KEY = PrebuiltMetrics.TOOL_TRAJECTORY_AVG_SCORE.value
# This evaluation is not very stable.
# This is always optional unless explicitly specified.
RESPONSE_EVALUATION_SCORE_KEY = "response_evaluation_score"
RESPONSE_MATCH_SCORE_KEY = "response_match_score"
SAFETY_V1_KEY = "safety_v1"
RESPONSE_EVALUATION_SCORE_KEY = PrebuiltMetrics.RESPONSE_EVALUATION_SCORE.value
RESPONSE_MATCH_SCORE_KEY = PrebuiltMetrics.RESPONSE_MATCH_SCORE.value
SAFETY_V1_KEY = PrebuiltMetrics.SAFETY_V1.value
ALLOWED_CRITERIA = [
TOOL_TRAJECTORY_SCORE_KEY,
@@ -56,7 +65,6 @@ ALLOWED_CRITERIA = [
SAFETY_V1_KEY,
]
QUERY_COLUMN = "query"
REFERENCE_COLUMN = "reference"
EXPECTED_TOOL_USE_COLUMN = "expected_tool_use"
@@ -73,6 +81,18 @@ def load_json(file_path: str) -> Union[Dict, List]:
return json.load(f)
class _EvalMetricResultWithInvocation(BaseModel):
"""EvalMetricResult along with both actual and expected invocation.
This is class is intentionally marked as private and is created for
convenience.
"""
actual_invocation: Invocation
expected_invocation: Invocation
eval_metric_result: EvalMetricResult
class AgentEvaluator:
"""An evaluator for Agents, mainly intended for helping with test cases."""
@@ -99,8 +119,8 @@ class AgentEvaluator:
agent_module: str,
eval_set: EvalSet,
criteria: dict[str, float],
num_runs=NUM_RUNS,
agent_name=None,
num_runs: int = NUM_RUNS,
agent_name: Optional[str] = None,
print_detailed_results: bool = True,
):
"""Evaluates an agent using the given EvalSet.
@@ -114,58 +134,45 @@ class AgentEvaluator:
respective thresholds.
num_runs: Number of times all entries in the eval dataset should be
assessed.
agent_name: The name of the agent.
agent_name: The name of the agent, if trying to evaluate something other
than root agent. If left empty or none, then root agent is evaluated.
print_detailed_results: Whether to print detailed results for each metric
evaluation.
"""
try:
from .evaluation_generator import EvaluationGenerator
except ModuleNotFoundError as e:
raise ModuleNotFoundError(MISSING_EVAL_DEPENDENCIES_MESSAGE) from e
eval_case_responses_list = await EvaluationGenerator.generate_responses(
agent_for_eval = AgentEvaluator._get_agent_for_eval(
module_name=agent_module, agent_name=agent_name
)
eval_metrics = [
EvalMetric(metric_name=n, threshold=t) for n, t in criteria.items()
]
# Step 1: Perform evals, basically inferencing and evaluation of metrics
eval_results_by_eval_id = await AgentEvaluator._get_eval_results_by_eval_id(
agent_for_eval=agent_for_eval,
eval_set=eval_set,
agent_module_path=agent_module,
repeat_num=num_runs,
agent_name=agent_name,
eval_metrics=eval_metrics,
num_runs=num_runs,
)
failures = []
# Step 2: Post-process the results!
for eval_case_responses in eval_case_responses_list:
actual_invocations = [
invocation
for invocations in eval_case_responses.responses
for invocation in invocations
]
expected_invocations = (
eval_case_responses.eval_case.conversation * num_runs
# We keep track of eval case failures, these are not infra failures but eval
# test failures. We track them and then report them towards the end.
failures: list[str] = []
for _, eval_results_per_eval_id in eval_results_by_eval_id.items():
eval_metric_results = (
AgentEvaluator._get_eval_metric_results_with_invocation(
eval_results_per_eval_id
)
)
failures_per_eval_case = AgentEvaluator._process_metrics_and_get_failures(
eval_metric_results=eval_metric_results,
print_detailed_results=print_detailed_results,
agent_module=agent_name,
)
for metric_name, threshold in criteria.items():
metric_evaluator = AgentEvaluator._get_metric_evaluator(
metric_name=metric_name, threshold=threshold
)
evaluation_result: EvaluationResult = (
metric_evaluator.evaluate_invocations(
actual_invocations=actual_invocations,
expected_invocations=expected_invocations,
)
)
if print_detailed_results:
AgentEvaluator._print_details(
evaluation_result=evaluation_result,
metric_name=metric_name,
threshold=threshold,
)
# Gather all the failures.
if evaluation_result.overall_eval_status != EvalStatus.PASSED:
failures.append(
f"{metric_name} for {agent_module} Failed. Expected {threshold},"
f" but got {evaluation_result.overall_score}."
)
failures.extend(failures_per_eval_case)
assert not failures, (
"Following are all the test failures. If you looking to get more"
@@ -386,31 +393,15 @@ class AgentEvaluator:
f" {sample}."
)
@staticmethod
def _get_metric_evaluator(metric_name: str, threshold: float) -> Evaluator:
try:
from .response_evaluator import ResponseEvaluator
from .safety_evaluator import SafetyEvaluatorV1
from .trajectory_evaluator import TrajectoryEvaluator
except ModuleNotFoundError as e:
raise ModuleNotFoundError(MISSING_EVAL_DEPENDENCIES_MESSAGE) from e
if metric_name == TOOL_TRAJECTORY_SCORE_KEY:
return TrajectoryEvaluator(threshold=threshold)
elif (
metric_name == RESPONSE_MATCH_SCORE_KEY
or metric_name == RESPONSE_EVALUATION_SCORE_KEY
):
return ResponseEvaluator(threshold=threshold, metric_name=metric_name)
elif metric_name == SAFETY_V1_KEY:
return SafetyEvaluatorV1(
eval_metric=EvalMetric(threshold=threshold, metric_name=metric_name)
)
raise ValueError(f"Unsupported eval metric: {metric_name}")
@staticmethod
def _print_details(
evaluation_result: EvaluationResult, metric_name: str, threshold: float
eval_metric_result_with_invocations: list[
_EvalMetricResultWithInvocation
],
overall_eval_status: EvalStatus,
overall_score: Optional[float],
metric_name: str,
threshold: float,
):
try:
from pandas import pandas as pd
@@ -418,16 +409,16 @@ class AgentEvaluator:
except ModuleNotFoundError as e:
raise ModuleNotFoundError(MISSING_EVAL_DEPENDENCIES_MESSAGE) from e
print(
f"Summary: `{evaluation_result.overall_eval_status}` for Metric:"
f"Summary: `{overall_eval_status}` for Metric:"
f" `{metric_name}`. Expected threshold: `{threshold}`, actual value:"
f" `{evaluation_result.overall_score}`."
f" `{overall_score}`."
)
data = []
for per_invocation_result in evaluation_result.per_invocation_results:
for per_invocation_result in eval_metric_result_with_invocations:
data.append({
"eval_status": per_invocation_result.eval_status,
"score": per_invocation_result.score,
"eval_status": per_invocation_result.eval_metric_result.eval_status,
"score": per_invocation_result.eval_metric_result.score,
"threshold": threshold,
"prompt": AgentEvaluator._convert_content_to_text(
per_invocation_result.expected_invocation.user_content
@@ -464,3 +455,196 @@ class AgentEvaluator:
return "\n".join([str(t) for t in intermediate_data.tool_uses])
return ""
@staticmethod
def _get_agent_for_eval(
module_name: str, agent_name: Optional[str] = None
) -> BaseAgent:
module_path = f"{module_name}"
agent_module = importlib.import_module(module_path)
root_agent = agent_module.agent.root_agent
agent_for_eval = root_agent
if agent_name:
agent_for_eval = root_agent.find_agent(agent_name)
assert agent_for_eval, f"Sub-Agent `{agent_name}` not found."
return agent_for_eval
@staticmethod
def _get_eval_sets_manager(
app_name: str, eval_set: EvalSet
) -> EvalSetsManager:
eval_sets_manager = InMemoryEvalSetsManager()
eval_sets_manager.create_eval_set(
app_name=app_name, eval_set_id=eval_set.eval_set_id
)
for eval_case in eval_set.eval_cases:
eval_sets_manager.add_eval_case(
app_name=app_name,
eval_set_id=eval_set.eval_set_id,
eval_case=eval_case,
)
return eval_sets_manager
@staticmethod
async def _get_eval_results_by_eval_id(
agent_for_eval: BaseAgent,
eval_set: EvalSet,
eval_metrics: list[EvalMetric],
num_runs: int,
) -> dict[str, list[EvalCaseResult]]:
"""Returns EvalCaseResults grouped by eval case id.
The grouping happens because of the "num_runs" argument, where for any value
greater than 1, we would have generated inferences num_runs times and so
by extension we would have evaluated metrics on each of those inferences.
"""
try:
from .base_eval_service import EvaluateConfig
from .base_eval_service import EvaluateRequest
from .base_eval_service import InferenceConfig
from .base_eval_service import InferenceRequest
from .local_eval_service import LocalEvalService
except ModuleNotFoundError as e:
raise ModuleNotFoundError(MISSING_EVAL_DEPENDENCIES_MESSAGE) from e
# It is okay to pick up this dummy name.
app_name = "test_app"
eval_service = LocalEvalService(
root_agent=agent_for_eval,
eval_sets_manager=AgentEvaluator._get_eval_sets_manager(
app_name=app_name, eval_set=eval_set
),
)
inference_requests = [
InferenceRequest(
app_name=app_name,
eval_set_id=eval_set.eval_set_id,
inference_config=InferenceConfig(),
)
] * num_runs # Repeat inference request num_runs times.
# Generate inferences
inference_results = []
for inference_request in inference_requests:
async for inference_result in eval_service.perform_inference(
inference_request=inference_request
):
inference_results.append(inference_result)
# Evaluate metrics
# As we perform more than one run for an eval case, we collect eval results
# by eval id.
eval_results_by_eval_id: dict[str, list[EvalCaseResult]] = {}
evaluate_request = EvaluateRequest(
inference_results=inference_results,
evaluate_config=EvaluateConfig(eval_metrics=eval_metrics),
)
async for eval_result in eval_service.evaluate(
evaluate_request=evaluate_request
):
eval_id = eval_result.eval_id
if eval_id not in eval_results_by_eval_id:
eval_results_by_eval_id[eval_id] = []
eval_results_by_eval_id[eval_id].append(eval_result)
return eval_results_by_eval_id
@staticmethod
def _get_eval_metric_results_with_invocation(
eval_results_per_eval_id: list[EvalCaseResult],
) -> dict[str, list[_EvalMetricResultWithInvocation]]:
"""Retruns _EvalMetricResultWithInvocation grouped by metric.
EvalCaseResult contain results for each metric per invocation.
This method flips it around and returns a structure that groups metric
results per invocation by eval metric.
This is a convenience function.
"""
eval_metric_results: dict[str, list[_EvalMetricResultWithInvocation]] = {}
# Go over the EvalCaseResult one by one, do note that at this stage all
# EvalCaseResult belong to the same eval id.
for eval_case_result in eval_results_per_eval_id:
# For the given eval_case_result, we go over metric results for each
# invocation. Do note that a single eval case can have more than one
# invocation and for each invocation there could be more than on eval
# metrics that were evaluated.
for (
eval_metrics_per_invocation
) in eval_case_result.eval_metric_result_per_invocation:
# Go over each eval_metric_result for an invocation.
for (
eval_metric_result
) in eval_metrics_per_invocation.eval_metric_results:
metric_name = eval_metric_result.metric_name
if metric_name not in eval_metric_results:
eval_metric_results[metric_name] = []
actual_invocation = eval_metrics_per_invocation.actual_invocation
expected_invocation = eval_metrics_per_invocation.expected_invocation
eval_metric_results[metric_name].append(
_EvalMetricResultWithInvocation(
actual_invocation=actual_invocation,
expected_invocation=expected_invocation,
eval_metric_result=eval_metric_result,
)
)
return eval_metric_results
@staticmethod
def _process_metrics_and_get_failures(
eval_metric_results: dict[str, list[_EvalMetricResultWithInvocation]],
print_detailed_results: bool,
agent_module: str,
) -> list[str]:
"""Returns a list of failures based on the score for each invocation."""
failures: list[str] = []
for (
metric_name,
eval_metric_results_with_invocations,
) in eval_metric_results.items():
threshold = eval_metric_results_with_invocations[
0
].eval_metric_result.threshold
scores = [
m.eval_metric_result.score
for m in eval_metric_results_with_invocations
if m.eval_metric_result.score
]
if scores:
overall_score = statistics.mean(scores)
overall_eval_status = (
EvalStatus.PASSED
if overall_score >= threshold
else EvalStatus.FAILED
)
else:
overall_score = None
overall_eval_status = EvalStatus.NOT_EVALUATED
# Gather all the failures.
if overall_eval_status != EvalStatus.PASSED:
if print_detailed_results:
AgentEvaluator._print_details(
eval_metric_result_with_invocations=eval_metric_results_with_invocations,
overall_eval_status=overall_eval_status,
overall_score=overall_score,
metric_name=metric_name,
threshold=threshold,
)
failures.append(
f"{metric_name} for {agent_module} Failed. Expected {threshold},"
f" but got {overall_score}."
)
return failures
src/google/adk/evaluation/local_eval_service.py
+1 -3
View File
@@ -114,8 +114,6 @@ class LocalEvalService(BaseEvalService):
if eval_case.eval_id in inference_request.eval_case_ids
]
root_agent = self._root_agent.clone()
semaphore = asyncio.Semaphore(
value=inference_request.inference_config.parallelism
)
@@ -126,7 +124,7 @@ class LocalEvalService(BaseEvalService):
app_name=inference_request.app_name,
eval_set_id=inference_request.eval_set_id,
eval_case=eval_case,
root_agent=root_agent,
root_agent=self._root_agent,
)
inference_results = [run_inference(eval_case) for eval_case in eval_cases]