diff --git a/agentplatform/_genai/_evals_common.py b/agentplatform/_genai/_evals_common.py index 608d5a1db7..c68c1d8d1f 100644 --- a/agentplatform/_genai/_evals_common.py +++ b/agentplatform/_genai/_evals_common.py @@ -557,22 +557,335 @@ def _resolve_inference_configs( def _is_gemini_agent_resource(agent: str) -> bool: - """Returns True if `agent` is a Gemini Agent resource name. + """Returns True if `agent` is a Gemini Agent resource name. + + A Gemini Agent resource name has the format + `projects/{project}/locations/{location}/agents/{agent}`, as opposed to an + Agent Engine resource name which uses `.../reasoningEngines/{id}`. + """ + parts = agent.split("/") + return ( + len(parts) == 6 + and parts[0] == "projects" + and parts[2] == "locations" + and parts[4] == "agents" + and bool(parts[1]) + and bool(parts[3]) + and bool(parts[5]) + ) + + +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: + # Append this event's text parts to the previous event. + 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 - A Gemini Agent resource name has the format - `projects/{project}/locations/{location}/agents/{agent}`, as opposed to an - Agent Engine resource name which uses `.../reasoningEngines/{id}`. - """ - parts = agent.split("/") - return ( - len(parts) == 6 - and parts[0] == "projects" - and parts[2] == "locations" - and parts[4] == "agents" - and bool(parts[1]) - and bool(parts[3]) - and bool(parts[5]) - ) + +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 from the Interactions API and + converts it to AgentData using ``_interaction_dict_to_agent_data``. + Returns a new list of EvaluationDataset objects with the resolved data + so that ``show()`` can render the System Topology and Conversation + Trace sections. + + Args: + api_client: The API client used to fetch interactions. + 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. + # Format: projects/{p}/locations/{l}/interactions/{id} + 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 + ) + # Fetch the interaction trace (conversation steps). + 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] + + # Fetch the agent config (instruction, tools, description) + # from the Agent API so the JS can render System Topology. + # This is best-effort; if it fails we still show the trace. + 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) + # The Agent proto uses snake_case json_names. + 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"): + # The Agent API returns tools with a `type` field + # (e.g. "code_execution", "google_search") that + # genai_types.Tool rejects (extra="forbid"). + # Strip `type` and filter out tools that become + # empty (built-in tools without function_declarations). + 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 consecutive text parts within each event so the + # trace renderer does not split a single response into + # multiple visual blocks. + _merge_text_parts_in_agent_data(agent_data) + + # Create a new EvalCase with the resolved 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( @@ -1803,131 +2116,137 @@ def _execute_evaluation( # type: ignore[no-untyped-def] evaluation_service_qps: Optional[float] = None, **kwargs, ) -> types.EvaluationResult: - """Evaluates a dataset using the provided metrics. - - Args: - api_client: The API client. - dataset: The dataset to evaluate. - metrics: The metrics to evaluate the dataset against. - dataset_schema: The schema of the dataset. - dest: The destination to save the evaluation results. - location: The location to use for the evaluation. If not specified, the - location configured in the client will be used. - evaluation_service_qps: The rate limit (queries per second) for calls - to the evaluation service. Defaults to 10. Increase this value if - your project has a higher EvaluateInstances API quota. - **kwargs: Extra arguments to pass to evaluation, such as `agent_info`. - - Returns: - The evaluation result. - """ + """Evaluates a dataset using the provided metrics. - if location: - api_client = _get_api_client_with_location(api_client, location) - - logger.info("Preparing dataset(s) and metrics...") - if isinstance(dataset, types.EvaluationDataset): - dataset_list = [dataset] - elif isinstance(dataset, list): - for item in dataset: - if not isinstance(item, types.EvaluationDataset): - raise TypeError( - f"Unsupported dataset type: {type(item)}. " - "Must be EvaluationDataset." - ) - dataset_list = dataset - else: + Args: + api_client: The API client. + dataset: The dataset to evaluate. + metrics: The metrics to evaluate the dataset against. + dataset_schema: The schema of the dataset. + dest: The destination to save the evaluation results. + location: The location to use for the evaluation. If not specified, the + location configured in the client will be used. + evaluation_service_qps: The rate limit (queries per second) for calls to + the evaluation service. Defaults to 10. Increase this value if your + project has a higher EvaluateInstances API quota. + **kwargs: Extra arguments to pass to evaluation, such as `agent_info`. + + Returns: + The evaluation result. + """ + + if location: + api_client = _get_api_client_with_location(api_client, location) + + logger.info("Preparing dataset(s) and metrics...") + if isinstance(dataset, types.EvaluationDataset): + dataset_list = [dataset] + elif isinstance(dataset, list): + for item in dataset: + if not isinstance(item, types.EvaluationDataset): raise TypeError( - f"Unsupported dataset type: {type(dataset)}. Must be an" - " EvaluationDataset or a list of EvaluationDataset." + f"Unsupported dataset type: {type(item)}. " + "Must be EvaluationDataset." ) - original_candidate_names = [ - ds.candidate_name or f"candidate_{i + 1}" for i, ds in enumerate(dataset_list) - ] - name_counts = collections.Counter(original_candidate_names) - deduped_candidate_names = [] - current_name_counts: collections.defaultdict[Any, int] = collections.defaultdict( - int - ) - - for name in original_candidate_names: - if name_counts[name] > 1: - current_name_counts[name] += 1 - deduped_candidate_names.append(f"{name} #{current_name_counts[name]}") - else: - deduped_candidate_names.append(name) - - loader = _evals_utils.EvalDatasetLoader(api_client=api_client) - - agent_info = kwargs.get("agent_info", None) - validated_agent_info = None - if agent_info: - if isinstance(agent_info, dict): - validated_agent_info = types.evals.AgentInfo.model_validate(agent_info) - elif isinstance(agent_info, types.evals.AgentInfo): - validated_agent_info = agent_info - else: - raise TypeError( - "agent_info values must be of type types.evals.AgentInfo or dict," - f" but got {type(agent_info)}'" - ) - - processed_eval_dataset, num_response_candidates = _resolve_dataset_inputs( - dataset=dataset_list, - dataset_schema=dataset_schema, - loader=loader, - agent_info=validated_agent_info, + dataset_list = dataset + else: + raise TypeError( + f"Unsupported dataset type: {type(dataset)}. Must be an" + " EvaluationDataset or a list of EvaluationDataset." ) + original_candidate_names = [ + ds.candidate_name or f"candidate_{i + 1}" + for i, ds in enumerate(dataset_list) + ] + name_counts = collections.Counter(original_candidate_names) + deduped_candidate_names = [] + current_name_counts: collections.defaultdict[Any, int] = ( + collections.defaultdict(int) + ) + + for name in original_candidate_names: + if name_counts[name] > 1: + current_name_counts[name] += 1 + deduped_candidate_names.append(f"{name} #{current_name_counts[name]}") + else: + deduped_candidate_names.append(name) - resolved_metrics = _resolve_metrics(metrics, api_client) - - evaluation_run_config = _evals_metric_handlers.EvaluationRunConfig( - evals_module=evals.Evals(api_client_=api_client), - dataset=processed_eval_dataset, - metrics=resolved_metrics, - num_response_candidates=num_response_candidates, - ) + loader = _evals_utils.EvalDatasetLoader(api_client=api_client) - logger.info("Running Metric Computation...") - t1 = time.perf_counter() - evaluation_result = _evals_metric_handlers.compute_metrics_and_aggregate( - evaluation_run_config, - evaluation_service_qps=evaluation_service_qps, + agent_info = kwargs.get("agent_info", None) + validated_agent_info = None + if agent_info: + if isinstance(agent_info, dict): + validated_agent_info = types.evals.AgentInfo.model_validate(agent_info) + elif isinstance(agent_info, types.evals.AgentInfo): + validated_agent_info = agent_info + else: + raise TypeError( + "agent_info values must be of type types.evals.AgentInfo or dict," + f" but got {type(agent_info)}'" + ) + + processed_eval_dataset, num_response_candidates = _resolve_dataset_inputs( + dataset=dataset_list, + dataset_schema=dataset_schema, + loader=loader, + agent_info=validated_agent_info, + ) + + resolved_metrics = _resolve_metrics(metrics, api_client) + + evaluation_run_config = _evals_metric_handlers.EvaluationRunConfig( + evals_module=evals.Evals(api_client_=api_client), + dataset=processed_eval_dataset, + metrics=resolved_metrics, + num_response_candidates=num_response_candidates, + ) + + logger.info("Running Metric Computation...") + t1 = time.perf_counter() + evaluation_result = _evals_metric_handlers.compute_metrics_and_aggregate( + evaluation_run_config, + evaluation_service_qps=evaluation_service_qps, + ) + 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 + + if not evaluation_result.metadata: + evaluation_result.metadata = types.EvaluationRunMetadata() + + evaluation_result.metadata.creation_timestamp = datetime.datetime.now( + datetime.timezone.utc + ) + + if deduped_candidate_names: + evaluation_result.metadata.candidate_names = deduped_candidate_names + + logger.info("Evaluation run completed.") + + if dest: + uploaded_path = _gcs_utils.GcsUtils( + api_client=api_client + ).upload_json_to_prefix( + data=evaluation_result.model_dump( + mode="json", + exclude_none=True, + exclude={"evaluation_dataset"}, + ), + gcs_dest_prefix=dest, + filename_prefix="evaluation_result", ) - t2 = time.perf_counter() - logger.info("Evaluation took: %f seconds", t2 - t1) - - evaluation_result.evaluation_dataset = dataset_list - evaluation_result.agent_info = validated_agent_info - - if not evaluation_result.metadata: - evaluation_result.metadata = types.EvaluationRunMetadata() - - evaluation_result.metadata.creation_timestamp = datetime.datetime.now( - datetime.timezone.utc + logger.info( + "Evaluation results uploaded successfully to GCS: %s", uploaded_path ) - - if deduped_candidate_names: - evaluation_result.metadata.candidate_names = deduped_candidate_names - - logger.info("Evaluation run completed.") - - if dest: - uploaded_path = _gcs_utils.GcsUtils( - api_client=api_client - ).upload_json_to_prefix( - data=evaluation_result.model_dump( - mode="json", - exclude_none=True, - exclude={"evaluation_dataset"}, - ), - gcs_dest_prefix=dest, - filename_prefix="evaluation_result", - ) - logger.info( - "Evaluation results uploaded successfully to GCS: %s", uploaded_path - ) - return evaluation_result + return evaluation_result def _get_session_inputs(row: pd.Series) -> types.evals.SessionInput: 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