From f414a79171d36d719b12d0a4ed3fea8bb455b19b Mon Sep 17 00:00:00 2001 From: A Vertex SDK engineer Date: Wed, 8 Jul 2026 11:18:21 -0700 Subject: [PATCH] feat: Support creating evaluation runs from pre-existing Interactions API data PiperOrigin-RevId: 944601748 --- agentplatform/_genai/_evals_common.py | 409 ++++++++++++ .../replays/test_create_evaluation_run.py | 36 ++ tests/unit/agentplatform/genai/test_evals.py | 599 ++++++++++++++++++ 3 files changed, 1044 insertions(+) diff --git a/agentplatform/_genai/_evals_common.py b/agentplatform/_genai/_evals_common.py index 608d5a1db7..91f5412455 100644 --- a/agentplatform/_genai/_evals_common.py +++ b/agentplatform/_genai/_evals_common.py @@ -32,6 +32,11 @@ import agentplatform from google.genai import types as genai_types from google.genai._api_client import BaseApiClient +from google.genai._gaos.types.interactions import interaction as interaction_types +from google.genai._gaos.types.interactions import functioncallstep +from google.genai._gaos.types.interactions import functionresultstep +from google.genai._gaos.types.interactions import modeloutputstep +from google.genai._gaos.types.interactions import userinputstep from google.genai.models import Models import pandas as pd from tqdm import tqdm @@ -428,6 +433,15 @@ def _resolve_dataset( ) -> types.EvaluationRunDataSource: """Resolves dataset for the evaluation run.""" if isinstance(dataset, types.EvaluationDataset): + # Resolve EvalCases with interactions_data_source by fetching + # each interaction and converting it to agent_data, then flowing + # through the normal DataFrame/GCS pipeline. + if dataset.eval_cases and _has_interactions_data_source(dataset.eval_cases): + resolved_cases = _resolve_interactions_to_eval_cases( + api_client, dataset.eval_cases + ) + dataset = types.EvaluationDataset(eval_cases=resolved_cases) + candidate_name = _get_candidate_name(dataset, parsed_agent_info) eval_df = dataset.eval_dataset_df if eval_df is None and dataset.eval_cases: @@ -575,6 +589,401 @@ def _is_gemini_agent_resource(agent: str) -> bool: ) +def _step_to_agent_event(step: Any) -> Optional[types.evals.AgentEvent]: + """Converts a typed GenAI SDK Interaction step to an AgentEvent. + + Uses ``isinstance`` checks against the GenAI SDK step classes so that + attribute access stays in sync with SDK/proto changes. + + Args: + step: A step from ``Interaction.steps`` (a GenAI SDK step type). + + Returns: + An AgentEvent, or ``None`` if the step type is not handled. + """ + if isinstance(step, userinputstep.UserInputStep): + return _text_step_to_event(step, author="user", role="user") + elif isinstance(step, modeloutputstep.ModelOutputStep): + return _text_step_to_event(step, author="agent", role="model") + elif isinstance(step, functioncallstep.FunctionCallStep): + return _function_call_step_to_event(step) + elif isinstance(step, functionresultstep.FunctionResultStep): + return _function_response_step_to_event(step) + else: + logger.info("Skipping unhandled interaction step type: %s", type(step).__name__) + return None + + +def _function_response_step_to_event( + step: functionresultstep.FunctionResultStep, +) -> types.evals.AgentEvent: + """Converts a FunctionResultStep to an AgentEvent.""" + result = step.result + if isinstance(result, dict): + result_str = json.dumps(result) + elif isinstance(result, str): + result_str = result + else: + result_str = str(result) if result is not None else "" + return types.evals.AgentEvent( # pytype: disable=missing-parameter + author="user", + content=genai_types.Content( + role="user", + parts=[ + genai_types.Part( + function_response=genai_types.FunctionResponse( + name=step.name or "", + response={"result": result_str}, + id=step.call_id or "", + ) + ) + ], + ), + ) + + +def _function_call_step_to_event( + step: functioncallstep.FunctionCallStep, +) -> types.evals.AgentEvent: + """Converts a FunctionCallStep to an AgentEvent.""" + return types.evals.AgentEvent( # pytype: disable=missing-parameter + author="agent", + content=genai_types.Content( + role="model", + parts=[ + genai_types.Part( + function_call=genai_types.FunctionCall( + name=step.name or "", + args=step.arguments or {}, + id=step.id or "", + ) + ) + ], + ), + ) + + +def _text_step_to_event( + step: Any, *, author: str, role: str +) -> Optional[types.evals.AgentEvent]: + """Converts a text-bearing step (UserInputStep / ModelOutputStep) to an AgentEvent. + + Args: + step: A GenAI SDK step with a ``content`` attribute. + author: The event author (``"user"`` or ``"agent"``). + role: The content role (``"user"`` or ``"model"``). + + Returns: + An AgentEvent, or ``None`` if no text parts were found. + """ + parts = [] + for content_item in step.content or []: + if getattr(content_item, "text", None): + parts.append(genai_types.Part(text=content_item.text)) + if not parts: + return None + return types.evals.AgentEvent( # pytype: disable=missing-parameter + author=author, + content=genai_types.Content(role=role, parts=parts), + ) + + +def _interaction_steps_to_events( + steps: list[Any], +) -> list[tuple[types.evals.AgentEvent, type]]: + """Converts a list of typed Interaction steps to AgentEvents. + + Each step is mapped via ``_step_to_agent_event``. Steps whose type is + not handled are skipped with a log message. The originating SDK step + class is returned alongside each event so callers can determine turn + boundaries without inspecting event content. + + Args: + steps: The ``steps`` list from a GenAI SDK ``Interaction`` object. + + Returns: + A list of ``(AgentEvent, step_class)`` tuples. + """ + events: list[tuple[types.evals.AgentEvent, type]] = [] + for step in steps: + event = _step_to_agent_event(step) + if event is not None: + events.append((event, type(step))) + return events + + +def _interaction_dict_to_agent_data( + interaction: dict[str, Any], +) -> types.evals.AgentData: + """Converts an Interaction API JSON response to an AgentData object. + + Parses the raw dict into a typed ``Interaction`` object (from the GenAI + SDK) so that step conversion uses ``isinstance`` checks and typed + attribute access. Steps are grouped into ConversationTurns -- each + ``UserInputStep`` starts a new turn, so multi-turn conversations + produce multiple turns. + + Args: + interaction: A dict from the Interactions API GET response. + + Returns: + An AgentData object with one or more ConversationTurns. + """ + typed_interaction = interaction_types.Interaction.model_validate(interaction) + all_events = _interaction_steps_to_events(typed_interaction.steps or []) + + # Group events into turns. Each UserInputStep starts a new turn. + grouped: list[list[types.evals.AgentEvent]] = [] + for event, step_type in all_events: + if not grouped or step_type is userinputstep.UserInputStep: + grouped.append([]) + grouped[-1].append(event) + + if not grouped: + return types.evals.AgentData( # pytype: disable=missing-parameter + turns=[ + types.evals.ConversationTurn( # pytype: disable=missing-parameter + turn_index=0, events=[] + ) + ] + ) + return types.evals.AgentData( # pytype: disable=missing-parameter + turns=[ + types.evals.ConversationTurn( # pytype: disable=missing-parameter + turn_index=i, events=events + ) + for i, events in enumerate(grouped) + ] + ) + + +def _merge_text_parts_in_agent_data( + agent_data: types.evals.AgentData, +) -> None: + """Merges consecutive text events and parts for cleaner trace display. + + The Interaction API may return multiple consecutive ``model_output`` + steps (one per paragraph) and/or multiple text content items within a + single step. ``_interaction_dict_to_agent_data`` maps each step to a + separate event, and each content item to a separate ``part``, causing + the trace renderer to display them as separate visual blocks. + + This function performs two merges: + + 1. **Event merge** -- consecutive events from the same author that + contain only text parts are collapsed into a single event. + 2. **Part merge** -- within each (possibly merged) event, consecutive + text-only parts are collapsed into a single part. + + Mutates ``agent_data`` in place. + + Args: + agent_data: An AgentData object to merge in place. + """ + for turn in agent_data.turns or []: + events = turn.events + if not events: + continue + + # --- Pass 1: merge consecutive text-only events from the same author --- + merged_events: list[types.evals.AgentEvent] = [] + for event in events: + parts = (event.content.parts if event.content else None) or [] + is_text_only = parts and all( + p.text is not None + and p.function_call is None + and p.function_response is None + for p in parts + ) + if ( + merged_events + and is_text_only + and event.author == merged_events[-1].author + ): + prev_content = merged_events[-1].content + prev_parts = (prev_content.parts if prev_content else None) or [] + prev_parts.extend(parts) + continue + merged_events.append(event) + turn.events = merged_events + + # --- Pass 2: merge consecutive text parts within each event --- + for event in turn.events: + content = event.content + if not content: + continue + parts = content.parts + if not parts or len(parts) <= 1: + continue + merged_parts: list[genai_types.Part] = [] + text_buffer: list[str] = [] + for part in parts: + if ( + part.text is not None + and part.function_call is None + and part.function_response is None + ): + text_buffer.append(part.text) + else: + if text_buffer: + merged_parts.append( + genai_types.Part(text="\n".join(text_buffer)) + ) + text_buffer = [] + merged_parts.append(part) + if text_buffer: + merged_parts.append(genai_types.Part(text="\n".join(text_buffer))) + content.parts = merged_parts + + +def _agent_tools_to_config_tools( + agent_tools: Optional[list[Any]], +) -> Optional[list[genai_types.Tool]]: + """Maps Gemini Agents API tools to ``genai_types.Tool`` for an AgentConfig.""" + if not agent_tools: + return None + tools: list[genai_types.Tool] = [] + for tool in agent_tools: + if not isinstance(tool, dict): + continue + tool_type = tool.get("type") + if tool_type == "google_search": + tools.append(genai_types.Tool(google_search=genai_types.GoogleSearch())) + elif tool_type == "code_execution": + tools.append( + genai_types.Tool(code_execution=genai_types.ToolCodeExecution()) + ) + elif tool_type == "url_context": + tools.append(genai_types.Tool(url_context=genai_types.UrlContext())) + else: + remainder = {k: v for k, v in tool.items() if k != "type"} + if remainder: + tools.append(genai_types.Tool.model_validate(remainder)) + return tools or None + + +def _fetch_agent_config_dict( + api_client: BaseApiClient, + agent_resource_name: str, +) -> types.evals.AgentConfig: + """Fetches an agent's config from the Agent API and returns an AgentConfig.""" + agent_short_id = agent_resource_name.split("/")[-1] or "agent" + + instruction: Optional[str] = None + description: Optional[str] = None + agent_type: Optional[str] = None + tools: Optional[list[genai_types.Tool]] = None + + try: + agent_resp = api_client.request("get", f"agents/{agent_short_id}", {}, None) + if agent_resp.body: + agent_dict = json.loads(agent_resp.body) + instruction = agent_dict.get("system_instruction") or None + description = agent_dict.get("description") or None + agent_type = agent_dict.get("base_agent") or None + tools = _agent_tools_to_config_tools(agent_dict.get("tools")) + except Exception as e: # pylint: disable=broad-exception-caught + logger.warning( + "Failed to fetch agent config for '%s' (continuing without it): %s", + agent_resource_name, + e, + ) + + return types.evals.AgentConfig( # pytype: disable=missing-parameter + agent_id=agent_short_id, + instruction=instruction, + description=description, + agent_type=agent_type, + tools=tools, + ) + + +def _has_interactions_data_source( + eval_cases: list[types.EvalCase], +) -> bool: + """Returns True if any EvalCase has interactions_data_source set.""" + return any(case.interactions_data_source is not None for case in eval_cases) + + +def _resolve_interactions_to_eval_cases( + api_client: BaseApiClient, + eval_cases: list[types.EvalCase], +) -> list[types.EvalCase]: + """Resolves EvalCases with interactions_data_source to agent_data. + + For each EvalCase that has interactions_data_source set, fetches the + Interaction via the SDK's interactions.get() API, converts the steps + to AgentData, and returns a new EvalCase with agent_data populated. + + Args: + api_client: The API client (must have an interactions module). + eval_cases: EvalCases with interactions_data_source set. + + Returns: + New list of EvalCases with agent_data populated from resolved + interactions. + + Raises: + ValueError: If eval_cases have missing interaction references. + """ + # Validate all cases up front before making any API calls. + for case in eval_cases: + ids = case.interactions_data_source + if ids is None: + raise ValueError( + "All eval_cases must have interactions_data_source set when" + " using interaction resolution. Found a case without it. Do" + " not mix interaction-based and prompt-based eval cases." + ) + if not ids.interaction: + raise ValueError( + "interactions_data_source.interaction is required. Each" + " EvalCase must reference an existing Interaction resource." + ) + + resolved_cases = [] + + for case in eval_cases: + ids = case.interactions_data_source + + # Extract the interaction short ID from the resource name. + # Handles both full resource names (projects/.../interactions/{id}) + # and bare IDs by always taking the last path component. + interaction_id = ids.interaction.split("/")[-1] + + logger.info("Fetching interaction: %s", ids.interaction) + path = f"interactions/{interaction_id}" + response = api_client.request("get", path, {}, None) + interaction_dict = {} if not response.body else json.loads(response.body) + + agent_data = _interaction_dict_to_agent_data(interaction_dict) + + # Best-effort: fetch the agent config (instruction, tools, + # description) from the Agent API so the display can render + # the System Topology section. + gemini_cfg = ids.gemini_agent_config + agent_name = gemini_cfg.gemini_agent if gemini_cfg else None + agent_config = _fetch_agent_config_dict(api_client, agent_name or "") + agent_data.agents = {agent_config.agent_id: agent_config} + + # Merge consecutive text events and parts so multi-paragraph + # responses render as a single block in the trace display. + _merge_text_parts_in_agent_data(agent_data) + + # Preserve all original EvalCase fields; only update agent_data + # and clear the now-resolved interactions_data_source. + resolved_cases.append( + case.model_copy( + update={ + "agent_data": agent_data, + "interactions_data_source": None, + } + ) + ) + + return resolved_cases + + def _add_evaluation_run_labels( labels: Optional[dict[str, str]] = None, agent: Optional[str] = None, diff --git a/tests/unit/agentplatform/genai/replays/test_create_evaluation_run.py b/tests/unit/agentplatform/genai/replays/test_create_evaluation_run.py index b2995b7cf3..b89d794026 100644 --- a/tests/unit/agentplatform/genai/replays/test_create_evaluation_run.py +++ b/tests/unit/agentplatform/genai/replays/test_create_evaluation_run.py @@ -709,6 +709,42 @@ def test_create_eval_run_with_metric_resource_name(mock_uuid4, client): # == INPUT_DF_WITH_CONTEXT_AND_HISTORY.iloc[i]["response"] # ) # assert evaluation_run.error is None +@mock.patch("uuid.uuid4") +def test_create_eval_run_with_interactions_data_source(mock_uuid4, client): + """Tests create_evaluation_run() with EvalCases using interactions_data_source. + + The SDK resolves each interaction client-side (GET interactions/{id} and + GET agents/{id}) before uploading the EvalCases to the evaluation pipeline. + """ + mock_uuid4.return_value = uuid.UUID("a1b2c3d4-e5f6-7890-abcd-ef1234567890") + client._api_client._http_options.api_version = "v1beta1" + interaction_id = "ChA2YzllYzk0MjY1NjZjODM5EAgaATAqBG1haW4" + gemini_agent = ( + "projects/model-evaluation-dev/locations/global/agents/test-agent-eval" + ) + eval_case = types.EvalCase( + interactions_data_source=types.InteractionsDataSource( + interaction=( + f"projects/model-evaluation-dev/locations/global" + f"/interactions/{interaction_id}" + ), + gemini_agent_config=types.GeminiAgentConfig( + gemini_agent=gemini_agent, + ), + ) + ) + evaluation_run = client.evals.create_evaluation_run( + name="test_interactions_data_source", + display_name="test_interactions_data_source", + dataset=types.EvaluationDataset(eval_cases=[eval_case]), + dest=GCS_DEST, + metrics=[GENERAL_QUALITY_METRIC], + ) + assert isinstance(evaluation_run, types.EvaluationRun) + assert evaluation_run.state == types.EvaluationRunState.PENDING + assert evaluation_run.error is None + + def test_create_eval_run_with_red_teaming_config(client): """Tests that create_evaluation_run() with red_teaming_config sends analysisConfigs.""" evaluation_run = client.evals.create_evaluation_run( diff --git a/tests/unit/agentplatform/genai/test_evals.py b/tests/unit/agentplatform/genai/test_evals.py index eda21f7e31..4dad5a3b9c 100644 --- a/tests/unit/agentplatform/genai/test_evals.py +++ b/tests/unit/agentplatform/genai/test_evals.py @@ -9563,6 +9563,230 @@ def test_resolve_dataset_preserves_conversation_history( assert "conversation_history" in ptd_values +class TestResolveDatasetWithInteractions: + """Tests for resolving interactions_data_source in _resolve_dataset.""" + + def setup_method(self): + self.mock_api_client = mock.Mock() + self.mock_api_client.project = "test-project" + self.mock_api_client.location = "us-central1" + + def test_has_interactions_data_source_true(self): + cases = [ + agentplatform_genai_types.EvalCase( + interactions_data_source=agentplatform_genai_types.InteractionsDataSource( + gemini_agent_config=agentplatform_genai_types.GeminiAgentConfig( + gemini_agent="projects/p/locations/l/agents/a" + ), + interaction="projects/p/locations/l/interactions/i1", + ) + ) + ] + assert _evals_common._has_interactions_data_source(cases) + + def test_has_interactions_data_source_false(self): + cases = [ + agentplatform_genai_types.EvalCase( + prompt=genai_types.Content( + parts=[genai_types.Part(text="test")] + ), + ) + ] + assert not _evals_common._has_interactions_data_source(cases) + + def test_resolve_rejects_mixed_cases(self): + """Mixing interaction-based and prompt-based cases raises ValueError.""" + cases = [ + agentplatform_genai_types.EvalCase( + interactions_data_source=agentplatform_genai_types.InteractionsDataSource( + interaction="projects/p/locations/l/interactions/i1", + ) + ), + agentplatform_genai_types.EvalCase( + prompt=genai_types.Content( + parts=[genai_types.Part(text="test")] + ), + ), + ] + with pytest.raises(ValueError, match="interactions_data_source"): + _evals_common._resolve_interactions_to_eval_cases( + self.mock_api_client, cases + ) + + def test_resolve_rejects_missing_interaction(self): + """EvalCase with interactions_data_source but no interaction raises.""" + cases = [ + agentplatform_genai_types.EvalCase( + interactions_data_source=agentplatform_genai_types.InteractionsDataSource( + gemini_agent_config=agentplatform_genai_types.GeminiAgentConfig( + gemini_agent="projects/p/locations/l/agents/a" + ), + ) + ), + ] + with pytest.raises(ValueError, match="interaction is required"): + _evals_common._resolve_interactions_to_eval_cases( + self.mock_api_client, cases + ) + + def test_interaction_dict_to_agent_data_text_conversation(self): + """Converts user_input + model_output steps to agent_data.""" + interaction_dict = { + "status": "completed", + "steps": [ + { + "type": "user_input", + "content": [{"type": "text", "text": "Hello agent"}], + }, + { + "type": "model_output", + "content": [ + {"type": "text", "text": "Hello! How can I help?"} + ], + }, + ] + } + + result = _evals_common._interaction_dict_to_agent_data( + interaction_dict + ) + + assert len(result.turns) == 1 + events = result.turns[0].events + assert len(events) == 2 + assert events[0].author == "user" + assert events[0].content.parts[0].text == "Hello agent" + assert events[1].author == "agent" + assert events[1].content.parts[0].text == "Hello! How can I help?" + + def test_interaction_dict_to_agent_data_with_tool_calls(self): + """Converts function_call + function_result steps.""" + interaction_dict = { + "status": "completed", + "steps": [ + { + "type": "function_call", + "name": "get_weather", + "arguments": {"city": "NYC"}, + "id": "call_1", + }, + { + "type": "function_result", + "name": "get_weather", + "call_id": "call_1", + "result": {"temp": "72F"}, + }, + ] + } + + result = _evals_common._interaction_dict_to_agent_data( + interaction_dict + ) + + events = result.turns[0].events + assert len(events) == 2 + fc_event = events[0] + assert fc_event.author == "agent" + assert fc_event.content.parts[0].function_call.name == "get_weather" + fr_event = events[1] + assert fr_event.content.parts[0].function_response.id == "call_1" + + def test_resolve_interactions_to_eval_cases_happy_path(self): + """Fetches an interaction and populates agent_data on the EvalCase.""" + interaction_body = json.dumps({ + "status": "completed", + "steps": [ + { + "type": "user_input", + "content": [{"type": "text", "text": "What is the weather?"}], + }, + { + "type": "model_output", + "content": [{"type": "text", "text": "It is sunny."}], + }, + ] + }) + agent_body = json.dumps({ + "system_instruction": "You are helpful.", + "base_agent": "gemini-2.0-flash", + }) + + def mock_request(method, path, *args, **kwargs): + resp = mock.MagicMock() + if path.startswith("interactions/"): + resp.body = interaction_body + elif path.startswith("agents/"): + resp.body = agent_body + else: + resp.body = None + return resp + + self.mock_api_client.request.side_effect = mock_request + + original_case = agentplatform_genai_types.EvalCase( + interactions_data_source=agentplatform_genai_types.InteractionsDataSource( + gemini_agent_config=agentplatform_genai_types.GeminiAgentConfig( + gemini_agent="projects/p/locations/l/agents/my-agent", + ), + interaction="projects/p/locations/l/interactions/i1", + ), + ) + + resolved = _evals_common._resolve_interactions_to_eval_cases( + self.mock_api_client, [original_case] + ) + + assert len(resolved) == 1 + result_case = resolved[0] + + # interactions_data_source should be cleared after resolution. + assert result_case.interactions_data_source is None + + # agent_data should be populated with the interaction content. + agent_data = result_case.agent_data + assert agent_data is not None + assert len(agent_data.turns) == 1 + events = agent_data.turns[0].events + assert len(events) == 2 + assert events[0].author == "user" + assert events[0].content.parts[0].text == "What is the weather?" + assert events[1].author == "agent" + assert events[1].content.parts[0].text == "It is sunny." + + # agents map should be populated with config from the Agent API. + assert "my-agent" in agent_data.agents + agent_config = agent_data.agents["my-agent"] + assert agent_config.agent_id == "my-agent" + assert agent_config.instruction == "You are helpful." + assert agent_config.agent_type == "gemini-2.0-flash" + + def test_resolve_preserves_original_eval_case_fields(self): + """Fields other than agent_data are preserved on the resolved EvalCase.""" + interaction_body = json.dumps({"status": "completed", "steps": []}) + + def mock_request(method, path, *args, **kwargs): + resp = mock.MagicMock() + resp.body = interaction_body + return resp + + self.mock_api_client.request.side_effect = mock_request + + original_case = agentplatform_genai_types.EvalCase( + interactions_data_source=agentplatform_genai_types.InteractionsDataSource( + interaction="projects/p/locations/l/interactions/i1", + ), + ) + + resolved = _evals_common._resolve_interactions_to_eval_cases( + self.mock_api_client, [original_case] + ) + + assert len(resolved) == 1 + # agent_data is populated and interactions_data_source cleared. + assert resolved[0].agent_data is not None + assert resolved[0].interactions_data_source is None + + class TestRateLimiter: """Tests for the RateLimiter class in _evals_utils.""" @@ -9989,3 +10213,378 @@ def test_create_evaluation_run_agent_engine_does_not_set_gemini(self): agent_run_config = self._agent_run_config(request_body) assert "gemini_agent_config" not in agent_run_config assert agent_run_config["agent_engine"] == _TEST_AGENT_ENGINE + + +class TestStepToAgentEvent: + """Tests for _step_to_agent_event using typed GenAI SDK step objects.""" + + def test_user_input_step(self): + from google.genai._gaos.types.interactions import textcontent # pylint: disable=g-import-not-at-top + from google.genai._gaos.types.interactions import userinputstep # pylint: disable=g-import-not-at-top + + step = userinputstep.UserInputStep( + content=[textcontent.TextContent(text="hello")], + ) + event = _evals_common._step_to_agent_event(step) + assert event is not None + assert event.author == "user" + assert event.content.parts[0].text == "hello" + + def test_model_output_step(self): + from google.genai._gaos.types.interactions import modeloutputstep # pylint: disable=g-import-not-at-top + from google.genai._gaos.types.interactions import textcontent # pylint: disable=g-import-not-at-top + + step = modeloutputstep.ModelOutputStep( + content=[textcontent.TextContent(text="world")], + ) + event = _evals_common._step_to_agent_event(step) + assert event is not None + assert event.author == "agent" + assert event.content.parts[0].text == "world" + + def test_function_call_step(self): + from google.genai._gaos.types.interactions import functioncallstep # pylint: disable=g-import-not-at-top + + step = functioncallstep.FunctionCallStep( + name="get_weather", + arguments={"city": "NYC"}, + id="call_1", + ) + event = _evals_common._step_to_agent_event(step) + assert event.content.parts[0].function_call.name == "get_weather" + + def test_function_result_step(self): + from google.genai._gaos.types.interactions import functionresultstep # pylint: disable=g-import-not-at-top + + step = functionresultstep.FunctionResultStep( + name="get_weather", + call_id="call_1", + result={"temp": "72F"}, + ) + event = _evals_common._step_to_agent_event(step) + assert event.author == "user" + assert event.content.parts[0].function_response.id == "call_1" + + def test_unknown_step_returns_none(self): + """An unrecognised step type returns None.""" + step = mock.MagicMock() + step.type = "some_future_step" + event = _evals_common._step_to_agent_event(step) + assert event is None + + def test_empty_text_content_returns_none(self): + from google.genai._gaos.types.interactions import userinputstep # pylint: disable=g-import-not-at-top + + step = userinputstep.UserInputStep(content=[]) + event = _evals_common._step_to_agent_event(step) + assert event is None + + +class TestInteractionDictToAgentData: + """Tests for _interaction_dict_to_agent_data.""" + + def test_single_turn_conversation(self): + """One user_input + model_output produces one turn.""" + interaction_dict = { + "status": "completed", + "steps": [ + { + "type": "user_input", + "content": [{"type": "text", "text": "Hello agent"}], + }, + { + "type": "model_output", + "content": [ + {"type": "text", "text": "Hello! How can I help?"} + ], + }, + ] + } + result = _evals_common._interaction_dict_to_agent_data( + interaction_dict + ) + assert len(result.turns) == 1 + events = result.turns[0].events + assert len(events) == 2 + assert events[0].author == "user" + assert events[0].content.parts[0].text == "Hello agent" + assert events[1].author == "agent" + assert events[1].content.parts[0].text == "Hello! How can I help?" + + def test_multi_turn_conversation(self): + """Multiple user_input steps produce multiple turns.""" + interaction_dict = { + "status": "completed", + "steps": [ + { + "type": "user_input", + "content": [{"type": "text", "text": "Turn 1 user"}], + }, + { + "type": "model_output", + "content": [{"type": "text", "text": "Turn 1 model"}], + }, + { + "type": "user_input", + "content": [{"type": "text", "text": "Turn 2 user"}], + }, + { + "type": "model_output", + "content": [{"type": "text", "text": "Turn 2 model"}], + }, + ] + } + result = _evals_common._interaction_dict_to_agent_data( + interaction_dict + ) + assert len(result.turns) == 2 + assert result.turns[0].turn_index == 0 + assert result.turns[1].turn_index == 1 + # Turn 1 events + assert result.turns[0].events[0].content.parts[0].text == "Turn 1 user" + # Turn 2 events + assert result.turns[1].events[0].content.parts[0].text == "Turn 2 user" + + def test_with_tool_calls(self): + """Function call/result in same turn as user_input.""" + interaction_dict = { + "status": "completed", + "steps": [ + { + "type": "user_input", + "content": [{"type": "text", "text": "What is the weather?"}], + }, + { + "type": "function_call", + "name": "get_weather", + "arguments": {"city": "NYC"}, + "id": "call_1", + }, + { + "type": "function_result", + "name": "get_weather", + "callId": "call_1", + "result": {"temp": "72F"}, + }, + { + "type": "model_output", + "content": [{"type": "text", "text": "It is 72F in NYC."}], + }, + ] + } + result = _evals_common._interaction_dict_to_agent_data( + interaction_dict + ) + # All in one turn since there's only one user_input. + assert len(result.turns) == 1 + events = result.turns[0].events + assert len(events) == 4 + assert events[1].content.parts[0].function_call.name == "get_weather" + + def test_empty_interaction(self): + """Empty interaction produces a single turn with no events.""" + result = _evals_common._interaction_dict_to_agent_data( + {"status": "completed", "steps": []} + ) + assert len(result.turns) == 1 + assert result.turns[0].events == [] + + def test_unknown_steps_skipped(self): + """Unrecognised step types are skipped, not included as events.""" + interaction_dict = { + "status": "completed", + "steps": [ + { + "type": "user_input", + "content": [{"type": "text", "text": "hello"}], + }, + {"type": "some_future_type", "data": "payload"}, + { + "type": "model_output", + "content": [{"type": "text", "text": "hi"}], + }, + ] + } + result = _evals_common._interaction_dict_to_agent_data( + interaction_dict + ) + events = result.turns[0].events + assert len(events) == 2 + + +class TestMergeTextPartsInAgentData: + """Tests for _merge_text_parts_in_agent_data.""" + + def _make_agent_data(self, turns_dict): + from agentplatform._genai.types import evals as evals_types # pylint: disable=g-import-not-at-top + return evals_types.AgentData.model_validate({"turns": turns_dict}) + + def test_consecutive_agent_events_merged(self): + """Multiple consecutive model_output events are merged into one.""" + agent_data = self._make_agent_data([{ + "turn_index": 0, + "events": [ + {"author": "agent", "content": { + "role": "model", + "parts": [{"text": "Paragraph 1."}], + }}, + {"author": "agent", "content": { + "role": "model", + "parts": [{"text": "Paragraph 2."}], + }}, + {"author": "agent", "content": { + "role": "model", + "parts": [{"text": "Paragraph 3."}], + }}, + ], + }]) + _evals_common._merge_text_parts_in_agent_data(agent_data) + events = agent_data.turns[0].events + assert len(events) == 1 + assert len(events[0].content.parts) == 1 + text = events[0].content.parts[0].text + assert "Paragraph 1." in text + assert "Paragraph 2." in text + assert "Paragraph 3." in text + + def test_different_authors_not_merged(self): + """Events from different authors are not merged.""" + agent_data = self._make_agent_data([{ + "turn_index": 0, + "events": [ + {"author": "user", "content": { + "role": "user", + "parts": [{"text": "hi"}], + }}, + {"author": "agent", "content": { + "role": "model", + "parts": [{"text": "hello"}], + }}, + ], + }]) + _evals_common._merge_text_parts_in_agent_data(agent_data) + events = agent_data.turns[0].events + assert len(events) == 2 + + def test_function_call_events_not_merged(self): + """Events with function_call parts are not merged with text events.""" + agent_data = self._make_agent_data([{ + "turn_index": 0, + "events": [ + {"author": "agent", "content": { + "role": "model", + "parts": [{"text": "Let me check."}], + }}, + {"author": "agent", "content": { + "role": "model", + "parts": [{"function_call": {"name": "search"}}], + }}, + ], + }]) + _evals_common._merge_text_parts_in_agent_data(agent_data) + events = agent_data.turns[0].events + assert len(events) == 2 + + def test_multiple_text_parts_within_event_merged(self): + """Multiple text parts within a single event are merged.""" + agent_data = self._make_agent_data([{ + "turn_index": 0, + "events": [ + {"author": "agent", "content": { + "role": "model", + "parts": [ + {"text": "Part A"}, + {"text": "Part B"}, + ], + }}, + ], + }]) + _evals_common._merge_text_parts_in_agent_data(agent_data) + parts = agent_data.turns[0].events[0].content.parts + assert len(parts) == 1 + assert "Part A" in parts[0].text + assert "Part B" in parts[0].text + + +class TestFetchAgentConfigDict: + """Tests for _fetch_agent_config_dict.""" + + def _make_api_response(self, body_dict): + resp = mock.MagicMock() + resp.body = json.dumps(body_dict) + return resp + + def test_extracts_full_config(self): + """Extracts instruction, description, agent_type, and tools.""" + agent_json = { + "system_instruction": "You are helpful.", + "description": "A helpful agent.", + "base_agent": "gemini-2.0-flash", + "tools": [ + {"type": "code_execution"}, + {"type": "google_search"}, + { + "type": "function", + "function_declarations": [ + {"name": "search", "description": "Search"} + ], + }, + ], + } + mock_api_client = mock.MagicMock() + mock_api_client.request.return_value = self._make_api_response( + agent_json + ) + + result = _evals_common._fetch_agent_config_dict( + mock_api_client, + "projects/p/locations/l/agents/my-agent", + ) + assert result.agent_id == "my-agent" + assert result.instruction == "You are helpful." + assert result.description == "A helpful agent." + assert result.agent_type == "gemini-2.0-flash" + # Built-in tools are mapped to typed Tool objects; function tool also included. + assert len(result.tools) == 3 + assert any(t.code_execution is not None for t in result.tools) + assert any(t.google_search is not None for t in result.tools) + assert any(t.function_declarations is not None for t in result.tools) + + def test_fetch_failure_returns_minimal_config(self): + """If the API call fails, returns just the agent_id.""" + mock_api_client = mock.MagicMock() + mock_api_client.request.side_effect = RuntimeError("not found") + + result = _evals_common._fetch_agent_config_dict( + mock_api_client, + "projects/p/locations/l/agents/broken-agent", + ) + assert result.agent_id == "broken-agent" + assert result.instruction is None + assert result.tools is None + + def test_empty_response_body_returns_minimal_config(self): + """If the response body is falsy, returns just the agent_id.""" + mock_api_client = mock.MagicMock() + resp = mock.MagicMock() + resp.body = None + mock_api_client.request.return_value = resp + + result = _evals_common._fetch_agent_config_dict( + mock_api_client, + "projects/p/locations/l/agents/my-agent", + ) + assert result.agent_id == "my-agent" + assert result.instruction is None + assert result.tools is None + + def test_empty_resource_name_uses_default_agent_id(self): + """An empty resource name falls back to 'agent' as the agent_id.""" + mock_api_client = mock.MagicMock() + mock_api_client.request.side_effect = RuntimeError("irrelevant") + + result = _evals_common._fetch_agent_config_dict( + mock_api_client, + "", + ) + assert result.agent_id == "agent"