From f6351d036fe380c19e1bf0889d5c76adb9e75f85 Mon Sep 17 00:00:00 2001 From: A Vertex SDK engineer Date: Wed, 8 Jul 2026 13:08:09 -0700 Subject: [PATCH] feat: Resolve Interactions API data for agent trace display in eval results. PiperOrigin-RevId: 944658614 --- agentplatform/_genai/_evals_common.py | 325 +++++++++++++++++++ tests/unit/agentplatform/genai/test_evals.py | 284 ++++++++++++++++ 2 files changed, 609 insertions(+) diff --git a/agentplatform/_genai/_evals_common.py b/agentplatform/_genai/_evals_common.py index 608d5a1db7..46a39bcd10 100644 --- a/agentplatform/_genai/_evals_common.py +++ b/agentplatform/_genai/_evals_common.py @@ -575,6 +575,326 @@ def _is_gemini_agent_resource(agent: str) -> bool: ) +def _interaction_dict_to_agent_data( + interaction: dict[str, Any], +) -> dict[str, Any]: + """Converts an Interaction API JSON response to an AgentData-compatible dict. + + Maps the flat list of Interaction steps (user_input, model_output, + function_call, function_result, etc.) into a single ConversationTurn + with AgentEvents, matching the AgentData structure expected by the + evaluation pipeline. + + NOTE: This function is shared with cl/944601748. When that CL lands + first, this copy should be removed during the rebase. + + Args: + interaction: A dict from the Interactions API GET response. + + Returns: + A dict matching the AgentData schema with a single turn containing + all events from the interaction. + """ + events = [] + for step in interaction.get("steps", []): + step_type = step.get("type") + + if step_type == "user_input": + parts = [] + for content_item in step.get("content", []): + if content_item.get("type") == "text": + parts.append({"text": content_item.get("text", "")}) + if parts: + events.append({ + "author": "user", + "content": {"role": "user", "parts": parts}, + }) + + elif step_type == "model_output": + parts = [] + for content_item in step.get("content", []): + if content_item.get("type") == "text": + parts.append({"text": content_item.get("text", "")}) + if parts: + events.append({ + "author": "agent", + "content": {"role": "model", "parts": parts}, + }) + + elif step_type == "function_call": + events.append({ + "author": "agent", + "content": { + "role": "model", + "parts": [{ + "function_call": { + "name": step.get("name", ""), + "args": step.get("arguments", {}), + "id": step.get("id", ""), + }, + }], + }, + }) + + elif step_type == "function_result": + result = step.get("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 "" + events.append({ + "author": "user", + "content": { + "role": "user", + "parts": [{ + "function_response": { + "name": step.get("name", ""), + "response": {"result": result_str}, + "id": step.get("callId", step.get("call_id", "")), + }, + }], + }, + }) + + agent_data: dict[str, Any] = { + "turns": [{ + "turn_index": 0, + "events": events, + }], + } + return agent_data + + +def _merge_text_parts_in_agent_data( + agent_data: dict[str, Any], +) -> 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: A dict matching the AgentData schema. + """ + for turn in agent_data.get("turns", []): + events = turn.get("events") + if not events: + continue + + # --- Pass 1: merge consecutive text-only events from the same author --- + merged_events: list[dict[str, Any]] = [] + for event in events: + parts = (event.get("content") or {}).get("parts", []) + is_text_only = parts and all( + "text" in p and len(p) == 1 for p in parts + ) + if ( + merged_events + and is_text_only + and event.get("author") == merged_events[-1].get("author") + ): + prev_parts = ( + merged_events[-1].get("content") or {} + ).get("parts", []) + prev_is_text_only = prev_parts and all( + "text" in p and len(p) == 1 for p in prev_parts + ) + if prev_is_text_only: + 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.get("content") + if not content: + continue + parts = content.get("parts") + if not parts or len(parts) <= 1: + continue + merged_parts: list[dict[str, Any]] = [] + text_buffer: list[str] = [] + for part in parts: + if "text" in part and len(part) == 1: + text_buffer.append(part["text"]) + else: + if text_buffer: + merged_parts.append({"text": "\n".join(text_buffer)}) + text_buffer = [] + merged_parts.append(part) + if text_buffer: + merged_parts.append({"text": "\n".join(text_buffer)}) + content["parts"] = merged_parts + + +def _resolve_interactions_for_display( + api_client: BaseApiClient, + dataset_list: list[types.EvaluationDataset], +) -> list[types.EvaluationDataset]: + """Resolves interactions_data_source on EvalCases for display. + + For each EvalCase that has ``interactions_data_source`` set but no + ``agent_data``, fetches the Interaction and Agent config from the + API, converts to AgentData, and populates it on the EvalCase so + that ``show()`` can render the System Topology and Conversation + Trace sections. + + Args: + api_client: The API client used to fetch interactions and agents. + dataset_list: The original evaluation datasets. + + Returns: + A list of EvaluationDatasets with agent_data populated from + resolved interactions. + """ + resolved_datasets = [] + for dataset in dataset_list: + if not dataset.eval_cases: + resolved_datasets.append(dataset) + continue + + resolved_cases = [] + any_resolved = False + for case in dataset.eval_cases: + ids = getattr(case, "interactions_data_source", None) + existing_agent_data = getattr(case, "agent_data", None) + if ids and not existing_agent_data: + interaction_name = getattr(ids, "interaction", None) + gemini_cfg = getattr(ids, "gemini_agent_config", None) + agent_name = ( + getattr(gemini_cfg, "gemini_agent", None) + if gemini_cfg else None + ) + + if interaction_name: + try: + # Extract the interaction ID from the resource name. + parts = interaction_name.split("/") + if len(parts) >= 6 and parts[4] == "interactions": + interaction_id = parts[5] + else: + interaction_id = interaction_name + + logger.info( + "Fetching interaction %s for display.", + interaction_name, + ) + path = f"interactions/{interaction_id}" + response = api_client.request("get", path, {}, None) + if not response.body: + logger.warning( + "Empty response fetching interaction %s.", + interaction_name, + ) + resolved_cases.append(case) + continue + interaction_dict = json.loads(response.body) + + agent_data = _interaction_dict_to_agent_data( + interaction_dict + ) + + # Derive the agent short name from the resource name. + agent_id = "agent" + if agent_name: + agent_parts = agent_name.split("/") + if len(agent_parts) >= 2: + agent_id = agent_parts[-1] + + # Best-effort: fetch agent config (instruction, + # tools, description) from the Agent API. + agent_config: dict[str, Any] = {"agent_id": agent_id} + if agent_name: + try: + agent_short_id = agent_name.split("/")[-1] + agent_resp = api_client.request( + "get", f"agents/{agent_short_id}", {}, None + ) + if agent_resp.body: + agent_dict = json.loads(agent_resp.body) + if agent_dict.get("system_instruction"): + agent_config["instruction"] = ( + agent_dict["system_instruction"] + ) + if agent_dict.get("description"): + agent_config["description"] = ( + agent_dict["description"] + ) + if agent_dict.get("base_agent"): + agent_config["agent_type"] = ( + agent_dict["base_agent"] + ) + if agent_dict.get("tools"): + cleaned_tools = [] + for tool in agent_dict["tools"]: + if isinstance(tool, dict): + tool = { + k: v + for k, v in tool.items() + if k != "type" + } + if tool: + cleaned_tools.append(tool) + if cleaned_tools: + agent_config["tools"] = ( + cleaned_tools + ) + except Exception as e: # pylint: disable=broad-exception-caught + logger.warning( + "Failed to fetch agent config for '%s'" + " (continuing without it): %s", + agent_name, + e, + ) + agent_data["agents"] = {agent_id: agent_config} + + _merge_text_parts_in_agent_data(agent_data) + + case_dict = case.model_dump( + mode="python", exclude_none=True + ) + case_dict["agent_data"] = agent_data + resolved_cases.append(types.EvalCase(**case_dict)) + any_resolved = True + continue + except Exception as e: # pylint: disable=broad-exception-caught + logger.warning( + "Failed to resolve interaction %s for display: %s", + interaction_name, + e, + ) + + resolved_cases.append(case) + + if any_resolved: + ds_dict = dataset.model_dump(mode="python", exclude_none=True) + ds_dict.pop("eval_cases", None) + resolved_datasets.append( + types.EvaluationDataset( + eval_cases=resolved_cases, **ds_dict + ) + ) + else: + resolved_datasets.append(dataset) + + return resolved_datasets + + def _add_evaluation_run_labels( labels: Optional[dict[str, str]] = None, agent: Optional[str] = None, @@ -1897,6 +2217,11 @@ def _execute_evaluation( # type: ignore[no-untyped-def] t2 = time.perf_counter() logger.info("Evaluation took: %f seconds", t2 - t1) + # Resolve interactions_data_source to agent_data for display. + # This fetches Interaction trace data client-side so that show() can + # render the System Topology and Conversation Trace sections. + dataset_list = _resolve_interactions_for_display(api_client, dataset_list) + evaluation_result.evaluation_dataset = dataset_list evaluation_result.agent_info = validated_agent_info diff --git a/tests/unit/agentplatform/genai/test_evals.py b/tests/unit/agentplatform/genai/test_evals.py index eda21f7e31..a030dd0e57 100644 --- a/tests/unit/agentplatform/genai/test_evals.py +++ b/tests/unit/agentplatform/genai/test_evals.py @@ -9989,3 +9989,287 @@ 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 TestResolveInteractionsForDisplay: + """Tests for _resolve_interactions_for_display.""" + + def _make_api_response(self, body_dict): + """Create a mock API response with a JSON body.""" + resp = mock.MagicMock() + resp.body = json.dumps(body_dict) + return resp + + def test_no_interactions_data_source_is_noop(self): + """Cases without interactions_data_source are returned as-is.""" + case = agentplatform_genai_types.EvalCase( + prompt=genai_types.Content( + parts=[genai_types.Part(text="hello")] + ), + ) + dataset = agentplatform_genai_types.EvaluationDataset( + eval_cases=[case] + ) + mock_api_client = mock.MagicMock() + result = _evals_common._resolve_interactions_for_display( + mock_api_client, [dataset] + ) + assert result[0].eval_cases[0] is case + mock_api_client.request.assert_not_called() + + def test_resolves_interaction_to_agent_data(self): + """When interactions_data_source is present, agent_data is populated.""" + case = agentplatform_genai_types.EvalCase( + interactions_data_source=agentplatform_genai_types.InteractionsDataSource( + interaction=( + "projects/p/locations/l/interactions/test-id" + ), + gemini_agent_config=agentplatform_genai_types.GeminiAgentConfig( + gemini_agent="projects/p/locations/l/agents/my-agent", + ), + ) + ) + dataset = agentplatform_genai_types.EvaluationDataset( + eval_cases=[case] + ) + + interaction_json = { + "steps": [ + { + "type": "user_input", + "content": [{"type": "text", "text": "hello"}], + }, + { + "type": "model_output", + "content": [{"type": "text", "text": "hi there"}], + }, + ] + } + agent_json = {"system_instruction": "Be helpful"} + + def mock_request(method, path, *args, **kwargs): + if "interactions/" in path: + return self._make_api_response(interaction_json) + if "agents/" in path: + return self._make_api_response(agent_json) + return self._make_api_response({}) + + mock_api_client = mock.MagicMock() + mock_api_client.request.side_effect = mock_request + + result = _evals_common._resolve_interactions_for_display( + mock_api_client, [dataset] + ) + resolved_case = result[0].eval_cases[0] + assert resolved_case.agent_data is not None + assert "my-agent" in resolved_case.agent_data.agents + + # Verify the interaction and agent were fetched. + assert mock_api_client.request.call_count == 2 + mock_api_client.request.assert_any_call( + "get", "interactions/test-id", {}, None + ) + mock_api_client.request.assert_any_call( + "get", "agents/my-agent", {}, None + ) + + def test_agents_map_populated_with_full_config(self): + """The agents dict includes instruction and tools from the Agent API.""" + case = agentplatform_genai_types.EvalCase( + interactions_data_source=agentplatform_genai_types.InteractionsDataSource( + interaction="projects/p/locations/l/interactions/i1", + gemini_agent_config=agentplatform_genai_types.GeminiAgentConfig( + gemini_agent="projects/p/locations/l/agents/weather-bot", + ), + ) + ) + dataset = agentplatform_genai_types.EvaluationDataset( + eval_cases=[case] + ) + interaction_json = {"steps": []} + agent_json = { + "system_instruction": "You are a weather assistant.", + "description": "Helps with weather queries.", + "base_agent": "gemini-2.0-flash", + "tools": [ + {"type": "code_execution"}, + {"type": "google_search"}, + { + "type": "function", + "function_declarations": [ + {"name": "get_weather", "description": "Get weather"} + ], + }, + ], + } + + def mock_request(method, path, *args, **kwargs): + if "interactions/" in path: + return self._make_api_response(interaction_json) + if "agents/" in path: + return self._make_api_response(agent_json) + return self._make_api_response({}) + + mock_api_client = mock.MagicMock() + mock_api_client.request.side_effect = mock_request + + result = _evals_common._resolve_interactions_for_display( + mock_api_client, [dataset] + ) + resolved_case = result[0].eval_cases[0] + agent_cfg = resolved_case.agent_data.agents["weather-bot"] + assert agent_cfg.agent_id == "weather-bot" + assert agent_cfg.instruction == "You are a weather assistant." + assert agent_cfg.description == "Helps with weather queries." + assert agent_cfg.agent_type == "gemini-2.0-flash" + # Built-in tools (code_execution, google_search) are filtered out; + # only the tool with function_declarations remains. + assert agent_cfg.tools is not None + assert len(agent_cfg.tools) == 1 + + def test_consecutive_model_output_steps_merged(self): + """Multiple consecutive model_output steps are merged into one event.""" + case = agentplatform_genai_types.EvalCase( + interactions_data_source=agentplatform_genai_types.InteractionsDataSource( + interaction="projects/p/locations/l/interactions/i1", + gemini_agent_config=agentplatform_genai_types.GeminiAgentConfig( + gemini_agent="projects/p/locations/l/agents/a", + ), + ) + ) + dataset = agentplatform_genai_types.EvaluationDataset( + eval_cases=[case] + ) + # Simulate multiple model_output steps (one per paragraph). + interaction_json = { + "steps": [ + { + "type": "model_output", + "content": [{"type": "text", "text": "Paragraph 1."}], + }, + { + "type": "model_output", + "content": [{"type": "text", "text": "Paragraph 2."}], + }, + { + "type": "model_output", + "content": [{"type": "text", "text": "Paragraph 3."}], + }, + ] + } + + def mock_request(method, path, *args, **kwargs): + if "interactions/" in path: + return self._make_api_response(interaction_json) + return self._make_api_response({}) + + mock_api_client = mock.MagicMock() + mock_api_client.request.side_effect = mock_request + + result = _evals_common._resolve_interactions_for_display( + mock_api_client, [dataset] + ) + resolved_case = result[0].eval_cases[0] + events = resolved_case.agent_data.turns[0].events + # All three model_output steps should be merged into one event. + assert len(events) == 1 + text_parts = [ + p for p in events[0].content.parts if hasattr(p, "text") and p.text + ] + assert len(text_parts) == 1 + assert "Paragraph 1." in text_parts[0].text + assert "Paragraph 2." in text_parts[0].text + assert "Paragraph 3." in text_parts[0].text + + def test_existing_agent_data_not_overwritten(self): + """Cases that already have agent_data are not resolved again.""" + existing_ad = agentplatform_genai_types.evals.AgentData( + agents={"existing": agentplatform_genai_types.evals.AgentConfig( + agent_id="existing" + )}, + ) + case = agentplatform_genai_types.EvalCase( + agent_data=existing_ad, + interactions_data_source=agentplatform_genai_types.InteractionsDataSource( + interaction="projects/p/locations/l/interactions/test-id", + ), + ) + dataset = agentplatform_genai_types.EvaluationDataset( + eval_cases=[case] + ) + + mock_api_client = mock.MagicMock() + result = _evals_common._resolve_interactions_for_display( + mock_api_client, [dataset] + ) + # Should not have called request since agent_data already exists. + mock_api_client.request.assert_not_called() + assert result[0].eval_cases[0].agent_data is existing_ad + + def test_interaction_fetch_failure_returns_original(self): + """If fetching the interaction fails, the original case is returned.""" + case = agentplatform_genai_types.EvalCase( + interactions_data_source=agentplatform_genai_types.InteractionsDataSource( + interaction="projects/p/locations/l/interactions/bad-id", + gemini_agent_config=agentplatform_genai_types.GeminiAgentConfig( + gemini_agent="projects/p/locations/l/agents/my-agent", + ), + ) + ) + dataset = agentplatform_genai_types.EvaluationDataset( + eval_cases=[case] + ) + + mock_api_client = mock.MagicMock() + mock_api_client.request.side_effect = RuntimeError("API error") + + result = _evals_common._resolve_interactions_for_display( + mock_api_client, [dataset] + ) + # Original case returned (no agent_data). + resolved_case = result[0].eval_cases[0] + assert resolved_case.agent_data is None + + def test_agent_fetch_failure_still_shows_trace(self): + """If fetching the agent config fails, trace is still populated.""" + case = agentplatform_genai_types.EvalCase( + interactions_data_source=agentplatform_genai_types.InteractionsDataSource( + interaction="projects/p/locations/l/interactions/i1", + gemini_agent_config=agentplatform_genai_types.GeminiAgentConfig( + gemini_agent="projects/p/locations/l/agents/bad-agent", + ), + ) + ) + dataset = agentplatform_genai_types.EvaluationDataset( + eval_cases=[case] + ) + interaction_json = { + "steps": [ + { + "type": "user_input", + "content": [{"type": "text", "text": "hi"}], + }, + ] + } + + def mock_request(method, path, *args, **kwargs): + if "interactions/" in path: + return self._make_api_response(interaction_json) + if "agents/" in path: + raise RuntimeError("Agent not found") + return self._make_api_response({}) + + mock_api_client = mock.MagicMock() + mock_api_client.request.side_effect = mock_request + + result = _evals_common._resolve_interactions_for_display( + mock_api_client, [dataset] + ) + resolved_case = result[0].eval_cases[0] + # Trace should still be populated even though agent config failed. + assert resolved_case.agent_data is not None + assert len(resolved_case.agent_data.turns) == 1 + # Agent config is minimal (just agent_id, no instruction/tools). + agent_cfg = resolved_case.agent_data.agents["bad-agent"] + assert agent_cfg.agent_id == "bad-agent" + assert agent_cfg.instruction is None