evals.py

# Copyright 2025 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

# Code generated by the Google Gen AI SDK generator DO NOT EDIT.

# Test this should trigger the diff test

import json
import logging
from typing import Any, Callable, Optional, Union, cast
from urllib.parse import urlencode
import uuid

from google.genai import _api_module
from google.genai import _common
from google.genai import types as genai_types
from google.genai._common import get_value_by_path as getv
from google.genai._common import set_value_by_path as setv
import pandas as pd

from . import _evals_common
from . import _evals_utils
from . import _transformers as t
from . import types
from .types import evals as evals_types

try:
    from google.adk.agents import LlmAgent
except ImportError:
    LlmAgent = None


logger = logging.getLogger("agentplatform_genai.evals")


def _CreateEvaluationExperimentParameters_to_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["display_name"]) is not None:
        setv(to_object, ["displayName"], getv(from_object, ["display_name"]))

    if getv(from_object, ["labels"]) is not None:
        setv(to_object, ["labels"], getv(from_object, ["labels"]))

    if getv(from_object, ["merge_strategy"]) is not None:
        setv(to_object, ["mergeStrategy"], getv(from_object, ["merge_strategy"]))

    if getv(from_object, ["metadata"]) is not None:
        setv(to_object, ["metadata"], getv(from_object, ["metadata"]))

    if getv(from_object, ["config"]) is not None:
        setv(to_object, ["config"], getv(from_object, ["config"]))

    return to_object


def _CreateEvaluationItemParameters_to_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["evaluation_item_type"]) is not None:
        setv(
            to_object,
            ["evaluationItemType"],
            getv(from_object, ["evaluation_item_type"]),
        )

    if getv(from_object, ["gcs_uri"]) is not None:
        setv(to_object, ["gcsUri"], getv(from_object, ["gcs_uri"]))

    if getv(from_object, ["display_name"]) is not None:
        setv(to_object, ["displayName"], getv(from_object, ["display_name"]))

    if getv(from_object, ["config"]) is not None:
        setv(to_object, ["config"], getv(from_object, ["config"]))

    return to_object


def _CreateEvaluationMetricParameters_to_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["display_name"]) is not None:
        setv(to_object, ["displayName"], getv(from_object, ["display_name"]))

    if getv(from_object, ["description"]) is not None:
        setv(to_object, ["description"], getv(from_object, ["description"]))

    if getv(from_object, ["metric"]) is not None:
        setv(
            to_object,
            ["metric"],
            t.t_metric_for_registry(getv(from_object, ["metric"])),
        )

    if getv(from_object, ["config"]) is not None:
        setv(to_object, ["config"], getv(from_object, ["config"]))

    return to_object


def _CreateEvaluationRunParameters_to_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["name"]) is not None:
        setv(to_object, ["name"], getv(from_object, ["name"]))

    if getv(from_object, ["display_name"]) is not None:
        setv(to_object, ["displayName"], getv(from_object, ["display_name"]))

    if getv(from_object, ["data_source"]) is not None:
        setv(to_object, ["dataSource"], getv(from_object, ["data_source"]))

    if getv(from_object, ["evaluation_config"]) is not None:
        setv(
            to_object,
            ["evaluationConfig"],
            _EvaluationRunConfig_to_vertex(
                getv(from_object, ["evaluation_config"]), to_object
            ),
        )

    if getv(from_object, ["labels"]) is not None:
        setv(to_object, ["labels"], getv(from_object, ["labels"]))

    if getv(from_object, ["inference_configs"]) is not None:
        setv(
            to_object,
            ["inferenceConfigs"],
            {
                k: _EvaluationRunInferenceConfig_to_vertex(v, to_object)
                for k, v in getv(from_object, ["inference_configs"]).items()
            },
        )

    if getv(from_object, ["config"]) is not None:
        setv(to_object, ["config"], getv(from_object, ["config"]))

    if getv(from_object, ["analysis_configs"]) is not None:
        setv(
            to_object,
            ["analysisConfigs"],
            [item for item in getv(from_object, ["analysis_configs"])],
        )

    return to_object


def _CreateEvaluationSetParameters_to_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["evaluation_items"]) is not None:
        setv(to_object, ["evaluationItems"], getv(from_object, ["evaluation_items"]))

    if getv(from_object, ["display_name"]) is not None:
        setv(to_object, ["displayName"], getv(from_object, ["display_name"]))

    if getv(from_object, ["config"]) is not None:
        setv(to_object, ["config"], getv(from_object, ["config"]))

    return to_object


def _CustomCodeExecutionSpec_from_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["evaluationFunction"]) is not None:
        setv(
            to_object,
            ["evaluation_function"],
            getv(from_object, ["evaluationFunction"]),
        )

    if getv(from_object, ["evaluation_function"]) is not None:
        setv(
            to_object,
            ["remote_custom_function"],
            getv(from_object, ["evaluation_function"]),
        )

    return to_object


def _CustomCodeExecutionSpec_to_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["evaluation_function"]) is not None:
        setv(
            to_object,
            ["evaluationFunction"],
            getv(from_object, ["evaluation_function"]),
        )

    if getv(from_object, ["remote_custom_function"]) is not None:
        setv(
            to_object,
            ["evaluation_function"],
            getv(from_object, ["remote_custom_function"]),
        )

    return to_object


def _DeleteEvaluationMetricParameters_to_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["metric_resource_name"]) is not None:
        setv(
            to_object,
            ["_url", "evaluation_metric"],
            getv(from_object, ["metric_resource_name"]),
        )

    if getv(from_object, ["config"]) is not None:
        setv(to_object, ["config"], getv(from_object, ["config"]))

    return to_object


def _EvaluateInstancesRequestParameters_to_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["bleu_input"]) is not None:
        setv(to_object, ["bleuInput"], getv(from_object, ["bleu_input"]))

    if getv(from_object, ["exact_match_input"]) is not None:
        setv(to_object, ["exactMatchInput"], getv(from_object, ["exact_match_input"]))

    if getv(from_object, ["rouge_input"]) is not None:
        setv(to_object, ["rougeInput"], getv(from_object, ["rouge_input"]))

    if getv(from_object, ["pointwise_metric_input"]) is not None:
        setv(
            to_object,
            ["pointwiseMetricInput"],
            getv(from_object, ["pointwise_metric_input"]),
        )

    if getv(from_object, ["pairwise_metric_input"]) is not None:
        setv(
            to_object,
            ["pairwiseMetricInput"],
            getv(from_object, ["pairwise_metric_input"]),
        )

    if getv(from_object, ["tool_call_valid_input"]) is not None:
        setv(
            to_object,
            ["toolCallValidInput"],
            getv(from_object, ["tool_call_valid_input"]),
        )

    if getv(from_object, ["tool_name_match_input"]) is not None:
        setv(
            to_object,
            ["toolNameMatchInput"],
            getv(from_object, ["tool_name_match_input"]),
        )

    if getv(from_object, ["tool_parameter_key_match_input"]) is not None:
        setv(
            to_object,
            ["toolParameterKeyMatchInput"],
            getv(from_object, ["tool_parameter_key_match_input"]),
        )

    if getv(from_object, ["tool_parameter_kv_match_input"]) is not None:
        setv(
            to_object,
            ["toolParameterKvMatchInput"],
            getv(from_object, ["tool_parameter_kv_match_input"]),
        )

    if getv(from_object, ["rubric_based_metric_input"]) is not None:
        setv(
            to_object,
            ["rubricBasedMetricInput"],
            _RubricBasedMetricInput_to_vertex(
                getv(from_object, ["rubric_based_metric_input"]), to_object
            ),
        )

    if getv(from_object, ["autorater_config"]) is not None:
        setv(to_object, ["autoraterConfig"], getv(from_object, ["autorater_config"]))

    if getv(from_object, ["metrics"]) is not None:
        setv(
            to_object,
            ["metrics"],
            [item for item in t.t_metrics(getv(from_object, ["metrics"]))],
        )

    if getv(from_object, ["instance"]) is not None:
        setv(
            to_object,
            ["instance"],
            _EvaluationInstance_to_vertex(getv(from_object, ["instance"]), to_object),
        )

    if getv(from_object, ["metric_sources"]) is not None:
        setv(
            to_object,
            ["metricSources"],
            [
                item
                for item in t.t_metric_sources(getv(from_object, ["metric_sources"]))
            ],
        )

    if getv(from_object, ["config"]) is not None:
        setv(to_object, ["config"], getv(from_object, ["config"]))

    return to_object


def _EvaluationInstance_to_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["prompt"]) is not None:
        setv(to_object, ["prompt"], getv(from_object, ["prompt"]))

    if getv(from_object, ["response"]) is not None:
        setv(to_object, ["response"], getv(from_object, ["response"]))

    if getv(from_object, ["reference"]) is not None:
        setv(to_object, ["reference"], getv(from_object, ["reference"]))

    if getv(from_object, ["other_data"]) is not None:
        setv(to_object, ["otherData"], getv(from_object, ["other_data"]))

    if getv(from_object, ["agent_data"]) is not None:
        setv(to_object, ["agent_eval_data"], getv(from_object, ["agent_data"]))

    if getv(from_object, ["rubric_groups"]) is not None:
        setv(to_object, ["rubricGroups"], getv(from_object, ["rubric_groups"]))

    if getv(from_object, ["interactions_data_source"]) is not None:
        setv(
            to_object,
            ["interactionsDataSource"],
            getv(from_object, ["interactions_data_source"]),
        )

    return to_object


def _EvaluationMetric_from_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["name"]) is not None:
        setv(to_object, ["name"], getv(from_object, ["name"]))

    if getv(from_object, ["displayName"]) is not None:
        setv(to_object, ["display_name"], getv(from_object, ["displayName"]))

    if getv(from_object, ["description"]) is not None:
        setv(to_object, ["description"], getv(from_object, ["description"]))

    if getv(from_object, ["metric"]) is not None:
        setv(
            to_object,
            ["metric"],
            _UnifiedMetric_from_vertex(getv(from_object, ["metric"]), to_object),
        )

    return to_object


def _EvaluationRunConfig_from_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["metrics"]) is not None:
        setv(
            to_object,
            ["metrics"],
            [
                _EvaluationRunMetric_from_vertex(item, to_object)
                for item in getv(from_object, ["metrics"])
            ],
        )

    if getv(from_object, ["outputConfig"]) is not None:
        setv(to_object, ["output_config"], getv(from_object, ["outputConfig"]))

    if getv(from_object, ["autoraterConfig"]) is not None:
        setv(to_object, ["autorater_config"], getv(from_object, ["autoraterConfig"]))

    if getv(from_object, ["promptTemplate"]) is not None:
        setv(to_object, ["prompt_template"], getv(from_object, ["promptTemplate"]))

    if getv(from_object, ["lossAnalysisConfig"]) is not None:
        setv(
            to_object,
            ["loss_analysis_config"],
            [item for item in getv(from_object, ["lossAnalysisConfig"])],
        )

    if getv(from_object, ["allowCrossRegionModel"]) is not None:
        setv(
            to_object,
            ["allow_cross_region_model"],
            getv(from_object, ["allowCrossRegionModel"]),
        )

    return to_object


def _EvaluationRunConfig_to_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["metrics"]) is not None:
        setv(
            to_object,
            ["metrics"],
            [
                _EvaluationRunMetric_to_vertex(item, to_object)
                for item in getv(from_object, ["metrics"])
            ],
        )

    if getv(from_object, ["output_config"]) is not None:
        setv(to_object, ["outputConfig"], getv(from_object, ["output_config"]))

    if getv(from_object, ["autorater_config"]) is not None:
        setv(to_object, ["autoraterConfig"], getv(from_object, ["autorater_config"]))

    if getv(from_object, ["prompt_template"]) is not None:
        setv(to_object, ["promptTemplate"], getv(from_object, ["prompt_template"]))

    if getv(from_object, ["loss_analysis_config"]) is not None:
        setv(
            to_object,
            ["lossAnalysisConfig"],
            [item for item in getv(from_object, ["loss_analysis_config"])],
        )

    if getv(from_object, ["allow_cross_region_model"]) is not None:
        setv(
            to_object,
            ["allowCrossRegionModel"],
            getv(from_object, ["allow_cross_region_model"]),
        )

    return to_object


def _EvaluationRunInferenceConfig_from_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["agentConfig"]) is not None:
        setv(to_object, ["agent_config"], getv(from_object, ["agentConfig"]))

    if getv(from_object, ["model"]) is not None:
        setv(to_object, ["model"], getv(from_object, ["model"]))

    if getv(from_object, ["promptTemplate"]) is not None:
        setv(to_object, ["prompt_template"], getv(from_object, ["promptTemplate"]))

    if getv(from_object, ["agentRunConfig"]) is not None:
        setv(to_object, ["agent_run_config"], getv(from_object, ["agentRunConfig"]))

    if getv(from_object, ["agents"]) is not None:
        setv(to_object, ["agent_configs"], getv(from_object, ["agents"]))

    return to_object


def _EvaluationRunInferenceConfig_to_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["agent_config"]) is not None:
        setv(to_object, ["agentConfig"], getv(from_object, ["agent_config"]))

    if getv(from_object, ["model"]) is not None:
        setv(to_object, ["model"], getv(from_object, ["model"]))

    if getv(from_object, ["prompt_template"]) is not None:
        setv(to_object, ["promptTemplate"], getv(from_object, ["prompt_template"]))

    if getv(from_object, ["agent_run_config"]) is not None:
        setv(to_object, ["agentRunConfig"], getv(from_object, ["agent_run_config"]))

    if getv(from_object, ["agent_configs"]) is not None:
        setv(to_object, ["agents"], getv(from_object, ["agent_configs"]))

    return to_object


def _EvaluationRunMetric_from_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["metric"]) is not None:
        setv(to_object, ["metric"], getv(from_object, ["metric"]))

    if getv(from_object, ["metricResourceName"]) is not None:
        setv(
            to_object,
            ["metric_resource_name"],
            getv(from_object, ["metricResourceName"]),
        )

    if getv(from_object, ["metricConfig"]) is not None:
        setv(
            to_object,
            ["metric_config"],
            _UnifiedMetric_from_vertex(getv(from_object, ["metricConfig"]), to_object),
        )

    return to_object


def _EvaluationRunMetric_to_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["metric"]) is not None:
        setv(to_object, ["metric"], getv(from_object, ["metric"]))

    if getv(from_object, ["metric_resource_name"]) is not None:
        setv(
            to_object,
            ["metricResourceName"],
            getv(from_object, ["metric_resource_name"]),
        )

    if getv(from_object, ["metric_config"]) is not None:
        setv(
            to_object,
            ["metricConfig"],
            _UnifiedMetric_to_vertex(getv(from_object, ["metric_config"]), to_object),
        )

    return to_object


def _EvaluationRun_from_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["name"]) is not None:
        setv(to_object, ["name"], getv(from_object, ["name"]))

    if getv(from_object, ["displayName"]) is not None:
        setv(to_object, ["display_name"], getv(from_object, ["displayName"]))

    if getv(from_object, ["metadata"]) is not None:
        setv(to_object, ["metadata"], getv(from_object, ["metadata"]))

    if getv(from_object, ["createTime"]) is not None:
        setv(to_object, ["create_time"], getv(from_object, ["createTime"]))

    if getv(from_object, ["completionTime"]) is not None:
        setv(to_object, ["completion_time"], getv(from_object, ["completionTime"]))

    if getv(from_object, ["state"]) is not None:
        setv(to_object, ["state"], getv(from_object, ["state"]))

    if getv(from_object, ["evaluationSetSnapshot"]) is not None:
        setv(
            to_object,
            ["evaluation_set_snapshot"],
            getv(from_object, ["evaluationSetSnapshot"]),
        )

    if getv(from_object, ["error"]) is not None:
        setv(to_object, ["error"], getv(from_object, ["error"]))

    if getv(from_object, ["dataSource"]) is not None:
        setv(to_object, ["data_source"], getv(from_object, ["dataSource"]))

    if getv(from_object, ["evaluationResults"]) is not None:
        setv(
            to_object,
            ["evaluation_run_results"],
            getv(from_object, ["evaluationResults"]),
        )

    if getv(from_object, ["evaluationConfig"]) is not None:
        setv(
            to_object,
            ["evaluation_config"],
            _EvaluationRunConfig_from_vertex(
                getv(from_object, ["evaluationConfig"]), to_object
            ),
        )

    if getv(from_object, ["inferenceConfigs"]) is not None:
        setv(
            to_object,
            ["inference_configs"],
            {
                k: _EvaluationRunInferenceConfig_from_vertex(v, to_object)
                for k, v in getv(from_object, ["inferenceConfigs"]).items()
            },
        )

    if getv(from_object, ["labels"]) is not None:
        setv(to_object, ["labels"], getv(from_object, ["labels"]))

    if getv(from_object, ["analysisConfigs"]) is not None:
        setv(
            to_object,
            ["analysis_configs"],
            [item for item in getv(from_object, ["analysisConfigs"])],
        )

    return to_object


def _GenerateInstanceRubricsRequest_to_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["contents"]) is not None:
        setv(to_object, ["contents"], getv(from_object, ["contents"]))

    if getv(from_object, ["predefined_rubric_generation_spec"]) is not None:
        setv(
            to_object,
            ["predefinedRubricGenerationSpec"],
            getv(from_object, ["predefined_rubric_generation_spec"]),
        )

    if getv(from_object, ["rubric_generation_spec"]) is not None:
        setv(
            to_object,
            ["rubricGenerationSpec"],
            getv(from_object, ["rubric_generation_spec"]),
        )

    if getv(from_object, ["metric_resource_name"]) is not None:
        setv(
            to_object,
            ["metricResourceName"],
            getv(from_object, ["metric_resource_name"]),
        )

    if getv(from_object, ["config"]) is not None:
        setv(to_object, ["config"], getv(from_object, ["config"]))

    return to_object


