feat: Support creating evaluation runs from pre-existing Interactions API data

PiperOrigin-RevId: 944601748

Author: vertex-sdk-bot

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: