Files
wehub-resource-sync ec2b666284
Continuous Integration / Pre-commit Linter (push) Has been cancelled
Continuous Integration / Mypy Check (Python 3.10) (push) Has been cancelled
Continuous Integration / Mypy Check (Python 3.11) (push) Has been cancelled
Continuous Integration / Mypy Check (Python 3.12) (push) Has been cancelled
Continuous Integration / Mypy Check (Python 3.13) (push) Has been cancelled
Continuous Integration / Unit Tests (Python 3.10) (push) Has been cancelled
Continuous Integration / Unit Tests (Python 3.11) (push) Has been cancelled
Continuous Integration / Unit Tests (Python 3.12) (push) Has been cancelled
Continuous Integration / Unit Tests (Python 3.13) (push) Has been cancelled
Continuous Integration / Unit Tests (Python 3.14) (push) Has been cancelled
Continuous Integration / A2A v0.3 Tests (Python 3.10) (push) Has been cancelled
Continuous Integration / A2A v0.3 Tests (Python 3.11) (push) Has been cancelled
Continuous Integration / A2A v0.3 Tests (Python 3.12) (push) Has been cancelled
Copybara PR Handler / close-imported-pr (push) Has been cancelled
Continuous Integration / A2A v0.3 Tests (Python 3.13) (push) Has been cancelled
Continuous Integration / A2A v0.3 Tests (Python 3.14) (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 13:25:13 +08:00

365 lines
13 KiB
Python

# Copyright 2026 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 json
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_case import InvocationEvent
from google.adk.evaluation.eval_case import InvocationEvents
from google.adk.evaluation.eval_rubrics import RubricScore
from google.adk.evaluation.evaluator import EvalStatus
from google.adk.evaluation.llm_as_judge_utils import get_average_rubric_score
from google.adk.evaluation.llm_as_judge_utils import get_eval_status
from google.adk.evaluation.llm_as_judge_utils import get_text_from_content
from google.adk.evaluation.llm_as_judge_utils import get_tool_calls_and_responses_as_json_str
from google.adk.evaluation.llm_as_judge_utils import get_tool_declarations_as_json_str
from google.genai import types as genai_types
def test_get_text_from_content_with_none():
"""Tests get_text_from_content with None as input."""
assert get_text_from_content(None) is None
def test_get_text_from_content_with_content_and_none_parts():
"""Tests get_text_from_content with Content that has None for parts."""
content = genai_types.Content(parts=None)
assert get_text_from_content(content) is None
def test_get_text_from_content_with_empty_parts():
"""Tests get_text_from_content with an empty parts list."""
content = genai_types.Content(parts=[])
assert get_text_from_content(content) == None
def test_get_text_from_content_with_parts_but_no_text():
"""Tests get_text_from_content with parts that do not contain text."""
content = genai_types.Content(
parts=[
genai_types.Part(
function_call=genai_types.FunctionCall(name="test_func")
)
]
)
assert get_text_from_content(content) == ""
def test_get_text_from_content_with_single_text_part():
"""Tests get_text_from_content with a single text part."""
content = genai_types.Content(parts=[genai_types.Part(text="Hello")])
assert get_text_from_content(content) == "Hello"
def test_get_text_from_content_with_multiple_text_parts():
"""Tests get_text_from_content with multiple text parts."""
content = genai_types.Content(
parts=[genai_types.Part(text="Hello"), genai_types.Part(text="World")]
)
assert get_text_from_content(content) == "Hello\nWorld"
def test_get_text_from_content_with_mixed_parts():
"""Tests get_text_from_content with a mix of text and non-text parts."""
content = genai_types.Content(
parts=[
genai_types.Part(text="Hello"),
genai_types.Part(
function_call=genai_types.FunctionCall(name="test_func")
),
genai_types.Part(text="World"),
]
)
assert get_text_from_content(content) == "Hello\nWorld"
def test_get_text_from_content_with_invocation_include_intermediate_responses_in_final():
"""Tests get_text_from_content on an Invocation with and without the flag."""
intermediate_text = "Let me check."
final_response_text = "Done."
invocation = Invocation(
user_content=genai_types.Content(parts=[genai_types.Part(text="user")]),
intermediate_data=InvocationEvents(
invocation_events=[
InvocationEvent(
author="agent",
content=genai_types.Content(
parts=[genai_types.Part(text=intermediate_text)]
),
),
InvocationEvent(
author="tool",
content=genai_types.Content(
parts=[
genai_types.Part(
function_call=genai_types.FunctionCall(name="t")
)
]
),
),
]
),
final_response=genai_types.Content(
parts=[genai_types.Part(text=final_response_text)]
),
)
# Flag off (default): only the final response text is returned.
assert get_text_from_content(invocation) == final_response_text
# Flag on: intermediate text is concatenated before the final response.
assert (
get_text_from_content(
invocation, include_intermediate_responses_in_final=True
)
== f"{intermediate_text}\n{final_response_text}"
)
def test_get_text_from_content_with_intermediate_data_full_response():
invocation = Invocation(
user_content=genai_types.Content(parts=[genai_types.Part(text="user")]),
intermediate_data=IntermediateData(
intermediate_responses=[
("agent", [genai_types.Part(text="legacy intro")]),
(
"tool",
[
genai_types.Part(
function_call=genai_types.FunctionCall(name="lookup")
)
],
),
]
),
final_response=genai_types.Content(
parts=[genai_types.Part(text="final answer")]
),
)
assert get_text_from_content(invocation) == "final answer"
assert (
get_text_from_content(
invocation, include_intermediate_responses_in_final=True
)
== "legacy intro\nfinal answer"
)
def test_get_eval_status_with_none_score():
"""Tests get_eval_status returns NOT_EVALUATED for a None score."""
assert get_eval_status(score=None, threshold=0.5) == EvalStatus.NOT_EVALUATED
def test_get_eval_status_when_score_is_greater_than_threshold():
"""Tests get_eval_status returns PASSED when score > threshold."""
assert get_eval_status(score=0.8, threshold=0.5) == EvalStatus.PASSED
def test_get_eval_status_when_score_is_equal_to_threshold():
"""Tests get_eval_status returns PASSED when score == threshold."""
assert get_eval_status(score=0.5, threshold=0.5) == EvalStatus.PASSED
def test_get_eval_status_when_score_is_less_than_threshold():
"""Tests get_eval_status returns FAILED when score < threshold."""
assert get_eval_status(score=0.4, threshold=0.5) == EvalStatus.FAILED
def test_get_average_rubric_score_with_empty_list():
"""Tests get_average_rubric_score returns None for an empty list."""
assert get_average_rubric_score([]) is None
def test_get_average_rubric_score_with_all_none_scores():
"""Tests get_average_rubric_score returns None when all scores are None."""
rubric_scores = [
RubricScore(rubric_id="1", score=None),
RubricScore(rubric_id="2", score=None),
]
assert get_average_rubric_score(rubric_scores) is None
def test_get_average_rubric_score_with_single_score():
"""Tests get_average_rubric_score with a single valid score."""
rubric_scores = [RubricScore(rubric_id="1", score=0.8)]
assert get_average_rubric_score(rubric_scores) == 0.8
def test_get_average_rubric_score_with_multiple_scores():
"""Tests get_average_rubric_score with multiple valid scores."""
rubric_scores = [
RubricScore(rubric_id="1", score=0.8),
RubricScore(rubric_id="2", score=0.6),
]
assert get_average_rubric_score(rubric_scores) == 0.7
def test_get_average_rubric_score_with_mixed_scores():
"""Tests get_average_rubric_score with a mix of valid and None scores."""
rubric_scores = [
RubricScore(rubric_id="1", score=0.8),
RubricScore(rubric_id="2", score=None),
RubricScore(rubric_id="3", score=0.6),
]
assert get_average_rubric_score(rubric_scores) == 0.7
def test_get_tool_declarations_as_json_str_with_no_agents():
"""Tests get_tool_declarations_as_json_str with no agents."""
app_details = AppDetails(agent_details={})
expected_json = {"tool_declarations": {}}
actual_json_str = get_tool_declarations_as_json_str(app_details)
assert json.loads(actual_json_str) == expected_json
def test_get_tool_declarations_as_json_str_with_agent_no_tools():
"""Tests get_tool_declarations_as_json_str with an agent that has no tools."""
agent_details = {"agent1": AgentDetails(name="agent1", tool_declarations=[])}
app_details = AppDetails(agent_details=agent_details)
expected_json = {"tool_declarations": {"agent1": []}}
actual_json_str = get_tool_declarations_as_json_str(app_details)
assert json.loads(actual_json_str) == expected_json
def test_get_tool_declarations_as_json_str_with_agent_with_tools():
"""Tests get_tool_declarations_as_json_str with an agent that has tools."""
tool1 = genai_types.Tool(
function_declarations=[
genai_types.FunctionDeclaration(
name="test_func", description="A test function."
)
]
)
agent_details = {
"agent1": AgentDetails(name="agent1", tool_declarations=[tool1])
}
app_details = AppDetails(agent_details=agent_details)
expected_json = {
"tool_declarations": {
"agent1": [{
"function_declarations": [{
"name": "test_func",
"description": "A test function.",
}]
}]
}
}
actual_json_str = get_tool_declarations_as_json_str(app_details)
assert json.loads(actual_json_str) == expected_json
def test_get_tool_declarations_as_json_str_with_multiple_agents():
"""Tests get_tool_declarations_as_json_str with multiple agents."""
tool1 = genai_types.Tool(
function_declarations=[
genai_types.FunctionDeclaration(
name="test_func1", description="A test function 1."
)
]
)
agent_details = {
"agent1": AgentDetails(name="agent1", tool_declarations=[tool1]),
"agent2": AgentDetails(name="agent2", tool_declarations=[]),
}
app_details = AppDetails(agent_details=agent_details)
expected_json = {
"tool_declarations": {
"agent1": [{
"function_declarations": [{
"name": "test_func1",
"description": "A test function 1.",
}]
}],
"agent2": [],
}
}
actual_json_str = get_tool_declarations_as_json_str(app_details)
assert json.loads(actual_json_str) == expected_json
def test_get_tool_calls_and_responses_as_json_str_with_none():
"""Tests get_tool_calls_and_responses_as_json_str with None."""
assert (
get_tool_calls_and_responses_as_json_str(None)
== "No intermediate steps were taken."
)
def test_get_tool_calls_and_responses_as_json_str_with_intermediate_data_no_tools():
"""Tests get_tool_calls_and_responses_as_json_str with IntermediateData and no tools."""
intermediate_data = IntermediateData(tool_uses=[], tool_responses=[])
assert (
get_tool_calls_and_responses_as_json_str(intermediate_data)
== "No intermediate steps were taken."
)
intermediate_data = InvocationEvents(invocation_events=[])
assert (
get_tool_calls_and_responses_as_json_str(intermediate_data)
== "No intermediate steps were taken."
)
def test_get_tool_calls_and_responses_as_json_str_with_invocation_events_multiple_calls():
"""Tests get_tool_calls_and_responses_as_json_str with multiple calls in InvocationEvents."""
tool_call1 = genai_types.FunctionCall(name="func1", args={}, id="call1")
tool_call2 = genai_types.FunctionCall(name="func2", args={}, id="call2")
tool_response1 = genai_types.FunctionResponse(
name="func1", response={"status": "ok"}, id="call1"
)
invocation_event1 = InvocationEvent(
author="agent",
content=genai_types.Content(
parts=[
genai_types.Part(function_call=tool_call1),
genai_types.Part(function_call=tool_call2),
]
),
)
invocation_event2 = InvocationEvent(
author="tool",
content=genai_types.Content(
parts=[genai_types.Part(function_response=tool_response1)]
),
)
intermediate_data = InvocationEvents(
invocation_events=[invocation_event1, invocation_event2]
)
json_str = get_tool_calls_and_responses_as_json_str(intermediate_data)
expected_json = {
"tool_calls_and_response": [
{
"step": 0,
"tool_call": {"name": "func1", "args": {}, "id": "call1"},
"tool_response": {
"name": "func1",
"response": {"status": "ok"},
"id": "call1",
},
},
{
"step": 1,
"tool_call": {"name": "func2", "args": {}, "id": "call2"},
"tool_response": "None",
},
]
}
assert json.loads(json_str) == expected_json