def _GenerateLossClustersParameters_to_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["location"]) is not None:
        setv(to_object, ["location"], getv(from_object, ["location"]))

    if getv(from_object, ["evaluation_set"]) is not None:
        setv(to_object, ["evaluationSet"], getv(from_object, ["evaluation_set"]))

    if getv(from_object, ["inline_results"]) is not None:
        setv(
            to_object,
            ["inlineResults", "evaluationResults"],
            [
                item
                for item in t.t_inline_results(getv(from_object, ["inline_results"]))
            ],
        )

    if getv(from_object, ["configs"]) is not None:
        setv(to_object, ["configs"], [item for item in getv(from_object, ["configs"])])

    if getv(from_object, ["config"]) is not None:
        setv(to_object, ["config"], getv(from_object, ["config"]))

    return to_object


def _GenerateUserScenariosParameters_to_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["location"]) is not None:
        setv(to_object, ["location"], getv(from_object, ["location"]))

    if getv(from_object, ["agents"]) is not None:
        setv(to_object, ["agents"], getv(from_object, ["agents"]))

    if getv(from_object, ["root_agent_id"]) is not None:
        setv(to_object, ["rootAgentId"], getv(from_object, ["root_agent_id"]))

    if getv(from_object, ["user_scenario_generation_config"]) is not None:
        setv(
            to_object,
            ["userScenarioGenerationConfig"],
            t.t_user_scenario_generation_config(
                getv(from_object, ["user_scenario_generation_config"])
            ),
        )

    if getv(from_object, ["config"]) is not None:
        setv(to_object, ["config"], getv(from_object, ["config"]))

    if getv(from_object, ["allow_cross_region_model"]) is not None:
        setv(
            to_object,
            ["allowCrossRegionModel"],
            getv(from_object, ["allow_cross_region_model"]),
        )

    if getv(from_object, ["gemini_agent_config"]) is not None:
        setv(
            to_object, ["geminiAgentConfig"], getv(from_object, ["gemini_agent_config"])
        )

    return to_object


def _GetEvaluationExperimentParameters_to_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["name"]) is not None:
        setv(to_object, ["_url", "name"], getv(from_object, ["name"]))

    if getv(from_object, ["config"]) is not None:
        setv(to_object, ["config"], getv(from_object, ["config"]))

    return to_object


def _GetEvaluationItemParameters_to_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["name"]) is not None:
        setv(to_object, ["_url", "name"], getv(from_object, ["name"]))

    if getv(from_object, ["config"]) is not None:
        setv(to_object, ["config"], getv(from_object, ["config"]))

    return to_object


def _GetEvaluationMetricParameters_to_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["metric_resource_name"]) is not None:
        setv(
            to_object,
            ["_url", "evaluation_metric"],
            getv(from_object, ["metric_resource_name"]),
        )

    if getv(from_object, ["config"]) is not None:
        setv(to_object, ["config"], getv(from_object, ["config"]))

    return to_object


def _GetEvaluationRunParameters_to_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["name"]) is not None:
        setv(to_object, ["_url", "name"], getv(from_object, ["name"]))

    if getv(from_object, ["config"]) is not None:
        setv(to_object, ["config"], getv(from_object, ["config"]))

    return to_object


def _GetEvaluationSetParameters_to_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["name"]) is not None:
        setv(to_object, ["_url", "name"], getv(from_object, ["name"]))

    if getv(from_object, ["config"]) is not None:
        setv(to_object, ["config"], getv(from_object, ["config"]))

    return to_object


def _ListEvaluationExperimentsConfig_to_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}

    if getv(from_object, ["page_size"]) is not None:
        setv(parent_object, ["_query", "pageSize"], getv(from_object, ["page_size"]))

    if getv(from_object, ["page_token"]) is not None:
        setv(parent_object, ["_query", "pageToken"], getv(from_object, ["page_token"]))

    if getv(from_object, ["filter"]) is not None:
        setv(parent_object, ["_query", "filter"], getv(from_object, ["filter"]))

    if getv(from_object, ["order_by"]) is not None:
        setv(parent_object, ["_query", "orderBy"], getv(from_object, ["order_by"]))

    return to_object


def _ListEvaluationExperimentsParameters_to_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["config"]) is not None:
        setv(
            to_object,
            ["config"],
            _ListEvaluationExperimentsConfig_to_vertex(
                getv(from_object, ["config"]), to_object
            ),
        )

    return to_object


def _ListEvaluationMetricsConfig_to_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}

    if getv(from_object, ["page_size"]) is not None:
        setv(parent_object, ["_query", "pageSize"], getv(from_object, ["page_size"]))

    if getv(from_object, ["page_token"]) is not None:
        setv(parent_object, ["_query", "pageToken"], getv(from_object, ["page_token"]))

    if getv(from_object, ["filter"]) is not None:
        setv(parent_object, ["_query", "filter"], getv(from_object, ["filter"]))

    if getv(from_object, ["order_by"]) is not None:
        setv(parent_object, ["_query", "orderBy"], getv(from_object, ["order_by"]))

    return to_object


def _ListEvaluationMetricsParameters_to_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["config"]) is not None:
        setv(
            to_object,
            ["config"],
            _ListEvaluationMetricsConfig_to_vertex(
                getv(from_object, ["config"]), to_object
            ),
        )

    return to_object


def _ListEvaluationMetricsResponse_from_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["sdkHttpResponse"]) is not None:
        setv(to_object, ["sdk_http_response"], getv(from_object, ["sdkHttpResponse"]))

    if getv(from_object, ["nextPageToken"]) is not None:
        setv(to_object, ["next_page_token"], getv(from_object, ["nextPageToken"]))

    if getv(from_object, ["evaluationMetrics"]) is not None:
        setv(
            to_object,
            ["evaluation_metrics"],
            [
                _EvaluationMetric_from_vertex(item, to_object)
                for item in getv(from_object, ["evaluationMetrics"])
            ],
        )

    return to_object


def _RubricBasedMetricInput_to_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["metric_spec"]) is not None:
        setv(
            to_object,
            ["metricSpec"],
            _RubricBasedMetricSpec_to_vertex(
                getv(from_object, ["metric_spec"]), to_object
            ),
        )

    if getv(from_object, ["instance"]) is not None:
        setv(to_object, ["instance"], getv(from_object, ["instance"]))

    return to_object


def _RubricBasedMetricSpec_to_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["metric_prompt_template"]) is not None:
        setv(
            to_object,
            ["metricPromptTemplate"],
            getv(from_object, ["metric_prompt_template"]),
        )

    if getv(from_object, ["judge_autorater_config"]) is not None:
        setv(
            to_object,
            ["judgeAutoraterConfig"],
            getv(from_object, ["judge_autorater_config"]),
        )

    if getv(from_object, ["inline_rubrics"]) is not None:
        setv(
            to_object,
            ["inline_rubrics", "rubrics"],
            getv(from_object, ["inline_rubrics"]),
        )

    if getv(from_object, ["rubric_group_key"]) is not None:
        setv(to_object, ["rubricGroupKey"], getv(from_object, ["rubric_group_key"]))

    if getv(from_object, ["rubric_generation_spec"]) is not None:
        setv(
            to_object,
            ["rubricGenerationSpec"],
            getv(from_object, ["rubric_generation_spec"]),
        )

    return to_object


def _UnifiedMetric_from_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["bleuSpec"]) is not None:
        setv(to_object, ["bleu_spec"], getv(from_object, ["bleuSpec"]))

    if getv(from_object, ["rougeSpec"]) is not None:
        setv(to_object, ["rouge_spec"], getv(from_object, ["rougeSpec"]))

    if getv(from_object, ["pointwiseMetricSpec"]) is not None:
        setv(
            to_object,
            ["pointwise_metric_spec"],
            getv(from_object, ["pointwiseMetricSpec"]),
        )

    if getv(from_object, ["llmBasedMetricSpec"]) is not None:
        setv(
            to_object,
            ["llm_based_metric_spec"],
            getv(from_object, ["llmBasedMetricSpec"]),
        )

    if getv(from_object, ["customCodeExecutionSpec"]) is not None:
        setv(
            to_object,
            ["custom_code_execution_spec"],
            _CustomCodeExecutionSpec_from_vertex(
                getv(from_object, ["customCodeExecutionSpec"]), to_object
            ),
        )

    if getv(from_object, ["predefinedMetricSpec"]) is not None:
        setv(
            to_object,
            ["predefined_metric_spec"],
            getv(from_object, ["predefinedMetricSpec"]),
        )

    if getv(from_object, ["computationBasedMetricSpec"]) is not None:
        setv(
            to_object,
            ["computation_based_metric_spec"],
            getv(from_object, ["computationBasedMetricSpec"]),
        )

    return to_object


def _UnifiedMetric_to_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["bleu_spec"]) is not None:
        setv(to_object, ["bleuSpec"], getv(from_object, ["bleu_spec"]))

    if getv(from_object, ["rouge_spec"]) is not None:
        setv(to_object, ["rougeSpec"], getv(from_object, ["rouge_spec"]))

    if getv(from_object, ["pointwise_metric_spec"]) is not None:
        setv(
            to_object,
            ["pointwiseMetricSpec"],
            getv(from_object, ["pointwise_metric_spec"]),
        )

    if getv(from_object, ["llm_based_metric_spec"]) is not None:
        setv(
            to_object,
            ["llmBasedMetricSpec"],
            getv(from_object, ["llm_based_metric_spec"]),
        )

    if getv(from_object, ["custom_code_execution_spec"]) is not None:
        setv(
            to_object,
            ["customCodeExecutionSpec"],
            _CustomCodeExecutionSpec_to_vertex(
                getv(from_object, ["custom_code_execution_spec"]), to_object
            ),
        )

    if getv(from_object, ["predefined_metric_spec"]) is not None:
        setv(
            to_object,
            ["predefinedMetricSpec"],
            getv(from_object, ["predefined_metric_spec"]),
        )

    if getv(from_object, ["computation_based_metric_spec"]) is not None:
        setv(
            to_object,
            ["computationBasedMetricSpec"],
            getv(from_object, ["computation_based_metric_spec"]),
        )

    return to_object


