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

chore: Remove deprecated convert_session_to_eval_format function

Browse Source 
This change removes the `convert_session_to_eval_format` function and its associated unit tests. New tests for `create_gcs_eval_managers_from_uri` are also added.

PiperOrigin-RevId: 819576620
This commit is contained in:
Ankur Sharma
2025-10-14 22:37:49 -07:00
committed by Copybara-Service
parent 05aa3fa38b
commit d82c492140
2 changed files with 41 additions and 493 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/google/adk/cli/utils/evals.py
-81
View File
@@ -14,15 +14,11 @@
from __future__ import annotations
import dataclasses
import os
from typing import Any
from typing import Tuple
from pydantic import alias_generators
from pydantic import BaseModel
from pydantic import ConfigDict
from typing_extensions import deprecated
from ...evaluation.eval_case import Invocation
from ...evaluation.evaluation_generator import EvaluationGenerator
@@ -43,83 +39,6 @@ class GcsEvalManagers(BaseModel):
eval_set_results_manager: GcsEvalSetResultsManager
@deprecated('Use convert_session_to_eval_invocations instead.')
def convert_session_to_eval_format(session: Session) -> list[dict[str, Any]]:
"""Converts a session data into eval format.
Args:
session: The session that should be converted.
Returns:
list: A single evaluation dataset in the required format.
"""
eval_case = []
events = session.events if session and session.events else []
for event in events:
if event.author == 'user':
if not event.content or not event.content.parts:
continue
# Extract user query
content = event.content
parts = content.parts
query = parts[0].text or ''
# Find the corresponding tool usage or response for the query
expected_tool_use = []
intermediate_agent_responses = []
# Check subsequent events to extract tool uses or responses for this turn.
for subsequent_event in events[events.index(event) + 1 :]:
event_author = subsequent_event.author or 'agent'
if event_author == 'user':
# We found an event where the author was the user. This means that a
# new turn has started. So close this turn here.
break
if not subsequent_event.content or not subsequent_event.content.parts:
continue
for subsequent_part in subsequent_event.content.parts:
# Some events have both function call and reference
if subsequent_part.function_call:
tool_name = subsequent_part.function_call.name or ''
tool_input = subsequent_part.function_call.args or {}
expected_tool_use.append({
'tool_name': tool_name,
'tool_input': tool_input,
})
elif subsequent_part.text:
# Also keep track of all the natural language responses that
# agent (or sub agents) generated.
intermediate_agent_responses.append(
{'author': event_author, 'text': subsequent_part.text}
)
# If we are here then either we are done reading all the events or we
# encountered an event that had content authored by the end-user.
# This, basically means an end of turn.
# We assume that the last natural language intermediate response is the
# final response from the agent/model. We treat that as a reference.
eval_case.append({
'query': query,
'expected_tool_use': expected_tool_use,
'expected_intermediate_agent_responses': intermediate_agent_responses[
:-1
],
'reference': (
intermediate_agent_responses[-1]['text']
if intermediate_agent_responses
else ''
),
})
return eval_case
def convert_session_to_eval_invocations(session: Session) -> list[Invocation]:
"""Converts a session data into a list of Invocation.
tests/unittests/cli/utils/test_evals.py
+41 -412
View File
@@ -14,421 +14,50 @@
"""Tests for utilities in eval."""
import os
from unittest import mock
from google.adk.cli.utils.evals import convert_session_to_eval_format
from google.adk.events.event import Event
from google.adk.sessions.session import Session
from google.genai import types
from google.adk.cli.utils import evals
from google.adk.evaluation.gcs_eval_set_results_manager import GcsEvalSetResultsManager
from google.adk.evaluation.gcs_eval_sets_manager import GcsEvalSetsManager
import pytest
def build_event(author: str, parts_content: list[dict]) -> Event:
"""Builds an Event object with specified parts."""
parts = []
for p_data in parts_content:
part_args = {}
if "text" in p_data:
part_args["text"] = p_data["text"]
if "func_name" in p_data:
part_args["function_call"] = types.FunctionCall(
name=p_data.get("func_name"), args=p_data.get("func_args")
)
# Add other part types here if needed for future tests
parts.append(types.Part(**part_args))
return Event(author=author, content=types.Content(parts=parts))
def test_convert_empty_session():
"""Test conversion function with empty events list in Session."""
# Pydantic models require mandatory fields for instantiation
session_empty_events = Session(
id="s1", app_name="app", user_id="u1", events=[]
@mock.patch.dict(os.environ, {'GOOGLE_CLOUD_PROJECT': 'test-project'})
@mock.patch(
'google.adk.cli.utils.evals.GcsEvalSetResultsManager',
autospec=True,
)
@mock.patch(
'google.adk.cli.utils.evals.GcsEvalSetsManager',
autospec=True,
)
def test_create_gcs_eval_managers_from_uri_success(
mock_gcs_eval_sets_manager, mock_gcs_eval_set_results_manager
):
mock_gcs_eval_sets_manager.return_value = mock.MagicMock(
spec=GcsEvalSetsManager
)
assert not convert_session_to_eval_format(session_empty_events)
def test_convert_none_session():
"""Test conversion function with None Session."""
assert not convert_session_to_eval_format(None)
def test_convert_session_skips_initial_non_user_events():
"""Test conversion function with only user events."""
events = [
build_event("model", [{"text": "Hello"}]),
build_event("user", [{"text": "How are you?"}]),
]
session = Session(id="s1", app_name="app", user_id="u1", events=events)
expected = [
{
"query": "How are you?",
"expected_tool_use": [],
"expected_intermediate_agent_responses": [],
"reference": "",
},
]
assert convert_session_to_eval_format(session) == expected
def test_convert_single_turn_text_only():
"""Test a single user query followed by a single agent text response."""
events = [
build_event("user", [{"text": "What is the time?"}]),
build_event("root_agent", [{"text": "It is 3 PM."}]),
]
session = Session(id="s1", app_name="app", user_id="u1", events=events)
expected = [{
"query": "What is the time?",
"expected_tool_use": [],
"expected_intermediate_agent_responses": [],
"reference": "It is 3 PM.",
}]
assert convert_session_to_eval_format(session) == expected
def test_convert_single_turn_tool_only():
"""Test a single user query followed by a single agent tool call."""
events = [
build_event("user", [{"text": "Get weather for Seattle"}]),
build_event(
"root_agent",
[{"func_name": "get_weather", "func_args": {"city": "Seattle"}}],
),
]
session = Session(id="s1", app_name="app", user_id="u1", events=events)
expected = [{
"query": "Get weather for Seattle",
"expected_tool_use": [
{"tool_name": "get_weather", "tool_input": {"city": "Seattle"}}
],
"expected_intermediate_agent_responses": [],
"reference": "",
}]
assert convert_session_to_eval_format(session) == expected
def test_convert_single_turn_multiple_tools_and_texts():
"""Test a turn with multiple agent responses (tools and text)."""
events = [
build_event("user", [{"text": "Do task A then task B"}]),
build_event(
"root_agent", [{"text": "Okay, starting task A."}]
), # Intermediate Text 1
build_event(
"root_agent", [{"func_name": "task_A", "func_args": {"param": 1}}]
), # Tool 1
build_event(
"root_agent", [{"text": "Task A done. Now starting task B."}]
), # Intermediate Text 2
build_event(
"another_agent", [{"func_name": "task_B", "func_args": {}}]
), # Tool 2
build_event(
"root_agent", [{"text": "All tasks completed."}]
), # Final Text (Reference)
]
session = Session(id="s1", app_name="app", user_id="u1", events=events)
expected = [{
"query": "Do task A then task B",
"expected_tool_use": [
{"tool_name": "task_A", "tool_input": {"param": 1}},
{"tool_name": "task_B", "tool_input": {}},
],
"expected_intermediate_agent_responses": [
{"author": "root_agent", "text": "Okay, starting task A."},
{
"author": "root_agent",
"text": "Task A done. Now starting task B.",
},
],
"reference": "All tasks completed.",
}]
assert convert_session_to_eval_format(session) == expected
def test_convert_multi_turn_session():
"""Test a session with multiple user/agent turns."""
events = [
build_event("user", [{"text": "Query 1"}]),
build_event("agent", [{"text": "Response 1"}]),
build_event("user", [{"text": "Query 2"}]),
build_event("agent", [{"func_name": "tool_X", "func_args": {}}]),
build_event("agent", [{"text": "Response 2"}]),
]
session = Session(id="s1", app_name="app", user_id="u1", events=events)
expected = [
{ # Turn 1
"query": "Query 1",
"expected_tool_use": [],
"expected_intermediate_agent_responses": [],
"reference": "Response 1",
},
{ # Turn 2
"query": "Query 2",
"expected_tool_use": [{"tool_name": "tool_X", "tool_input": {}}],
"expected_intermediate_agent_responses": [],
"reference": "Response 2",
},
]
assert convert_session_to_eval_format(session) == expected
def test_convert_agent_event_multiple_parts():
"""Test an agent event with both text and tool call parts."""
events = [
build_event("user", [{"text": "Do something complex"}]),
# Build event with multiple dicts in parts_content list
build_event(
"agent",
[
{"text": "Okay, doing it."},
{"func_name": "complex_tool", "func_args": {"value": True}},
],
),
build_event("agent", [{"text": "Finished."}]),
]
session = Session(id="s1", app_name="app", user_id="u1", events=events)
expected = [{
"query": "Do something complex",
"expected_tool_use": [
{"tool_name": "complex_tool", "tool_input": {"value": True}}
],
"expected_intermediate_agent_responses": [{
"author": "agent",
"text": "Okay, doing it.",
}], # Text from first part of agent event
"reference": "Finished.", # Text from second agent event
}]
assert convert_session_to_eval_format(session) == expected
def test_convert_handles_missing_content_or_parts():
"""Test that events missing content or parts are skipped gracefully."""
events = [
build_event("user", [{"text": "Query 1"}]),
Event(author="agent", content=None), # Agent event missing content
build_event("agent", [{"text": "Response 1"}]),
Event(author="user", content=None), # User event missing content
build_event("user", [{"text": "Query 2"}]),
Event(
author="agent", content=types.Content(parts=[])
), # Agent event with empty parts list
build_event("agent", [{"text": "Response 2"}]),
# User event with content but no parts (or None parts)
Event(author="user", content=types.Content(parts=None)),
build_event("user", [{"text": "Query 3"}]),
build_event("agent", [{"text": "Response 3"}]),
]
session = Session(id="s1", app_name="app", user_id="u1", events=events)
expected = [
{ # Turn 1 (from Query 1)
"query": "Query 1",
"expected_tool_use": [],
"expected_intermediate_agent_responses": [],
"reference": "Response 1",
},
{ # Turn 2 (from Query 2 - user event with None content was skipped)
"query": "Query 2",
"expected_tool_use": [],
"expected_intermediate_agent_responses": [],
"reference": "Response 2",
},
{ # Turn 3 (from Query 3 - user event with None parts was skipped)
"query": "Query 3",
"expected_tool_use": [],
"expected_intermediate_agent_responses": [],
"reference": "Response 3",
},
]
assert convert_session_to_eval_format(session) == expected
def test_convert_handles_missing_tool_name_or_args():
"""Test tool calls with missing name or args."""
events = [
build_event("user", [{"text": "Call tools"}]),
# Event where FunctionCall has name=None
Event(
author="agent",
content=types.Content(
parts=[
types.Part(
function_call=types.FunctionCall(name=None, args={"a": 1})
)
]
),
),
# Event where FunctionCall has args=None
Event(
author="agent",
content=types.Content(
parts=[
types.Part(
function_call=types.FunctionCall(name="tool_B", args=None)
)
]
),
),
# Event where FunctionCall part exists but FunctionCall object is None
# (should skip)
Event(
author="agent",
content=types.Content(
parts=[types.Part(function_call=None, text="some text")]
),
),
build_event("agent", [{"text": "Done"}]),
]
session = Session(id="s1", app_name="app", user_id="u1", events=events)
expected = [{
"query": "Call tools",
"expected_tool_use": [
{"tool_name": "", "tool_input": {"a": 1}}, # Defaults name to ""
{"tool_name": "tool_B", "tool_input": {}}, # Defaults args to {}
],
"expected_intermediate_agent_responses": [{
"author": "agent",
"text": "some text",
}], # Text part from the event where function_call was None
"reference": "Done",
}]
assert convert_session_to_eval_format(session) == expected
def test_convert_handles_missing_user_query_text():
"""Test user event where the first part has no text."""
events = [
# Event where user part has text=None
Event(
author="user", content=types.Content(parts=[types.Part(text=None)])
),
build_event("agent", [{"text": "Response 1"}]),
# Event where user part has text=""
build_event("user", [{"text": ""}]),
build_event("agent", [{"text": "Response 2"}]),
]
session = Session(id="s1", app_name="app", user_id="u1", events=events)
expected = [
{
"query": "", # Defaults to "" if text is None
"expected_tool_use": [],
"expected_intermediate_agent_responses": [],
"reference": "Response 1",
},
{
"query": "", # Defaults to "" if text is ""
"expected_tool_use": [],
"expected_intermediate_agent_responses": [],
"reference": "Response 2",
},
]
assert convert_session_to_eval_format(session) == expected
def test_convert_handles_empty_agent_text():
"""Test agent responses with empty string text."""
events = [
build_event("user", [{"text": "Query"}]),
build_event("agent", [{"text": "Okay"}]),
build_event("agent", [{"text": ""}]), # Empty text
build_event("agent", [{"text": "Done"}]),
]
session = Session(id="s1", app_name="app", user_id="u1", events=events)
expected = [{
"query": "Query",
"expected_tool_use": [],
"expected_intermediate_agent_responses": [
{"author": "agent", "text": "Okay"},
],
"reference": "Done",
}]
assert convert_session_to_eval_format(session) == expected
def test_convert_complex_sample_session():
"""Test using the complex sample session provided earlier."""
events = [
build_event("user", [{"text": "What can you do?"}]),
build_event(
"root_agent",
[{"text": "I can roll dice and check if numbers are prime. \n"}],
),
build_event(
"user",
[{
"text": (
"Roll a 8 sided dice and then check if 90 is a prime number"
" or not."
)
}],
),
build_event(
"root_agent",
[{
"func_name": "transfer_to_agent",
"func_args": {"agent_name": "roll_agent"},
}],
),
# Skipping FunctionResponse events as they don't have text/functionCall
# parts used by converter
build_event(
"roll_agent", [{"func_name": "roll_die", "func_args": {"sides": 8}}]
),
# Skipping FunctionResponse
build_event(
"roll_agent",
[
{"text": "I rolled a 2. Now, I'll check if 90 is prime. \n\n"},
{
"func_name": "transfer_to_agent",
"func_args": {"agent_name": "prime_agent"},
},
],
),
# Skipping FunctionResponse
build_event(
"prime_agent",
[{"func_name": "check_prime", "func_args": {"nums": [90]}}],
),
# Skipping FunctionResponse
build_event("prime_agent", [{"text": "90 is not a prime number. \n"}]),
]
session = Session(
id="some_id",
app_name="hello_world_ma",
user_id="user",
events=events,
mock_gcs_eval_set_results_manager.return_value = mock.MagicMock(
spec=GcsEvalSetResultsManager
)
expected = [
{
"query": "What can you do?",
"expected_tool_use": [],
"expected_intermediate_agent_responses": [],
"reference": "I can roll dice and check if numbers are prime. \n",
},
{
"query": (
"Roll a 8 sided dice and then check if 90 is a prime number or"
" not."
),
"expected_tool_use": [
{
"tool_name": "transfer_to_agent",
"tool_input": {"agent_name": "roll_agent"},
},
{"tool_name": "roll_die", "tool_input": {"sides": 8}},
{
"tool_name": "transfer_to_agent",
"tool_input": {"agent_name": "prime_agent"},
}, # From combined event
{"tool_name": "check_prime", "tool_input": {"nums": [90]}},
],
"expected_intermediate_agent_responses": [{
"author": "roll_agent",
"text": "I rolled a 2. Now, I'll check if 90 is prime. \n\n",
}], # Text from combined event
"reference": "90 is not a prime number. \n",
},
]
actual = convert_session_to_eval_format(session)
assert actual == expected
managers = evals.create_gcs_eval_managers_from_uri('gs://test-bucket')
assert managers is not None
mock_gcs_eval_sets_manager.assert_called_once_with(
bucket_name='test-bucket', project='test-project'
)
mock_gcs_eval_set_results_manager.assert_called_once_with(
bucket_name='test-bucket', project='test-project'
)
assert managers.eval_sets_manager == mock_gcs_eval_sets_manager.return_value
assert (
managers.eval_set_results_manager
== mock_gcs_eval_set_results_manager.return_value
)
def test_create_gcs_eval_managers_from_uri_failure():
with pytest.raises(ValueError):
evals.create_gcs_eval_managers_from_uri('unsupported-uri')