diff --git a/agentplatform/_genai/_evals_common.py b/agentplatform/_genai/_evals_common.py index 608d5a1db7..9c9b5ae910 100644 --- a/agentplatform/_genai/_evals_common.py +++ b/agentplatform/_genai/_evals_common.py @@ -420,6 +420,311 @@ def _extract_response_from_completed_trace( return event_dicts +def _has_interactions_data_source( + eval_cases: list[types.EvalCase], +) -> bool: + """Returns True if any EvalCase has interactions_data_source set.""" + return any( + getattr(case, "interactions_data_source", None) is not None + for case in eval_cases + ) + + +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. + + 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_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 ID from the resource name. + # Format: projects/{p}/locations/{l}/interactions/{id} + parts = ids.interaction.split("/") + if len(parts) >= 6 and parts[4] == "interactions": + interaction_id = parts[5] + else: + interaction_id = ids.interaction + + 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) + + # Derive agent short name from the resource name. + gemini_cfg = getattr(ids, "gemini_agent_config", None) + agent_name = ( + getattr(gemini_cfg, "gemini_agent", None) + if gemini_cfg else None + ) + agent_id = "agent" + if agent_name: + agent_parts = agent_name.split("/") + if len(agent_parts) >= 2: + agent_id = agent_parts[-1] + + # Best-effort: fetch the agent config (instruction, tools, + # description) from the Agent API so the display can render + # the System Topology section. + 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"): + # Strip the ``type`` field from each tool + # (e.g. "code_execution", "google_search") + # since genai_types.Tool rejects it + # (extra="forbid"), 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 events and parts so multi-paragraph + # responses render as a single block in the trace display. + _merge_text_parts_in_agent_data(agent_data) + + resolved_cases.append(types.EvalCase(agent_data=agent_data)) + + return resolved_cases + + def _resolve_dataset( api_client: BaseApiClient, dataset: Union[types.EvaluationRunDataSource, types.EvaluationDataset], @@ -428,6 +733,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: diff --git a/tests/unit/agentplatform/genai/test_evals.py b/tests/unit/agentplatform/genai/test_evals.py index eda21f7e31..708a7d1825 100644 --- a/tests/unit/agentplatform/genai/test_evals.py +++ b/tests/unit/agentplatform/genai/test_evals.py @@ -9563,6 +9563,139 @@ 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 = { + "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 = { + "steps": [ + { + "type": "function_call", + "name": "get_weather", + "arguments": {"city": "NYC"}, + "id": "call_1", + }, + { + "type": "function_result", + "name": "get_weather", + "callId": "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" + ) + + class TestRateLimiter: """Tests for the RateLimiter class in _evals_utils."""