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: Add Rubric based tool use metric

Browse Source 
The PR does two main things:
1) Introduces a new rubric based tool use metric
2) Given that we now have two rubric based metric, we refactor and create a new RubricBasedEvaluator interface.

PiperOrigin-RevId: 811983514
This commit is contained in:
Ankur Sharma
2025-09-26 15:47:42 -07:00
committed by Copybara-Service
parent a959653cf3
commit c984b9e552
10 changed files with 1312 additions and 681 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/google/adk/evaluation/common.py
+1 -1
View File
@@ -22,6 +22,6 @@ class EvalBaseModel(pydantic.BaseModel):
model_config = pydantic.ConfigDict(
alias_generator=alias_generators.to_camel,
populate_by_name=True,
extra='forbid',
extra="forbid",
arbitrary_types_allowed=True,
)
src/google/adk/evaluation/eval_metrics.py
+2
View File
@@ -52,6 +52,8 @@ class PrebuiltMetrics(Enum):
"rubric_based_final_response_quality_v1"
)
RUBRIC_BASED_TOOL_USE_QUALITY_V1 = "rubric_based_tool_use_quality_v1"
MetricName: TypeAlias = Union[str, PrebuiltMetrics]
Threshold: TypeAlias = float
src/google/adk/evaluation/llm_as_judge.py
+2
View File
@@ -26,6 +26,7 @@ from ..models.llm_request import LlmRequest
from ..models.llm_response import LlmResponse
from ..models.registry import LLMRegistry
from ..utils.context_utils import Aclosing
from ..utils.feature_decorator import experimental
from .common import EvalBaseModel
from .eval_case import Invocation
from .eval_metrics import BaseCriterion
@@ -42,6 +43,7 @@ class AutoRaterScore(EvalBaseModel):
rubric_scores: Optional[list[RubricScore]] = None
@experimental
class LlmAsJudge(Evaluator):
"""Evaluator based on a LLM.
src/google/adk/evaluation/metric_evaluator_registry.py
+5
View File
@@ -25,6 +25,7 @@ from .evaluator import Evaluator
from .final_response_match_v2 import FinalResponseMatchV2Evaluator
from .response_evaluator import ResponseEvaluator
from .rubric_based_final_response_quality_v1 import RubricBasedFinalResponseQualityV1Evaluator
from .rubric_based_tool_use_quality_v1 import RubricBasedToolUseV1Evaluator
from .safety_evaluator import SafetyEvaluatorV1
from .trajectory_evaluator import TrajectoryEvaluator
@@ -116,6 +117,10 @@ def _get_default_metric_evaluator_registry() -> MetricEvaluatorRegistry:
metric_info=RubricBasedFinalResponseQualityV1Evaluator.get_metric_info(),
evaluator=RubricBasedFinalResponseQualityV1Evaluator,
)
metric_evaluator_registry.register_evaluator(
metric_info=RubricBasedToolUseV1Evaluator.get_metric_info(),
evaluator=RubricBasedToolUseV1Evaluator,
)
return metric_evaluator_registry
src/google/adk/evaluation/rubric_based_evaluator.py
+390
View File
@@ -0,0 +1,390 @@
# 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
import abc
import logging
import re
from typing import Optional
from typing_extensions import override
from ..models.llm_response import LlmResponse
from ..utils.feature_decorator import experimental
from .common import EvalBaseModel
from .eval_metrics import BaseCriterion
from .eval_metrics import EvalMetric
from .eval_rubrics import Rubric
from .eval_rubrics import RubricScore
from .evaluator import EvaluationResult
from .evaluator import PerInvocationResult
from .llm_as_judge import AutoRaterScore
from .llm_as_judge import LlmAsJudge
from .llm_as_judge_utils import get_average_rubric_score
from .llm_as_judge_utils import get_eval_status
from .llm_as_judge_utils import get_text_from_content
logger = logging.getLogger("google_adk." + __name__)
class RubricResponse(EvalBaseModel):
"""Internal data model to represent a rubric's response from the auto-rater."""
property_text: Optional[str] = None
rationale: Optional[str] = None
score: Optional[float] = None
class AutoRaterResponseParser(abc.ABC):
"""An interface for parsing auto rater's response."""
@abc.abstractmethod
def parse(self, auto_rater_response: str) -> list[RubricResponse]:
"""Parses the auto rater's response."""
raise NotImplementedError
_PROPERTY_PATTERN = r"(?<=Property: )(.*)"
_RATIONALE_PATTERN = r"(?<=Rationale: )(.*)"
_VERDICT_PATTERN = r"(?<=Verdict: )(.*)"
class DefaultAutoRaterResponseParser(AutoRaterResponseParser):
"""The default implementation of the AutoRaterResponseParser."""
def parse(self, auto_rater_response: str) -> list[RubricResponse]:
"""Returns a list of RubricResponse parsed from the AutoRater's response."""
properties = re.findall(_PROPERTY_PATTERN, auto_rater_response)
rationales = re.findall(_RATIONALE_PATTERN, auto_rater_response)
scores = []
for verdict in re.findall(_VERDICT_PATTERN, auto_rater_response):
if "yes" in verdict.lower():
score = 1.0
elif "no" in verdict.lower():
score = 0.0
else:
score = None
scores.append(score)
rubric_responses = []
for p, r, s in zip(properties, rationales, scores):
rubric_responses.append(
RubricResponse(property_text=p.strip(), rationale=r.strip(), score=s)
)
return rubric_responses
class PerInvocationResultsAggregator(abc.ABC):
"""An interface for aggregating per invocation samples.
AutoRaters that are backed by an LLM are known to have certain degree of
unreliabilty to their responses. In order to counter that we sample the
autorater more than once for a single invocation.
The aggregator helps convert those multiple samples into a single result.
"""
@abc.abstractmethod
def aggregate(
self,
per_invocation_samples: list[PerInvocationResult],
threshold: float,
) -> PerInvocationResult:
"""Aggregates per invocation samples into a single result."""
raise NotImplementedError
class MajorityVotePerInvocationResultsAggregator(
PerInvocationResultsAggregator
):
"""Aggregates per invocation samples using majority vote."""
def aggregate(
self,
per_invocation_samples: list[PerInvocationResult],
threshold: float,
) -> PerInvocationResult:
"""Returns a combined result for the invocation using majority vote.
This method takes all those samples for a single invocation and combines
them to genreate one single result for the invocation.
This method specifically uses majority vote to aggregate scores for a
rubric. Take following Invocation and Rubric for example:
Invocation:
User: Is it going to be cold in Seattle tomorrow?
Weather Agent: No, it will be moderately warm as predicted temperature
for Seattle, WA tomorrow is 88F.
Rubric: Agent's reponse was concise and to the point.
We will sample the AutoRater 5 times, and the AutoRater responds
with (skipping the rationale field for now):
Sample 1:
Verdict: Yes
Sample 2:
Verdict: No
Sample 3:
Verdict: Yes
Sample 4:
Verdict: Yes
Sample 5:
Verdict: No
This method will use majority vote and combine the results of 5 samples
into one, and it will report "Yes" as the final verdict.
"""
score_category_by_rubric_id = {}
# We go over each rubric for each sample, and categorize the rubric into
# one of the following buckets:
# - Bucket 0: No score was generated for the rubric
# - Bucket 1: Score was generated and it was positive (1.0)
# - Bucket 2: Score was generated and it was negative (0.0)
for sample in per_invocation_samples:
if not sample.rubric_scores:
continue
for rubric_score in sample.rubric_scores:
rubric_id = rubric_score.rubric_id
if rubric_id not in score_category_by_rubric_id:
score_category_by_rubric_id[rubric_id] = ([], [], [])
if rubric_score.score is None: # No score
score_category_by_rubric_id[rubric_id][0].append(rubric_score)
elif rubric_score.score == 1.0: # Positive Result
score_category_by_rubric_id[rubric_id][1].append(rubric_score)
else: # Negative result
score_category_by_rubric_id[rubric_id][2].append(rubric_score)
aggregated_rubric_scores = []
for rubric_id in score_category_by_rubric_id:
no_scores, positives, negatives = score_category_by_rubric_id[rubric_id]
if not positives and not negatives:
# There has to be at least a no score rubric!
aggregated_rubric_scores.append(no_scores[0])
# This is where we are taking a majority vote.
elif len(positives) > len(negatives):
aggregated_rubric_scores.append(positives[0])
else:
aggregated_rubric_scores.append(negatives[0])
aggregated_overall_score = get_average_rubric_score(
aggregated_rubric_scores
)
return PerInvocationResult(
actual_invocation=per_invocation_samples[0].actual_invocation,
expected_invocation=per_invocation_samples[0].expected_invocation,
score=aggregated_overall_score,
rubric_scores=aggregated_rubric_scores,
eval_status=get_eval_status(aggregated_overall_score, threshold),
)
class InvocationResultsSummarizer(abc.ABC):
"""An interface for summarizing per invocation results."""
@abc.abstractmethod
def summarize(
self, per_invocation_results: list[PerInvocationResult], threshold: float
) -> EvaluationResult:
"""Summaries per invocation results into a single result."""
raise NotImplementedError
class MeanInvocationResultsSummarizer(InvocationResultsSummarizer):
"""Summarizes per invocation results using mean score."""
def summarize(
self, per_invocation_results: list[PerInvocationResult], threshold: float
) -> EvaluationResult:
"""Summarizes per invocation evaluation results into a single score.
A single eval case can have multiple invocations and the eval metric is
assessed for each invocation. But, we do want to summarize and make a
statement on how the eval case as a whole performed on the metric.
This method helps us aggreate rubric scores across invocation.
This method calculates the mean score of a rubric across several
invocations.
"""
unaggregated_rubric_scores = [] # Later used to calculate average.
# Collect rubric scores by id, so that we can calculate average score
# for each rubric id.
rubric_scores_by_id = {}
for sample in per_invocation_results:
if not sample.rubric_scores:
continue
for rubric_score in sample.rubric_scores:
rubric_id = rubric_score.rubric_id
if rubric_id not in rubric_scores_by_id:
rubric_scores_by_id[rubric_id] = []
rubric_scores_by_id[rubric_id].append(rubric_score)
unaggregated_rubric_scores.append(rubric_score)
aggregated_rubric_scores = []
for rubric_id, rubric_scores in rubric_scores_by_id.items():
overall_score = get_average_rubric_score(rubric_scores)
aggregated_rubric_scores.append(
RubricScore(
rubric_id=rubric_id,
score=overall_score,
# There is no real way for us generate a rationale here, so we
# make is clear to the consumer of the result.
rationale=(
"This is an aggregated score derived from individual entries."
" Please refer to individual entries in each invocation for"
" actual rationale from the model."
),
)
)
# Use unaggregate rubric score to calculate overall score.
aggregated_overall_score = get_average_rubric_score(
unaggregated_rubric_scores
)
return EvaluationResult(
overall_score=aggregated_overall_score,
overall_eval_status=get_eval_status(
aggregated_overall_score, threshold
),
per_invocation_results=per_invocation_results,
overall_rubric_scores=aggregated_rubric_scores,
)
def _normalize_text(text: str) -> str:
"""Returns a normalized version of the passed in text."""
if not isinstance(text, str):
return ""
return text.lower().strip()
@experimental
class RubricBasedEvaluator(LlmAsJudge):
"""A base class for rubric based evaluators."""
def __init__(
self,
eval_metric: EvalMetric,
criterion_type: type[BaseCriterion],
auto_rater_response_parser: AutoRaterResponseParser = (
DefaultAutoRaterResponseParser()
),
per_invocation_results_aggregator: PerInvocationResultsAggregator = (
MajorityVotePerInvocationResultsAggregator()
),
invocation_results_summarizer: InvocationResultsSummarizer = (
MeanInvocationResultsSummarizer()
),
):
"""Initializes the RubricBasedEvaluator.
Args:
eval_metric: The evaluation metric configuration.
criterion_type: The type of the criterion used for this evaluator.
auto_rater_response_parser: An object that parses the auto-rater's
response text and extracts rubric scores.
per_invocation_results_aggregator: An object that aggregates multiple
samples for a single invocation into a single result. This is useful in
cases where the auto-rater is an LLM and multiple samples are generated
to account for the unreliability of the LLM.
invocation_results_summarizer: An object that summarizes the results of
all invocations in an eval case into a single result.
"""
super().__init__(
eval_metric,
criterion_type=criterion_type,
)
self._auto_rater_prompt_template = ""
self._auto_rater_response_parser = auto_rater_response_parser
self._per_invocation_results_aggregator = per_invocation_results_aggregator
self._invocation_results_summarizer = invocation_results_summarizer
assert self._criterion.rubrics, "Rubrics are required."
self._rubrics: list[Rubric] = self._criterion.rubrics
self._normalized_rubric_to_id_map = {
_normalize_text(r.rubric_content.text_property): r.rubric_id
for r in self._rubrics
}
@override
def convert_auto_rater_response_to_score(
self, auto_rater_response: LlmResponse
) -> AutoRaterScore:
"""Returns an AutoRaterScore generated from AutoRater's response."""
response_text = get_text_from_content(auto_rater_response.content)
rubric_responses = self._auto_rater_response_parser.parse(response_text)
rubric_scores = []
for rubric_response in rubric_responses:
normalized_rubric = _normalize_text(rubric_response.property_text)
rubric_id = self._normalized_rubric_to_id_map.get(normalized_rubric, None)
if rubric_id:
rubric_scores.append(
RubricScore(
rubric_id=rubric_id,
rationale=rubric_response.rationale,
score=rubric_response.score,
)
)
else:
logger.warning(
f"Rubric {rubric_response.property_text} not found in the rubrics"
" provided to the metric."
)
aggregated_score = get_average_rubric_score(rubric_scores)
return AutoRaterScore(score=aggregated_score, rubric_scores=rubric_scores)
@override
def aggregate_per_invocation_samples(
self,
per_invocation_samples: list[PerInvocationResult],
) -> PerInvocationResult:
"""Returns a combined result by aggregating multiple samples for the same invocation.
AutoRaters that are backed by an LLM are known to have certain degree of
unreliabilty to their responses. In order to counter that we sample the
autorater more than once for a single invocation.
The aggregator helps convert those multiple samples into a single result.
"""
return self._per_invocation_results_aggregator.aggregate(
per_invocation_samples, self._eval_metric.threshold
)
@override
def aggregate_invocation_results(
self, per_invocation_results: list[PerInvocationResult]
) -> EvaluationResult:
"""Summarizes per invocation evaluation results into a single score."""
return self._invocation_results_summarizer.summarize(
per_invocation_results, self._eval_metric.threshold
)
src/google/adk/evaluation/rubric_based_final_response_quality_v1.py
+2 -254
View File
@@ -15,15 +15,11 @@
from __future__ import annotations
import logging
import re
from typing import ClassVar
from typing import Optional
from typing_extensions import override
from ..models.llm_response import LlmResponse
from ..utils.feature_decorator import experimental
from .common import EvalBaseModel
from .eval_case import Invocation
from .eval_case import InvocationEvents
from .eval_metrics import EvalMetric
@@ -32,17 +28,10 @@ from .eval_metrics import MetricInfo
from .eval_metrics import MetricValueInfo
from .eval_metrics import PrebuiltMetrics
from .eval_metrics import RubricsBasedCriterion
from .eval_rubrics import Rubric
from .eval_rubrics import RubricScore
from .evaluator import EvaluationResult
from .evaluator import PerInvocationResult
from .llm_as_judge import AutoRaterScore
from .llm_as_judge import LlmAsJudge
from .llm_as_judge_utils import get_average_rubric_score
from .llm_as_judge_utils import get_eval_status
from .llm_as_judge_utils import get_text_from_content
from .llm_as_judge_utils import get_tool_calls_and_responses_as_json_str
from .llm_as_judge_utils import get_tool_declarations_as_json_str
from .rubric_based_evaluator import RubricBasedEvaluator
logger = logging.getLogger("google_adk." + __name__)
@@ -241,55 +230,8 @@ Verdict: yes
"""
_PROPERTY_PATTERN = r"(?<=Property: )(.*)"
_RATIONALE_PATTERN = r"(?<=Rationale: )(.*)"
_VERDICT_PATTERN = r"(?<=Verdict: )(.*)"
class _RubricResponse(EvalBaseModel):
"""Internal data model to represent a rubric's response from the auto-rater."""
property_text: Optional[str] = None
rationale: Optional[str] = None
score: Optional[float] = None
def _normalize_text(text: str) -> str:
"""Returns a normalized version of the passed in text."""
if not isinstance(text, str):
return ""
return text.lower().strip()
def _parse_auto_rater_response(
auto_rater_response: str,
) -> list[_RubricResponse]:
"""Returns a list of _RubricResponse parsed from the AutoRater's response."""
properties = re.findall(_PROPERTY_PATTERN, auto_rater_response)
rationales = re.findall(_RATIONALE_PATTERN, auto_rater_response)
scores = []
for verdict in re.findall(_VERDICT_PATTERN, auto_rater_response):
if "yes" in verdict.lower():
score = 1.0
elif "no" in verdict.lower():
score = 0.0
else:
score = None
scores.append(score)
rubric_responses = []
for p, r, s in zip(properties, rationales, scores):
rubric_responses.append(
_RubricResponse(property_text=p.strip(), rationale=r.strip(), score=s)
)
return rubric_responses
@experimental
class RubricBasedFinalResponseQualityV1Evaluator(LlmAsJudge):
class RubricBasedFinalResponseQualityV1Evaluator(RubricBasedEvaluator):
"""An Evaluator for rubric based assessment of the agent's final response using a LLM.
The evaluator uses a set of rubrics to assess the quality of the agent's
@@ -323,15 +265,6 @@ class RubricBasedFinalResponseQualityV1Evaluator(LlmAsJudge):
_RUBRIC_BASED_FINAL_RESPONSE_QUALITY_V1_PROMPT
)
assert self._criterion.rubrics, "Rubrics are required."
self._rubrics: list[Rubric] = self._criterion.rubrics
self._normalized_rubric_to_id_map = {
_normalize_text(r.rubric_content.text_property): r.rubric_id
for r in self._rubrics
}
@staticmethod
def get_metric_info() -> MetricInfo:
return MetricInfo(
@@ -387,188 +320,3 @@ class RubricBasedFinalResponseQualityV1Evaluator(LlmAsJudge):
)
return auto_rater_prompt
@override
def convert_auto_rater_response_to_score(
self, auto_rater_response: LlmResponse
) -> AutoRaterScore:
"""Returns an AutoRaterScore generated from AutoRater's response."""
response_text = get_text_from_content(auto_rater_response.content)
rubric_responses = _parse_auto_rater_response(response_text)
rubric_scores = []
for rubric_response in rubric_responses:
normalized_rubric = _normalize_text(rubric_response.property_text)
rubric_id = self._normalized_rubric_to_id_map.get(normalized_rubric, None)
if rubric_id:
rubric_scores.append(
RubricScore(
rubric_id=rubric_id,
rationale=rubric_response.rationale,
score=rubric_response.score,
)
)
else:
logger.warning(
f"Rubric {rubric_response.property_text} not found in the rubrics"
" provided to the metric."
)
aggregated_score = get_average_rubric_score(rubric_scores)
return AutoRaterScore(score=aggregated_score, rubric_scores=rubric_scores)
@override
def aggregate_per_invocation_samples(
self,
per_invocation_samples: list[PerInvocationResult],
) -> PerInvocationResult:
"""Returns a combined result for the invocation.
This AutoRater is backed by an LLM that are known to have certain degree of
unreliabilty to their responses. In order to counter that we sample the
autorater more than once for a single invocation.
This method takes all those samples for a single invocation and combines
them to genreate one single result for the invocation.
This method specifically uses majority vote to aggregate scores for a
rubric. Take following Invocation and Rubric for example:
Invocation:
User: Is it going to be cold in Seattle tomorrow?
Weather Agent: No, it will be moderately warm as predicted temperature
for Seattle, WA tomorrow is 88F.
Rubric: Agent's reponse was concise and to the point.
We will sample the AutoRater 5 times, and the AutoRater responds
with (skipping the rationale field for now):
Sample 1:
Verdict: Yes
Sample 2:
Verdict: No
Sample 3:
Verdict: Yes
Sample 4:
Verdict: Yes
Sample 5:
Verdict: No
This method will use majority vote and combine the results of 5 samples
into one, and it will report "Yes" as the final verdict.
"""
score_category_by_rubric_id = {}
# We go over each rubric for each sample, and categorize the rubric into
# one of the following buckets:
# - Bucket 0: No score was generated for the rubric
# - Bucket 1: Score was generated and it was positive (1.0)
# - Bucket 2: Score was generated and it was negative (0.0)
for sample in per_invocation_samples:
if not sample.rubric_scores:
continue
for rubric_score in sample.rubric_scores:
rubric_id = rubric_score.rubric_id
if rubric_id not in score_category_by_rubric_id:
score_category_by_rubric_id[rubric_id] = ([], [], [])
if rubric_score.score is None: # No score
score_category_by_rubric_id[rubric_id][0].append(rubric_score)
elif rubric_score.score == 1.0: # Positive Result
score_category_by_rubric_id[rubric_id][1].append(rubric_score)
else: # Negative result
score_category_by_rubric_id[rubric_id][2].append(rubric_score)
aggregated_rubric_scores = []
for rubric_id in score_category_by_rubric_id:
no_scores, positives, negatives = score_category_by_rubric_id[rubric_id]
if not positives and not negatives:
# There has to be at least a no score rubric!
aggregated_rubric_scores.append(no_scores[0])
# This is where we are taking a majority vote.
elif len(positives) > len(negatives):
aggregated_rubric_scores.append(positives[0])
else:
aggregated_rubric_scores.append(negatives[0])
aggregated_overall_score = get_average_rubric_score(
aggregated_rubric_scores
)
return PerInvocationResult(
actual_invocation=per_invocation_samples[0].actual_invocation,
expected_invocation=per_invocation_samples[0].expected_invocation,
score=aggregated_overall_score,
rubric_scores=aggregated_rubric_scores,
eval_status=get_eval_status(
aggregated_overall_score, self._eval_metric.threshold
),
)
@override
def aggregate_invocation_results(
self, per_invocation_results: list[PerInvocationResult]
) -> EvaluationResult:
"""Aggregates per invocation evaluation results into a single score.
A single eval case can have multiple invocations and the eval metric is
assessed for each invocation. But, we do want to make an aggregate
statement on how the eval case as a whole performed on the metric.
This method helps us aggreate rubric scores across invocation.
Do note that the aggregation strategy used here is different from the one
that is used in `aggregate_per_invocation_samples` method, where we used
majority vote. In this method, we actually calculate the mean score of a
rubric across several invocations, as majority score would be misleading.
"""
unaggregated_rubric_scores = [] # Later used to calculate average.
# Collect rubric scores by id, so that we can calculate average score
# for each rubric id.
rubric_scores_by_id = {}
for sample in per_invocation_results:
if not sample.rubric_scores:
continue
for rubric_score in sample.rubric_scores:
rubric_id = rubric_score.rubric_id
if rubric_id not in rubric_scores_by_id:
rubric_scores_by_id[rubric_id] = []
rubric_scores_by_id[rubric_id].append(rubric_score)
unaggregated_rubric_scores.append(rubric_score)
aggregated_rubric_scores = []
for rubric_id, rubric_scores in rubric_scores_by_id.items():
overall_score = get_average_rubric_score(rubric_scores)
aggregated_rubric_scores.append(
RubricScore(
rubric_id=rubric_id,
score=overall_score,
# There is no real way for us generate a rationale here, so we
# make is clear to the consumer of the result.
rationale=(
"This is an aggregated score derived from individual entries."
" Please refer to individual entries in each invocation for"
" actual rationale from the model."
),
)
)
# Use unaggregate rubric score to calculate overall score.
aggregated_overall_score = get_average_rubric_score(
unaggregated_rubric_scores
)
return EvaluationResult(
overall_score=aggregated_overall_score,
overall_eval_status=get_eval_status(
aggregated_overall_score, self._eval_metric.threshold
),
per_invocation_results=per_invocation_results,
overall_rubric_scores=aggregated_rubric_scores,
)
src/google/adk/evaluation/rubric_based_tool_use_quality_v1.py
+206
View File
@@ -0,0 +1,206 @@
# 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
import logging
from typing import ClassVar
from typing_extensions import override
from ..utils.feature_decorator import experimental
from .eval_case import Invocation
from .eval_metrics import EvalMetric
from .eval_metrics import Interval
from .eval_metrics import MetricInfo
from .eval_metrics import MetricValueInfo
from .eval_metrics import PrebuiltMetrics
from .eval_metrics import RubricsBasedCriterion
from .llm_as_judge_utils import get_text_from_content
from .llm_as_judge_utils import get_tool_calls_and_responses_as_json_str
from .llm_as_judge_utils import get_tool_declarations_as_json_str
from .rubric_based_evaluator import RubricBasedEvaluator
logger = logging.getLogger("google_adk." + __name__)
_RUBRIC_BASED_TOOL_USE_QUALITY_V1_PROMPT = """# Mission
- Your mission is to evaluate the quality of responses generated by an AI agent. You will be presented with a user prompt (<user_prompt>), the agent's response (<response>) to that user prompt, and a set of properties (<property>) that you must use to objectively assess the validity of the agent's response.
- Only use the properties provided. Do not make up new properties.
- IMPORTANT: Assess all of the provided properties. Do not drop any of the properties from your response.
- The primary focus of this rating task is to check correctness of the agent's responses w.r.t. each of the properties.
# Rubric
"yes": The agent's response fulfilled the property or the property is not applicable to the response.
"no": The agent's response did not fulfill the property.
# For each property started with a new line, follow these steps:
STEP 1: Repeat the property, word for word, without making any changes. Keep everything including punctuation and capitalization as-is.
STEP 2: Determine the steps needed to **exactly**, **precisely** and **completely** determine whether the agent's response fulfilled the property.
STEP 3: Follow the steps outlined in STEP 2, thinking out loud.
STEP 4: Review the thoughts and the original property.
STEP 5: Output the final verdict.
Property: [[Repeat the property in STEP 1 again.]]
Rationale: [[Explain your reasoning for the verdict.]]
Verdict: [[yes|no]]
# Output format (repeat this format for every property started with a new line):
STEP 1: ...
STEP 2: ...
STEP 3: ...
STEP 4: ...
STEP 5: ...
Property: ...
Rationale: ...
Verdict: ...
# Example output 1
STEP 1: Does the agent run function call 'default_api.grammar_check'?
STEP 2: I need to check if the agent runs the function call with exact function name as 'default_api.grammar_check'.
STEP 3: The response includes a function call 'default_api.grammar_check'.
STEP 4: The function call format and the function name are correct.
STEP 5: yes
Property: Does the agent run function call 'default_api.grammar_check'?
Rationale: The agent's response contains the function call 'default_api.grammar_check' within a proper code block and with the correct function name.
Verdict: yes
STEP 1: Does the agent provide function call 'default_api.grammar_check' with input parameter 'sentence' that is valid compared to the reference 'sentence'= 'the dog walks on the a park' and based on the following guideline? Guideline for 'sentence': 'The wording can differ. The agent response is valid if it conveys similar core content as the reference response. Less efficient and minor inaccurate phrasing is acceptable. The default value is None, if the reference response includes this parameter with value equal to the default value but it is not provided in the agent response, then evaluate it as valid.'
STEP 2: I need to check if the function call 'default_api.grammar_check' includes the parameter 'sentence' and whether the value assigned to 'sentence' is valid according to the provided guideline. The reference value is 'the dog walks on the a park'. According to the guideline, the wording can differ as long as the core content is similar.
STEP 3: The agent's response includes the function call `default_api.grammar_check(sentence="the dog walks on the a park")`. The parameter 'sentence' is present, and the value assigned to it is "the dog walks on the a park", which is identical to the reference value.
STEP 4: The parameter 'sentence' is present and its value is exactly the same as the reference value.
STEP 5: yes
Property: Does the agent provide function call 'default_api.grammar_check' with input parameter 'sentence' that is valid compared to the reference 'sentence'= 'the dog walks on the a park' and based on the following guideline? Guideline for 'sentence': 'The wording can differ. The agent response is valid if it conveys similar core content as the reference response. Less efficient and minor inaccurate phrasing is acceptable. The default value is None, if the reference response includes this parameter with value equal to the default value but it is not provided in the agent response, then evaluate it as valid.'
Rationale: The agent's response includes the 'sentence' parameter in the function call 'default_api.grammar_check', and the value assigned to it is exactly the same as the reference value, thus satisfying the given guideline.
Verdict: yes
# Example output 2
STEP 1: Does the agent run function call 'default_api.search_via_perplexity'?
STEP 2: I need to check if the agent runs the function call with exact function name as 'default_api.search_via_perplexity'.
STEP 3: The response includes a function call `default_api.get_web_search_results`, which does not match 'default_api.search_via_perplexity'.
STEP 4: The function name does not match.
STEP 5: no
Property: Does the agent run function call 'default_api.search_via_perplexity'?
Rationale: The agent called 'default_api.get_web_search_results', not 'default_api.search_via_perplexity'.
Verdict: no
STEP 1: Does the agent provide function call 'default_api.search_via_perplexity' with input parameter 'keyword' that is valid compared to the reference 'keyword'= 'GPT-4o vs GPT-3.5 cost comparison' and based on the following guideline? Guideline for 'keyword': 'The wording can differ. The agent response is valid if it conveys similar core content as the reference response. Less efficient and minor inaccurate phrasing is acceptable.'
STEP 2: Since the previous property is no, this property is not applicable.
STEP 3: N/A
STEP 4: N/A
STEP 5: yes
Property: Does the agent provide function call 'default_api.search_via_perplexity' with input parameter 'keyword' that is valid compared to the reference 'keyword'= 'GPT-4o vs GPT-3.5 cost comparison' and based on the following guideline? Guideline for 'keyword': 'The wording can differ. The agent response is valid if it conveys similar core content as the reference response. Less efficient and minor inaccurate phrasing is acceptable.'
Rationale: The agent did not use the function call 'default_api.search_via_perplexity'.
Verdict: yes
# Available tools, user input, response and properties:
<available_tools>
{tool_declarations}
</available_tools>
<user_prompt>
{user_input}
</user_prompt>
<response>
{tool_usage}
</response>
<properties>
{rubrics}
</properties>
REMEMBER: Your answer will help improve the AI agent. It is important to determine the fulfillment of the properties correctly. Even answering "no" will improve the agent! Respond in pure text, not json.
IMPORTANT: Make sure for each of the property listed, follow the example steps and output "Property: ..." on a new line and "Verdict: ..." on another new line.
"""
@experimental
class RubricBasedToolUseV1Evaluator(RubricBasedEvaluator):
"""An Evaluator for rubric based assessment of the agent's usage of Tools.
Example: Lets take an example of a Weather Agent that has access to two tools:
1: GeoCoding Tool: Coverts a city name, address or zip code into geographic
cordinates.
2: GetWeather Tool: Gets weather for the next 10 days for the given geographic
cordinates.
For this agent, one can create following Rubrics that could focus on tool use
Rubric 1: A call is made to GeoCoding Tool.
Rubric 2: A call is made to GetWeather Tool.
Rubric 3: The call to GetWeather Tool happens after the GeoCoding Tool.
Rubric 4: The input to GeoCoding Tool can be mapped back to user prompt.
Rubric 5: The input to GetWeather Tool comes from the output of GeoCoding
Tool.)
For each rubric, this evaluator will generate a confidence score between 0
and 1, where 0 means that agent's response did not satisfy the rubric at all
and 1 means complete adherence. Value closer to 1 are desirable.
A combined score using individual rubric confidences will also be generated.
Like individual rubric confidence scores, the range for this value will be
between 0 and 1, and it will have the same interpretation.
"""
criterion_type: ClassVar[type[RubricsBasedCriterion]] = RubricsBasedCriterion
def __init__(self, eval_metric: EvalMetric):
super().__init__(
eval_metric,
criterion_type=RubricBasedToolUseV1Evaluator.criterion_type,
)
self._auto_rater_prompt_template = _RUBRIC_BASED_TOOL_USE_QUALITY_V1_PROMPT
@staticmethod
def get_metric_info() -> MetricInfo:
return MetricInfo(
metric_name=PrebuiltMetrics.RUBRIC_BASED_TOOL_USE_QUALITY_V1.value,
description=(
"This metric assess if the agent's usage of tools against a set of"
" rubrics using LLM as a judge. Value range for this metric is"
" [0,1], with values closer to 1 more desirable."
),
metric_value_info=MetricValueInfo(
interval=Interval(min_value=0.0, max_value=1.0)
),
)
@override
def format_auto_rater_prompt(
self, actual_invocation: Invocation, _: Invocation
) -> str:
"""Returns the autorater prompt."""
user_input = get_text_from_content(actual_invocation.user_content)
tool_usage = get_tool_calls_and_responses_as_json_str(
actual_invocation.intermediate_data
)
rubrics = "\n* ".join(
[r.rubric_content.text_property for r in self._rubrics]
)
app_details = actual_invocation.app_details
tool_declarations = "Agent has no tools."
if app_details:
tool_declarations = get_tool_declarations_as_json_str(app_details)
return self._auto_rater_prompt_template.format(
tool_declarations=tool_declarations,
user_input=user_input,
tool_usage=tool_usage,
rubrics=rubrics,
)
tests/unittests/evaluation/test_rubric_based_evaluator.py
+554
View File
File diff suppressed because it is too large Load Diff
tests/unittests/evaluation/test_rubric_based_final_response_quality_v1.py
-426
View File
@@ -28,12 +28,9 @@ from google.adk.evaluation.eval_rubrics import Rubric
from google.adk.evaluation.eval_rubrics import RubricContent
from google.adk.evaluation.eval_rubrics import RubricScore
from google.adk.evaluation.evaluator import EvalStatus
from google.adk.evaluation.evaluator import EvaluationResult
from google.adk.evaluation.evaluator import PerInvocationResult
from google.adk.evaluation.llm_as_judge_utils import get_average_rubric_score
from google.adk.evaluation.rubric_based_final_response_quality_v1 import _parse_auto_rater_response
from google.adk.evaluation.rubric_based_final_response_quality_v1 import RubricBasedFinalResponseQualityV1Evaluator
from google.adk.models.llm_response import LlmResponse
from google.genai import types as genai_types
import pytest
@@ -225,426 +222,3 @@ def test_format_auto_rater_prompt_with_intermediate_data_no_tools(
prompt = evaluator.format_auto_rater_prompt(invocation, None)
assert "No intermediate steps were taken." in prompt
def test_parse_auto_rater_response_with_empty_string():
"""Tests _parse_auto_rater_response with an empty string."""
assert _parse_auto_rater_response("") == []
def test_parse_auto_rater_response_with_malformed_string():
"""Tests _parse_auto_rater_response with a malformed string."""
response = "This is just some random text without the expected format."
assert _parse_auto_rater_response(response) == []
def test_parse_auto_rater_response_with_single_yes_verdict():
"""Tests _parse_auto_rater_response with a single 'yes' verdict."""
response = """
Property: Is the response good?
Rationale: It was good.
Verdict: yes
"""
parsed = _parse_auto_rater_response(response)
assert len(parsed) == 1
assert parsed[0].property_text == "Is the response good?"
assert parsed[0].rationale == "It was good."
assert parsed[0].score == 1.0
def test_parse_auto_rater_response_with_single_no_verdict():
"""Tests _parse_auto_rater_response with a single 'no' verdict."""
response = """
Property: Is the response bad?
Rationale: It was bad.
Verdict: no
"""
parsed = _parse_auto_rater_response(response)
assert len(parsed) == 1
assert parsed[0].property_text == "Is the response bad?"
assert parsed[0].rationale == "It was bad."
assert parsed[0].score == 0.0
def test_parse_auto_rater_response_with_invalid_verdict():
"""Tests _parse_auto_rater_response with an invalid verdict."""
response = """
Property: Is it unclear?
Rationale: I cannot tell.
Verdict: maybe
"""
parsed = _parse_auto_rater_response(response)
assert len(parsed) == 1
assert parsed[0].property_text == "Is it unclear?"
assert parsed[0].rationale == "I cannot tell."
assert parsed[0].score is None
def test_parse_auto_rater_response_with_multiple_verdicts():
"""Tests _parse_auto_rater_response with multiple verdicts."""
response = """
Property: Is the response good?
Rationale: It was good.
Verdict: yes
Property: Is the response bad?
Rationale: It was not bad.
Verdict: no
"""
parsed = _parse_auto_rater_response(response)
assert len(parsed) == 2
assert parsed[0].property_text == "Is the response good?"
assert parsed[0].rationale == "It was good."
assert parsed[0].score == 1.0
assert parsed[1].property_text == "Is the response bad?"
assert parsed[1].rationale == "It was not bad."
assert parsed[1].score == 0.0
def test_parse_auto_rater_response_with_incomplete_entry():
"""Tests _parse_auto_rater_response with an incomplete entry."""
response = """
Property: Is the response good?
Rationale: It was good.
Verdict: yes
Property: Is the response bad?
Rationale: It was not bad.
""" # Missing Verdict
parsed = _parse_auto_rater_response(response)
assert len(parsed) == 1 # zip will only create one item
assert parsed[0].property_text == "Is the response good?"
def test_parse_auto_rater_response_with_case_insensitive_verdict():
"""Tests _parse_auto_rater_response is case-insensitive for verdicts."""
response = """
Property: Is the response good?
Rationale: It was good.
Verdict: Yes
Property: Is the response bad?
Rationale: It was bad.
Verdict: NO
"""
parsed = _parse_auto_rater_response(response)
assert len(parsed) == 2
assert parsed[0].score == 1.0
assert parsed[1].score == 0.0
def test_convert_auto_rater_response_to_score_with_empty_response(
evaluator: RubricBasedFinalResponseQualityV1Evaluator,
):
"""Tests convert_auto_rater_response_to_score with an empty response."""
response = LlmResponse(
content=genai_types.Content(parts=[genai_types.Part(text="")])
)
auto_rater_score = evaluator.convert_auto_rater_response_to_score(response)
assert auto_rater_score.score is None
assert auto_rater_score.rubric_scores == []
def test_convert_auto_rater_response_to_score_with_malformed_response(
evaluator: RubricBasedFinalResponseQualityV1Evaluator,
):
"""Tests convert_auto_rater_response_to_score with a malformed response."""
response = LlmResponse(
content=genai_types.Content(
parts=[genai_types.Part(text="This is not a valid format.")]
)
)
auto_rater_score = evaluator.convert_auto_rater_response_to_score(response)
assert auto_rater_score.score is None
assert auto_rater_score.rubric_scores == []
def test_convert_auto_rater_response_to_score_with_mixed_verdicts(
evaluator: RubricBasedFinalResponseQualityV1Evaluator,
):
"""Tests convert_auto_rater_response_to_score with mixed verdicts."""
response_text = """
Property: Is the response good?
Rationale: It was good.
Verdict: yes
Property: Is the response bad?
Rationale: It was bad.
Verdict: no
"""
response = LlmResponse(
content=genai_types.Content(parts=[genai_types.Part(text=response_text)])
)
auto_rater_score = evaluator.convert_auto_rater_response_to_score(response)
assert auto_rater_score.score == 0.5
assert len(auto_rater_score.rubric_scores) == 2
assert auto_rater_score.rubric_scores[0].score == 1.0
assert auto_rater_score.rubric_scores[1].score == 0.0
def test_convert_auto_rater_response_to_score_with_invalid_verdict(
evaluator: RubricBasedFinalResponseQualityV1Evaluator,
):
"""Tests convert_auto_rater_response_to_score with an invalid verdict."""
response_text = """
Property: Is the response good?
Rationale: It was good.
Verdict: yes
Property: Is the response bad?
Rationale: I cannot tell.
Verdict: invalid
"""
response = LlmResponse(
content=genai_types.Content(parts=[genai_types.Part(text=response_text)])
)
auto_rater_score = evaluator.convert_auto_rater_response_to_score(response)
assert auto_rater_score.score == 1.0
assert len(auto_rater_score.rubric_scores) == 2
assert auto_rater_score.rubric_scores[0].score == 1.0
assert auto_rater_score.rubric_scores[1].score is None
def test_convert_auto_rater_response_to_score_with_unknown_property(
evaluator: RubricBasedFinalResponseQualityV1Evaluator,
):
"""Tests convert_auto_rater_response_to_score with an unknown property."""
response_text = """
Property: Is the response amazing?
Rationale: It was amazing.
Verdict: yes
"""
response = LlmResponse(
content=genai_types.Content(parts=[genai_types.Part(text=response_text)])
)
auto_rater_score = evaluator.convert_auto_rater_response_to_score(response)
assert auto_rater_score.score is None
assert len(auto_rater_score.rubric_scores) == 0
def test_aggregate_per_invocation_samples_with_no_rubric_scores(
evaluator: RubricBasedFinalResponseQualityV1Evaluator,
):
"""Tests aggregation when samples have no rubric scores."""
samples = [
_create_per_invocation_result([]),
_create_per_invocation_result([]),
]
result = evaluator.aggregate_per_invocation_samples(samples)
assert result.score is None
assert result.rubric_scores == []
def test_aggregate_per_invocation_samples_with_majority_positive(
evaluator: RubricBasedFinalResponseQualityV1Evaluator,
):
"""Tests aggregation with a majority of positive scores."""
samples = [
_create_per_invocation_result([RubricScore(rubric_id="1", score=1.0)]),
_create_per_invocation_result([RubricScore(rubric_id="1", score=1.0)]),
_create_per_invocation_result([RubricScore(rubric_id="1", score=0.0)]),
]
result = evaluator.aggregate_per_invocation_samples(samples)
assert result.score == 1.0
assert len(result.rubric_scores) == 1
assert result.rubric_scores[0].rubric_id == "1"
assert result.rubric_scores[0].score == 1.0
def test_aggregate_per_invocation_samples_with_majority_negative(
evaluator: RubricBasedFinalResponseQualityV1Evaluator,
):
"""Tests aggregation with a majority of negative scores."""
samples = [
_create_per_invocation_result([RubricScore(rubric_id="1", score=1.0)]),
_create_per_invocation_result([RubricScore(rubric_id="1", score=0.0)]),
_create_per_invocation_result([RubricScore(rubric_id="1", score=0.0)]),
]
result = evaluator.aggregate_per_invocation_samples(samples)
assert result.score == 0.0
assert len(result.rubric_scores) == 1
assert result.rubric_scores[0].rubric_id == "1"
assert result.rubric_scores[0].score == 0.0
def test_aggregate_per_invocation_samples_with_tie_verdicts(
evaluator: RubricBasedFinalResponseQualityV1Evaluator,
):
"""Tests aggregation with a tie, where negative should win."""
samples = [
_create_per_invocation_result([RubricScore(rubric_id="1", score=1.0)]),
_create_per_invocation_result([RubricScore(rubric_id="1", score=0.0)]),
]
result = evaluator.aggregate_per_invocation_samples(samples)
assert result.score == 0.0
assert len(result.rubric_scores) == 1
assert result.rubric_scores[0].rubric_id == "1"
assert result.rubric_scores[0].score == 0.0
def test_aggregate_per_invocation_samples_with_all_none_scores(
evaluator: RubricBasedFinalResponseQualityV1Evaluator,
):
"""Tests aggregation when all samples have a score of None."""
samples = [
_create_per_invocation_result(
[RubricScore(rubric_id="1", score=None, rationale="r1")]
),
_create_per_invocation_result(
[RubricScore(rubric_id="1", score=None, rationale="r2")]
),
]
result = evaluator.aggregate_per_invocation_samples(samples)
assert result.score is None
assert len(result.rubric_scores) == 1
assert result.rubric_scores[0].rubric_id == "1"
assert result.rubric_scores[0].score is None
assert result.rubric_scores[0].rationale == "r1"
def test_aggregate_per_invocation_samples_with_multiple_rubrics(
evaluator: RubricBasedFinalResponseQualityV1Evaluator,
):
"""Tests aggregation with multiple rubrics."""
samples = [
_create_per_invocation_result([
RubricScore(rubric_id="1", score=1.0),
RubricScore(rubric_id="2", score=0.0),
]),
_create_per_invocation_result([
RubricScore(rubric_id="1", score=1.0),
RubricScore(rubric_id="2", score=0.0),
]),
_create_per_invocation_result([
RubricScore(rubric_id="1", score=0.0),
RubricScore(rubric_id="2", score=1.0),
]),
]
result = evaluator.aggregate_per_invocation_samples(samples)
assert result.score == 0.5
assert len(result.rubric_scores) == 2
rubric1_score = next(
(s for s in result.rubric_scores if s.rubric_id == "1"), None
)
rubric2_score = next(
(s for s in result.rubric_scores if s.rubric_id == "2"), None
)
assert rubric1_score is not None
assert rubric1_score.score == 1.0
assert rubric2_score is not None
assert rubric2_score.score == 0.0
def test_aggregate_invocation_results_with_empty_list(
evaluator: RubricBasedFinalResponseQualityV1Evaluator,
):
"""Tests aggregate_invocation_results with an empty list."""
result = evaluator.aggregate_invocation_results([])
assert isinstance(result, EvaluationResult)
assert result.overall_score is None
assert result.overall_rubric_scores == []
assert result.per_invocation_results == []
def test_aggregate_invocation_results_with_no_rubric_scores(
evaluator: RubricBasedFinalResponseQualityV1Evaluator,
):
"""Tests aggregate_invocation_results with samples that have no rubric scores."""
invocations = [
_create_per_invocation_result([]),
_create_per_invocation_result([]),
]
result = evaluator.aggregate_invocation_results(invocations)
assert result.overall_score is None
assert result.overall_rubric_scores == []
assert result.per_invocation_results == invocations
def test_aggregate_invocation_results_with_single_invocation(
evaluator: RubricBasedFinalResponseQualityV1Evaluator,
):
"""Tests aggregate_invocation_results with a single invocation result."""
invocations = [
_create_per_invocation_result([
RubricScore(rubric_id="1", score=1.0),
RubricScore(rubric_id="2", score=0.0),
])
]
result = evaluator.aggregate_invocation_results(invocations)
assert result.overall_score == 0.5
assert len(result.overall_rubric_scores) == 2
rubric1_score = next(
s for s in result.overall_rubric_scores if s.rubric_id == "1"
)
rubric2_score = next(
s for s in result.overall_rubric_scores if s.rubric_id == "2"
)
assert rubric1_score.score == 1.0
assert rubric2_score.score == 0.0
def test_aggregate_invocation_results_with_multiple_invocations_single_rubric(
evaluator: RubricBasedFinalResponseQualityV1Evaluator,
):
"""Tests aggregate_invocation_results with multiple invocations for a single rubric."""
invocations = [
_create_per_invocation_result([RubricScore(rubric_id="1", score=1.0)]),
_create_per_invocation_result([RubricScore(rubric_id="1", score=0.0)]),
_create_per_invocation_result([RubricScore(rubric_id="1", score=1.0)]),
]
result = evaluator.aggregate_invocation_results(invocations)
assert result.overall_score == pytest.approx(2 / 3)
assert len(result.overall_rubric_scores) == 1
assert result.overall_rubric_scores[0].rubric_id == "1"
assert result.overall_rubric_scores[0].score == pytest.approx(2 / 3)
def test_aggregate_invocation_results_with_multiple_invocations_and_rubrics(
evaluator: RubricBasedFinalResponseQualityV1Evaluator,
):
"""Tests aggregate_invocation_results with multiple invocations and rubrics."""
invocations = [
_create_per_invocation_result([
RubricScore(rubric_id="1", score=1.0),
RubricScore(rubric_id="2", score=0.0),
]),
_create_per_invocation_result([
RubricScore(rubric_id="1", score=0.0),
RubricScore(rubric_id="2", score=1.0),
]),
]
result = evaluator.aggregate_invocation_results(invocations)
assert result.overall_score == 0.5
assert len(result.overall_rubric_scores) == 2
rubric1_score = next(
s for s in result.overall_rubric_scores if s.rubric_id == "1"
)
rubric2_score = next(
s for s in result.overall_rubric_scores if s.rubric_id == "2"
)
assert rubric1_score.score == 0.5
assert rubric2_score.score == 0.5
def test_aggregate_invocation_results_with_none_scores(
evaluator: RubricBasedFinalResponseQualityV1Evaluator,
):
"""Tests aggregate_invocation_results with some None scores."""
invocations = [
_create_per_invocation_result([
RubricScore(rubric_id="1", score=1.0),
RubricScore(rubric_id="2", score=None),
]),
_create_per_invocation_result([
RubricScore(rubric_id="1", score=0.0),
RubricScore(rubric_id="2", score=1.0),
]),
]
result = evaluator.aggregate_invocation_results(invocations)
assert result.overall_score == pytest.approx(2 / 3)
assert len(result.overall_rubric_scores) == 2
rubric1_score = next(
s for s in result.overall_rubric_scores if s.rubric_id == "1"
)
rubric2_score = next(
s for s in result.overall_rubric_scores if s.rubric_id == "2"
)
assert rubric1_score.score == 0.5
assert rubric2_score.score == 1.0
tests/unittests/evaluation/test_rubric_based_tool_use_quality_v1.py
+150
View File
@@ -0,0 +1,150 @@
# 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 google.adk.evaluation.app_details import AgentDetails
from google.adk.evaluation.app_details import AppDetails
from google.adk.evaluation.eval_case import IntermediateData
from google.adk.evaluation.eval_case import Invocation
from google.adk.evaluation.eval_metrics import EvalMetric
from google.adk.evaluation.eval_metrics import JudgeModelOptions
from google.adk.evaluation.eval_metrics import PrebuiltMetrics
from google.adk.evaluation.eval_metrics import RubricsBasedCriterion
from google.adk.evaluation.eval_rubrics import Rubric
from google.adk.evaluation.eval_rubrics import RubricContent
from google.adk.evaluation.rubric_based_tool_use_quality_v1 import RubricBasedToolUseV1Evaluator
from google.genai import types as genai_types
import pytest
@pytest.fixture
def evaluator() -> RubricBasedToolUseV1Evaluator:
"""Returns a RubricBasedToolUseV1Evaluator."""
rubrics = [
Rubric(
rubric_id="1",
rubric_content=RubricContent(
text_property="Did the agent use the correct tool?"
),
),
Rubric(
rubric_id="2",
rubric_content=RubricContent(
text_property="Were the tool parameters correct?"
),
),
]
judge_model_options = JudgeModelOptions(
judge_model_config=None,
num_samples=3,
)
criterion = RubricsBasedCriterion(
threshold=0.5, rubrics=rubrics, judge_model_options=judge_model_options
)
metric = EvalMetric(
metric_name=PrebuiltMetrics.RUBRIC_BASED_TOOL_USE_QUALITY_V1.value,
threshold=0.5,
criterion=criterion,
)
return RubricBasedToolUseV1Evaluator(metric)
def test_format_auto_rater_prompt_with_basic_invocation(
evaluator: RubricBasedToolUseV1Evaluator,
):
"""Tests format_auto_rater_prompt with a basic invocation."""
invocation = Invocation(
user_content=genai_types.Content(
parts=[genai_types.Part(text="User input here.")]
),
)
prompt = evaluator.format_auto_rater_prompt(invocation, None)
assert "User input here." in prompt
assert "Did the agent use the correct tool?" in prompt
assert "Were the tool parameters correct?" in prompt
assert "<available_tools>\nAgent has no tools.\n</available_tools>" in prompt
assert "<response>\nNo intermediate steps were taken.\n</response>" in prompt
def test_format_auto_rater_prompt_with_app_details(
evaluator: RubricBasedToolUseV1Evaluator,
):
"""Tests format_auto_rater_prompt with app_details in invocation."""
tool = genai_types.Tool(
function_declarations=[
genai_types.FunctionDeclaration(
name="test_func", description="A test function."
)
]
)
app_details = AppDetails(
agent_details={
"agent1": AgentDetails(
name="agent1",
tool_declarations=[tool],
)
},
)
invocation = Invocation(
user_content=genai_types.Content(
parts=[genai_types.Part(text="User input here.")]
),
app_details=app_details,
)
prompt = evaluator.format_auto_rater_prompt(invocation, None)
assert '"name": "test_func"' in prompt
assert '"description": "A test function."' in prompt
def test_format_auto_rater_prompt_with_intermediate_data(
evaluator: RubricBasedToolUseV1Evaluator,
):
"""Tests format_auto_rater_prompt with intermediate_data in invocation."""
tool_call = genai_types.FunctionCall(
name="test_func", args={"arg1": "val1"}, id="call1"
)
tool_response = genai_types.FunctionResponse(
name="test_func", response={"result": "ok"}, id="call1"
)
intermediate_data = IntermediateData(
tool_uses=[tool_call], tool_responses=[tool_response]
)
invocation = Invocation(
user_content=genai_types.Content(
parts=[genai_types.Part(text="User input here.")]
),
intermediate_data=intermediate_data,
)
prompt = evaluator.format_auto_rater_prompt(invocation, None)
assert '"step": 0' in prompt
assert '"tool_call":' in prompt
assert '"name": "test_func"' in prompt
assert '"tool_response":' in prompt
assert '"result": "ok"' in prompt
def test_get_metric_info(evaluator: RubricBasedToolUseV1Evaluator):
"""Tests the get_metric_info method."""
metric_info = evaluator.get_metric_info()
assert (
metric_info.metric_name
== PrebuiltMetrics.RUBRIC_BASED_TOOL_USE_QUALITY_V1.value
)
assert "agent's usage of tools" in metric_info.description
assert metric_info.metric_value_info.interval.min_value == 0.0
assert metric_info.metric_value_info.interval.max_value == 1.0