class Evals(_api_module.BaseModule):

    def create_evaluation_experiment(
        self,
        *,
        display_name: Optional[str] = None,
        labels: Optional[dict[str, str]] = None,
        merge_strategy: Optional[types.EvaluationExperimentMergeStrategy] = None,
        metadata: Optional[dict[str, Any]] = None,
        config: Optional[types.CreateEvaluationExperimentConfigOrDict] = None,
    ) -> types.EvaluationExperiment:
        """
        Creates an EvaluationExperiment.

        Args:
          display_name: The display name of the evaluation experiment.
          labels: Labels for the evaluation experiment.
          merge_strategy: Merge strategy for the evaluation experiment.
          metadata: Metadata about the evaluation experiment, can be used by the
            caller to store additional tracking information about the experiment.
          config: Optional configuration for the create operation.

        Returns:
          The created evaluation experiment.

        .. code-block:: python

          eval_experiment = client.evals.create_evaluation_experiment(
              display_name="my-experiment"
          )

        """

        parameter_model = types._CreateEvaluationExperimentParameters(
            display_name=display_name,
            labels=labels,
            merge_strategy=merge_strategy,
            metadata=metadata,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError(
                "This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
            )
        else:
            request_dict = _CreateEvaluationExperimentParameters_to_vertex(
                parameter_model
            )
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "evaluationExperiments".format_map(request_url_dict)
            else:
                path = "evaluationExperiments"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = self._api_client.request("post", path, request_dict, http_options)

        response_dict = {} if not response.body else json.loads(response.body)

        return_value = types.EvaluationExperiment._from_response(
            response=response_dict,
            kwargs=(
                {
                    "config": {
                        "response_schema": getattr(
                            parameter_model.config, "response_schema", None
                        ),
                        "response_json_schema": getattr(
                            parameter_model.config, "response_json_schema", None
                        ),
                        "include_all_fields": getattr(
                            parameter_model.config, "include_all_fields", None
                        ),
                    }
                }
                if getattr(parameter_model, "config", None)
                else {}
            ),
        )

        self._api_client._verify_response(return_value)
        return return_value

    def _create_evaluation_item(
        self,
        *,
        evaluation_item_type: str,
        gcs_uri: str,
        display_name: Optional[str] = None,
        config: Optional[types.CreateEvaluationItemConfigOrDict] = None,
    ) -> types.EvaluationItem:
        """
        Creates an EvaluationItem.
        """

        parameter_model = types._CreateEvaluationItemParameters(
            evaluation_item_type=evaluation_item_type,
            gcs_uri=gcs_uri,
            display_name=display_name,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError(
                "This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
            )
        else:
            request_dict = _CreateEvaluationItemParameters_to_vertex(parameter_model)
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "evaluationItems".format_map(request_url_dict)
            else:
                path = "evaluationItems"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = self._api_client.request("post", path, request_dict, http_options)

        response_dict = {} if not response.body else json.loads(response.body)

        return_value = types.EvaluationItem._from_response(
            response=response_dict,
            kwargs=(
                {
                    "config": {
                        "response_schema": getattr(
                            parameter_model.config, "response_schema", None
                        ),
                        "response_json_schema": getattr(
                            parameter_model.config, "response_json_schema", None
                        ),
                        "include_all_fields": getattr(
                            parameter_model.config, "include_all_fields", None
                        ),
                    }
                }
                if getattr(parameter_model, "config", None)
                else {}
            ),
        )

        self._api_client._verify_response(return_value)
        return return_value

    def _create_evaluation_metric(
        self,
        *,
        display_name: Optional[str] = None,
        description: Optional[str] = None,
        metric: Optional[types.MetricOrDict] = None,
        config: Optional[types.CreateEvaluationMetricConfigOrDict] = None,
    ) -> types.EvaluationMetric:
        """
        Creates an EvaluationMetric.
        """

        parameter_model = types._CreateEvaluationMetricParameters(
            display_name=display_name,
            description=description,
            metric=metric,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError(
                "This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
            )
        else:
            request_dict = _CreateEvaluationMetricParameters_to_vertex(parameter_model)
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "evaluationMetrics".format_map(request_url_dict)
            else:
                path = "evaluationMetrics"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = self._api_client.request("post", path, request_dict, http_options)

        response_dict = {} if not response.body else json.loads(response.body)

        if self._api_client.vertexai:
            response_dict = _EvaluationMetric_from_vertex(response_dict)

        return_value = types.EvaluationMetric._from_response(
            response=response_dict,
            kwargs=(
                {
                    "config": {
                        "response_schema": getattr(
                            parameter_model.config, "response_schema", None
                        ),
                        "response_json_schema": getattr(
                            parameter_model.config, "response_json_schema", None
                        ),
                        "include_all_fields": getattr(
                            parameter_model.config, "include_all_fields", None
                        ),
                    }
                }
                if getattr(parameter_model, "config", None)
                else {}
            ),
        )

        self._api_client._verify_response(return_value)
        return return_value

    def _create_evaluation_run(
        self,
        *,
        name: Optional[str] = None,
        display_name: Optional[str] = None,
        data_source: types.EvaluationRunDataSourceOrDict,
        evaluation_config: types.EvaluationRunConfigOrDict,
        labels: Optional[dict[str, str]] = None,
        inference_configs: Optional[
            dict[str, types.EvaluationRunInferenceConfigOrDict]
        ] = None,
        config: Optional[types.CreateEvaluationRunConfigOrDict] = None,
        analysis_configs: Optional[list[types.AnalysisConfigOrDict]] = None,
    ) -> types.EvaluationRun:
        """
        Creates an EvaluationRun.
        """

        parameter_model = types._CreateEvaluationRunParameters(
            name=name,
            display_name=display_name,
            data_source=data_source,
            evaluation_config=evaluation_config,
            labels=labels,
            inference_configs=inference_configs,
            config=config,
            analysis_configs=analysis_configs,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError(
                "This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
            )
        else:
            request_dict = _CreateEvaluationRunParameters_to_vertex(parameter_model)
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "evaluationRuns".format_map(request_url_dict)
            else:
                path = "evaluationRuns"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = self._api_client.request("post", path, request_dict, http_options)

        response_dict = {} if not response.body else json.loads(response.body)

        if self._api_client.vertexai:
            response_dict = _EvaluationRun_from_vertex(response_dict)

        return_value = types.EvaluationRun._from_response(
            response=response_dict,
            kwargs=(
                {
                    "config": {
                        "response_schema": getattr(
                            parameter_model.config, "response_schema", None
                        ),
                        "response_json_schema": getattr(
                            parameter_model.config, "response_json_schema", None
                        ),
                        "include_all_fields": getattr(
                            parameter_model.config, "include_all_fields", None
                        ),
                    }
                }
                if getattr(parameter_model, "config", None)
                else {}
            ),
        )

        self._api_client._verify_response(return_value)
        return return_value

    def _create_evaluation_set(
        self,
        *,
        evaluation_items: list[str],
        display_name: Optional[str] = None,
        config: Optional[types.CreateEvaluationSetConfigOrDict] = None,
    ) -> types.EvaluationSet:
        """
        Creates an EvaluationSet.
        """

        parameter_model = types._CreateEvaluationSetParameters(
            evaluation_items=evaluation_items,
            display_name=display_name,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError(
                "This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
            )
        else:
            request_dict = _CreateEvaluationSetParameters_to_vertex(parameter_model)
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "evaluationSets".format_map(request_url_dict)
            else:
                path = "evaluationSets"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = self._api_client.request("post", path, request_dict, http_options)

        response_dict = {} if not response.body else json.loads(response.body)

        return_value = types.EvaluationSet._from_response(
            response=response_dict,
            kwargs=(
                {
                    "config": {
                        "response_schema": getattr(
                            parameter_model.config, "response_schema", None
                        ),
                        "response_json_schema": getattr(
                            parameter_model.config, "response_json_schema", None
                        ),
                        "include_all_fields": getattr(
                            parameter_model.config, "include_all_fields", None
                        ),
                    }
                }
                if getattr(parameter_model, "config", None)
                else {}
            ),
        )

        self._api_client._verify_response(return_value)
        return return_value

    def _delete_evaluation_metric(
        self,
        *,
        metric_resource_name: str,
        config: Optional[types.DeleteEvaluationMetricConfigOrDict] = None,
    ) -> types.DeleteEvaluationMetricOperation:
        """
        Deletes an EvaluationMetric.
        """

        parameter_model = types._DeleteEvaluationMetricParameters(
            metric_resource_name=metric_resource_name,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError(
                "This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
            )
        else:
            request_dict = _DeleteEvaluationMetricParameters_to_vertex(parameter_model)
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "{evaluation_metric}".format_map(request_url_dict)
            else:
                path = "{evaluation_metric}"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = self._api_client.request("delete", path, request_dict, http_options)

        response_dict = {} if not response.body else json.loads(response.body)

        return_value = types.DeleteEvaluationMetricOperation._from_response(
            response=response_dict,
            kwargs=(
                {
                    "config": {
                        "response_schema": getattr(
                            parameter_model.config, "response_schema", None
                        ),
                        "response_json_schema": getattr(
                            parameter_model.config, "response_json_schema", None
                        ),
                        "include_all_fields": getattr(
                            parameter_model.config, "include_all_fields", None
                        ),
                    }
                }
                if getattr(parameter_model, "config", None)
                else {}
            ),
        )

        self._api_client._verify_response(return_value)
        return return_value

    def _evaluate_instances(
        self,
        *,
        bleu_input: Optional[types.BleuInputOrDict] = None,
        exact_match_input: Optional[types.ExactMatchInputOrDict] = None,
        rouge_input: Optional[types.RougeInputOrDict] = None,
        pointwise_metric_input: Optional[types.PointwiseMetricInputOrDict] = None,
        pairwise_metric_input: Optional[types.PairwiseMetricInputOrDict] = None,
        tool_call_valid_input: Optional[types.ToolCallValidInputOrDict] = None,
        tool_name_match_input: Optional[types.ToolNameMatchInputOrDict] = None,
        tool_parameter_key_match_input: Optional[
            types.ToolParameterKeyMatchInputOrDict
        ] = None,
        tool_parameter_kv_match_input: Optional[
            types.ToolParameterKVMatchInputOrDict
        ] = None,
        rubric_based_metric_input: Optional[types.RubricBasedMetricInputOrDict] = None,
        autorater_config: Optional[genai_types.AutoraterConfigOrDict] = None,
        metrics: Optional[list[types.MetricOrDict]] = None,
        instance: Optional[types.EvaluationInstanceOrDict] = None,
        metric_sources: Optional[list[types.MetricSourceOrDict]] = None,
        config: Optional[types.EvaluateInstancesConfigOrDict] = None,
    ) -> types.EvaluateInstancesResponse:
        """
        Evaluates instances based on a given metric.
        """

        parameter_model = types._EvaluateInstancesRequestParameters(
            bleu_input=bleu_input,
            exact_match_input=exact_match_input,
            rouge_input=rouge_input,
            pointwise_metric_input=pointwise_metric_input,
            pairwise_metric_input=pairwise_metric_input,
            tool_call_valid_input=tool_call_valid_input,
            tool_name_match_input=tool_name_match_input,
            tool_parameter_key_match_input=tool_parameter_key_match_input,
            tool_parameter_kv_match_input=tool_parameter_kv_match_input,
            rubric_based_metric_input=rubric_based_metric_input,
            autorater_config=autorater_config,
            metrics=metrics,
            instance=instance,
            metric_sources=metric_sources,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError(
                "This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
            )
        else:
            request_dict = _EvaluateInstancesRequestParameters_to_vertex(
                parameter_model
            )
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = ":evaluateInstances".format_map(request_url_dict)
            else:
                path = ":evaluateInstances"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = self._api_client.request("post", path, request_dict, http_options)

        response_dict = {} if not response.body else json.loads(response.body)

        return_value = types.EvaluateInstancesResponse._from_response(
            response=response_dict,
            kwargs=(
                {
                    "config": {
                        "response_schema": getattr(
                            parameter_model.config, "response_schema", None
                        ),
                        "response_json_schema": getattr(
                            parameter_model.config, "response_json_schema", None
                        ),
                        "include_all_fields": getattr(
                            parameter_model.config, "include_all_fields", None
                        ),
                    }
                }
                if getattr(parameter_model, "config", None)
                else {}
            ),
        )

        self._api_client._verify_response(return_value)
        return return_value

    def _generate_user_scenarios(
        self,
        *,
        location: Optional[str] = None,
        agents: Optional[dict[str, evals_types.AgentConfigOrDict]] = None,
        root_agent_id: Optional[str] = None,
        user_scenario_generation_config: Optional[
            evals_types.UserScenarioGenerationConfigOrDict
        ] = None,
        config: Optional[types.GenerateUserScenariosConfigOrDict] = None,
        allow_cross_region_model: Optional[bool] = None,
        gemini_agent_config: Optional[types.GeminiAgentConfigOrDict] = None,
    ) -> types.GenerateUserScenariosResponse:
        """
        Generates user scenarios for agent evaluation.
        """

        parameter_model = types._GenerateUserScenariosParameters(
            location=location,
            agents=agents,
            root_agent_id=root_agent_id,
            user_scenario_generation_config=user_scenario_generation_config,
            config=config,
            allow_cross_region_model=allow_cross_region_model,
            gemini_agent_config=gemini_agent_config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError(
                "This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
            )
        else:
            request_dict = _GenerateUserScenariosParameters_to_vertex(parameter_model)
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = ":generateUserScenarios".format_map(request_url_dict)
            else:
                path = ":generateUserScenarios"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = self._api_client.request("post", path, request_dict, http_options)

        response_dict = {} if not response.body else json.loads(response.body)

        return_value = types.GenerateUserScenariosResponse._from_response(
            response=response_dict,
            kwargs=(
                {
                    "config": {
                        "response_schema": getattr(
                            parameter_model.config, "response_schema", None
                        ),
                        "response_json_schema": getattr(
                            parameter_model.config, "response_json_schema", None
                        ),
                        "include_all_fields": getattr(
                            parameter_model.config, "include_all_fields", None
                        ),
                    }
                }
                if getattr(parameter_model, "config", None)
                else {}
            ),
        )

        self._api_client._verify_response(return_value)
        return return_value

    def _generate_loss_clusters(
        self,
        *,
        location: Optional[str] = None,
        evaluation_set: Optional[str] = None,
        inline_results: Optional[list[types.EvaluationResultOrDict]] = None,
        configs: Optional[list[types.LossAnalysisConfigOrDict]] = None,
        config: Optional[types.GenerateLossClustersConfigOrDict] = None,
    ) -> types.GenerateLossClustersOperation:
        """
        Generates loss clusters from evaluation results.
        """

        parameter_model = types._GenerateLossClustersParameters(
            location=location,
            evaluation_set=evaluation_set,
            inline_results=inline_results,
            configs=configs,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError(
                "This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
            )
        else:
            request_dict = _GenerateLossClustersParameters_to_vertex(parameter_model)
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = ":generateLossClusters".format_map(request_url_dict)
            else:
                path = ":generateLossClusters"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = self._api_client.request("post", path, request_dict, http_options)

        response_dict = {} if not response.body else json.loads(response.body)

        return_value = types.GenerateLossClustersOperation._from_response(
            response=response_dict,
            kwargs=(
                {
                    "config": {
                        "response_schema": getattr(
                            parameter_model.config, "response_schema", None
                        ),
                        "response_json_schema": getattr(
                            parameter_model.config, "response_json_schema", None
                        ),
                        "include_all_fields": getattr(
                            parameter_model.config, "include_all_fields", None
                        ),
                    }
                }
                if getattr(parameter_model, "config", None)
                else {}
            ),
        )

        self._api_client._verify_response(return_value)
        return return_value

    def _generate_rubrics(
        self,
        *,
        contents: list[genai_types.ContentOrDict],
        predefined_rubric_generation_spec: Optional[
            genai_types.PredefinedMetricSpecOrDict
        ] = None,
        rubric_generation_spec: Optional[genai_types.RubricGenerationSpecOrDict] = None,
        metric_resource_name: Optional[str] = None,
        config: Optional[types.RubricGenerationConfigOrDict] = None,
    ) -> types.GenerateInstanceRubricsResponse:
        """
        Generates rubrics for a given prompt.
        """

        parameter_model = types._GenerateInstanceRubricsRequest(
            contents=contents,
            predefined_rubric_generation_spec=predefined_rubric_generation_spec,
            rubric_generation_spec=rubric_generation_spec,
            metric_resource_name=metric_resource_name,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError(
                "This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
            )
        else:
            request_dict = _GenerateInstanceRubricsRequest_to_vertex(parameter_model)
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = ":generateInstanceRubrics".format_map(request_url_dict)
            else:
                path = ":generateInstanceRubrics"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = self._api_client.request("post", path, request_dict, http_options)

        response_dict = {} if not response.body else json.loads(response.body)

        return_value = types.GenerateInstanceRubricsResponse._from_response(
            response=response_dict,
            kwargs=(
                {
                    "config": {
                        "response_schema": getattr(
                            parameter_model.config, "response_schema", None
                        ),
                        "response_json_schema": getattr(
                            parameter_model.config, "response_json_schema", None
                        ),
                        "include_all_fields": getattr(
                            parameter_model.config, "include_all_fields", None
                        ),
                    }
                }
                if getattr(parameter_model, "config", None)
                else {}
            ),
        )

        self._api_client._verify_response(return_value)
        return return_value

    def get_evaluation_experiment(
        self,
        *,
        name: str,
        config: Optional[types.GetEvaluationExperimentConfigOrDict] = None,
    ) -> types.EvaluationExperiment:
        """
        Retrieves an EvaluationExperiment from the resource name.

        Args:
          name: The resource name of the EvaluationExperiment. Format:
            `projects/{project}/locations/{location}/evaluationExperiments/{evaluation_experiment}`
          config: The optional configuration for the get operation.

        Returns:
          The evaluation experiment.

        Example Usage:

        .. code-block:: python

          eval_experiment = client.evals.get_evaluation_experiment(
              name="projects/my-project/locations/us-central1/evaluationExperiments/my-experiment"
          )

        """

        parameter_model = types._GetEvaluationExperimentParameters(
            name=name,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError(
                "This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
            )
        else:
            request_dict = _GetEvaluationExperimentParameters_to_vertex(parameter_model)
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "{name}".format_map(request_url_dict)
            else:
                path = "{name}"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = self._api_client.request("get", path, request_dict, http_options)

        response_dict = {} if not response.body else json.loads(response.body)

        return_value = types.EvaluationExperiment._from_response(
            response=response_dict,
            kwargs=(
                {
                    "config": {
                        "response_schema": getattr(
                            parameter_model.config, "response_schema", None
                        ),
                        "response_json_schema": getattr(
                            parameter_model.config, "response_json_schema", None
                        ),
                        "include_all_fields": getattr(
                            parameter_model.config, "include_all_fields", None
                        ),
                    }
                }
                if getattr(parameter_model, "config", None)
                else {}
            ),
        )

        self._api_client._verify_response(return_value)
        return return_value

    def _get_evaluation_metric(
        self,
        *,
        metric_resource_name: str,
        config: Optional[types.GetEvaluationMetricConfigOrDict] = None,
    ) -> types.EvaluationMetric:
        """
        Retrieves an EvaluationMetric from the resource name.
        """

        parameter_model = types._GetEvaluationMetricParameters(
            metric_resource_name=metric_resource_name,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError(
                "This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
            )
        else:
            request_dict = _GetEvaluationMetricParameters_to_vertex(parameter_model)
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "{evaluation_metric}".format_map(request_url_dict)
            else:
                path = "{evaluation_metric}"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = self._api_client.request("get", path, request_dict, http_options)

        response_dict = {} if not response.body else json.loads(response.body)

        if self._api_client.vertexai:
            response_dict = _EvaluationMetric_from_vertex(response_dict)

        return_value = types.EvaluationMetric._from_response(
            response=response_dict,
            kwargs=(
                {
                    "config": {
                        "response_schema": getattr(
                            parameter_model.config, "response_schema", None
                        ),
                        "response_json_schema": getattr(
                            parameter_model.config, "response_json_schema", None
                        ),
                        "include_all_fields": getattr(
                            parameter_model.config, "include_all_fields", None
                        ),
                    }
                }
                if getattr(parameter_model, "config", None)
                else {}
            ),
        )

        self._api_client._verify_response(return_value)
        return return_value

    def _get_evaluation_run(
        self, *, name: str, config: Optional[types.GetEvaluationRunConfigOrDict] = None
    ) -> types.EvaluationRun:
        """
        Retrieves an EvaluationRun from the resource name.
        """

        parameter_model = types._GetEvaluationRunParameters(
            name=name,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError(
                "This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
            )
        else:
            request_dict = _GetEvaluationRunParameters_to_vertex(parameter_model)
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "evaluationRuns/{name}".format_map(request_url_dict)
            else:
                path = "evaluationRuns/{name}"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = self._api_client.request("get", path, request_dict, http_options)

        response_dict = {} if not response.body else json.loads(response.body)

        if self._api_client.vertexai:
            response_dict = _EvaluationRun_from_vertex(response_dict)

        return_value = types.EvaluationRun._from_response(
            response=response_dict,
            kwargs=(
                {
                    "config": {
                        "response_schema": getattr(
                            parameter_model.config, "response_schema", None
                        ),
                        "response_json_schema": getattr(
                            parameter_model.config, "response_json_schema", None
                        ),
                        "include_all_fields": getattr(
                            parameter_model.config, "include_all_fields", None
                        ),
                    }
                }
                if getattr(parameter_model, "config", None)
                else {}
            ),
        )

        self._api_client._verify_response(return_value)
        return return_value

    def _get_evaluation_set(
        self, *, name: str, config: Optional[types.GetEvaluationSetConfigOrDict] = None
    ) -> types.EvaluationSet:
        """
        Retrieves an EvaluationSet from the resource name.
        """

        parameter_model = types._GetEvaluationSetParameters(
            name=name,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError(
                "This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
            )
        else:
            request_dict = _GetEvaluationSetParameters_to_vertex(parameter_model)
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "evaluationSets/{name}".format_map(request_url_dict)
            else:
                path = "evaluationSets/{name}"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = self._api_client.request("get", path, request_dict, http_options)

        response_dict = {} if not response.body else json.loads(response.body)

        return_value = types.EvaluationSet._from_response(
            response=response_dict,
            kwargs=(
                {
                    "config": {
                        "response_schema": getattr(
                            parameter_model.config, "response_schema", None
                        ),
                        "response_json_schema": getattr(
                            parameter_model.config, "response_json_schema", None
                        ),
                        "include_all_fields": getattr(
                            parameter_model.config, "include_all_fields", None
                        ),
                    }
                }
                if getattr(parameter_model, "config", None)
                else {}
            ),
        )

        self._api_client._verify_response(return_value)
        return return_value

    def _get_evaluation_item(
        self, *, name: str, config: Optional[types.GetEvaluationItemConfigOrDict] = None
    ) -> types.EvaluationItem:
        """
        Retrieves an EvaluationItem from the resource name.
        """

        parameter_model = types._GetEvaluationItemParameters(
            name=name,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError(
                "This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
            )
        else:
            request_dict = _GetEvaluationItemParameters_to_vertex(parameter_model)
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "evaluationItems/{name}".format_map(request_url_dict)
            else:
                path = "evaluationItems/{name}"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = self._api_client.request("get", path, request_dict, http_options)

        response_dict = {} if not response.body else json.loads(response.body)

        return_value = types.EvaluationItem._from_response(
            response=response_dict,
            kwargs=(
                {
                    "config": {
                        "response_schema": getattr(
                            parameter_model.config, "response_schema", None
                        ),
                        "response_json_schema": getattr(
                            parameter_model.config, "response_json_schema", None
                        ),
                        "include_all_fields": getattr(
                            parameter_model.config, "include_all_fields", None
                        ),
                    }
                }
                if getattr(parameter_model, "config", None)
                else {}
            ),
        )

        self._api_client._verify_response(return_value)
        return return_value

    def list_evaluation_experiments(
        self, *, config: Optional[types.ListEvaluationExperimentsConfigOrDict] = None
    ) -> types.ListEvaluationExperimentsResponse:
        """
        Lists EvaluationExperiments.

        Args:
          config: Optional configuration for the list operation, including
            pagination (``page_size``, ``page_token``), ``filter``, and
            ``order_by``.

        Returns:
          The list evaluation experiments response.

        Example Usage:

        .. code-block:: python

          response = client.evals.list_evaluation_experiments(
              config={"filter": 'display_name="my-experiment"'}
          )

        """

        parameter_model = types._ListEvaluationExperimentsParameters(
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError(
                "This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
            )
        else:
            request_dict = _ListEvaluationExperimentsParameters_to_vertex(
                parameter_model
            )
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "evaluationExperiments".format_map(request_url_dict)
            else:
                path = "evaluationExperiments"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = self._api_client.request("get", path, request_dict, http_options)

        response_dict = {} if not response.body else json.loads(response.body)

        return_value = types.ListEvaluationExperimentsResponse._from_response(
            response=response_dict,
            kwargs=(
                {
                    "config": {
                        "response_schema": getattr(
                            parameter_model.config, "response_schema", None
                        ),
                        "response_json_schema": getattr(
                            parameter_model.config, "response_json_schema", None
                        ),
                        "include_all_fields": getattr(
                            parameter_model.config, "include_all_fields", None
                        ),
                    }
                }
                if getattr(parameter_model, "config", None)
                else {}
            ),
        )

        self._api_client._verify_response(return_value)
        return return_value

    def _list_evaluation_metrics(
        self, *, config: Optional[types.ListEvaluationMetricsConfigOrDict] = None
    ) -> types.ListEvaluationMetricsResponse:
        """
        Lists EvaluationMetrics.
        """

        parameter_model = types._ListEvaluationMetricsParameters(
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError(
                "This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
            )
        else:
            request_dict = _ListEvaluationMetricsParameters_to_vertex(parameter_model)
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "evaluationMetrics".format_map(request_url_dict)
            else:
                path = "evaluationMetrics"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = self._api_client.request("get", path, request_dict, http_options)

        response_dict = {} if not response.body else json.loads(response.body)

        if self._api_client.vertexai:
            response_dict = _ListEvaluationMetricsResponse_from_vertex(response_dict)

        return_value = types.ListEvaluationMetricsResponse._from_response(
            response=response_dict,
            kwargs=(
                {
                    "config": {
                        "response_schema": getattr(
                            parameter_model.config, "response_schema", None
                        ),
                        "response_json_schema": getattr(
                            parameter_model.config, "response_json_schema", None
                        ),
                        "include_all_fields": getattr(
                            parameter_model.config, "include_all_fields", None
                        ),
                    }
                }
                if getattr(parameter_model, "config", None)
                else {}
            ),
        )

        self._api_client._verify_response(return_value)
        return return_value

    def evaluate_instances(
        self,
        *,
        metric_config: types._EvaluateInstancesRequestParameters,
    ) -> types.EvaluateInstancesResponse:
        """Evaluates an instance of a model."""

        if isinstance(metric_config, types._EvaluateInstancesRequestParameters):
            metric_config = metric_config.model_dump()  # type: ignore[assignment]
        else:
            metric_config = dict(metric_config)

        return self._evaluate_instances(
            **metric_config,
        )

    def run_inference(
        self,
        *,
        src: Union[str, pd.DataFrame, types.EvaluationDataset],
        model: Optional[Union[str, Callable[[Any], Any]]] = None,
        agent: Optional[Union[str, types.AgentEngine, LlmAgent]] = None,
        location: Optional[str] = None,
        config: Optional[types.EvalRunInferenceConfigOrDict] = None,
    ) -> types.EvaluationDataset:
        """Runs inference on a dataset for evaluation.

        Args:
          src: The source of the dataset. Can be a string (path to a local file,
                a GCS path, or a BigQuery table), a Pandas DataFrame, or an
                EvaluationDataset object. An EvaluationDataset may have either
                ``eval_dataset_df`` or ``eval_cases`` populated. When
                ``eval_cases`` with ``agent_data`` is provided, the last user
                event in the turns is used as the current prompt and prior
                events are replayed as session history for local ADK agents.
          model: Optional type is experimental and may change in future versions.
                The model to use for inference, optional for agent evaluations.
              - For Google Gemini models, provide the model name string (e.g., "gemini-2.5-flash").
              - For third-party models via LiteLLM, use the format "provider/model_name"
                (e.g., "openai/gpt-4o"). Ensure the necessary API key (e.g., OPENAI_API_KEY)
                is set as an environment variable.
              - For custom logic, provide a callable function that accepts a prompt and
                returns a response.
          agent: This field is experimental and may change in future versions
                The agent engine used or local agent to run agent, optional for non-agent evaluations.
              - agent engine resource name in str type, with format
                `projects/{project}/locations/{location}/reasoningEngines/{reasoning_engine_id}`,
                run_inference will fetch the agent engine from the resource name.
              - Or `types.AgentEngine` object.
              - Or ADK agent in LlMAgent type.
          location: The location to use for the inference. If not specified, the
                location configured in the client will be used. If specified,
                this will override the location set in `agentplatform.Client` only
                for this API call.
          config: The optional configuration for the inference run. Must be a dict or
              `types.EvalRunInferenceConfig` type.
                - dest: The destination path for storage of the inference results.
                - prompt_template: The template string to use for constructing prompts.
                - generate_content_config: The config for the Gemini generate content call.
                - allow_cross_region_model: Opt-in flag to authorize cross-region routing for LLM models.

        Returns:
          The evaluation dataset.
        """
        if not config:
            config = types.EvalRunInferenceConfig()
        if isinstance(config, dict):
            config = types.EvalRunInferenceConfig.model_validate(config)

        if isinstance(src, types.EvaluationDataset):
            if src.eval_dataset_df is not None:
                src = src.eval_dataset_df
            elif src.eval_cases:
                src = _evals_common._eval_cases_to_dataframe(src.eval_cases)
            else:
                raise ValueError(
                    "EvaluationDataset must have eval_dataset_df or eval_cases"
                    " populated."
                )

        agent_engine_instance = None
        agent_instance = None
        gemini_agent_instance = None
        if agent:
            if isinstance(agent, str) and _evals_common._is_gemini_agent_resource(
                agent
            ):
                gemini_agent_instance = agent
            elif isinstance(agent, str) or isinstance(agent, types.AgentEngine):
                agent_engine_instance = agent
            else:
                agent_instance = agent

        return _evals_common._execute_inference(  # type: ignore[no-any-return]
            api_client=self._api_client,
            model=model,
            agent_engine=agent_engine_instance,
            agent=agent_instance,
            gemini_agent=gemini_agent_instance,
            src=src,
            dest=config.dest,
            prompt_template=config.prompt_template,
            location=location,
            config=config.generate_content_config,
            user_simulator_config=getattr(config, "user_simulator_config", None),
            allow_cross_region_model=getattr(config, "allow_cross_region_model", False),
        )

    def evaluate(
        self,
        *,
        dataset: Union[
            pd.DataFrame,
            types.EvaluationDatasetOrDict,
            list[types.EvaluationDatasetOrDict],
        ],
        metrics: Optional[list[types.MetricOrDict]] = None,
        agent: Optional[str] = None,
        location: Optional[str] = None,
        config: Optional[types.EvaluateMethodConfigOrDict] = None,
        **kwargs: Any,
    ) -> types.EvaluationResult:
        """Evaluates candidate responses in the provided dataset(s) using the specified metrics.

        Args:
          dataset: The dataset(s) to evaluate. Can be a pandas DataFrame, a single
            `types.EvaluationDataset` or a list of `types.EvaluationDataset`. To
            evaluate existing interactions, provide a dataset with an
            `interaction_id` column; each interaction is resolved by the backend
            using `agent` to populate the agent data for evaluation.
          metrics: The list of metrics to use for evaluation.
          agent: Optional Gemini Agents API agent resource name
            (`projects/{project}/locations/{location}/agents/{agent}`). Required
            when the dataset contains an `interaction_id` column: the backend uses
            it to resolve the Agent config for each referenced interaction.
          location: The location to use for the evaluation service. If not specified,
             the location configured in the client will be used. If specified,
             this will override the location set in `agentplatform.Client` only for
             this API call.
          config: Optional configuration for the evaluation. Can be a dictionary or a
            `types.EvaluateMethodConfig` object.
            - dataset_schema: Schema to use for the dataset. If not specified, the
              dataset schema will be inferred from the dataset automatically.
            - dest: Destination path for storing evaluation results.
            - 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 not config:
            config = types.EvaluateMethodConfig()
        if isinstance(config, dict):
            config = types.EvaluateMethodConfig.model_validate(config)

        if isinstance(dataset, pd.DataFrame):
            dataset = types.EvaluationDataset(eval_dataset_df=dataset)

        if isinstance(dataset, list):
            dataset = [
                (
                    types.EvaluationDataset.model_validate(ds_item)
                    if isinstance(ds_item, dict)
                    else ds_item
                )
                for ds_item in dataset
            ]
        else:
            if isinstance(dataset, dict):
                dataset = types.EvaluationDataset.model_validate(dataset)
        if metrics is None:
            metrics = [types.Metric(name="general_quality_v1")]

        # TODO: Replace kwargs with agent_info after the experimental phase.
        if kwargs:
            logger.warning(
                "`kwargs` attribute in `evaluate` method is experimental and may change in future versions."
            )

        return _evals_common._execute_evaluation(
            api_client=self._api_client,
            dataset=dataset,
            metrics=metrics,
            agent=agent,
            dataset_schema=config.dataset_schema,
            dest=config.dest,
            location=location,
            evaluation_service_qps=getattr(config, "evaluation_service_qps", None),
            **kwargs,
        )

    def batch_evaluate(
        self,
        *,
        dataset: types.EvaluationDatasetOrDict,
        metrics: list[types.MetricOrDict],
        dest: str,
        config: Optional[types.EvaluateDatasetConfigOrDict] = None,
    ) -> types.EvaluateDatasetOperation:
        """Evaluates a dataset based on a set of given metrics."""

        resolved_metrics = _evals_common._resolve_metrics(metrics, self._api_client)
        output_config = genai_types.OutputConfig(
            gcs_destination=genai_types.GcsDestination(output_uri_prefix=dest)
        )
        parameter_model = types.EvaluateDatasetRequestParameters(
            dataset=dataset,
            metrics=resolved_metrics,
            output_config=output_config,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError("This method is only supported in the Vertex AI client.")
        else:
            request_dict = _evals_utils.BatchEvaluateRequestPreparer.EvaluateDatasetRequestParameters_to_vertex(
                parameter_model
            )
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = ":evaluateDataset".format_map(request_url_dict)
            else:
                path = ":evaluateDataset"

        request_dict = _evals_utils.BatchEvaluateRequestPreparer.prepare_metric_payload(
            request_dict, resolved_metrics
        )

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = self._api_client.request("post", path, request_dict, http_options)

        response_dict = "" if not response.body else json.loads(response.body)

        if self._api_client.vertexai:
            response_dict = _evals_utils.BatchEvaluateRequestPreparer.EvaluateDatasetOperation_from_vertex(
                response_dict
            )

        return_value = types.EvaluateDatasetOperation._from_response(
            response=response_dict, kwargs=parameter_model.model_dump()
        )
        self._api_client._verify_response(return_value)

        return return_value

    def generate_rubrics(
        self,
        *,
        src: Union[str, "pd.DataFrame", types.EvaluationDataset],
        rubric_group_name: str,
        prompt_template: Optional[str] = None,
        generator_model_config: Optional["genai_types.AutoraterConfigOrDict"] = None,
        rubric_content_type: Optional["types.RubricContentType"] = None,
        rubric_type_ontology: Optional[list[str]] = None,
        predefined_spec_name: Optional[Union[str, "types.PrebuiltMetric"]] = None,
        metric_spec_parameters: Optional[dict[str, Any]] = None,
        metric: Optional[types.MetricOrDict] = None,
        config: Optional[types.RubricGenerationConfigOrDict] = None,
    ) -> types.EvaluationDataset:
        """Generates rubrics for each prompt in the source and adds them as a new column
        structured as a dictionary.

        You can generate rubrics by providing either:
          1. A `metric` to use a pre-registered metric resource.
          2. A `predefined_spec_name` to use a Vertex AI backend recipe.
          3. A `prompt_template` along with other configuration parameters
             (`generator_model_config`, `rubric_content_type`, `rubric_type_ontology`)
             for custom rubric generation.
        with `metric` taking precedence over `predefined_spec_name`,
        and `predefined_spec_name` taking precedence over `prompt_template`

        These two modes are mutually exclusive.

        Args:
            src: The source of the prompts. Can be a string (path to a local
              file, a GCS path, or a BigQuery table), a Pandas DataFrame, or
              an EvaluationDataset object. The loaded data must contain either
              a 'prompt' column (for text) or a 'request' column (for text or
              multimodal Gemini Content).
            rubric_group_name: Name for the key within the dictionary in the new
              column.
            prompt_template: Optional. Template for the rubric generation prompt. Used for
              custom rubric generation. Mutually exclusive with `predefined_spec_name`.
              If using a 'prompt' column, use {prompt} as the placeholder. If using a
              'request' column, this template is passed to the service along
              with the content.
            generator_model_config: Optional. Configuration for the model used
              in custom rubric generation. Only used if `prompt_template` is provided.
              e.g., {"autorater_model": "gemini-2.5-flash"}.
            rubric_content_type: Optional. The type of rubric content to be
              generated. Only used if `prompt_template` is provided.
            rubric_type_ontology: Optional. A pre-defined list of allowed types
              for generated rubrics. Only used if `prompt_template` is provided.
            predefined_spec_name: Optional. The name of a Predefined Metric to use
                for rubric generation (e.g., "general_quality_v1") or a types.PrebuiltMetric object.
                Mutually exclusive with `prompt_template` and its related parameters.
            metric_spec_parameters: Optional. Parameters for the Predefined Metric,
                used to customize rubric generation. Only used if `predefined_spec_name` is set.
                Example: {"guidelines": ["The response must be in Japanese."]}
            metric: Optional. A types.Metric object containing a metric_resource_name,
                or a resource name string. If provided, this will take precedence over
                predefined_spec_name and prompt_template.
            config: Optional. Configuration for the rubric generation process.

        Returns:
            An `EvaluationDataset` with an added column named `rubric_groups` in its
            `eval_dataset_df`. Each cell in this column contains a dictionary like:
            {rubric_group_name: [list[Rubric]]}.
        """
        if isinstance(src, types.EvaluationDataset):
            if src.eval_dataset_df is not None:
                prompts_df = src.eval_dataset_df
            elif src.eval_cases:
                prompts_df = _evals_common._eval_cases_to_dataframe(src.eval_cases)
            else:
                raise ValueError(
                    "EvaluationDataset must have eval_dataset_df or eval_cases"
                    " populated."
                )
        elif isinstance(src, (str, pd.DataFrame)):
            try:
                prompts_df = _evals_common._load_dataframe(self._api_client, src)
            except Exception as e:
                raise ValueError(
                    f"Failed to load prompt dataset from source: {src}. Error: {e}"
                )
        else:
            raise TypeError(
                "Unsupported type for src. Must be str, pd.DataFrame, or types.EvaluationDataset."
            )

        if "prompt" not in prompts_df.columns and "request" not in prompts_df.columns:
            raise ValueError("Loaded dataset must have a 'prompt' or 'request' column.")
        if not rubric_group_name:
            raise ValueError("rubric_group_name cannot be empty.")

        input_column = "request" if "request" in prompts_df.columns else "prompt"
        logger.info(
            "Generating rubrics for %d prompts from column '%s', group: '%s'...",
            len(prompts_df),
            input_column,
            rubric_group_name,
        )
        all_rubric_groups: list[dict[str, list[types.Rubric]]] = []

        actual_metric_resource_name = None
        if metric:
            if isinstance(metric, str) and metric.startswith("projects/"):
                actual_metric_resource_name = metric
            else:
                metric_obj = (
                    types.Metric.model_validate(metric)
                    if isinstance(metric, dict)
                    else metric
                )
                actual_metric_resource_name = getattr(
                    metric_obj, "metric_resource_name", None
                )
                if not actual_metric_resource_name:
                    raise ValueError(
                        "The provided Metric object must have metric_resource_name set."
                    )

        rubric_gen_spec = None
        predefined_spec = None

        if actual_metric_resource_name:
            # Precedence: Registered metric resource overrides everything else.
            predefined_spec = None
            rubric_gen_spec = None
        elif predefined_spec_name:
            if prompt_template:
                logger.warning(
                    "prompt_template is ignored when predefined_spec_name is provided."
                )
            if generator_model_config:
                logger.warning(
                    "generator_model_config is ignored when predefined_spec_name is provided."
                )
            if rubric_content_type:
                logger.warning(
                    "rubric_content_type is ignored when predefined_spec_name is provided."
                )
            if rubric_type_ontology:
                logger.warning(
                    "rubric_type_ontology is ignored when predefined_spec_name is provided."
                )

            if isinstance(predefined_spec_name, str):
                actual_predefined_spec_name = predefined_spec_name
            elif hasattr(
                predefined_spec_name, "resolve"
            ):  # Check if it's LazyLoadedPrebuiltMetric
                try:
                    resolved_metric = predefined_spec_name.resolve(self._api_client)
                    actual_predefined_spec_name = resolved_metric.name
                except Exception as e:
                    raise ValueError(f"Failed to resolve PrebuiltMetric: {e}")
            else:
                raise TypeError(
                    "predefined_spec_name must be a string or types.PrebuiltMetric"
                )

            if not actual_predefined_spec_name:
                raise ValueError(
                    "Could not determine metric_spec_name from predefined_spec_name"
                )

            predefined_spec = genai_types.PredefinedMetricSpec(
                metric_spec_name=actual_predefined_spec_name,
                metric_spec_parameters=metric_spec_parameters,
            )
        elif prompt_template:
            if metric_spec_parameters:
                logger.warning(
                    "metric_spec_parameters is ignored when prompt_template is provided."
                )
            spec_dict = {
                "prompt_template": prompt_template,
                "rubric_content_type": rubric_content_type,
                "rubric_type_ontology": rubric_type_ontology,
                "generator_model_config": generator_model_config,
            }
            spec_dict = {k: v for k, v in spec_dict.items() if v is not None}
            rubric_gen_spec = genai_types.RubricGenerationSpec.model_validate(spec_dict)
        else:
            raise ValueError(
                "Either metric, predefined_spec_name or prompt_template must be provided."
            )

        for _, row in prompts_df.iterrows():
            input_data = row[input_column]
            if isinstance(input_data, str):
                contents = [
                    genai_types.Content(parts=[genai_types.Part(text=input_data)])
                ]
            elif isinstance(input_data, list):
                contents = input_data
            else:
                logger.warning(
                    f"Skipping row: Unexpected input format in column '{input_column}'."
                )
                all_rubric_groups.append({rubric_group_name: []})
                continue

            try:
                response = self._generate_rubrics(
                    contents=contents,
                    rubric_generation_spec=rubric_gen_spec,
                    predefined_rubric_generation_spec=predefined_spec,
                    metric_resource_name=actual_metric_resource_name,
                    config=config,
                )
                rubric_group = {rubric_group_name: response.generated_rubrics}
                all_rubric_groups.append(rubric_group)
            except Exception as e:
                logger.error(
                    "Rubric generation failed for input: %s... Error: %s",
                    str(input_data)[:50],
                    e,
                    exc_info=True,
                )
                all_rubric_groups.append({rubric_group_name: []})

        prompts_with_rubrics = prompts_df.copy()
        prompts_with_rubrics["rubric_groups"] = all_rubric_groups
        logger.info(
            f"Rubric generation complete. Added column 'rubric_groups' with key '{rubric_group_name}'."
        )
        return types.EvaluationDataset(eval_dataset_df=prompts_with_rubrics)

    def get_evaluation_run(
        self,
        *,
        name: str,
        include_evaluation_items: bool = False,
        config: Optional[types.GetEvaluationRunConfigOrDict] = None,
    ) -> types.EvaluationRun:
        """Retrieves an EvaluationRun from the resource name.
        Args:
          name: The resource name of the EvaluationRun. Format:
            `projects/{project}/locations/{location}/evaluationRuns/{evaluation_run}`
          include_evaluation_items: Whether to include the evaluation items in the
            response.
          config: The optional configuration for the evaluation run. Must be a dict or
              `types.GetEvaluationRunConfigOrDict` type.

        Returns:
          The evaluation run.
        Raises:
          ValueError: If the name is empty or invalid.
        """
        if not name:
            raise ValueError("name cannot be empty.")
        if name.startswith("projects/"):
            name = name.split("/")[-1]
        result = self._get_evaluation_run(name=name, config=config)
        if include_evaluation_items:
            eval_result, eval_item_map = _evals_common._convert_evaluation_run_results(
                self._api_client,
                result.evaluation_run_results,
                result.inference_configs,
            )
            result.evaluation_item_results = eval_result
            # Bypass pydantic validation (extra='forbid') for this internal field.
            object.__setattr__(result, "_eval_item_map", eval_item_map)
        return result

    def create_evaluation_run(
        self,
        *,
        dataset: Union[types.EvaluationRunDataSource, types.EvaluationDataset],
        dest: str,
        metrics: list[types.EvaluationRunMetricOrDict],
        name: Optional[str] = None,
        display_name: Optional[str] = None,
        agent_info: Optional[evals_types.AgentInfoOrDict] = None,
        agent: Optional[str] = None,
        user_simulator_config: Optional[evals_types.UserSimulatorConfigOrDict] = None,
        inference_configs: Optional[
            dict[str, types.EvaluationRunInferenceConfigOrDict]
        ] = None,
        labels: Optional[dict[str, str]] = None,
        loss_analysis_metrics: Optional[list[Union[str, types.MetricOrDict]]] = None,
        loss_analysis_configs: Optional[list[types.LossAnalysisConfigOrDict]] = None,
        red_teaming_config: Optional[types.RedTeamingAnalysisConfigOrDict] = None,
        config: Optional[types.CreateEvaluationRunConfigOrDict] = None,
    ) -> types.EvaluationRun:
        """Creates an EvaluationRun.

        Args:
          dataset: The dataset to evaluate. Either an EvaluationRunDataSource or an EvaluationDataset.
          dest: The GCS URI prefix to write the evaluation results to.
          metrics: The list of metrics to evaluate.
          name: The name of the evaluation run.
          display_name: The display name of the evaluation run.
          agent_info: The agent info to evaluate. Mutually exclusive with
              `inference_configs`.
          agent: The agent resource name in str type. Accepts either an Agent
              Engine resource name
              `projects/{project}/locations/{location}/reasoningEngines/{reasoning_engine_id}`
              or a Gemini Agent (Vertex AI Agent) resource name
              `projects/{project}/locations/{location}/agents/{agent}`. When a Gemini
              Agent resource is provided, the backend scrapes the agent to produce
              agent responses. If an Agent Engine resource name is provided, runs
              inference with the deployed agent to get agent responses for evaluation.
              The `agent` parameter is required if `agent_info` is provided.
          user_simulator_config: The user simulator configuration for agent evaluation.
              If `agent_info` is provided without `inference_configs`, this config is used
              to automatically construct the inference configuration. If not specified,
              or if `max_turn` is not set, `max_turn` defaults to 5.
              The `model_name` inside this config can be either a full model path or a
              short model name, e.g. `gemini-3-preview-flash`.
          inference_configs: The candidate to inference config map for the evaluation run.
              The key is the candidate name, and the value is the inference config.
              If provided, `agent_info` must be None. If omitted and `agent_info` is provided,
              this will be automatically constructed using `agent_info` and `user_simulator_config`.
              The `model` field of an inference config accepts a short Gemini model
              name (e.g. `gemini-2.5-flash`), which is automatically expanded to a
              fully-qualified resource name using the client's project and location,
              or an already fully-qualified publisher-model or endpoint resource
              name.
              Example:
              {"candidate-1": types.EvaluationRunInferenceConfig(model="gemini-2.5-flash")}
          labels: The labels to apply to the evaluation run.
          loss_analysis_metrics: This field is experimental and may change in future
              versions. Optional list of metrics to run loss analysis on. The
              candidate is auto-inferred from ``inference_configs`` or
              ``agent_info`` when there is exactly one candidate. Each metric can be
              a string (e.g., ``"multi_turn_task_success_v1"``), a ``Metric``
              object, or a ``RubricMetric`` enum
              (e.g., ``types.RubricMetric.MULTI_TURN_TASK_SUCCESS``). Loss analysis
              runs after metric calculation completes.
              Mutually exclusive with ``loss_analysis_configs``.
              Example::

                  loss_analysis_metrics=[
                      types.RubricMetric.MULTI_TURN_TASK_SUCCESS,
                      types.RubricMetric.MULTI_TURN_TOOL_USE_QUALITY,
                  ]
          loss_analysis_configs: This field is experimental and may change in future
              versions. Optional list of ``LossAnalysisConfig`` objects for full
              control over loss analysis, including explicit candidate and
              advanced options like ``predefined_taxonomy`` and
              ``max_top_cluster_count``. Mutually exclusive with
              ``loss_analysis_metrics``.
          config: The configuration for the evaluation run.
            - allow_cross_region_model: Allows the evaluation run to use cross
              region models. When this flag is set, the service may route traffic to
              other regions if a model is unavailable in the current region (e.g.,
              to a `global`endpoint). If a fully-qualified model endpoint resource
              name with a different region than the run location is provided
              elsewhere in the runconfig, this flag must be set to true or the
              request will fail.

        Returns:
            The created evaluation run.
        """
        if loss_analysis_metrics and loss_analysis_configs:
            raise ValueError(
                "At most one of loss_analysis_metrics or loss_analysis_configs"
                " can be provided."
            )
        if agent_info and inference_configs:
            raise ValueError(
                "At most one of agent_info or inference_configs can be provided."
            )
        parsed_agent_info = (
            evals_types.AgentInfo.model_validate(agent_info)
            if isinstance(agent_info, dict)
            else (agent_info or evals_types.AgentInfo())
        )

        if not config:
            config = types.CreateEvaluationRunConfig()
        if isinstance(config, dict):
            config = types.CreateEvaluationRunConfig.model_validate(config)

        # Auto-construct inference_configs when agent_info is explicitly
        # provided (existing behavior) OR an agent resource is provided
        # (allows omitting agent_info for both Gemini Agents and Agent
        # Engine). The server skips inference per-item when a
        # CandidateResponse with a matching candidate name already exists,
        # so it is safe to always send inference_configs.
        _should_auto_infer = not inference_configs and (agent_info or agent)
        if _should_auto_infer:
            if not parsed_agent_info.name:
                # Prefer the dataset's candidate_name (set by run_inference)
                # so the inference_configs key matches the CandidateResponse
                # and the server correctly skips already-completed items.
                if (
                    isinstance(dataset, types.EvaluationDataset)
                    and dataset.candidate_name
                ):
                    parsed_agent_info.name = dataset.candidate_name
                else:
                    parsed_agent_info.name = _evals_common._DEFAULT_CANDIDATE_NAME
            parsed_user_simulator_config = (
                evals_types.UserSimulatorConfig.model_validate(user_simulator_config)
                if isinstance(user_simulator_config, dict)
                else (user_simulator_config or evals_types.UserSimulatorConfig())
            )
            if getattr(parsed_user_simulator_config, "max_turn", None) is None:
                parsed_user_simulator_config.max_turn = 5

            candidate_name = (
                parsed_agent_info.name or _evals_common._DEFAULT_CANDIDATE_NAME
            )
            if agent and _evals_common._is_gemini_agent_resource(agent):
                agent_run_config = types.AgentRunConfig(
                    gemini_agent_config=types.GeminiAgentConfig(gemini_agent=agent),
                    user_simulator_config=parsed_user_simulator_config,
                )
            else:
                agent_run_config = types.AgentRunConfig(
                    agent_engine=agent,
                    user_simulator_config=parsed_user_simulator_config,
                )
            inference_configs = {
                candidate_name: types.EvaluationRunInferenceConfig(
                    agent_configs=parsed_agent_info.agents,
                    agent_run_config=agent_run_config,
                )
            }

        if isinstance(dataset, types.EvaluationDataset):
            _evals_utils._validate_dataset_agent_data(dataset, inference_configs)
        resolved_dataset = _evals_common._resolve_dataset(
            self._api_client, dataset, dest, parsed_agent_info
        )
        output_config = genai_types.OutputConfig(
            gcs_destination=genai_types.GcsDestination(output_uri_prefix=dest)
        )
        resolved_metrics = _evals_common._resolve_evaluation_run_metrics(
            metrics, self._api_client
        )
        resolved_loss_configs = _evals_utils._resolve_eval_run_loss_configs(
            loss_analysis_metrics=loss_analysis_metrics,
            loss_analysis_configs=loss_analysis_configs,
            inference_configs=inference_configs,
        )
        resolved_analysis_configs = _evals_utils._resolve_red_teaming_config(
            red_teaming_config
        )
        evaluation_config = types.EvaluationRunConfig(
            output_config=output_config,
            metrics=resolved_metrics,
            loss_analysis_config=resolved_loss_configs,
            allow_cross_region_model=getattr(config, "allow_cross_region_model", None),
        )
        resolved_inference_configs = _evals_common._resolve_inference_configs(
            self._api_client, resolved_dataset, inference_configs, parsed_agent_info
        )
        resolved_labels = _evals_common._add_evaluation_run_labels(labels, agent)
        resolved_name = name or f"evaluation_run_{uuid.uuid4()}"
        return self._create_evaluation_run(
            name=resolved_name,
            display_name=display_name or resolved_name,
            data_source=resolved_dataset,
            evaluation_config=evaluation_config,
            inference_configs=resolved_inference_configs,
            analysis_configs=resolved_analysis_configs,
            labels=resolved_labels,
            config=config,
        )

    def get_evaluation_set(
        self,
        *,
        name: str,
        config: Optional[types.GetEvaluationSetConfigOrDict] = None,
    ) -> types.EvaluationSet:
        """Retrieves an EvaluationSet from the resource name.

        Args:
          name: The resource name of the EvaluationSet. Format:
            `projects/{project}/locations/{location}/evaluationSets/{evaluation_set}`
          config: The optional configuration for the evaluation set. Must be a dict or
              `types.GetEvaluationSetConfigOrDict` type.

        Returns:
          The evaluation set.
        """

        if not name:
            raise ValueError("name cannot be empty.")
        if name.startswith("projects/"):
            name = name.split("/")[-1]
        return self._get_evaluation_set(name=name, config=config)

    def get_evaluation_item(
        self,
        *,
        name: str,
        config: Optional[types.GetEvaluationItemConfigOrDict] = None,
    ) -> types.EvaluationItem:
        """Retrieves an EvaluationItem from the resource name.

        Args:
          name: The resource name of the EvaluationItem. Format:
            `projects/{project}/locations/{location}/evaluationItems/{evaluation_item}`
          config: The optional configuration for the evaluation item. Must be a dict or
              `types.GetEvaluationItemConfigOrDict` type.

        Returns:
          The evaluation item.
        """
        if not name:
            raise ValueError("name cannot be empty.")
        if name.startswith("projects/"):
            name = name.split("/")[-1]
        result = self._get_evaluation_item(name=name, config=config)
        if (
            result.gcs_uri
            and result.evaluation_item_type == types.EvaluationItemType.RESULT
        ):
            result.evaluation_response = (
                _evals_common._convert_gcs_to_evaluation_item_result(
                    self._api_client, result.gcs_uri
                )
            )
        elif (
            result.gcs_uri
            and result.evaluation_item_type == types.EvaluationItemType.REQUEST
        ):
            result.evaluation_request = (
                _evals_common._convert_gcs_to_evaluation_item_request(
                    self._api_client, result.gcs_uri
                )
            )
        return result

    def create_evaluation_item(
        self,
        *,
        evaluation_item_type: types.EvaluationItemType,
        gcs_uri: str,
        display_name: Optional[str] = None,
        config: Optional[types.CreateEvaluationItemConfigOrDict] = None,
    ) -> types.EvaluationItem:
        """Creates an EvaluationItem.

        Args:
          evaluation_item_type: The type of the evaluation item.
          gcs_uri: The GCS URI of the evaluation item.
          display_name: The display name of the evaluation item.
          config: The optional configuration for the evaluation item. Must be a dict or
              `types.CreateEvaluationItemConfigOrDict` type.

        Returns:
          The evaluation item.
        """
        return self._create_evaluation_item(
            evaluation_item_type=evaluation_item_type,
            gcs_uri=gcs_uri,
            display_name=display_name,
            config=config,
        )

    def create_evaluation_set(
        self,
        *,
        evaluation_items: list[str],
        display_name: Optional[str] = None,
        config: Optional[types.CreateEvaluationSetConfigOrDict] = None,
    ) -> types.EvaluationSet:
        """Creates an EvaluationSet.

        Args:
          evaluation_items: The list of evaluation item names. Format:
            `projects/{project}/locations/{location}/evaluationItems/{evaluation_item}`
          display_name: The display name of the evaluation set.
          config: The optional configuration for the evaluation set. Must be a dict or
              `types.CreateEvaluationSetConfigOrDict` type.

        Returns:
          The evaluation set.
        """
        return self._create_evaluation_set(
            evaluation_items=evaluation_items,
            display_name=display_name,
            config=config,
        )

    def generate_conversation_scenarios(
        self,
        *,
        agent_info: Optional[evals_types.AgentInfoOrDict] = None,
        agent: Optional[str] = None,
        config: evals_types.UserScenarioGenerationConfigOrDict,
        allow_cross_region_model: Optional[bool] = None,
    ) -> types.EvaluationDataset:
        """Generates an evaluation dataset with user scenarios,
           which helps to generate conversations between a simulated user
           and the agent under test.

        Exactly one of `agent_info` or `agent` must be provided. When `agent` is
        a Gemini Agents API agent resource name, the agent config is resolved
        server-side via ``gemini_agent_config``.

        Args:
            agent_info: The agent info to generate user scenarios for. Mutually
                exclusive with `agent`.
            agent: A Gemini Agents API agent resource name
                (`projects/{p}/locations/{l}/agents/{name}`). When provided, the
                server resolves the agent config directly. Mutually exclusive
                with `agent_info`.
            config: Configuration for generating user scenarios.
            allow_cross_region_model: Opt-in flag to authorize cross-region
                routing for model inference.

        Returns:
            An EvaluationDataset containing the generated user scenarios.
        """
        if agent is not None and agent_info is not None:
            raise ValueError(
                "Only one of `agent` or `agent_info` may be provided, not both."
            )
        if agent is None and agent_info is None:
            raise ValueError("One of `agent` or `agent_info` must be provided.")
        if agent is not None:
            if not _evals_common._is_gemini_agent_resource(agent):
                raise ValueError(
                    "`agent` must be a Gemini Agents API agent resource name of the"
                    " form projects/{project}/locations/{location}/agents/{agent}."
                )
            response = self._generate_user_scenarios(
                gemini_agent_config=types.GeminiAgentConfig(gemini_agent=agent),
                user_scenario_generation_config=config,
                allow_cross_region_model=allow_cross_region_model,
            )
        else:
            parsed_agent_info = (
                evals_types.AgentInfo.model_validate(agent_info)
                if isinstance(agent_info, dict)
                else agent_info
            )
            response = self._generate_user_scenarios(
                agents=parsed_agent_info.agents,
                root_agent_id=parsed_agent_info.root_agent_id,
                user_scenario_generation_config=config,
                allow_cross_region_model=allow_cross_region_model,
            )
        return _evals_utils._postprocess_user_scenarios_response(response)

    def generate_loss_clusters(
        self,
        *,
        eval_result: types.EvaluationResult,
        metric: Optional[Union[str, types.MetricOrDict]] = None,
        candidate: Optional[str] = None,
        config: Optional[types.LossAnalysisConfigOrDict] = None,
    ) -> types.GenerateLossClustersResponse:
        """Generates loss clusters from evaluation results.

        Analyzes "Pass/Fail" signals from rubric-based autoraters and groups
        them into semantic "Loss Patterns" (e.g., "Hallucination of Action").

        This method calls the GenerateLossClusters LRO and polls until
        completion, returning the results directly.

        If ``metric`` or ``candidate`` are not provided, they will be
        auto-inferred from ``eval_result`` when unambiguous (i.e., when the
        eval result contains exactly one metric or one candidate). For
        multi-metric or multi-candidate evaluations, provide them explicitly.

        Available candidate names can be found in
        ``eval_result.metadata.candidate_names``.

        Note: This API is only available in the ``global`` region.

        Args:
            eval_result: The EvaluationResult object returned from
                client.evals.evaluate().
            metric: The metric to analyze. Can be a metric name string
                (e.g., "multi_turn_task_success_v1"), a Metric object, or a
                RubricMetric enum (e.g., types.RubricMetric.MULTI_TURN_TASK_SUCCESS).
                If not provided and config does not specify it, auto-inferred
                from eval_result.
            candidate: The candidate to analyze. If not provided and config
                does not specify it, auto-inferred from eval_result.
            config: Optional LossAnalysisConfig with additional options
                (predefined_taxonomy, max_top_cluster_count). Can also
                specify metric/candidate, but explicit arguments take
                precedence.

        Returns:
            A GenerateLossClustersResponse containing the analysis results.
            Call .show() to visualize, or access .results for individual
            LossAnalysisResult objects (each with their own .show()).
        """
        metric_name = _evals_utils._resolve_metric_name(metric)
        parsed_config = (
            types.LossAnalysisConfig.model_validate(config)
            if isinstance(config, dict)
            else config
        )
        resolved_config = _evals_utils._resolve_loss_analysis_config(
            eval_result=eval_result,
            config=parsed_config,
            metric=metric_name,
            candidate=candidate,
        )
        operation = self._generate_loss_clusters(
            inline_results=[eval_result],
            configs=[resolved_config],
        )
        completed = _evals_utils._poll_operation(
            api_client=self._api_client,
            operation=operation,
        )
        if completed.error:
            raise RuntimeError(f"Loss analysis operation failed: {completed.error}")
        if completed.response is None:
            raise RuntimeError(
                "Loss analysis operation completed but returned no response."
            )
        _evals_utils._enrich_loss_response_with_rubric_descriptions(
            completed.response, eval_result
        )
        return completed.response

    def create_evaluation_metric(
        self,
        *,
        display_name: Optional[str] = None,
        description: Optional[str] = None,
        metric: Optional[types.MetricOrDict] = None,
        config: Optional[types.CreateEvaluationMetricConfigOrDict] = None,
    ) -> str:
        """Creates an EvaluationMetric."""
        if metric and not isinstance(metric, dict):
            # metric is now Metric | LazyLoadedPrebuiltMetric (RubricMetric)
            # Mypy correctly narrows the type here, so cast is not needed.
            resolved_metrics = _evals_common._resolve_metrics(
                [metric], self._api_client
            )
            metric = resolved_metrics[0]

        # Add fallback logic for display_name
        if display_name is None and metric:
            if isinstance(metric, dict):
                display_name = metric.get("name")
            else:
                display_name = getattr(metric, "name", None)

        result = self._create_evaluation_metric(
            display_name=display_name,
            description=description,
            metric=metric,
            config=config,
        )
        # result.name is Optional[str], but we know it's always returned on creation
        return cast(str, result.name)

    def get_evaluation_metric(
        self,
        *,
        metric_resource_name: str,
        config: Optional[types.GetEvaluationMetricConfigOrDict] = None,
    ) -> types.EvaluationMetric:
        """Retrieves an EvaluationMetric from the resource name."""
        return self._get_evaluation_metric(
            metric_resource_name=metric_resource_name,
            config=config,
        )

    def list_evaluation_metrics(
        self,
        *,
        filter: Optional[str] = None,
        order_by: Optional[str] = None,
        config: Optional[types.ListEvaluationMetricsConfigOrDict] = None,
    ) -> types.ListEvaluationMetricsResponse:
        """Lists EvaluationMetrics.

        Args:
          filter: An expression for filtering the results of the request. For
            field names both snake_case and camelCase are supported. For more
            information about filter syntax, see
            `AIP-160 <https://google.aip.dev/160>`_.
            Example: ``'display_name="my_metric"'``.
          order_by: A comma-separated list of fields to order by, sorted in
            ascending order by default. Use ``desc`` after a field name for
            descending. Example: ``"create_time desc"``.
          config: Optional configuration for the list operation, including
            pagination (``page_size``, ``page_token``), ``filter``, and
            ``order_by``. Top-level ``filter`` and ``order_by`` arguments
            take precedence over values set in ``config``.

        Returns:
          The list evaluation metrics response.
        """
        if config is None:
            config = types.ListEvaluationMetricsConfig()
        if isinstance(config, dict):
            config = types.ListEvaluationMetricsConfig.model_validate(config)
        if filter is not None:
            config.filter = filter
        if order_by is not None:
            config.order_by = order_by
        return self._list_evaluation_metrics(
            config=config,
        )

    def delete_evaluation_metric(
        self,
        *,
        metric_resource_name: str,
        config: Optional[types.DeleteEvaluationMetricConfigOrDict] = None,
    ) -> None:
        """Deletes an EvaluationMetric.

        Args:
          metric_resource_name: The resource name of the EvaluationMetric to delete.
            Format:
            `projects/{project}/locations/{location}/evaluationMetrics/{evaluation_metric}`
          config: The optional configuration for the delete operation.
        """
        self._delete_evaluation_metric(
            metric_resource_name=metric_resource_name,
            config=config,
        )


class AsyncEvals(_api_module.BaseModule):

    async def create_evaluation_experiment(
        self,
        *,
        display_name: Optional[str] = None,
        labels: Optional[dict[str, str]] = None,
        merge_strategy: Optional[types.EvaluationExperimentMergeStrategy] = None,
        metadata: Optional[dict[str, Any]] = None,
        config: Optional[types.CreateEvaluationExperimentConfigOrDict] = None,
    ) -> types.EvaluationExperiment:
        """
        Creates an EvaluationExperiment.

        Args:
          display_name: The display name of the evaluation experiment.
          labels: Labels for the evaluation experiment.
          merge_strategy: Merge strategy for the evaluation experiment.
          metadata: Metadata about the evaluation experiment, can be used by the
            caller to store additional tracking information about the experiment.
          config: Optional configuration for the create operation.

        Returns:
          The created evaluation experiment.

        .. code-block:: python

          eval_experiment = client.evals.create_evaluation_experiment(
              display_name="my-experiment"
          )

        """

        parameter_model = types._CreateEvaluationExperimentParameters(
            display_name=display_name,
            labels=labels,
            merge_strategy=merge_strategy,
            metadata=metadata,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError(
                "This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
            )
        else:
            request_dict = _CreateEvaluationExperimentParameters_to_vertex(
                parameter_model
            )
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "evaluationExperiments".format_map(request_url_dict)
            else:
                path = "evaluationExperiments"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = await self._api_client.async_request(
            "post", path, request_dict, http_options
        )

        response_dict = {} if not response.body else json.loads(response.body)

        return_value = types.EvaluationExperiment._from_response(
            response=response_dict,
            kwargs=(
                {
                    "config": {
                        "response_schema": getattr(
                            parameter_model.config, "response_schema", None
                        ),
                        "response_json_schema": getattr(
                            parameter_model.config, "response_json_schema", None
                        ),
                        "include_all_fields": getattr(
                            parameter_model.config, "include_all_fields", None
                        ),
                    }
                }
                if getattr(parameter_model, "config", None)
                else {}
            ),
        )

        self._api_client._verify_response(return_value)
        return return_value

    async def _create_evaluation_item(
        self,
        *,
        evaluation_item_type: str,
        gcs_uri: str,
        display_name: Optional[str] = None,
        config: Optional[types.CreateEvaluationItemConfigOrDict] = None,
    ) -> types.EvaluationItem:
        """
        Creates an EvaluationItem.
        """

        parameter_model = types._CreateEvaluationItemParameters(
            evaluation_item_type=evaluation_item_type,
            gcs_uri=gcs_uri,
            display_name=display_name,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError(
                "This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
            )
        else:
            request_dict = _CreateEvaluationItemParameters_to_vertex(parameter_model)
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "evaluationItems".format_map(request_url_dict)
            else:
                path = "evaluationItems"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = await self._api_client.async_request(
            "post", path, request_dict, http_options
        )

        response_dict = {} if not response.body else json.loads(response.body)

        return_value = types.EvaluationItem._from_response(
            response=response_dict,
            kwargs=(
                {
                    "config": {
                        "response_schema": getattr(
                            parameter_model.config, "response_schema", None
                        ),
                        "response_json_schema": getattr(
                            parameter_model.config, "response_json_schema", None
                        ),
                        "include_all_fields": getattr(
                            parameter_model.config, "include_all_fields", None
                        ),
                    }
                }
                if getattr(parameter_model, "config", None)
                else {}
            ),
        )

        self._api_client._verify_response(return_value)
        return return_value

    async def _create_evaluation_metric(
        self,
        *,
        display_name: Optional[str] = None,
        description: Optional[str] = None,
        metric: Optional[types.MetricOrDict] = None,
        config: Optional[types.CreateEvaluationMetricConfigOrDict] = None,
    ) -> types.EvaluationMetric:
        """
        Creates an EvaluationMetric.
        """

        parameter_model = types._CreateEvaluationMetricParameters(
            display_name=display_name,
            description=description,
            metric=metric,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError(
                "This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
            )
        else:
            request_dict = _CreateEvaluationMetricParameters_to_vertex(parameter_model)
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "evaluationMetrics".format_map(request_url_dict)
            else:
                path = "evaluationMetrics"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = await self._api_client.async_request(
            "post", path, request_dict, http_options
        )

        response_dict = {} if not response.body else json.loads(response.body)

        if self._api_client.vertexai:
            response_dict = _EvaluationMetric_from_vertex(response_dict)

        return_value = types.EvaluationMetric._from_response(
            response=response_dict,
            kwargs=(
                {
                    "config": {
                        "response_schema": getattr(
                            parameter_model.config, "response_schema", None
                        ),
                        "response_json_schema": getattr(
                            parameter_model.config, "response_json_schema", None
                        ),
                        "include_all_fields": getattr(
                            parameter_model.config, "include_all_fields", None
                        ),
                    }
                }
                if getattr(parameter_model, "config", None)
                else {}
            ),
        )

        self._api_client._verify_response(return_value)
        return return_value

    async def _create_evaluation_run(
        self,
        *,
        name: Optional[str] = None,
        display_name: Optional[str] = None,
        data_source: types.EvaluationRunDataSourceOrDict,
        evaluation_config: types.EvaluationRunConfigOrDict,
        labels: Optional[dict[str, str]] = None,
        inference_configs: Optional[
            dict[str, types.EvaluationRunInferenceConfigOrDict]
        ] = None,
        config: Optional[types.CreateEvaluationRunConfigOrDict] = None,
        analysis_configs: Optional[list[types.AnalysisConfigOrDict]] = None,
    ) -> types.EvaluationRun:
        """
        Creates an EvaluationRun.
        """

        parameter_model = types._CreateEvaluationRunParameters(
            name=name,
            display_name=display_name,
            data_source=data_source,
            evaluation_config=evaluation_config,
            labels=labels,
            inference_configs=inference_configs,
            config=config,
            analysis_configs=analysis_configs,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError(
                "This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
            )
        else:
            request_dict = _CreateEvaluationRunParameters_to_vertex(parameter_model)
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "evaluationRuns".format_map(request_url_dict)
            else:
                path = "evaluationRuns"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = await self._api_client.async_request(
            "post", path, request_dict, http_options
        )

        response_dict = {} if not response.body else json.loads(response.body)

        if self._api_client.vertexai:
            response_dict = _EvaluationRun_from_vertex(response_dict)

        return_value = types.EvaluationRun._from_response(
            response=response_dict,
            kwargs=(
                {
                    "config": {
                        "response_schema": getattr(
                            parameter_model.config, "response_schema", None
                        ),
                        "response_json_schema": getattr(
                            parameter_model.config, "response_json_schema", None
                        ),
                        "include_all_fields": getattr(
                            parameter_model.config, "include_all_fields", None
                        ),
                    }
                }
                if getattr(parameter_model, "config", None)
                else {}
            ),
        )

        self._api_client._verify_response(return_value)
        return return_value

    async def _create_evaluation_set(
        self,
        *,
        evaluation_items: list[str],
        display_name: Optional[str] = None,
        config: Optional[types.CreateEvaluationSetConfigOrDict] = None,
    ) -> types.EvaluationSet:
        """
        Creates an EvaluationSet.
        """

        parameter_model = types._CreateEvaluationSetParameters(
            evaluation_items=evaluation_items,
            display_name=display_name,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError(
                "This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
            )
        else:
            request_dict = _CreateEvaluationSetParameters_to_vertex(parameter_model)
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "evaluationSets".format_map(request_url_dict)
            else:
                path = "evaluationSets"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = await self._api_client.async_request(
            "post", path, request_dict, http_options
        )

        response_dict = {} if not response.body else json.loads(response.body)

        return_value = types.EvaluationSet._from_response(
            response=response_dict,
            kwargs=(
                {
                    "config": {
                        "response_schema": getattr(
                            parameter_model.config, "response_schema", None
                        ),
                        "response_json_schema": getattr(
                            parameter_model.config, "response_json_schema", None
                        ),
                        "include_all_fields": getattr(
                            parameter_model.config, "include_all_fields", None
                        ),
                    }
                }
                if getattr(parameter_model, "config", None)
                else {}
            ),
        )

        self._api_client._verify_response(return_value)
        return return_value

    async def _delete_evaluation_metric(
        self,
        *,
        metric_resource_name: str,
        config: Optional[types.DeleteEvaluationMetricConfigOrDict] = None,
    ) -> types.DeleteEvaluationMetricOperation:
        """
        Deletes an EvaluationMetric.
        """

        parameter_model = types._DeleteEvaluationMetricParameters(
            metric_resource_name=metric_resource_name,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError(
                "This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
            )
        else:
            request_dict = _DeleteEvaluationMetricParameters_to_vertex(parameter_model)
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "{evaluation_metric}".format_map(request_url_dict)
            else:
                path = "{evaluation_metric}"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = await self._api_client.async_request(
            "delete", path, request_dict, http_options
        )

        response_dict = {} if not response.body else json.loads(response.body)

        return_value = types.DeleteEvaluationMetricOperation._from_response(
            response=response_dict,
            kwargs=(
                {
                    "config": {
                        "response_schema": getattr(
                            parameter_model.config, "response_schema", None
                        ),
                        "response_json_schema": getattr(
                            parameter_model.config, "response_json_schema", None
                        ),
                        "include_all_fields": getattr(
                            parameter_model.config, "include_all_fields", None
                        ),
                    }
                }
                if getattr(parameter_model, "config", None)
                else {}
            ),
        )

        self._api_client._verify_response(return_value)
        return return_value

    async def _evaluate_instances(
        self,
        *,
        bleu_input: Optional[types.BleuInputOrDict] = None,
        exact_match_input: Optional[types.ExactMatchInputOrDict] = None,
        rouge_input: Optional[types.RougeInputOrDict] = None,
        pointwise_metric_input: Optional[types.PointwiseMetricInputOrDict] = None,
        pairwise_metric_input: Optional[types.PairwiseMetricInputOrDict] = None,
        tool_call_valid_input: Optional[types.ToolCallValidInputOrDict] = None,
        tool_name_match_input: Optional[types.ToolNameMatchInputOrDict] = None,
        tool_parameter_key_match_input: Optional[
            types.ToolParameterKeyMatchInputOrDict
        ] = None,
        tool_parameter_kv_match_input: Optional[
            types.ToolParameterKVMatchInputOrDict
        ] = None,
        rubric_based_metric_input: Optional[types.RubricBasedMetricInputOrDict] = None,
        autorater_config: Optional[genai_types.AutoraterConfigOrDict] = None,
        metrics: Optional[list[types.MetricOrDict]] = None,
        instance: Optional[types.EvaluationInstanceOrDict] = None,
        metric_sources: Optional[list[types.MetricSourceOrDict]] = None,
        config: Optional[types.EvaluateInstancesConfigOrDict] = None,
    ) -> types.EvaluateInstancesResponse:
        """
        Evaluates instances based on a given metric.
        """

        parameter_model = types._EvaluateInstancesRequestParameters(
            bleu_input=bleu_input,
            exact_match_input=exact_match_input,
            rouge_input=rouge_input,
            pointwise_metric_input=pointwise_metric_input,
            pairwise_metric_input=pairwise_metric_input,
            tool_call_valid_input=tool_call_valid_input,
            tool_name_match_input=tool_name_match_input,
            tool_parameter_key_match_input=tool_parameter_key_match_input,
            tool_parameter_kv_match_input=tool_parameter_kv_match_input,
            rubric_based_metric_input=rubric_based_metric_input,
            autorater_config=autorater_config,
            metrics=metrics,
            instance=instance,
            metric_sources=metric_sources,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError(
                "This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
            )
        else:
            request_dict = _EvaluateInstancesRequestParameters_to_vertex(
                parameter_model
            )
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = ":evaluateInstances".format_map(request_url_dict)
            else:
                path = ":evaluateInstances"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = await self._api_client.async_request(
            "post", path, request_dict, http_options
        )

        response_dict = {} if not response.body else json.loads(response.body)

        return_value = types.EvaluateInstancesResponse._from_response(
            response=response_dict,
            kwargs=(
                {
                    "config": {
                        "response_schema": getattr(
                            parameter_model.config, "response_schema", None
                        ),
                        "response_json_schema": getattr(
                            parameter_model.config, "response_json_schema", None
                        ),
                        "include_all_fields": getattr(
                            parameter_model.config, "include_all_fields", None
                        ),
                    }
                }
                if getattr(parameter_model, "config", None)
                else {}
            ),
        )

        self._api_client._verify_response(return_value)
        return return_value

    async def _generate_user_scenarios(
        self,
        *,
        location: Optional[str] = None,
        agents: Optional[dict[str, evals_types.AgentConfigOrDict]] = None,
        root_agent_id: Optional[str] = None,
        user_scenario_generation_config: Optional[
            evals_types.UserScenarioGenerationConfigOrDict
        ] = None,
        config: Optional[types.GenerateUserScenariosConfigOrDict] = None,
        allow_cross_region_model: Optional[bool] = None,
        gemini_agent_config: Optional[types.GeminiAgentConfigOrDict] = None,
    ) -> types.GenerateUserScenariosResponse:
        """
        Generates user scenarios for agent evaluation.
        """

        parameter_model = types._GenerateUserScenariosParameters(
            location=location,
            agents=agents,
            root_agent_id=root_agent_id,
            user_scenario_generation_config=user_scenario_generation_config,
            config=config,
            allow_cross_region_model=allow_cross_region_model,
            gemini_agent_config=gemini_agent_config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError(
                "This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
            )
        else:
            request_dict = _GenerateUserScenariosParameters_to_vertex(parameter_model)
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = ":generateUserScenarios".format_map(request_url_dict)
            else:
                path = ":generateUserScenarios"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = await self._api_client.async_request(
            "post", path, request_dict, http_options
        )

        response_dict = {} if not response.body else json.loads(response.body)

        return_value = types.GenerateUserScenariosResponse._from_response(
            response=response_dict,
            kwargs=(
                {
                    "config": {
                        "response_schema": getattr(
                            parameter_model.config, "response_schema", None
                        ),
                        "response_json_schema": getattr(
                            parameter_model.config, "response_json_schema", None
                        ),
                        "include_all_fields": getattr(
                            parameter_model.config, "include_all_fields", None
                        ),
                    }
                }
                if getattr(parameter_model, "config", None)
                else {}
            ),
        )

        self._api_client._verify_response(return_value)
        return return_value

    async def _generate_loss_clusters(
        self,
        *,
        location: Optional[str] = None,
        evaluation_set: Optional[str] = None,
        inline_results: Optional[list[types.EvaluationResultOrDict]] = None,
        configs: Optional[list[types.LossAnalysisConfigOrDict]] = None,
        config: Optional[types.GenerateLossClustersConfigOrDict] = None,
    ) -> types.GenerateLossClustersOperation:
        """
        Generates loss clusters from evaluation results.
        """

        parameter_model = types._GenerateLossClustersParameters(
            location=location,
            evaluation_set=evaluation_set,
            inline_results=inline_results,
            configs=configs,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError(
                "This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
            )
        else:
            request_dict = _GenerateLossClustersParameters_to_vertex(parameter_model)
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = ":generateLossClusters".format_map(request_url_dict)
            else:
                path = ":generateLossClusters"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = await self._api_client.async_request(
            "post", path, request_dict, http_options
        )

        response_dict = {} if not response.body else json.loads(response.body)

        return_value = types.GenerateLossClustersOperation._from_response(
            response=response_dict,
            kwargs=(
                {
                    "config": {
                        "response_schema": getattr(
                            parameter_model.config, "response_schema", None
                        ),
                        "response_json_schema": getattr(
                            parameter_model.config, "response_json_schema", None
                        ),
                        "include_all_fields": getattr(
                            parameter_model.config, "include_all_fields", None
                        ),
                    }
                }
                if getattr(parameter_model, "config", None)
                else {}
            ),
        )

        self._api_client._verify_response(return_value)
        return return_value

    async def _generate_rubrics(
        self,
        *,
        contents: list[genai_types.ContentOrDict],
        predefined_rubric_generation_spec: Optional[
            genai_types.PredefinedMetricSpecOrDict
        ] = None,
        rubric_generation_spec: Optional[genai_types.RubricGenerationSpecOrDict] = None,
        metric_resource_name: Optional[str] = None,
        config: Optional[types.RubricGenerationConfigOrDict] = None,
    ) -> types.GenerateInstanceRubricsResponse:
        """
        Generates rubrics for a given prompt.
        """

        parameter_model = types._GenerateInstanceRubricsRequest(
            contents=contents,
            predefined_rubric_generation_spec=predefined_rubric_generation_spec,
            rubric_generation_spec=rubric_generation_spec,
            metric_resource_name=metric_resource_name,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError(
                "This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
            )
        else:
            request_dict = _GenerateInstanceRubricsRequest_to_vertex(parameter_model)
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = ":generateInstanceRubrics".format_map(request_url_dict)
            else:
                path = ":generateInstanceRubrics"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = await self._api_client.async_request(
            "post", path, request_dict, http_options
        )

        response_dict = {} if not response.body else json.loads(response.body)

        return_value = types.GenerateInstanceRubricsResponse._from_response(
            response=response_dict,
            kwargs=(
                {
                    "config": {
                        "response_schema": getattr(
                            parameter_model.config, "response_schema", None
                        ),
                        "response_json_schema": getattr(
                            parameter_model.config, "response_json_schema", None
                        ),
                        "include_all_fields": getattr(
                            parameter_model.config, "include_all_fields", None
                        ),
                    }
                }
                if getattr(parameter_model, "config", None)
                else {}
            ),
        )

        self._api_client._verify_response(return_value)
        return return_value

    async def get_evaluation_experiment(
        self,
        *,
        name: str,
        config: Optional[types.GetEvaluationExperimentConfigOrDict] = None,
    ) -> types.EvaluationExperiment:
        """
        Retrieves an EvaluationExperiment from the resource name.

        Args:
          name: The resource name of the EvaluationExperiment. Format:
            `projects/{project}/locations/{location}/evaluationExperiments/{evaluation_experiment}`
          config: The optional configuration for the get operation.

        Returns:
          The evaluation experiment.

        Example Usage:

        .. code-block:: python

          eval_experiment = client.evals.get_evaluation_experiment(
              name="projects/my-project/locations/us-central1/evaluationExperiments/my-experiment"
          )

        """

        parameter_model = types._GetEvaluationExperimentParameters(
            name=name,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError(
                "This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
            )
        else:
            request_dict = _GetEvaluationExperimentParameters_to_vertex(parameter_model)
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "{name}".format_map(request_url_dict)
            else:
                path = "{name}"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = await self._api_client.async_request(
            "get", path, request_dict, http_options
        )

        response_dict = {} if not response.body else json.loads(response.body)

        return_value = types.EvaluationExperiment._from_response(
            response=response_dict,
            kwargs=(
                {
                    "config": {
                        "response_schema": getattr(
                            parameter_model.config, "response_schema", None
                        ),
                        "response_json_schema": getattr(
                            parameter_model.config, "response_json_schema", None
                        ),
                        "include_all_fields": getattr(
                            parameter_model.config, "include_all_fields", None
                        ),
                    }
                }
                if getattr(parameter_model, "config", None)
                else {}
            ),
        )

        self._api_client._verify_response(return_value)
        return return_value

    async def _get_evaluation_metric(
        self,
        *,
        metric_resource_name: str,
        config: Optional[types.GetEvaluationMetricConfigOrDict] = None,
    ) -> types.EvaluationMetric:
        """
        Retrieves an EvaluationMetric from the resource name.
        """

        parameter_model = types._GetEvaluationMetricParameters(
            metric_resource_name=metric_resource_name,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError(
                "This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
            )
        else:
            request_dict = _GetEvaluationMetricParameters_to_vertex(parameter_model)
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "{evaluation_metric}".format_map(request_url_dict)
            else:
                path = "{evaluation_metric}"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = await self._api_client.async_request(
            "get", path, request_dict, http_options
        )

        response_dict = {} if not response.body else json.loads(response.body)

        if self._api_client.vertexai:
            response_dict = _EvaluationMetric_from_vertex(response_dict)

        return_value = types.EvaluationMetric._from_response(
            response=response_dict,
            kwargs=(
                {
                    "config": {
                        "response_schema": getattr(
                            parameter_model.config, "response_schema", None
                        ),
                        "response_json_schema": getattr(
                            parameter_model.config, "response_json_schema", None
                        ),
                        "include_all_fields": getattr(
                            parameter_model.config, "include_all_fields", None
                        ),
                    }
                }
                if getattr(parameter_model, "config", None)
                else {}
            ),
        )

        self._api_client._verify_response(return_value)
        return return_value

    async def _get_evaluation_run(
        self, *, name: str, config: Optional[types.GetEvaluationRunConfigOrDict] = None
    ) -> types.EvaluationRun:
        """
        Retrieves an EvaluationRun from the resource name.
        """

        parameter_model = types._GetEvaluationRunParameters(
            name=name,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError(
                "This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
            )
        else:
            request_dict = _GetEvaluationRunParameters_to_vertex(parameter_model)
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "evaluationRuns/{name}".format_map(request_url_dict)
            else:
                path = "evaluationRuns/{name}"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = await self._api_client.async_request(
            "get", path, request_dict, http_options
        )

        response_dict = {} if not response.body else json.loads(response.body)

        if self._api_client.vertexai:
            response_dict = _EvaluationRun_from_vertex(response_dict)

        return_value = types.EvaluationRun._from_response(
            response=response_dict,
            kwargs=(
                {
                    "config": {
                        "response_schema": getattr(
                            parameter_model.config, "response_schema", None
                        ),
                        "response_json_schema": getattr(
                            parameter_model.config, "response_json_schema", None
                        ),
                        "include_all_fields": getattr(
                            parameter_model.config, "include_all_fields", None
                        ),
                    }
                }
                if getattr(parameter_model, "config", None)
                else {}
            ),
        )

        self._api_client._verify_response(return_value)
        return return_value

    async def _get_evaluation_set(
        self, *, name: str, config: Optional[types.GetEvaluationSetConfigOrDict] = None
    ) -> types.EvaluationSet:
        """
        Retrieves an EvaluationSet from the resource name.
        """

        parameter_model = types._GetEvaluationSetParameters(
            name=name,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError(
                "This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
            )
        else:
            request_dict = _GetEvaluationSetParameters_to_vertex(parameter_model)
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "evaluationSets/{name}".format_map(request_url_dict)
            else:
                path = "evaluationSets/{name}"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = await self._api_client.async_request(
            "get", path, request_dict, http_options
        )

        response_dict = {} if not response.body else json.loads(response.body)

        return_value = types.EvaluationSet._from_response(
            response=response_dict,
            kwargs=(
                {
                    "config": {
                        "response_schema": getattr(
                            parameter_model.config, "response_schema", None
                        ),
                        "response_json_schema": getattr(
                            parameter_model.config, "response_json_schema", None
                        ),
                        "include_all_fields": getattr(
                            parameter_model.config, "include_all_fields", None
                        ),
                    }
                }
                if getattr(parameter_model, "config", None)
                else {}
            ),
        )

        self._api_client._verify_response(return_value)
        return return_value

    async def _get_evaluation_item(
        self, *, name: str, config: Optional[types.GetEvaluationItemConfigOrDict] = None
    ) -> types.EvaluationItem:
        """
        Retrieves an EvaluationItem from the resource name.
        """

        parameter_model = types._GetEvaluationItemParameters(
            name=name,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError(
                "This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
            )
        else:
            request_dict = _GetEvaluationItemParameters_to_vertex(parameter_model)
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "evaluationItems/{name}".format_map(request_url_dict)
            else:
                path = "evaluationItems/{name}"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = await self._api_client.async_request(
            "get", path, request_dict, http_options
        )

        response_dict = {} if not response.body else json.loads(response.body)

        return_value = types.EvaluationItem._from_response(
            response=response_dict,
            kwargs=(
                {
                    "config": {
                        "response_schema": getattr(
                            parameter_model.config, "response_schema", None
                        ),
                        "response_json_schema": getattr(
                            parameter_model.config, "response_json_schema", None
                        ),
                        "include_all_fields": getattr(
                            parameter_model.config, "include_all_fields", None
                        ),
                    }
                }
                if getattr(parameter_model, "config", None)
                else {}
            ),
        )

        self._api_client._verify_response(return_value)
        return return_value

    async def list_evaluation_experiments(
        self, *, config: Optional[types.ListEvaluationExperimentsConfigOrDict] = None
    ) -> types.ListEvaluationExperimentsResponse:
        """
        Lists EvaluationExperiments.

        Args:
          config: Optional configuration for the list operation, including
            pagination (``page_size``, ``page_token``), ``filter``, and
            ``order_by``.

        Returns:
          The list evaluation experiments response.

        Example Usage:

        .. code-block:: python

          response = client.evals.list_evaluation_experiments(
              config={"filter": 'display_name="my-experiment"'}
          )

        """

        parameter_model = types._ListEvaluationExperimentsParameters(
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError(
                "This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
            )
        else:
            request_dict = _ListEvaluationExperimentsParameters_to_vertex(
                parameter_model
            )
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "evaluationExperiments".format_map(request_url_dict)
            else:
                path = "evaluationExperiments"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = await self._api_client.async_request(
            "get", path, request_dict, http_options
        )

        response_dict = {} if not response.body else json.loads(response.body)

        return_value = types.ListEvaluationExperimentsResponse._from_response(
            response=response_dict,
            kwargs=(
                {
                    "config": {
                        "response_schema": getattr(
                            parameter_model.config, "response_schema", None
                        ),
                        "response_json_schema": getattr(
                            parameter_model.config, "response_json_schema", None
                        ),
                        "include_all_fields": getattr(
                            parameter_model.config, "include_all_fields", None
                        ),
                    }
                }
                if getattr(parameter_model, "config", None)
                else {}
            ),
        )

        self._api_client._verify_response(return_value)
        return return_value

    async def _list_evaluation_metrics(
        self, *, config: Optional[types.ListEvaluationMetricsConfigOrDict] = None
    ) -> types.ListEvaluationMetricsResponse:
        """
        Lists EvaluationMetrics.
        """

        parameter_model = types._ListEvaluationMetricsParameters(
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError(
                "This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
            )
        else:
            request_dict = _ListEvaluationMetricsParameters_to_vertex(parameter_model)
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "evaluationMetrics".format_map(request_url_dict)
            else:
                path = "evaluationMetrics"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = await self._api_client.async_request(
            "get", path, request_dict, http_options
        )

        response_dict = {} if not response.body else json.loads(response.body)

        if self._api_client.vertexai:
            response_dict = _ListEvaluationMetricsResponse_from_vertex(response_dict)

        return_value = types.ListEvaluationMetricsResponse._from_response(
            response=response_dict,
            kwargs=(
                {
                    "config": {
                        "response_schema": getattr(
                            parameter_model.config, "response_schema", None
                        ),
                        "response_json_schema": getattr(
                            parameter_model.config, "response_json_schema", None
                        ),
                        "include_all_fields": getattr(
                            parameter_model.config, "include_all_fields", None
                        ),
                    }
                }
                if getattr(parameter_model, "config", None)
                else {}
            ),
        )

        self._api_client._verify_response(return_value)
        return return_value

    async def batch_evaluate(
        self,
        *,
        dataset: types.EvaluationDatasetOrDict,
        metrics: list[types.MetricOrDict],
        dest: str,
        config: Optional[types.EvaluateDatasetConfigOrDict] = None,
    ) -> types.EvaluateDatasetOperation:
        """Evaluates a dataset based on a set of given metrics."""
        resolved_metrics = _evals_common._resolve_metrics(metrics, self._api_client)
        output_config = genai_types.OutputConfig(
            gcs_destination=genai_types.GcsDestination(output_uri_prefix=dest)
        )
        parameter_model = types.EvaluateDatasetRequestParameters(
            dataset=dataset,
            metrics=resolved_metrics,
            output_config=output_config,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError("This method is only supported in the Vertex AI client.")
        else:
            request_dict = _evals_utils.BatchEvaluateRequestPreparer.EvaluateDatasetRequestParameters_to_vertex(
                parameter_model
            )
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = ":evaluateDataset".format_map(request_url_dict)
            else:
                path = ":evaluateDataset"

        request_dict = _evals_utils.BatchEvaluateRequestPreparer.prepare_metric_payload(
            request_dict, resolved_metrics
        )

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = await self._api_client.async_request(
            "post", path, request_dict, http_options
        )

        response_dict = "" if not response.body else json.loads(response.body)

        if self._api_client.vertexai:
            response_dict = _evals_utils.BatchEvaluateRequestPreparer.EvaluateDatasetOperation_from_vertex(
                response_dict
            )

        return_value = types.EvaluateDatasetOperation._from_response(
            response=response_dict, kwargs=parameter_model.model_dump()
        )
        self._api_client._verify_response(return_value)

        return return_value

    async def evaluate_instances(
        self,
        *,
        metric_config: types._EvaluateInstancesRequestParameters,
    ) -> types.EvaluateInstancesResponse:
        """Evaluates an instance of a model."""

        if isinstance(metric_config, types._EvaluateInstancesRequestParameters):
            metric_config = metric_config.model_dump()  # type: ignore[assignment]
        else:
            metric_config = dict(metric_config)

        result = await self._evaluate_instances(
            **metric_config,
        )

        return result

    async def get_evaluation_run(
        self,
        *,
        name: str,
        include_evaluation_items: bool = False,
        config: Optional[types.GetEvaluationRunConfigOrDict] = None,
    ) -> types.EvaluationRun:
        """Retrieves the EvaluationRun from the resource name.
        Args:
          name: The resource name of the EvaluationRun. Format:
            `projects/{project}/locations/{location}/evaluationRuns/{evaluation_run}`
          include_evaluation_items: Whether to include the evaluation items in the
            response.
          config: The optional configuration for the evaluation run. Must be a dict or
              `types.GetEvaluationRunConfigOrDict` type.

        Returns:
          The evaluation run.
        Raises:
          ValueError: If the name is empty or invalid.
        """
        if not name:
            raise ValueError("name cannot be empty.")
        if name.startswith("projects/"):
            name = name.split("/")[-1]
        result = await self._get_evaluation_run(name=name, config=config)
        if include_evaluation_items:
            eval_result, eval_item_map = (
                await _evals_common._convert_evaluation_run_results_async(
                    self._api_client,
                    result.evaluation_run_results,
                    result.inference_configs,
                )
            )
            result.evaluation_item_results = eval_result
            # Bypass pydantic validation (extra='forbid') for this internal field.
            object.__setattr__(result, "_eval_item_map", eval_item_map)

        return result

    async def create_evaluation_run(
        self,
        *,
        dataset: Union[types.EvaluationRunDataSource, types.EvaluationDataset],
        dest: str,
        metrics: list[types.EvaluationRunMetricOrDict],
        name: Optional[str] = None,
        display_name: Optional[str] = None,
        agent_info: Optional[evals_types.AgentInfo] = None,
        agent: Optional[str] = None,
        user_simulator_config: Optional[evals_types.UserSimulatorConfigOrDict] = None,
        inference_configs: Optional[
            dict[str, types.EvaluationRunInferenceConfigOrDict]
        ] = None,
        red_teaming_config: Optional[types.RedTeamingAnalysisConfigOrDict] = None,
        labels: Optional[dict[str, str]] = None,
        loss_analysis_metrics: Optional[list[Union[str, types.MetricOrDict]]] = None,
        loss_analysis_configs: Optional[list[types.LossAnalysisConfigOrDict]] = None,
        config: Optional[types.CreateEvaluationRunConfigOrDict] = None,
    ) -> types.EvaluationRun:
        """Creates an EvaluationRun.

        Args:
          dataset: The dataset to evaluate. Either an EvaluationRunDataSource or an EvaluationDataset.
          dest: The GCS URI prefix to write the evaluation results to.
          metrics: The list of metrics to evaluate.
          name: The name of the evaluation run.
          display_name: The display name of the evaluation run.
          agent_info: The agent info to evaluate. Mutually exclusive with
              `inference_configs`.
          agent: The agent resource name in str type. Accepts either an Agent
              Engine resource name
              `projects/{project}/locations/{location}/reasoningEngines/{reasoning_engine_id}`
              or a Gemini Agent (Vertex AI Agent) resource name
              `projects/{project}/locations/{location}/agents/{agent}`. When a Gemini
              Agent resource is provided, the backend scrapes the agent to produce
              agent responses. If an Agent Engine resource name is provided, runs
              inference with the deployed agent to get agent responses for evaluation.
              The `agent` parameter is required if `agent_info` is provided.
          user_simulator_config: The user simulator configuration for agent evaluation.
              If `agent_info` is provided without `inference_configs`, this config is used
              to automatically construct the inference configuration. If not specified,
              or if `max_turn` is not set, `max_turn` defaults to 5.
              The `model_name` inside this config can be either a full model path or a
              short model name, e.g. `gemini-3-preview-flash`.
          inference_configs: The candidate to inference config map for the evaluation run.
              The key is the candidate name, and the value is the inference config.
              If provided, `agent_info` must be None. If omitted and `agent_info` is provided,
              this will be automatically constructed using `agent_info` and `user_simulator_config`.
              The `model` field of an inference config accepts a short Gemini model
              name (e.g. `gemini-2.5-flash`), which is automatically expanded to a
              fully-qualified resource name using the client's project and location,
              or an already fully-qualified publisher-model or endpoint resource
              name.
              Example:
              {"candidate-1": types.EvaluationRunInferenceConfig(model="gemini-2.5-flash")}
          red_teaming_config: This field is experimental and may change in future
              versions. Optional configuration for automated Agent Red Teaming
              analysis. Specifies attack categories and vulnerable tools to
              test. When provided, the server runs a red teaming pipeline
              instead of standard evaluation metrics.
          labels: The labels to apply to the evaluation run.
          loss_analysis_metrics: This field is experimental and may change in future
              versions. Optional list of metrics to run loss analysis on. The
              candidate is auto-inferred from ``inference_configs`` or
              ``agent_info`` when there is exactly one candidate. Each metric can be
              a string (e.g., ``"multi_turn_task_success_v1"``), a ``Metric``
              object, or a ``RubricMetric`` enum
              (e.g., ``types.RubricMetric.MULTI_TURN_TASK_SUCCESS``). Loss analysis
              runs after metric calculation completes.
              Mutually exclusive with ``loss_analysis_configs``.
              Example::

                  loss_analysis_metrics=[
                      types.RubricMetric.MULTI_TURN_TASK_SUCCESS,
                      types.RubricMetric.MULTI_TURN_TOOL_USE_QUALITY,
                  ]
          loss_analysis_configs: This field is experimental and may change in future
              versions. Optional list of ``LossAnalysisConfig`` objects for full
              control over loss analysis, including explicit candidate and
              advanced options like ``predefined_taxonomy`` and
              ``max_top_cluster_count``. Mutually exclusive with
              ``loss_analysis_metrics``.
          config: The configuration for the evaluation run.
            - allow_cross_region_model: Opt-in flag to authorize cross-region
              routing for model inference. Applies to both scraping and evaluation.

        Returns:
            The created evaluation run.
        """
        if loss_analysis_metrics and loss_analysis_configs:
            raise ValueError(
                "At most one of loss_analysis_metrics or loss_analysis_configs"
                " can be provided."
            )
        if agent_info and inference_configs:
            raise ValueError(
                "At most one of agent_info or inference_configs can be provided."
            )
        parsed_agent_info = (
            evals_types.AgentInfo.model_validate(agent_info)
            if isinstance(agent_info, dict)
            else (agent_info or evals_types.AgentInfo())
        )

        if not config:
            config = types.CreateEvaluationRunConfig()
        if isinstance(config, dict):
            config = types.CreateEvaluationRunConfig.model_validate(config)

        # Auto-construct inference_configs when agent_info is explicitly
        # provided (existing behavior) OR an agent resource is provided
        # (allows omitting agent_info for both Gemini Agents and Agent
        # Engine). The server skips inference per-item when a
        # CandidateResponse with a matching candidate name already exists,
        # so it is safe to always send inference_configs.
        _should_auto_infer = not inference_configs and (agent_info or agent)
        if _should_auto_infer:
            if not parsed_agent_info.name:
                # Prefer the dataset's candidate_name (set by run_inference)
                # so the inference_configs key matches the CandidateResponse
                # and the server correctly skips already-completed items.
                if (
                    isinstance(dataset, types.EvaluationDataset)
                    and dataset.candidate_name
                ):
                    parsed_agent_info.name = dataset.candidate_name
                else:
                    parsed_agent_info.name = _evals_common._DEFAULT_CANDIDATE_NAME
            parsed_user_simulator_config = (
                evals_types.UserSimulatorConfig.model_validate(user_simulator_config)
                if isinstance(user_simulator_config, dict)
                else (user_simulator_config or evals_types.UserSimulatorConfig())
            )
            if getattr(parsed_user_simulator_config, "max_turn", None) is None:
                parsed_user_simulator_config.max_turn = 5

            candidate_name = (
                parsed_agent_info.name or _evals_common._DEFAULT_CANDIDATE_NAME
            )
            if agent and _evals_common._is_gemini_agent_resource(agent):
                agent_run_config = types.AgentRunConfig(
                    gemini_agent_config=types.GeminiAgentConfig(gemini_agent=agent),
                    user_simulator_config=parsed_user_simulator_config,
                )
            else:
                agent_run_config = types.AgentRunConfig(
                    agent_engine=agent,
                    user_simulator_config=parsed_user_simulator_config,
                )
            inference_configs = {
                candidate_name: types.EvaluationRunInferenceConfig(
                    agent_configs=parsed_agent_info.agents,
                    agent_run_config=agent_run_config,
                )
            }

        if isinstance(dataset, types.EvaluationDataset):
            _evals_utils._validate_dataset_agent_data(dataset, inference_configs)
        resolved_dataset = _evals_common._resolve_dataset(
            self._api_client, dataset, dest, parsed_agent_info
        )
        output_config = genai_types.OutputConfig(
            gcs_destination=genai_types.GcsDestination(output_uri_prefix=dest)
        )
        resolved_metrics = _evals_common._resolve_evaluation_run_metrics(
            metrics, self._api_client
        )
        resolved_loss_configs = _evals_utils._resolve_eval_run_loss_configs(
            loss_analysis_metrics=loss_analysis_metrics,
            loss_analysis_configs=loss_analysis_configs,
            inference_configs=inference_configs,
        )
        resolved_analysis_configs = _evals_utils._resolve_red_teaming_config(
            red_teaming_config
        )
        evaluation_config = types.EvaluationRunConfig(
            output_config=output_config,
            metrics=resolved_metrics,
            loss_analysis_config=resolved_loss_configs,
            allow_cross_region_model=getattr(config, "allow_cross_region_model", None),
        )
        resolved_inference_configs = _evals_common._resolve_inference_configs(
            self._api_client, resolved_dataset, inference_configs, parsed_agent_info
        )
        resolved_labels = _evals_common._add_evaluation_run_labels(labels, agent)
        resolved_name = name or f"evaluation_run_{uuid.uuid4()}"

        result = await self._create_evaluation_run(
            name=resolved_name,
            display_name=display_name or resolved_name,
            data_source=resolved_dataset,
            evaluation_config=evaluation_config,
            inference_configs=resolved_inference_configs,
            analysis_configs=resolved_analysis_configs,
            labels=resolved_labels,
            config=config,
        )

        return result

    async def get_evaluation_set(
        self,
        *,
        name: str,
        config: Optional[types.GetEvaluationSetConfigOrDict] = None,
    ) -> types.EvaluationSet:
        """Retrieves an EvaluationSet from the resource name.

        Args:
          name: The resource name of the EvaluationSet. Format:
            `projects/{project}/locations/{location}/evaluationSets/{evaluation_set}`
          config: The optional configuration for the evaluation set. Must be a dict or
              `types.GetEvaluationSetConfigOrDict` type.

        Returns:
          The evaluation set.
        """
        if not name:
            raise ValueError("name cannot be empty.")
        if name.startswith("projects/"):
            name = name.split("/")[-1]
        result = await self._get_evaluation_set(name=name, config=config)

        return result

    async def get_evaluation_item(
        self,
        *,
        name: str,
        config: Optional[types.GetEvaluationItemConfigOrDict] = None,
    ) -> types.EvaluationItem:
        """Retrieves an EvaluationItem from the resource name.

        Args:
          name: The resource name of the EvaluationItem. Format:
            `projects/{project}/locations/{location}/evaluationItems/{evaluation_item}`
          config: The optional configuration for the evaluation item. Must be a dict or
              `types.GetEvaluationItemConfigOrDict` type.

        Returns:
          The evaluation item.
        """
        if not name:
            raise ValueError("name cannot be empty.")
        if name.startswith("projects/"):
            name = name.split("/")[-1]
        result = await self._get_evaluation_item(name=name, config=config)
        if (
            result.gcs_uri
            and result.evaluation_item_type == types.EvaluationItemType.RESULT
        ):
            result.evaluation_response = (
                _evals_common._convert_gcs_to_evaluation_item_result(
                    self._api_client, result.gcs_uri
                )
            )
        elif (
            result.gcs_uri
            and result.evaluation_item_type == types.EvaluationItemType.REQUEST
        ):
            result.evaluation_request = (
                _evals_common._convert_gcs_to_evaluation_item_request(
                    self._api_client, result.gcs_uri
                )
            )

        return result

    async def create_evaluation_item(
        self,
        *,
        evaluation_item_type: types.EvaluationItemType,
        gcs_uri: str,
        display_name: Optional[str] = None,
        config: Optional[types.CreateEvaluationItemConfigOrDict] = None,
    ) -> types.EvaluationItem:
        """Creates an EvaluationItem.

        Args:
          evaluation_item_type: The type of the evaluation item.
          gcs_uri: The GCS URI of the evaluation item.
          display_name: The display name of the evaluation item.
          config: The optional configuration for the evaluation item. Must be a dict or
              `types.CreateEvaluationItemConfigOrDict` type.

        Returns:
          The evaluation item.
        """
        result = await self._create_evaluation_item(
            evaluation_item_type=evaluation_item_type,
            gcs_uri=gcs_uri,
            display_name=display_name,
            config=config,
        )
        return result

    async def create_evaluation_set(
        self,
        *,
        evaluation_items: list[str],
        display_name: Optional[str] = None,
        config: Optional[types.CreateEvaluationSetConfigOrDict] = None,
    ) -> types.EvaluationSet:
        """Creates an EvaluationSet.

        Args:
          evaluation_items: The list of evaluation item names. Format:
            `projects/{project}/locations/{location}/evaluationItems/{evaluation_item}`
          display_name: The display name of the evaluation set.
          config: The optional configuration for the evaluation set. Must be a dict or
              `types.CreateEvaluationSetConfigOrDict` type.

        Returns:
          The evaluation set.
        """
        result = await self._create_evaluation_set(
            evaluation_items=evaluation_items,
            display_name=display_name,
            config=config,
        )
        return result

    async def generate_conversation_scenarios(
        self,
        *,
        agent_info: Optional[evals_types.AgentInfoOrDict] = None,
        agent: Optional[str] = None,
        config: evals_types.UserScenarioGenerationConfigOrDict,
        allow_cross_region_model: Optional[bool] = None,
    ) -> types.EvaluationDataset:
        """Generates an evaluation dataset with user scenarios,
           which helps to generate conversations between a simulated user
           and the agent under test.

        Exactly one of `agent_info` or `agent` must be provided. When `agent` is
        a Gemini Agents API agent resource name, the agent config is resolved
        server-side via ``gemini_agent_config``.

        Args:
            agent_info: The agent info to generate user scenarios for. Mutually
                exclusive with `agent`.
            agent: A Gemini Agents API agent resource name
                (`projects/{p}/locations/{l}/agents/{name}`). When provided, the
                server resolves the agent config directly. Mutually exclusive
                with `agent_info`.
            config: Configuration for generating user scenarios.
            allow_cross_region_model: Opt-in flag to authorize cross-region
                routing for model inference.

        Returns:
            An EvaluationDataset containing the generated user scenarios.
        """
        if agent is not None and agent_info is not None:
            raise ValueError(
                "Only one of `agent` or `agent_info` may be provided, not both."
            )
        if agent is None and agent_info is None:
            raise ValueError("One of `agent` or `agent_info` must be provided.")
        if agent is not None:
            if not _evals_common._is_gemini_agent_resource(agent):
                raise ValueError(
                    "`agent` must be a Gemini Agents API agent resource name of the"
                    " form projects/{project}/locations/{location}/agents/{agent}."
                )
            response = await self._generate_user_scenarios(
                gemini_agent_config=types.GeminiAgentConfig(gemini_agent=agent),
                user_scenario_generation_config=config,
                allow_cross_region_model=allow_cross_region_model,
            )
        else:
            parsed_agent_info = (
                evals_types.AgentInfo.model_validate(agent_info)
                if isinstance(agent_info, dict)
                else agent_info
            )
            response = await self._generate_user_scenarios(
                agents=parsed_agent_info.agents,
                root_agent_id=parsed_agent_info.root_agent_id,
                user_scenario_generation_config=config,
                allow_cross_region_model=allow_cross_region_model,
            )
        return _evals_utils._postprocess_user_scenarios_response(response)

    async def generate_loss_clusters(
        self,
        *,
        eval_result: types.EvaluationResult,
        metric: Optional[Union[str, types.MetricOrDict]] = None,
        candidate: Optional[str] = None,
        config: Optional[types.LossAnalysisConfigOrDict] = None,
    ) -> types.GenerateLossClustersResponse:
        """Generates loss clusters from evaluation results.

        Analyzes "Pass/Fail" signals from rubric-based autoraters and groups
        them into semantic "Loss Patterns" (e.g., "Hallucination of Action").

        This method calls the GenerateLossClusters LRO and polls until
        completion, returning the results directly.

        If ``metric`` or ``candidate`` are not provided, they will be
        auto-inferred from ``eval_result`` when unambiguous (i.e., when the
        eval result contains exactly one metric or one candidate). For
        multi-metric or multi-candidate evaluations, provide them explicitly.

        Available candidate names can be found in
        ``eval_result.metadata.candidate_names``.

        Note: This API is only available in the ``global`` region.

        Args:
            eval_result: The EvaluationResult object returned from
                client.evals.evaluate().
            metric: The metric to analyze. Can be a metric name string
                (e.g., "multi_turn_task_success_v1"), a Metric object, or a
                RubricMetric enum (e.g., types.RubricMetric.MULTI_TURN_TASK_SUCCESS).
                If not provided and config does not specify it, auto-inferred
                from eval_result.
            candidate: The candidate to analyze. If not provided and config
                does not specify it, auto-inferred from eval_result.
            config: Optional LossAnalysisConfig with additional options
                (predefined_taxonomy, max_top_cluster_count). Can also
                specify metric/candidate, but explicit arguments take
                precedence.

        Returns:
            A GenerateLossClustersResponse containing the analysis results.
            Call .show() to visualize, or access .results for individual
            LossAnalysisResult objects (each with their own .show()).
        """
        metric_name = _evals_utils._resolve_metric_name(metric)
        parsed_config = (
            types.LossAnalysisConfig.model_validate(config)
            if isinstance(config, dict)
            else config
        )
        resolved_config = _evals_utils._resolve_loss_analysis_config(
            eval_result=eval_result,
            config=parsed_config,
            metric=metric_name,
            candidate=candidate,
        )
        operation = await self._generate_loss_clusters(
            inline_results=[eval_result],
            configs=[resolved_config],
        )
        completed = await _evals_utils._poll_operation_async(
            api_client=self._api_client,
            operation=operation,
        )
        if completed.error:
            raise RuntimeError(f"Loss analysis operation failed: {completed.error}")
        if completed.response is None:
            raise RuntimeError(
                "Loss analysis operation completed but returned no response."
            )
        _evals_utils._enrich_loss_response_with_rubric_descriptions(
            completed.response, eval_result
        )
        return completed.response

    async def create_evaluation_metric(
        self,
        *,
        display_name: Optional[str] = None,
        description: Optional[str] = None,
        metric: Optional[types.MetricOrDict] = None,
        config: Optional[types.CreateEvaluationMetricConfigOrDict] = None,
    ) -> str:
        """Creates an EvaluationMetric."""
        if metric and not isinstance(metric, dict):
            resolved_metrics = _evals_common._resolve_metrics(
                [metric], self._api_client
            )
            metric = resolved_metrics[0]

        # Add fallback logic for display_name
        if display_name is None and metric:
            if isinstance(metric, dict):
                display_name = metric.get("name")
            else:
                display_name = getattr(metric, "name", None)

        result = await self._create_evaluation_metric(
            display_name=display_name,
            description=description,
            metric=metric,
            config=config,
        )
        return cast(str, result.name)

    async def get_evaluation_metric(
        self,
        *,
        metric_resource_name: str,
        config: Optional[types.GetEvaluationMetricConfigOrDict] = None,
    ) -> types.EvaluationMetric:
        """Retrieves an EvaluationMetric from the resource name."""
        return await self._get_evaluation_metric(
            metric_resource_name=metric_resource_name,
            config=config,
        )

    async def list_evaluation_metrics(
        self,
        *,
        filter: Optional[str] = None,
        order_by: Optional[str] = None,
        config: Optional[types.ListEvaluationMetricsConfigOrDict] = None,
    ) -> types.ListEvaluationMetricsResponse:
        """Lists EvaluationMetrics.

        Args:
          filter: An expression for filtering the results of the request. For
            field names both snake_case and camelCase are supported. For more
            information about filter syntax, see
            `AIP-160 <https://google.aip.dev/160>`_.
            Example: ``'display_name="my_metric"'``.
          order_by: A comma-separated list of fields to order by, sorted in
            ascending order by default. Use ``desc`` after a field name for
            descending. Example: ``"create_time desc"``.
          config: Optional configuration for the list operation, including
            pagination (``page_size``, ``page_token``), ``filter``, and
            ``order_by``. Top-level ``filter`` and ``order_by`` arguments
            take precedence over values set in ``config``.

        Returns:
          The list evaluation metrics response.
        """
        if config is None:
            config = types.ListEvaluationMetricsConfig()
        if isinstance(config, dict):
            config = types.ListEvaluationMetricsConfig.model_validate(config)
        if filter is not None:
            config.filter = filter
        if order_by is not None:
            config.order_by = order_by
        return await self._list_evaluation_metrics(
            config=config,
        )

    async def delete_evaluation_metric(
        self,
        *,
        metric_resource_name: str,
        config: Optional[types.DeleteEvaluationMetricConfigOrDict] = None,
    ) -> None:
        """Deletes an EvaluationMetric.

        Args:
          metric_resource_name: The resource name of the EvaluationMetric to delete.
            Format:
            `projects/{project}/locations/{location}/evaluationMetrics/{evaluation_metric}`
          config: The optional configuration for the delete operation.
        """
        await self._delete_evaluation_metric(
            metric_resource_name=metric_resource_name,
            config=config,
        )