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

PiperOrigin-RevId: 944601748

Author: vertex-sdk-bot

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
@@ -420,6 +425,326 @@ def _extract_response_from_completed_trace(
     return event_dicts
 
 
+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."""
+    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 _step_to_agent_event(step: Any) -> Optional[types.evals.AgentEvent]:
+    """Converts a typed GenAI SDK Interaction step to an AgentEvent."""
+    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 _interaction_steps_to_events(
+    steps: list[Any],
+) -> list[tuple[types.evals.AgentEvent, type]]:
+    """Converts a list of typed Interaction steps to AgentEvents."""
+    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, then groups steps
+    into ConversationTurns (each UserInputStep starts a new turn).
+    """
+    typed_interaction = interaction_types.Interaction.model_validate(interaction)
+    all_events = _interaction_steps_to_events(typed_interaction.steps or [])
+
+    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 same-author text events and text parts 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 _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
+            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(genai_types.Tool.model_validate(tool))
+                if cleaned_tools:
+                    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_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 _resolve_dataset(
     api_client: BaseApiClient,
     dataset: Union[types.EvaluationRunDataSource, types.EvaluationDataset],
@@ -428,6 +753,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: