# 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. import json import logging from typing import Any, Optional, Union from urllib.parse import urlencode from . import _api_module from . import _common from . import _transformers as t from . import types from ._common import get_value_by_path as getv from ._common import set_value_by_path as setv from .pagers import AsyncPager, Pager logger = logging.getLogger('google_genai.tunings') def _AutoraterConfig_from_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['samplingCount']) is not None: setv(to_object, ['sampling_count'], getv(from_object, ['samplingCount'])) if getv(from_object, ['flipEnabled']) is not None: setv(to_object, ['flip_enabled'], getv(from_object, ['flipEnabled'])) if getv(from_object, ['autoraterModel']) is not None: setv(to_object, ['autorater_model'], getv(from_object, ['autoraterModel'])) if getv(from_object, ['generationConfig']) is not None: setv( to_object, ['generation_config'], _GenerationConfig_from_vertex( getv(from_object, ['generationConfig']), to_object, root_object ), ) return to_object def _AutoraterConfig_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['sampling_count']) is not None: setv(to_object, ['samplingCount'], getv(from_object, ['sampling_count'])) if getv(from_object, ['flip_enabled']) is not None: setv(to_object, ['flipEnabled'], getv(from_object, ['flip_enabled'])) if getv(from_object, ['autorater_model']) is not None: setv(to_object, ['autoraterModel'], getv(from_object, ['autorater_model'])) if getv(from_object, ['generation_config']) is not None: setv( to_object, ['generationConfig'], _GenerationConfig_to_vertex( getv(from_object, ['generation_config']), to_object, root_object ), ) return to_object def _CancelTuningJobParameters_to_mldev( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = 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'])) return to_object def _CancelTuningJobParameters_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = 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'])) return to_object def _CancelTuningJobResponse_from_mldev( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = 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']) ) return to_object def _CancelTuningJobResponse_from_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = 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']) ) return to_object def _CodeExecutionResult_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['outcome']) is not None: setv(to_object, ['outcome'], getv(from_object, ['outcome'])) if getv(from_object, ['output']) is not None: setv(to_object, ['output'], getv(from_object, ['output'])) if getv(from_object, ['id']) is not None: raise ValueError( 'id parameter is only supported in Gemini Developer API mode, not in' ' Gemini Enterprise Agent Platform mode.' ) return to_object def _CompositeReinforcementTuningRewardConfigWeightedRewardConfig_from_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['rewardConfig']) is not None: setv( to_object, ['reward_config'], _SingleReinforcementTuningRewardConfig_from_vertex( getv(from_object, ['rewardConfig']), to_object, root_object ), ) if getv(from_object, ['weight']) is not None: setv(to_object, ['weight'], getv(from_object, ['weight'])) return to_object def _CompositeReinforcementTuningRewardConfigWeightedRewardConfig_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['reward_config']) is not None: setv( to_object, ['rewardConfig'], _SingleReinforcementTuningRewardConfig_to_vertex( getv(from_object, ['reward_config']), to_object, root_object ), ) if getv(from_object, ['weight']) is not None: setv(to_object, ['weight'], getv(from_object, ['weight'])) return to_object def _CompositeReinforcementTuningRewardConfig_from_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['weightedRewardConfigs']) is not None: setv( to_object, ['weighted_reward_configs'], [ _CompositeReinforcementTuningRewardConfigWeightedRewardConfig_from_vertex( item, to_object, root_object ) for item in getv(from_object, ['weightedRewardConfigs']) ], ) return to_object def _CompositeReinforcementTuningRewardConfig_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['weighted_reward_configs']) is not None: setv( to_object, ['weightedRewardConfigs'], [ _CompositeReinforcementTuningRewardConfigWeightedRewardConfig_to_vertex( item, to_object, root_object ) for item in getv(from_object, ['weighted_reward_configs']) ], ) return to_object def _Content_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['parts']) is not None: setv( to_object, ['parts'], [ _Part_to_vertex(item, to_object, root_object) for item in getv(from_object, ['parts']) ], ) if getv(from_object, ['role']) is not None: setv(to_object, ['role'], getv(from_object, ['role'])) return to_object def _CreateTuningJobConfig_to_mldev( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['validation_dataset']) is not None: raise ValueError( 'validation_dataset parameter is only supported in Gemini Enterprise' ' Agent Platform mode, not in Gemini Developer API mode.' ) if getv(from_object, ['tuned_model_display_name']) is not None: setv( parent_object, ['displayName'], getv(from_object, ['tuned_model_display_name']), ) if getv(from_object, ['description']) is not None: raise ValueError( 'description parameter is only supported in Gemini Enterprise Agent' ' Platform mode, not in Gemini Developer API mode.' ) if getv(from_object, ['epoch_count']) is not None: setv( parent_object, ['tuningTask', 'hyperparameters', 'epochCount'], getv(from_object, ['epoch_count']), ) if getv(from_object, ['learning_rate_multiplier']) is not None: setv( to_object, ['tuningTask', 'hyperparameters', 'learningRateMultiplier'], getv(from_object, ['learning_rate_multiplier']), ) if getv(from_object, ['export_last_checkpoint_only']) is not None: raise ValueError( 'export_last_checkpoint_only parameter is only supported in Gemini' ' Enterprise Agent Platform mode, not in Gemini Developer API mode.' ) if getv(from_object, ['pre_tuned_model_checkpoint_id']) is not None: raise ValueError( 'pre_tuned_model_checkpoint_id parameter is only supported in Gemini' ' Enterprise Agent Platform mode, not in Gemini Developer API mode.' ) if getv(from_object, ['adapter_size']) is not None: raise ValueError( 'adapter_size parameter is only supported in Gemini Enterprise Agent' ' Platform mode, not in Gemini Developer API mode.' ) if getv(from_object, ['tuning_mode']) is not None: raise ValueError( 'tuning_mode parameter is only supported in Gemini Enterprise Agent' ' Platform mode, not in Gemini Developer API mode.' ) if getv(from_object, ['custom_base_model']) is not None: raise ValueError( 'custom_base_model parameter is only supported in Gemini Enterprise' ' Agent Platform mode, not in Gemini Developer API mode.' ) if getv(from_object, ['batch_size']) is not None: setv( parent_object, ['tuningTask', 'hyperparameters', 'batchSize'], getv(from_object, ['batch_size']), ) if getv(from_object, ['learning_rate']) is not None: setv( parent_object, ['tuningTask', 'hyperparameters', 'learningRate'], getv(from_object, ['learning_rate']), ) if getv(from_object, ['evaluation_config']) is not None: raise ValueError( 'evaluation_config parameter is only supported in Gemini Enterprise' ' Agent Platform mode, not in Gemini Developer API mode.' ) if getv(from_object, ['labels']) is not None: raise ValueError( 'labels parameter is only supported in Gemini Enterprise Agent Platform' ' mode, not in Gemini Developer API mode.' ) if getv(from_object, ['beta']) is not None: raise ValueError( 'beta parameter is only supported in Gemini Enterprise Agent Platform' ' mode, not in Gemini Developer API mode.' ) if getv(from_object, ['base_teacher_model']) is not None: raise ValueError( 'base_teacher_model parameter is only supported in Gemini Enterprise' ' Agent Platform mode, not in Gemini Developer API mode.' ) if getv(from_object, ['tuned_teacher_model_source']) is not None: raise ValueError( 'tuned_teacher_model_source parameter is only supported in Gemini' ' Enterprise Agent Platform mode, not in Gemini Developer API mode.' ) if getv(from_object, ['sft_loss_weight_multiplier']) is not None: raise ValueError( 'sft_loss_weight_multiplier parameter is only supported in Gemini' ' Enterprise Agent Platform mode, not in Gemini Developer API mode.' ) if getv(from_object, ['output_uri']) is not None: raise ValueError( 'output_uri parameter is only supported in Gemini Enterprise Agent' ' Platform mode, not in Gemini Developer API mode.' ) if getv(from_object, ['encryption_spec']) is not None: raise ValueError( 'encryption_spec parameter is only supported in Gemini Enterprise Agent' ' Platform mode, not in Gemini Developer API mode.' ) if getv(from_object, ['reward_config']) is not None: raise ValueError( 'reward_config parameter is only supported in Gemini Enterprise Agent' ' Platform mode, not in Gemini Developer API mode.' ) if getv(from_object, ['composite_reward_config']) is not None: raise ValueError( 'composite_reward_config parameter is only supported in Gemini' ' Enterprise Agent Platform mode, not in Gemini Developer API mode.' ) if getv(from_object, ['samples_per_prompt']) is not None: raise ValueError( 'samples_per_prompt parameter is only supported in Gemini Enterprise' ' Agent Platform mode, not in Gemini Developer API mode.' ) if getv(from_object, ['evaluate_interval']) is not None: raise ValueError( 'evaluate_interval parameter is only supported in Gemini Enterprise' ' Agent Platform mode, not in Gemini Developer API mode.' ) if getv(from_object, ['checkpoint_interval']) is not None: raise ValueError( 'checkpoint_interval parameter is only supported in Gemini Enterprise' ' Agent Platform mode, not in Gemini Developer API mode.' ) if getv(from_object, ['max_output_tokens']) is not None: raise ValueError( 'max_output_tokens parameter is only supported in Gemini Enterprise' ' Agent Platform mode, not in Gemini Developer API mode.' ) if getv(from_object, ['thinking_level']) is not None: raise ValueError( 'thinking_level parameter is only supported in Gemini Enterprise Agent' ' Platform mode, not in Gemini Developer API mode.' ) if getv(from_object, ['validation_dataset_uri']) is not None: raise ValueError( 'validation_dataset_uri parameter is only supported in Gemini' ' Enterprise Agent Platform mode, not in Gemini Developer API mode.' ) return to_object def _CreateTuningJobConfig_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} discriminator = getv(root_object, ['config', 'method']) if discriminator is None: discriminator = 'SUPERVISED_FINE_TUNING' if discriminator == 'SUPERVISED_FINE_TUNING': if getv(from_object, ['validation_dataset']) is not None: setv( parent_object, ['supervisedTuningSpec'], _TuningValidationDataset_to_vertex( getv(from_object, ['validation_dataset']), to_object, root_object ), ) elif discriminator == 'PREFERENCE_TUNING': if getv(from_object, ['validation_dataset']) is not None: setv( parent_object, ['preferenceOptimizationSpec'], _TuningValidationDataset_to_vertex( getv(from_object, ['validation_dataset']), to_object, root_object ), ) elif discriminator == 'DISTILLATION': if getv(from_object, ['validation_dataset']) is not None: setv( parent_object, ['distillationSpec'], _TuningValidationDataset_to_vertex( getv(from_object, ['validation_dataset']), to_object, root_object ), ) elif discriminator == 'REINFORCEMENT_TUNING': if getv(from_object, ['validation_dataset']) is not None: setv( parent_object, ['reinforcementTuningSpec'], _TuningValidationDataset_to_vertex( getv(from_object, ['validation_dataset']), to_object, root_object ), ) if getv(from_object, ['tuned_model_display_name']) is not None: setv( parent_object, ['tunedModelDisplayName'], getv(from_object, ['tuned_model_display_name']), ) if getv(from_object, ['description']) is not None: setv(parent_object, ['description'], getv(from_object, ['description'])) discriminator = getv(root_object, ['config', 'method']) if discriminator is None: discriminator = 'SUPERVISED_FINE_TUNING' if discriminator == 'SUPERVISED_FINE_TUNING': if getv(from_object, ['epoch_count']) is not None: setv( parent_object, ['supervisedTuningSpec', 'hyperParameters', 'epochCount'], getv(from_object, ['epoch_count']), ) elif discriminator == 'PREFERENCE_TUNING': if getv(from_object, ['epoch_count']) is not None: setv( parent_object, ['preferenceOptimizationSpec', 'hyperParameters', 'epochCount'], getv(from_object, ['epoch_count']), ) elif discriminator == 'DISTILLATION': if getv(from_object, ['epoch_count']) is not None: setv( parent_object, ['distillationSpec', 'hyperParameters', 'epochCount'], getv(from_object, ['epoch_count']), ) elif discriminator == 'REINFORCEMENT_TUNING': if getv(from_object, ['epoch_count']) is not None: setv( parent_object, ['reinforcementTuningSpec', 'hyperParameters', 'epochCount'], getv(from_object, ['epoch_count']), ) discriminator = getv(root_object, ['config', 'method']) if discriminator is None: discriminator = 'SUPERVISED_FINE_TUNING' if discriminator == 'SUPERVISED_FINE_TUNING': if getv(from_object, ['learning_rate_multiplier']) is not None: setv( parent_object, ['supervisedTuningSpec', 'hyperParameters', 'learningRateMultiplier'], getv(from_object, ['learning_rate_multiplier']), ) elif discriminator == 'PREFERENCE_TUNING': if getv(from_object, ['learning_rate_multiplier']) is not None: setv( parent_object, [ 'preferenceOptimizationSpec', 'hyperParameters', 'learningRateMultiplier', ], getv(from_object, ['learning_rate_multiplier']), ) elif discriminator == 'DISTILLATION': if getv(from_object, ['learning_rate_multiplier']) is not None: setv( parent_object, ['distillationSpec', 'hyperParameters', 'learningRateMultiplier'], getv(from_object, ['learning_rate_multiplier']), ) elif discriminator == 'REINFORCEMENT_TUNING': if getv(from_object, ['learning_rate_multiplier']) is not None: setv( parent_object, [ 'reinforcementTuningSpec', 'hyperParameters', 'learningRateMultiplier', ], getv(from_object, ['learning_rate_multiplier']), ) discriminator = getv(root_object, ['config', 'method']) if discriminator is None: discriminator = 'SUPERVISED_FINE_TUNING' if discriminator == 'SUPERVISED_FINE_TUNING': if getv(from_object, ['export_last_checkpoint_only']) is not None: setv( parent_object, ['supervisedTuningSpec', 'exportLastCheckpointOnly'], getv(from_object, ['export_last_checkpoint_only']), ) elif discriminator == 'PREFERENCE_TUNING': if getv(from_object, ['export_last_checkpoint_only']) is not None: setv( parent_object, ['preferenceOptimizationSpec', 'exportLastCheckpointOnly'], getv(from_object, ['export_last_checkpoint_only']), ) elif discriminator == 'DISTILLATION': if getv(from_object, ['export_last_checkpoint_only']) is not None: setv( parent_object, ['distillationSpec', 'exportLastCheckpointOnly'], getv(from_object, ['export_last_checkpoint_only']), ) discriminator = getv(root_object, ['config', 'method']) if discriminator is None: discriminator = 'SUPERVISED_FINE_TUNING' if discriminator == 'SUPERVISED_FINE_TUNING': if getv(from_object, ['adapter_size']) is not None: setv( parent_object, ['supervisedTuningSpec', 'hyperParameters', 'adapterSize'], getv(from_object, ['adapter_size']), ) elif discriminator == 'PREFERENCE_TUNING': if getv(from_object, ['adapter_size']) is not None: setv( parent_object, ['preferenceOptimizationSpec', 'hyperParameters', 'adapterSize'], getv(from_object, ['adapter_size']), ) elif discriminator == 'DISTILLATION': if getv(from_object, ['adapter_size']) is not None: setv( parent_object, ['distillationSpec', 'hyperParameters', 'adapterSize'], getv(from_object, ['adapter_size']), ) elif discriminator == 'REINFORCEMENT_TUNING': if getv(from_object, ['adapter_size']) is not None: setv( parent_object, ['reinforcementTuningSpec', 'hyperParameters', 'adapterSize'], getv(from_object, ['adapter_size']), ) discriminator = getv(root_object, ['config', 'method']) if discriminator is None: discriminator = 'SUPERVISED_FINE_TUNING' if discriminator == 'SUPERVISED_FINE_TUNING': if getv(from_object, ['tuning_mode']) is not None: setv( parent_object, ['supervisedTuningSpec', 'tuningMode'], getv(from_object, ['tuning_mode']), ) elif discriminator == 'DISTILLATION': if getv(from_object, ['tuning_mode']) is not None: setv( parent_object, ['distillationSpec', 'tuningMode'], getv(from_object, ['tuning_mode']), ) if getv(from_object, ['custom_base_model']) is not None: setv( parent_object, ['customBaseModel'], getv(from_object, ['custom_base_model']), ) discriminator = getv(root_object, ['config', 'method']) if discriminator is None: discriminator = 'SUPERVISED_FINE_TUNING' if discriminator == 'SUPERVISED_FINE_TUNING': if getv(from_object, ['batch_size']) is not None: setv( parent_object, ['supervisedTuningSpec', 'hyperParameters', 'batchSize'], getv(from_object, ['batch_size']), ) elif discriminator == 'DISTILLATION': if getv(from_object, ['batch_size']) is not None: setv( parent_object, ['distillationSpec', 'hyperParameters', 'batchSize'], getv(from_object, ['batch_size']), ) elif discriminator == 'REINFORCEMENT_TUNING': if getv(from_object, ['batch_size']) is not None: setv( parent_object, ['reinforcementTuningSpec', 'hyperParameters', 'batchSize'], getv(from_object, ['batch_size']), ) discriminator = getv(root_object, ['config', 'method']) if discriminator is None: discriminator = 'SUPERVISED_FINE_TUNING' if discriminator == 'SUPERVISED_FINE_TUNING': if getv(from_object, ['learning_rate']) is not None: setv( parent_object, ['supervisedTuningSpec', 'hyperParameters', 'learningRate'], getv(from_object, ['learning_rate']), ) elif discriminator == 'DISTILLATION': if getv(from_object, ['learning_rate']) is not None: setv( parent_object, ['distillationSpec', 'hyperParameters', 'learningRate'], getv(from_object, ['learning_rate']), ) discriminator = getv(root_object, ['config', 'method']) if discriminator is None: discriminator = 'SUPERVISED_FINE_TUNING' if discriminator == 'SUPERVISED_FINE_TUNING': if getv(from_object, ['evaluation_config']) is not None: setv( parent_object, ['supervisedTuningSpec', 'evaluationConfig'], _EvaluationConfig_to_vertex( getv(from_object, ['evaluation_config']), to_object, root_object ), ) elif discriminator == 'PREFERENCE_TUNING': if getv(from_object, ['evaluation_config']) is not None: setv( parent_object, ['preferenceOptimizationSpec', 'evaluationConfig'], _EvaluationConfig_to_vertex( getv(from_object, ['evaluation_config']), to_object, root_object ), ) elif discriminator == 'DISTILLATION': if getv(from_object, ['evaluation_config']) is not None: setv( parent_object, ['distillationSpec', 'evaluationConfig'], _EvaluationConfig_to_vertex( getv(from_object, ['evaluation_config']), to_object, root_object ), ) if getv(from_object, ['labels']) is not None: setv(parent_object, ['labels'], getv(from_object, ['labels'])) if getv(from_object, ['beta']) is not None: setv( parent_object, ['preferenceOptimizationSpec', 'hyperParameters', 'beta'], getv(from_object, ['beta']), ) if getv(from_object, ['base_teacher_model']) is not None: setv( parent_object, ['distillationSpec', 'baseTeacherModel'], getv(from_object, ['base_teacher_model']), ) if getv(from_object, ['tuned_teacher_model_source']) is not None: setv( parent_object, ['distillationSpec', 'tunedTeacherModelSource'], getv(from_object, ['tuned_teacher_model_source']), ) if getv(from_object, ['sft_loss_weight_multiplier']) is not None: setv( parent_object, ['distillationSpec', 'hyperParameters', 'sftLossWeightMultiplier'], getv(from_object, ['sft_loss_weight_multiplier']), ) if getv(from_object, ['output_uri']) is not None: setv(parent_object, ['outputUri'], getv(from_object, ['output_uri'])) if getv(from_object, ['encryption_spec']) is not None: setv( parent_object, ['encryptionSpec'], getv(from_object, ['encryption_spec']), ) if getv(from_object, ['reward_config']) is not None: setv( parent_object, ['reinforcementTuningSpec', 'singleRewardConfig'], _SingleReinforcementTuningRewardConfig_to_vertex( getv(from_object, ['reward_config']), to_object, root_object ), ) if getv(from_object, ['composite_reward_config']) is not None: setv( parent_object, ['reinforcementTuningSpec', 'compositeRewardConfig'], _CompositeReinforcementTuningRewardConfig_to_vertex( getv(from_object, ['composite_reward_config']), to_object, root_object, ), ) if getv(from_object, ['samples_per_prompt']) is not None: setv( parent_object, ['reinforcementTuningSpec', 'hyperParameters', 'samplesPerPrompt'], getv(from_object, ['samples_per_prompt']), ) if getv(from_object, ['evaluate_interval']) is not None: setv( parent_object, ['reinforcementTuningSpec', 'hyperParameters', 'evaluateInterval'], getv(from_object, ['evaluate_interval']), ) if getv(from_object, ['checkpoint_interval']) is not None: setv( parent_object, ['reinforcementTuningSpec', 'hyperParameters', 'checkpointInterval'], getv(from_object, ['checkpoint_interval']), ) if getv(from_object, ['max_output_tokens']) is not None: setv( parent_object, ['reinforcementTuningSpec', 'hyperParameters', 'maxOutputTokens'], getv(from_object, ['max_output_tokens']), ) if getv(from_object, ['thinking_level']) is not None: setv( parent_object, ['reinforcementTuningSpec', 'hyperParameters', 'thinkingLevel'], getv(from_object, ['thinking_level']), ) if getv(from_object, ['validation_dataset_uri']) is not None: setv( parent_object, ['reinforcementTuningSpec', 'validationDatasetUri'], getv(from_object, ['validation_dataset_uri']), ) return to_object def _CreateTuningJobParametersPrivate_to_mldev( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['base_model']) is not None: setv(to_object, ['baseModel'], getv(from_object, ['base_model'])) if getv(from_object, ['pre_tuned_model']) is not None: setv(to_object, ['preTunedModel'], getv(from_object, ['pre_tuned_model'])) if getv(from_object, ['training_dataset']) is not None: _TuningDataset_to_mldev( getv(from_object, ['training_dataset']), to_object, root_object ) if getv(from_object, ['config']) is not None: _CreateTuningJobConfig_to_mldev( getv(from_object, ['config']), to_object, root_object ) return to_object def _CreateTuningJobParametersPrivate_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['base_model']) is not None: setv(to_object, ['baseModel'], getv(from_object, ['base_model'])) if getv(from_object, ['pre_tuned_model']) is not None: setv(to_object, ['preTunedModel'], getv(from_object, ['pre_tuned_model'])) if getv(from_object, ['training_dataset']) is not None: _TuningDataset_to_vertex( getv(from_object, ['training_dataset']), to_object, root_object ) if getv(from_object, ['config']) is not None: _CreateTuningJobConfig_to_vertex( getv(from_object, ['config']), to_object, root_object ) return to_object def _DistillationHyperParameters_from_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['adapterSize']) is not None: setv(to_object, ['adapter_size'], getv(from_object, ['adapterSize'])) if getv(from_object, ['epochCount']) is not None: setv(to_object, ['epoch_count'], getv(from_object, ['epochCount'])) if getv(from_object, ['learningRateMultiplier']) is not None: setv( to_object, ['learning_rate_multiplier'], getv(from_object, ['learningRateMultiplier']), ) if getv(from_object, ['generationConfig']) is not None: setv( to_object, ['generation_config'], _GenerationConfig_from_vertex( getv(from_object, ['generationConfig']), to_object, root_object ), ) if getv(from_object, ['learningRate']) is not None: setv(to_object, ['learning_rate'], getv(from_object, ['learningRate'])) if getv(from_object, ['batchSize']) is not None: setv(to_object, ['batch_size'], getv(from_object, ['batchSize'])) return to_object def _DistillationSamplingSpec_from_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['baseTeacherModel']) is not None: setv( to_object, ['base_teacher_model'], getv(from_object, ['baseTeacherModel']), ) if getv(from_object, ['tunedTeacherModelSource']) is not None: setv( to_object, ['tuned_teacher_model_source'], getv(from_object, ['tunedTeacherModelSource']), ) if getv(from_object, ['validationDatasetUri']) is not None: setv( to_object, ['validation_dataset_uri'], getv(from_object, ['validationDatasetUri']), ) if getv(from_object, ['promptDatasetUri']) is not None: setv( to_object, ['prompt_dataset_uri'], getv(from_object, ['promptDatasetUri']), ) if getv(from_object, ['hyperparameters']) is not None: setv( to_object, ['hyperparameters'], _DistillationHyperParameters_from_vertex( getv(from_object, ['hyperparameters']), to_object, root_object ), ) return to_object def _DistillationSpec_from_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['promptDatasetUri']) is not None: setv( to_object, ['prompt_dataset_uri'], getv(from_object, ['promptDatasetUri']), ) if getv(from_object, ['baseTeacherModel']) is not None: setv( to_object, ['base_teacher_model'], getv(from_object, ['baseTeacherModel']), ) if getv(from_object, ['hyperParameters']) is not None: setv( to_object, ['hyper_parameters'], _DistillationHyperParameters_from_vertex( getv(from_object, ['hyperParameters']), to_object, root_object ), ) if getv(from_object, ['pipelineRootDirectory']) is not None: setv( to_object, ['pipeline_root_directory'], getv(from_object, ['pipelineRootDirectory']), ) if getv(from_object, ['studentModel']) is not None: setv(to_object, ['student_model'], getv(from_object, ['studentModel'])) if getv(from_object, ['trainingDatasetUri']) is not None: setv( to_object, ['training_dataset_uri'], getv(from_object, ['trainingDatasetUri']), ) if getv(from_object, ['tunedTeacherModelSource']) is not None: setv( to_object, ['tuned_teacher_model_source'], getv(from_object, ['tunedTeacherModelSource']), ) if getv(from_object, ['validationDatasetUri']) is not None: setv( to_object, ['validation_dataset_uri'], getv(from_object, ['validationDatasetUri']), ) if getv(from_object, ['tuningMode']) is not None: setv(to_object, ['tuning_mode'], getv(from_object, ['tuningMode'])) return to_object def _EvaluationConfig_from_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['metrics']) is not None: setv(to_object, ['metrics'], t.t_metrics(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'], _AutoraterConfig_from_vertex( getv(from_object, ['autoraterConfig']), to_object, root_object ), ) if getv(from_object, ['inferenceGenerationConfig']) is not None: setv( to_object, ['inference_generation_config'], _GenerationConfig_from_vertex( getv(from_object, ['inferenceGenerationConfig']), to_object, root_object, ), ) return to_object def _EvaluationConfig_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['metrics']) is not None: setv(to_object, ['metrics'], t.t_metrics(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'], _AutoraterConfig_to_vertex( getv(from_object, ['autorater_config']), to_object, root_object ), ) if getv(from_object, ['inference_generation_config']) is not None: setv( to_object, ['inferenceGenerationConfig'], _GenerationConfig_to_vertex( getv(from_object, ['inference_generation_config']), to_object, root_object, ), ) return to_object def _ExecutableCode_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['code']) is not None: setv(to_object, ['code'], getv(from_object, ['code'])) if getv(from_object, ['language']) is not None: setv(to_object, ['language'], getv(from_object, ['language'])) if getv(from_object, ['id']) is not None: raise ValueError( 'id parameter is only supported in Gemini Developer API mode, not in' ' Gemini Enterprise Agent Platform mode.' ) return to_object def _GenerationConfig_from_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['modelConfig']) is not None: setv( to_object, ['model_selection_config'], getv(from_object, ['modelConfig']), ) if getv(from_object, ['responseJsonSchema']) is not None: setv( to_object, ['response_json_schema'], getv(from_object, ['responseJsonSchema']), ) if getv(from_object, ['audioTimestamp']) is not None: setv(to_object, ['audio_timestamp'], getv(from_object, ['audioTimestamp'])) if getv(from_object, ['candidateCount']) is not None: setv(to_object, ['candidate_count'], getv(from_object, ['candidateCount'])) if getv(from_object, ['enableAffectiveDialog']) is not None: setv( to_object, ['enable_affective_dialog'], getv(from_object, ['enableAffectiveDialog']), ) if getv(from_object, ['frequencyPenalty']) is not None: setv( to_object, ['frequency_penalty'], getv(from_object, ['frequencyPenalty']), ) if getv(from_object, ['logprobs']) is not None: setv(to_object, ['logprobs'], getv(from_object, ['logprobs'])) if getv(from_object, ['maxOutputTokens']) is not None: setv( to_object, ['max_output_tokens'], getv(from_object, ['maxOutputTokens']) ) if getv(from_object, ['mediaResolution']) is not None: setv( to_object, ['media_resolution'], getv(from_object, ['mediaResolution']) ) if getv(from_object, ['presencePenalty']) is not None: setv( to_object, ['presence_penalty'], getv(from_object, ['presencePenalty']) ) if getv(from_object, ['responseLogprobs']) is not None: setv( to_object, ['response_logprobs'], getv(from_object, ['responseLogprobs']), ) if getv(from_object, ['responseMimeType']) is not None: setv( to_object, ['response_mime_type'], getv(from_object, ['responseMimeType']), ) if getv(from_object, ['responseModalities']) is not None: setv( to_object, ['response_modalities'], getv(from_object, ['responseModalities']), ) if getv(from_object, ['responseSchema']) is not None: setv(to_object, ['response_schema'], getv(from_object, ['responseSchema'])) if getv(from_object, ['routingConfig']) is not None: setv(to_object, ['routing_config'], getv(from_object, ['routingConfig'])) if getv(from_object, ['seed']) is not None: setv(to_object, ['seed'], getv(from_object, ['seed'])) if getv(from_object, ['speechConfig']) is not None: setv(to_object, ['speech_config'], getv(from_object, ['speechConfig'])) if getv(from_object, ['stopSequences']) is not None: setv(to_object, ['stop_sequences'], getv(from_object, ['stopSequences'])) if getv(from_object, ['temperature']) is not None: setv(to_object, ['temperature'], getv(from_object, ['temperature'])) if getv(from_object, ['thinkingConfig']) is not None: setv(to_object, ['thinking_config'], getv(from_object, ['thinkingConfig'])) if getv(from_object, ['topK']) is not None: setv(to_object, ['top_k'], getv(from_object, ['topK'])) if getv(from_object, ['topP']) is not None: setv(to_object, ['top_p'], getv(from_object, ['topP'])) return to_object def _GenerationConfig_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['model_selection_config']) is not None: setv( to_object, ['modelConfig'], getv(from_object, ['model_selection_config']), ) if getv(from_object, ['response_json_schema']) is not None: setv( to_object, ['responseJsonSchema'], getv(from_object, ['response_json_schema']), ) if getv(from_object, ['audio_timestamp']) is not None: setv(to_object, ['audioTimestamp'], getv(from_object, ['audio_timestamp'])) if getv(from_object, ['candidate_count']) is not None: setv(to_object, ['candidateCount'], getv(from_object, ['candidate_count'])) if getv(from_object, ['enable_affective_dialog']) is not None: setv( to_object, ['enableAffectiveDialog'], getv(from_object, ['enable_affective_dialog']), ) if getv(from_object, ['frequency_penalty']) is not None: setv( to_object, ['frequencyPenalty'], getv(from_object, ['frequency_penalty']), ) if getv(from_object, ['logprobs']) is not None: setv(to_object, ['logprobs'], getv(from_object, ['logprobs'])) if getv(from_object, ['max_output_tokens']) is not None: setv( to_object, ['maxOutputTokens'], getv(from_object, ['max_output_tokens']) ) if getv(from_object, ['media_resolution']) is not None: setv( to_object, ['mediaResolution'], getv(from_object, ['media_resolution']) ) if getv(from_object, ['presence_penalty']) is not None: setv( to_object, ['presencePenalty'], getv(from_object, ['presence_penalty']) ) if getv(from_object, ['response_logprobs']) is not None: setv( to_object, ['responseLogprobs'], getv(from_object, ['response_logprobs']), ) if getv(from_object, ['response_mime_type']) is not None: setv( to_object, ['responseMimeType'], getv(from_object, ['response_mime_type']), ) if getv(from_object, ['response_modalities']) is not None: setv( to_object, ['responseModalities'], getv(from_object, ['response_modalities']), ) if getv(from_object, ['response_schema']) is not None: setv(to_object, ['responseSchema'], getv(from_object, ['response_schema'])) if getv(from_object, ['routing_config']) is not None: setv(to_object, ['routingConfig'], getv(from_object, ['routing_config'])) if getv(from_object, ['seed']) is not None: setv(to_object, ['seed'], getv(from_object, ['seed'])) if getv(from_object, ['speech_config']) is not None: setv( to_object, ['speechConfig'], _SpeechConfig_to_vertex( getv(from_object, ['speech_config']), to_object, root_object ), ) if getv(from_object, ['stop_sequences']) is not None: setv(to_object, ['stopSequences'], getv(from_object, ['stop_sequences'])) if getv(from_object, ['temperature']) is not None: setv(to_object, ['temperature'], getv(from_object, ['temperature'])) if getv(from_object, ['thinking_config']) is not None: setv(to_object, ['thinkingConfig'], getv(from_object, ['thinking_config'])) if getv(from_object, ['top_k']) is not None: setv(to_object, ['topK'], getv(from_object, ['top_k'])) if getv(from_object, ['top_p']) is not None: setv(to_object, ['topP'], getv(from_object, ['top_p'])) if getv(from_object, ['enable_enhanced_civic_answers']) is not None: raise ValueError( 'enable_enhanced_civic_answers parameter is only supported in Gemini' ' Developer API mode, not in Gemini Enterprise Agent Platform mode.' ) return to_object def _GetTuningJobParameters_to_mldev( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = 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'])) return to_object def _GetTuningJobParameters_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = 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'])) return to_object def _ListTuningJobsConfig_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = 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'])) return to_object def _ListTuningJobsParameters_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['config']) is not None: _ListTuningJobsConfig_to_vertex( getv(from_object, ['config']), to_object, root_object ) return to_object def _ListTuningJobsResponse_from_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = 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, ['tuningJobs']) is not None: setv( to_object, ['tuning_jobs'], [ _TuningJob_from_vertex(item, to_object, root_object) for item in getv(from_object, ['tuningJobs']) ], ) return to_object def _MultiSpeakerVoiceConfig_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['speaker_voice_configs']) is not None: setv( to_object, ['speakerVoiceConfigs'], [ _SpeakerVoiceConfig_to_vertex(item, to_object, root_object) for item in getv(from_object, ['speaker_voice_configs']) ], ) return to_object def _Part_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['media_resolution']) is not None: setv( to_object, ['mediaResolution'], getv(from_object, ['media_resolution']) ) if getv(from_object, ['code_execution_result']) is not None: setv( to_object, ['codeExecutionResult'], _CodeExecutionResult_to_vertex( getv(from_object, ['code_execution_result']), to_object, root_object ), ) if getv(from_object, ['executable_code']) is not None: setv( to_object, ['executableCode'], _ExecutableCode_to_vertex( getv(from_object, ['executable_code']), to_object, root_object ), ) if getv(from_object, ['file_data']) is not None: setv(to_object, ['fileData'], getv(from_object, ['file_data'])) if getv(from_object, ['function_call']) is not None: setv(to_object, ['functionCall'], getv(from_object, ['function_call'])) if getv(from_object, ['function_response']) is not None: setv( to_object, ['functionResponse'], getv(from_object, ['function_response']), ) if getv(from_object, ['inline_data']) is not None: setv(to_object, ['inlineData'], getv(from_object, ['inline_data'])) if getv(from_object, ['text']) is not None: setv(to_object, ['text'], getv(from_object, ['text'])) if getv(from_object, ['thought']) is not None: setv(to_object, ['thought'], getv(from_object, ['thought'])) if getv(from_object, ['thought_signature']) is not None: setv( to_object, ['thoughtSignature'], getv(from_object, ['thought_signature']), ) if getv(from_object, ['video_metadata']) is not None: setv(to_object, ['videoMetadata'], getv(from_object, ['video_metadata'])) if getv(from_object, ['tool_call']) is not None: raise ValueError( 'tool_call parameter is only supported in Gemini Developer API mode,' ' not in Gemini Enterprise Agent Platform mode.' ) if getv(from_object, ['tool_response']) is not None: raise ValueError( 'tool_response parameter is only supported in Gemini Developer API' ' mode, not in Gemini Enterprise Agent Platform mode.' ) if getv(from_object, ['part_metadata']) is not None: raise ValueError( 'part_metadata parameter is only supported in Gemini Developer API' ' mode, not in Gemini Enterprise Agent Platform mode.' ) return to_object def _ReinforcementTuningAutoraterScorer_from_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['autoraterConfig']) is not None: setv( to_object, ['autorater_config'], _AutoraterConfig_from_vertex( getv(from_object, ['autoraterConfig']), to_object, root_object ), ) if getv(from_object, ['autoraterPrompt']) is not None: setv( to_object, ['autorater_prompt'], getv(from_object, ['autoraterPrompt']) ) if getv(from_object, ['autoraterResponseParseConfig']) is not None: setv( to_object, ['autorater_response_parse_config'], getv(from_object, ['autoraterResponseParseConfig']), ) if getv(from_object, ['parsedResponseConversionScorer']) is not None: setv( to_object, ['parsed_response_conversion_scorer'], getv(from_object, ['parsedResponseConversionScorer']), ) if getv(from_object, ['exactMatchScorer']) is not None: setv( to_object, ['exact_match_scorer'], getv(from_object, ['exactMatchScorer']), ) return to_object def _ReinforcementTuningAutoraterScorer_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['autorater_config']) is not None: setv( to_object, ['autoraterConfig'], _AutoraterConfig_to_vertex( getv(from_object, ['autorater_config']), to_object, root_object ), ) if getv(from_object, ['autorater_prompt']) is not None: setv( to_object, ['autoraterPrompt'], getv(from_object, ['autorater_prompt']) ) if getv(from_object, ['autorater_response_parse_config']) is not None: setv( to_object, ['autoraterResponseParseConfig'], getv(from_object, ['autorater_response_parse_config']), ) if getv(from_object, ['parsed_response_conversion_scorer']) is not None: setv( to_object, ['parsedResponseConversionScorer'], getv(from_object, ['parsed_response_conversion_scorer']), ) if getv(from_object, ['exact_match_scorer']) is not None: setv( to_object, ['exactMatchScorer'], getv(from_object, ['exact_match_scorer']), ) return to_object def _ReinforcementTuningExample_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['contents']) is not None: setv( to_object, ['contents'], [ _Content_to_vertex(item, to_object, root_object) for item in getv(from_object, ['contents']) ], ) if getv(from_object, ['references']) is not None: setv(to_object, ['references'], getv(from_object, ['references'])) if getv(from_object, ['system_instruction']) is not None: setv( to_object, ['systemInstruction'], _Content_to_vertex( getv(from_object, ['system_instruction']), to_object, root_object ), ) return to_object def _ReinforcementTuningSpec_from_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['compositeRewardConfig']) is not None: setv( to_object, ['composite_reward_config'], _CompositeReinforcementTuningRewardConfig_from_vertex( getv(from_object, ['compositeRewardConfig']), to_object, root_object ), ) if getv(from_object, ['trainingDatasetUri']) is not None: setv( to_object, ['training_dataset_uri'], getv(from_object, ['trainingDatasetUri']), ) if getv(from_object, ['validationDatasetUri']) is not None: setv( to_object, ['validation_dataset_uri'], getv(from_object, ['validationDatasetUri']), ) if getv(from_object, ['hyperParameters']) is not None: setv( to_object, ['hyper_parameters'], getv(from_object, ['hyperParameters']) ) if getv(from_object, ['singleRewardConfig']) is not None: setv( to_object, ['single_reward_config'], _SingleReinforcementTuningRewardConfig_from_vertex( getv(from_object, ['singleRewardConfig']), to_object, root_object ), ) return to_object def _ReplicatedVoiceConfig_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['mime_type']) is not None: setv(to_object, ['mimeType'], getv(from_object, ['mime_type'])) if getv(from_object, ['voice_sample_audio']) is not None: setv( to_object, ['voiceSampleAudio'], getv(from_object, ['voice_sample_audio']), ) if getv(from_object, ['consent_audio']) is not None: raise ValueError( 'consent_audio parameter is only supported in Gemini Developer API' ' mode, not in Gemini Enterprise Agent Platform mode.' ) if getv(from_object, ['voice_consent_signature']) is not None: raise ValueError( 'voice_consent_signature parameter is only supported in Gemini' ' Developer API mode, not in Gemini Enterprise Agent Platform mode.' ) return to_object def _SingleReinforcementTuningRewardConfig_from_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['autoraterScorer']) is not None: setv( to_object, ['autorater_scorer'], _ReinforcementTuningAutoraterScorer_from_vertex( getv(from_object, ['autoraterScorer']), to_object, root_object ), ) if getv(from_object, ['rewardName']) is not None: setv(to_object, ['reward_name'], getv(from_object, ['rewardName'])) if getv(from_object, ['parseResponseConfig']) is not None: setv( to_object, ['parse_response_config'], getv(from_object, ['parseResponseConfig']), ) if getv(from_object, ['codeExecutionRewardScorer']) is not None: setv( to_object, ['code_execution_reward_scorer'], getv(from_object, ['codeExecutionRewardScorer']), ) if getv(from_object, ['stringMatchRewardScorer']) is not None: setv( to_object, ['string_match_reward_scorer'], getv(from_object, ['stringMatchRewardScorer']), ) if getv(from_object, ['cloudRunRewardScorer']) is not None: setv( to_object, ['cloud_run_reward_scorer'], getv(from_object, ['cloudRunRewardScorer']), ) return to_object def _SingleReinforcementTuningRewardConfig_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['autorater_scorer']) is not None: setv( to_object, ['autoraterScorer'], _ReinforcementTuningAutoraterScorer_to_vertex( getv(from_object, ['autorater_scorer']), to_object, root_object ), ) if getv(from_object, ['reward_name']) is not None: setv(to_object, ['rewardName'], getv(from_object, ['reward_name'])) if getv(from_object, ['parse_response_config']) is not None: setv( to_object, ['parseResponseConfig'], getv(from_object, ['parse_response_config']), ) if getv(from_object, ['code_execution_reward_scorer']) is not None: setv( to_object, ['codeExecutionRewardScorer'], getv(from_object, ['code_execution_reward_scorer']), ) if getv(from_object, ['string_match_reward_scorer']) is not None: setv( to_object, ['stringMatchRewardScorer'], getv(from_object, ['string_match_reward_scorer']), ) if getv(from_object, ['cloud_run_reward_scorer']) is not None: setv( to_object, ['cloudRunRewardScorer'], getv(from_object, ['cloud_run_reward_scorer']), ) return to_object def _SpeakerVoiceConfig_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['speaker']) is not None: setv(to_object, ['speaker'], getv(from_object, ['speaker'])) if getv(from_object, ['voice_config']) is not None: setv( to_object, ['voiceConfig'], _VoiceConfig_to_vertex( getv(from_object, ['voice_config']), to_object, root_object ), ) return to_object def _SpeechConfig_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['voice_config']) is not None: setv( to_object, ['voiceConfig'], _VoiceConfig_to_vertex( getv(from_object, ['voice_config']), to_object, root_object ), ) if getv(from_object, ['language_code']) is not None: setv(to_object, ['languageCode'], getv(from_object, ['language_code'])) if getv(from_object, ['multi_speaker_voice_config']) is not None: setv( to_object, ['multiSpeakerVoiceConfig'], _MultiSpeakerVoiceConfig_to_vertex( getv(from_object, ['multi_speaker_voice_config']), to_object, root_object, ), ) return to_object def _TunedModel_from_mldev( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['name']) is not None: setv(to_object, ['model'], getv(from_object, ['name'])) if getv(from_object, ['name']) is not None: setv(to_object, ['endpoint'], getv(from_object, ['name'])) return to_object def _TuningDataset_to_mldev( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['gcs_uri']) is not None: raise ValueError( 'gcs_uri parameter is only supported in Gemini Enterprise Agent' ' Platform mode, not in Gemini Developer API mode.' ) if getv(from_object, ['vertex_dataset_resource']) is not None: raise ValueError( 'vertex_dataset_resource parameter is only supported in Gemini' ' Enterprise Agent Platform mode, not in Gemini Developer API mode.' ) if getv(from_object, ['examples']) is not None: setv( to_object, ['examples', 'examples'], [item for item in getv(from_object, ['examples'])], ) return to_object def _TuningDataset_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} discriminator = getv(root_object, ['config', 'method']) if discriminator is None: discriminator = 'SUPERVISED_FINE_TUNING' if discriminator == 'SUPERVISED_FINE_TUNING': if getv(from_object, ['gcs_uri']) is not None: setv( parent_object, ['supervisedTuningSpec', 'trainingDatasetUri'], getv(from_object, ['gcs_uri']), ) elif discriminator == 'PREFERENCE_TUNING': if getv(from_object, ['gcs_uri']) is not None: setv( parent_object, ['preferenceOptimizationSpec', 'trainingDatasetUri'], getv(from_object, ['gcs_uri']), ) elif discriminator == 'DISTILLATION': if getv(from_object, ['gcs_uri']) is not None: setv( parent_object, ['distillationSpec', 'promptDatasetUri'], getv(from_object, ['gcs_uri']), ) elif discriminator == 'REINFORCEMENT_TUNING': if getv(from_object, ['gcs_uri']) is not None: setv( parent_object, ['reinforcementTuningSpec', 'trainingDatasetUri'], getv(from_object, ['gcs_uri']), ) discriminator = getv(root_object, ['config', 'method']) if discriminator is None: discriminator = 'SUPERVISED_FINE_TUNING' if discriminator == 'SUPERVISED_FINE_TUNING': if getv(from_object, ['vertex_dataset_resource']) is not None: setv( parent_object, ['supervisedTuningSpec', 'trainingDatasetUri'], getv(from_object, ['vertex_dataset_resource']), ) elif discriminator == 'PREFERENCE_TUNING': if getv(from_object, ['vertex_dataset_resource']) is not None: setv( parent_object, ['preferenceOptimizationSpec', 'trainingDatasetUri'], getv(from_object, ['vertex_dataset_resource']), ) elif discriminator == 'DISTILLATION': if getv(from_object, ['vertex_dataset_resource']) is not None: setv( parent_object, ['distillationSpec', 'promptDatasetUri'], getv(from_object, ['vertex_dataset_resource']), ) elif discriminator == 'REINFORCEMENT_TUNING': if getv(from_object, ['vertex_dataset_resource']) is not None: setv( parent_object, ['reinforcementTuningSpec', 'trainingDatasetUri'], getv(from_object, ['vertex_dataset_resource']), ) if getv(from_object, ['examples']) is not None: raise ValueError( 'examples parameter is only supported in Gemini Developer API mode, not' ' in Gemini Enterprise Agent Platform mode.' ) return to_object def _TuningJob_from_mldev( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = 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, ['name']) is not None: setv(to_object, ['name'], getv(from_object, ['name'])) if getv(from_object, ['state']) is not None: setv( to_object, ['state'], t.t_tuning_job_status(getv(from_object, ['state'])), ) if getv(from_object, ['createTime']) is not None: setv(to_object, ['create_time'], getv(from_object, ['createTime'])) if getv(from_object, ['tuningTask', 'startTime']) is not None: setv( to_object, ['start_time'], getv(from_object, ['tuningTask', 'startTime']), ) if getv(from_object, ['tuningTask', 'completeTime']) is not None: setv( to_object, ['end_time'], getv(from_object, ['tuningTask', 'completeTime']), ) if getv(from_object, ['updateTime']) is not None: setv(to_object, ['update_time'], getv(from_object, ['updateTime'])) if getv(from_object, ['description']) is not None: setv(to_object, ['description'], getv(from_object, ['description'])) if getv(from_object, ['baseModel']) is not None: setv(to_object, ['base_model'], getv(from_object, ['baseModel'])) if getv(from_object, ['_self']) is not None: setv( to_object, ['tuned_model'], _TunedModel_from_mldev( getv(from_object, ['_self']), to_object, root_object ), ) return to_object def _TuningJob_from_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = 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, ['name']) is not None: setv(to_object, ['name'], getv(from_object, ['name'])) if getv(from_object, ['state']) is not None: setv( to_object, ['state'], t.t_tuning_job_status(getv(from_object, ['state'])), ) if getv(from_object, ['createTime']) is not None: setv(to_object, ['create_time'], getv(from_object, ['createTime'])) if getv(from_object, ['startTime']) is not None: setv(to_object, ['start_time'], getv(from_object, ['startTime'])) if getv(from_object, ['endTime']) is not None: setv(to_object, ['end_time'], getv(from_object, ['endTime'])) if getv(from_object, ['updateTime']) is not None: setv(to_object, ['update_time'], getv(from_object, ['updateTime'])) if getv(from_object, ['error']) is not None: setv(to_object, ['error'], getv(from_object, ['error'])) if getv(from_object, ['description']) is not None: setv(to_object, ['description'], getv(from_object, ['description'])) if getv(from_object, ['baseModel']) is not None: setv(to_object, ['base_model'], getv(from_object, ['baseModel'])) if getv(from_object, ['tunedModel']) is not None: setv(to_object, ['tuned_model'], getv(from_object, ['tunedModel'])) if getv(from_object, ['preTunedModel']) is not None: setv(to_object, ['pre_tuned_model'], getv(from_object, ['preTunedModel'])) if getv(from_object, ['supervisedTuningSpec']) is not None: setv( to_object, ['supervised_tuning_spec'], getv(from_object, ['supervisedTuningSpec']), ) if getv(from_object, ['preferenceOptimizationSpec']) is not None: setv( to_object, ['preference_optimization_spec'], getv(from_object, ['preferenceOptimizationSpec']), ) if getv(from_object, ['distillationSpec']) is not None: setv( to_object, ['distillation_spec'], _DistillationSpec_from_vertex( getv(from_object, ['distillationSpec']), to_object, root_object ), ) if getv(from_object, ['reinforcementTuningSpec']) is not None: setv( to_object, ['reinforcement_tuning_spec'], _ReinforcementTuningSpec_from_vertex( getv(from_object, ['reinforcementTuningSpec']), to_object, root_object, ), ) if getv(from_object, ['tuningDataStats']) is not None: setv( to_object, ['tuning_data_stats'], getv(from_object, ['tuningDataStats']) ) if getv(from_object, ['encryptionSpec']) is not None: setv(to_object, ['encryption_spec'], getv(from_object, ['encryptionSpec'])) if getv(from_object, ['partnerModelTuningSpec']) is not None: setv( to_object, ['partner_model_tuning_spec'], getv(from_object, ['partnerModelTuningSpec']), ) if getv(from_object, ['evaluationConfig']) is not None: setv( to_object, ['evaluation_config'], _EvaluationConfig_from_vertex( getv(from_object, ['evaluationConfig']), to_object, root_object ), ) if getv(from_object, ['customBaseModel']) is not None: setv( to_object, ['custom_base_model'], getv(from_object, ['customBaseModel']) ) if getv(from_object, ['evaluateDatasetRuns']) is not None: setv( to_object, ['evaluate_dataset_runs'], [item for item in getv(from_object, ['evaluateDatasetRuns'])], ) if getv(from_object, ['experiment']) is not None: setv(to_object, ['experiment'], getv(from_object, ['experiment'])) if getv(from_object, ['fullFineTuningSpec']) is not None: setv( to_object, ['full_fine_tuning_spec'], getv(from_object, ['fullFineTuningSpec']), ) if getv(from_object, ['labels']) is not None: setv(to_object, ['labels'], getv(from_object, ['labels'])) if getv(from_object, ['outputUri']) is not None: setv(to_object, ['output_uri'], getv(from_object, ['outputUri'])) if getv(from_object, ['pipelineJob']) is not None: setv(to_object, ['pipeline_job'], getv(from_object, ['pipelineJob'])) if getv(from_object, ['serviceAccount']) is not None: setv(to_object, ['service_account'], getv(from_object, ['serviceAccount'])) if getv(from_object, ['tunedModelDisplayName']) is not None: setv( to_object, ['tuned_model_display_name'], getv(from_object, ['tunedModelDisplayName']), ) if getv(from_object, ['tuningJobState']) is not None: setv(to_object, ['tuning_job_state'], getv(from_object, ['tuningJobState'])) if getv(from_object, ['veoTuningSpec']) is not None: setv(to_object, ['veo_tuning_spec'], getv(from_object, ['veoTuningSpec'])) if getv(from_object, ['tuningJobMetadata']) is not None: setv( to_object, ['tuning_job_metadata'], getv(from_object, ['tuningJobMetadata']), ) if getv(from_object, ['veoLoraTuningSpec']) is not None: setv( to_object, ['veo_lora_tuning_spec'], getv(from_object, ['veoLoraTuningSpec']), ) if getv(from_object, ['distillationSamplingSpec']) is not None: setv( to_object, ['distillation_sampling_spec'], _DistillationSamplingSpec_from_vertex( getv(from_object, ['distillationSamplingSpec']), to_object, root_object, ), ) return to_object def _TuningOperation_from_mldev( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = 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, ['name']) is not None: setv(to_object, ['name'], getv(from_object, ['name'])) if getv(from_object, ['metadata']) is not None: setv(to_object, ['metadata'], getv(from_object, ['metadata'])) if getv(from_object, ['done']) is not None: setv(to_object, ['done'], getv(from_object, ['done'])) if getv(from_object, ['error']) is not None: setv(to_object, ['error'], getv(from_object, ['error'])) return to_object def _TuningValidationDataset_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['gcs_uri']) is not None: setv(to_object, ['validationDatasetUri'], getv(from_object, ['gcs_uri'])) if getv(from_object, ['vertex_dataset_resource']) is not None: setv( to_object, ['validationDatasetUri'], getv(from_object, ['vertex_dataset_resource']), ) return to_object def _ValidateRewardParameters_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['parent']) is not None: setv(to_object, ['_url', 'parent'], getv(from_object, ['parent'])) if getv(from_object, ['sample_response']) is not None: setv( to_object, ['sampleResponse'], _Content_to_vertex( getv(from_object, ['sample_response']), to_object, root_object ), ) if getv(from_object, ['example']) is not None: setv( to_object, ['example'], _ReinforcementTuningExample_to_vertex( getv(from_object, ['example']), to_object, root_object ), ) if getv(from_object, ['single_reward_config']) is not None: setv( to_object, ['singleRewardConfig'], _SingleReinforcementTuningRewardConfig_to_vertex( getv(from_object, ['single_reward_config']), to_object, root_object ), ) if getv(from_object, ['composite_reward_config']) is not None: setv( to_object, ['compositeRewardConfig'], _CompositeReinforcementTuningRewardConfig_to_vertex( getv(from_object, ['composite_reward_config']), to_object, root_object, ), ) return to_object def _ValidateRewardResponse_from_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = 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, ['overallReward']) is not None: setv(to_object, ['overall_reward'], getv(from_object, ['overallReward'])) if getv(from_object, ['error']) is not None: setv(to_object, ['error'], getv(from_object, ['error'])) if getv(from_object, ['rewardInfoDetails']) is not None: setv( to_object, ['reward_info_details'], {k: v for k, v in getv(from_object, ['rewardInfoDetails']).items()}, ) return to_object def _VoiceConfig_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, root_object: Optional[Union[dict[str, Any], object]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['replicated_voice_config']) is not None: setv( to_object, ['replicatedVoiceConfig'], _ReplicatedVoiceConfig_to_vertex( getv(from_object, ['replicated_voice_config']), to_object, root_object, ), ) if getv(from_object, ['prebuilt_voice_config']) is not None: setv( to_object, ['prebuiltVoiceConfig'], getv(from_object, ['prebuilt_voice_config']), ) return to_object class Tunings(_api_module.BaseModule): def _get( self, *, name: str, config: Optional[types.GetTuningJobConfigOrDict] = None, ) -> types.TuningJob: """Gets a TuningJob. Args: name: The resource name of the tuning job. Returns: A TuningJob object. """ parameter_model = types._GetTuningJobParameters( name=name, config=config, ) request_url_dict: Optional[dict[str, str]] if self._api_client.vertexai: request_dict = _GetTuningJobParameters_to_vertex( parameter_model, None, parameter_model ) request_url_dict = request_dict.get('_url') if request_url_dict: path = '{name}'.format_map(request_url_dict) else: path = '{name}' else: request_dict = _GetTuningJobParameters_to_mldev( parameter_model, None, 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) if self._api_client.vertexai: response_dict = _TuningJob_from_vertex( response_dict, None, parameter_model ) if not self._api_client.vertexai: response_dict = _TuningJob_from_mldev( response_dict, None, parameter_model ) return_value = types.TuningJob._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 {}, ) return_value.sdk_http_response = types.HttpResponse( headers=response.headers ) self._api_client._verify_response(return_value) return return_value def _list( self, *, config: Optional[types.ListTuningJobsConfigOrDict] = None ) -> types.ListTuningJobsResponse: parameter_model = types._ListTuningJobsParameters( 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 = _ListTuningJobsParameters_to_vertex( parameter_model, None, parameter_model ) request_url_dict = request_dict.get('_url') if request_url_dict: path = 'tuningJobs'.format_map(request_url_dict) else: path = 'tuningJobs' 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 = _ListTuningJobsResponse_from_vertex( response_dict, None, parameter_model ) return_value = types.ListTuningJobsResponse._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 {}, ) return_value.sdk_http_response = types.HttpResponse( headers=response.headers ) self._api_client._verify_response(return_value) return return_value def cancel( self, *, name: str, config: Optional[types.CancelTuningJobConfigOrDict] = None, ) -> types.CancelTuningJobResponse: """Cancels a tuning job. Args: name (str): TuningJob resource name. """ parameter_model = types._CancelTuningJobParameters( name=name, config=config, ) request_url_dict: Optional[dict[str, str]] if self._api_client.vertexai: request_dict = _CancelTuningJobParameters_to_vertex( parameter_model, None, parameter_model ) request_url_dict = request_dict.get('_url') if request_url_dict: path = '{name}:cancel'.format_map(request_url_dict) else: path = '{name}:cancel' else: request_dict = _CancelTuningJobParameters_to_mldev( parameter_model, None, parameter_model ) request_url_dict = request_dict.get('_url') if request_url_dict: path = '{name}:cancel'.format_map(request_url_dict) else: path = '{name}:cancel' 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 = _CancelTuningJobResponse_from_vertex( response_dict, None, parameter_model ) if not self._api_client.vertexai: response_dict = _CancelTuningJobResponse_from_mldev( response_dict, None, parameter_model ) return_value = types.CancelTuningJobResponse._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 {}, ) return_value.sdk_http_response = types.HttpResponse( headers=response.headers ) self._api_client._verify_response(return_value) return return_value def _tune( self, *, base_model: Optional[str] = None, pre_tuned_model: Optional[types.PreTunedModelOrDict] = None, training_dataset: types.TuningDatasetOrDict, config: Optional[types.CreateTuningJobConfigOrDict] = None, ) -> types.TuningJob: """Creates a tuning job and returns the TuningJob object. Args: base_model: The name of the model to tune. training_dataset: The training dataset to use. config: The configuration to use for the tuning job. Returns: A TuningJob object. """ parameter_model = types._CreateTuningJobParametersPrivate( base_model=base_model, pre_tuned_model=pre_tuned_model, training_dataset=training_dataset, 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 = _CreateTuningJobParametersPrivate_to_vertex( parameter_model, None, parameter_model ) request_url_dict = request_dict.get('_url') if request_url_dict: path = 'tuningJobs'.format_map(request_url_dict) else: path = 'tuningJobs' 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 = _TuningJob_from_vertex( response_dict, None, parameter_model ) return_value = types.TuningJob._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 {}, ) return_value.sdk_http_response = types.HttpResponse( headers=response.headers ) self._api_client._verify_response(return_value) return return_value def _tune_mldev( self, *, base_model: Optional[str] = None, pre_tuned_model: Optional[types.PreTunedModelOrDict] = None, training_dataset: types.TuningDatasetOrDict, config: Optional[types.CreateTuningJobConfigOrDict] = None, ) -> types.TuningOperation: """Creates a tuning job and returns the TuningJob object. Args: base_model: The name of the model to tune. training_dataset: The training dataset to use. config: The configuration to use for the tuning job. Returns: A TuningJob operation. """ parameter_model = types._CreateTuningJobParametersPrivate( base_model=base_model, pre_tuned_model=pre_tuned_model, training_dataset=training_dataset, config=config, ) request_url_dict: Optional[dict[str, str]] if self._api_client.vertexai: raise ValueError( 'This method is only supported in Gemini Developer API mode, not in' ' Gemini Enterprise Agent Platform mode.' ) else: request_dict = _CreateTuningJobParametersPrivate_to_mldev( parameter_model, None, parameter_model ) request_url_dict = request_dict.get('_url') if request_url_dict: path = 'tunedModels'.format_map(request_url_dict) else: path = 'tunedModels' 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 not self._api_client.vertexai: response_dict = _TuningOperation_from_mldev( response_dict, None, parameter_model ) return_value = types.TuningOperation._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 {}, ) return_value.sdk_http_response = types.HttpResponse( headers=response.headers ) self._api_client._verify_response(return_value) return return_value def validate_reward( self, *, parent: str, sample_response: types.ContentOrDict, example: types.ReinforcementTuningExampleOrDict, single_reward_config: Optional[ types.SingleReinforcementTuningRewardConfigOrDict ] = None, composite_reward_config: Optional[ types.CompositeReinforcementTuningRewardConfigOrDict ] = None, config: Optional[types.ValidateRewardConfigOrDict] = None, ) -> types.ValidateRewardResponse: """Validates a reinforcement tuning reward configuration. Allows users to validate a reinforcement tuning reward configuration against a sample response and example before creating a reinforcement tuning job, so that they can iterate on the reward configuration without having to create a tuning job each time. Args: parent: The resource name of the Location to validate the reward in, e.g. `projects/{project}/locations/{location}`. sample_response: The sample response for validating the reward configuration. example: The example to validate the reward configuration. single_reward_config: Single reward function configuration for reinforcement tuning. Mutually exclusive with composite_reward_config. composite_reward_config: Composite reward function configuration for reinforcement tuning. Mutually exclusive with single_reward_config. config: Optional parameters for the request. Returns: A ValidateRewardResponse with the computed reward(s). """ parameter_model = types._ValidateRewardParameters( parent=parent, sample_response=sample_response, example=example, single_reward_config=single_reward_config, composite_reward_config=composite_reward_config, 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 = _ValidateRewardParameters_to_vertex( parameter_model, None, parameter_model ) request_url_dict = request_dict.get('_url') if request_url_dict: path = ( '{parent}/tuningJobs:validateReinforcementTuningReward'.format_map( request_url_dict ) ) else: path = '{parent}/tuningJobs:validateReinforcementTuningReward' 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 = _ValidateRewardResponse_from_vertex( response_dict, None, parameter_model ) return_value = types.ValidateRewardResponse._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 {}, ) return_value.sdk_http_response = types.HttpResponse( headers=response.headers ) self._api_client._verify_response(return_value) return return_value def get( self, *, name: str, config: Optional[types.GetTuningJobConfigOrDict] = None, ) -> types.TuningJob: job = self._get(name=name, config=config) if ( job.experiment and self._api_client.vertexai and self._api_client.project is not None ): _IpythonUtils.display_experiment_button( experiment=job.experiment, project=self._api_client.project, ) return job @_common.experimental_warning( "The SDK's tuning implementation is experimental, " 'and may change in future versions.', ) def tune( self, *, base_model: str, training_dataset: types.TuningDatasetOrDict, config: Optional[types.CreateTuningJobConfigOrDict] = None, ) -> types.TuningJob: if self._api_client.vertexai: if base_model.startswith('projects/'): # Pre-tuned model checkpoint_id = None if config: checkpoint_id = getattr(config, 'pre_tuned_model_checkpoint_id', None) pre_tuned_model = types.PreTunedModel( tuned_model_name=base_model, checkpoint_id=checkpoint_id ) tuning_job = self._tune( pre_tuned_model=pre_tuned_model, training_dataset=training_dataset, config=config, ) else: validated_evaluation_config: Optional[types.EvaluationConfig] = None if ( config is not None and getattr(config, 'evaluation_config', None) is not None ): evaluation_config = getattr(config, 'evaluation_config') if isinstance(evaluation_config, dict): evaluation_config = types.EvaluationConfig(**evaluation_config) if ( not evaluation_config.metrics or not evaluation_config.output_config ): raise ValueError( 'Evaluation config must have at least one metric and an output' ' config.' ) for i in range(len(evaluation_config.metrics)): if isinstance(evaluation_config.metrics[i], dict): evaluation_config.metrics[i] = types.Metric.model_validate( evaluation_config.metrics[i] ) if isinstance(config, dict): config['evaluation_config'] = evaluation_config else: config.evaluation_config = evaluation_config validated_evaluation_config = evaluation_config tuning_job = self._tune( base_model=base_model, training_dataset=training_dataset, config=config, ) if ( config is not None and getattr(config, 'evaluation_config', None) is not None ): tuning_job.evaluation_config = validated_evaluation_config else: operation = self._tune_mldev( base_model=base_model, training_dataset=training_dataset, config=config, ) if operation.metadata is not None and 'tunedModel' in operation.metadata: tuned_model_name = operation.metadata['tunedModel'] else: if operation.name is None: raise ValueError('Operation name is required.') tuned_model_name = operation.name.partition('/operations/')[0] tuning_job = types.TuningJob( name=tuned_model_name, state=types.JobState.JOB_STATE_QUEUED, ) if tuning_job.name and self._api_client.vertexai: _IpythonUtils.display_model_tuning_button( tuning_job_resource=tuning_job.name ) return tuning_job def list( self, *, config: Optional[types.ListTuningJobsConfigOrDict] = None ) -> Pager[types.TuningJob]: """Lists `TuningJob` objects. Args: config: The configuration for the list request. Returns: A Pager object that contains one page of tuning jobs. When iterating over the pager, it automatically fetches the next page if there are more. Usage: .. code-block:: python for tuning_job in client.tunings.list(): print(tuning_job.name) """ list_request = self._list return Pager( 'tuning_jobs', list_request, self._list(config=config), config, ) class AsyncTunings(_api_module.BaseModule): async def _get( self, *, name: str, config: Optional[types.GetTuningJobConfigOrDict] = None, ) -> types.TuningJob: """Gets a TuningJob. Args: name: The resource name of the tuning job. Returns: A TuningJob object. """ parameter_model = types._GetTuningJobParameters( name=name, config=config, ) request_url_dict: Optional[dict[str, str]] if self._api_client.vertexai: request_dict = _GetTuningJobParameters_to_vertex( parameter_model, None, parameter_model ) request_url_dict = request_dict.get('_url') if request_url_dict: path = '{name}'.format_map(request_url_dict) else: path = '{name}' else: request_dict = _GetTuningJobParameters_to_mldev( parameter_model, None, 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) if self._api_client.vertexai: response_dict = _TuningJob_from_vertex( response_dict, None, parameter_model ) if not self._api_client.vertexai: response_dict = _TuningJob_from_mldev( response_dict, None, parameter_model ) return_value = types.TuningJob._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 {}, ) return_value.sdk_http_response = types.HttpResponse( headers=response.headers ) self._api_client._verify_response(return_value) return return_value async def _list( self, *, config: Optional[types.ListTuningJobsConfigOrDict] = None ) -> types.ListTuningJobsResponse: parameter_model = types._ListTuningJobsParameters( 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 = _ListTuningJobsParameters_to_vertex( parameter_model, None, parameter_model ) request_url_dict = request_dict.get('_url') if request_url_dict: path = 'tuningJobs'.format_map(request_url_dict) else: path = 'tuningJobs' 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 = _ListTuningJobsResponse_from_vertex( response_dict, None, parameter_model ) return_value = types.ListTuningJobsResponse._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 {}, ) return_value.sdk_http_response = types.HttpResponse( headers=response.headers ) self._api_client._verify_response(return_value) return return_value async def cancel( self, *, name: str, config: Optional[types.CancelTuningJobConfigOrDict] = None, ) -> types.CancelTuningJobResponse: """Cancels a tuning job asynchronously. Args: name (str): A TuningJob resource name. """ parameter_model = types._CancelTuningJobParameters( name=name, config=config, ) request_url_dict: Optional[dict[str, str]] if self._api_client.vertexai: request_dict = _CancelTuningJobParameters_to_vertex( parameter_model, None, parameter_model ) request_url_dict = request_dict.get('_url') if request_url_dict: path = '{name}:cancel'.format_map(request_url_dict) else: path = '{name}:cancel' else: request_dict = _CancelTuningJobParameters_to_mldev( parameter_model, None, parameter_model ) request_url_dict = request_dict.get('_url') if request_url_dict: path = '{name}:cancel'.format_map(request_url_dict) else: path = '{name}:cancel' 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 = _CancelTuningJobResponse_from_vertex( response_dict, None, parameter_model ) if not self._api_client.vertexai: response_dict = _CancelTuningJobResponse_from_mldev( response_dict, None, parameter_model ) return_value = types.CancelTuningJobResponse._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 {}, ) return_value.sdk_http_response = types.HttpResponse( headers=response.headers ) self._api_client._verify_response(return_value) return return_value async def _tune( self, *, base_model: Optional[str] = None, pre_tuned_model: Optional[types.PreTunedModelOrDict] = None, training_dataset: types.TuningDatasetOrDict, config: Optional[types.CreateTuningJobConfigOrDict] = None, ) -> types.TuningJob: """Creates a tuning job and returns the TuningJob object. Args: base_model: The name of the model to tune. training_dataset: The training dataset to use. config: The configuration to use for the tuning job. Returns: A TuningJob object. """ parameter_model = types._CreateTuningJobParametersPrivate( base_model=base_model, pre_tuned_model=pre_tuned_model, training_dataset=training_dataset, 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 = _CreateTuningJobParametersPrivate_to_vertex( parameter_model, None, parameter_model ) request_url_dict = request_dict.get('_url') if request_url_dict: path = 'tuningJobs'.format_map(request_url_dict) else: path = 'tuningJobs' 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 = _TuningJob_from_vertex( response_dict, None, parameter_model ) return_value = types.TuningJob._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 {}, ) return_value.sdk_http_response = types.HttpResponse( headers=response.headers ) self._api_client._verify_response(return_value) return return_value async def _tune_mldev( self, *, base_model: Optional[str] = None, pre_tuned_model: Optional[types.PreTunedModelOrDict] = None, training_dataset: types.TuningDatasetOrDict, config: Optional[types.CreateTuningJobConfigOrDict] = None, ) -> types.TuningOperation: """Creates a tuning job and returns the TuningJob object. Args: base_model: The name of the model to tune. training_dataset: The training dataset to use. config: The configuration to use for the tuning job. Returns: A TuningJob operation. """ parameter_model = types._CreateTuningJobParametersPrivate( base_model=base_model, pre_tuned_model=pre_tuned_model, training_dataset=training_dataset, config=config, ) request_url_dict: Optional[dict[str, str]] if self._api_client.vertexai: raise ValueError( 'This method is only supported in Gemini Developer API mode, not in' ' Gemini Enterprise Agent Platform mode.' ) else: request_dict = _CreateTuningJobParametersPrivate_to_mldev( parameter_model, None, parameter_model ) request_url_dict = request_dict.get('_url') if request_url_dict: path = 'tunedModels'.format_map(request_url_dict) else: path = 'tunedModels' 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 not self._api_client.vertexai: response_dict = _TuningOperation_from_mldev( response_dict, None, parameter_model ) return_value = types.TuningOperation._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 {}, ) return_value.sdk_http_response = types.HttpResponse( headers=response.headers ) self._api_client._verify_response(return_value) return return_value async def validate_reward( self, *, parent: str, sample_response: types.ContentOrDict, example: types.ReinforcementTuningExampleOrDict, single_reward_config: Optional[ types.SingleReinforcementTuningRewardConfigOrDict ] = None, composite_reward_config: Optional[ types.CompositeReinforcementTuningRewardConfigOrDict ] = None, config: Optional[types.ValidateRewardConfigOrDict] = None, ) -> types.ValidateRewardResponse: """Validates a reinforcement tuning reward configuration. Allows users to validate a reinforcement tuning reward configuration against a sample response and example before creating a reinforcement tuning job, so that they can iterate on the reward configuration without having to create a tuning job each time. Args: parent: The resource name of the Location to validate the reward in, e.g. `projects/{project}/locations/{location}`. sample_response: The sample response for validating the reward configuration. example: The example to validate the reward configuration. single_reward_config: Single reward function configuration for reinforcement tuning. Mutually exclusive with composite_reward_config. composite_reward_config: Composite reward function configuration for reinforcement tuning. Mutually exclusive with single_reward_config. config: Optional parameters for the request. Returns: A ValidateRewardResponse with the computed reward(s). """ parameter_model = types._ValidateRewardParameters( parent=parent, sample_response=sample_response, example=example, single_reward_config=single_reward_config, composite_reward_config=composite_reward_config, 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 = _ValidateRewardParameters_to_vertex( parameter_model, None, parameter_model ) request_url_dict = request_dict.get('_url') if request_url_dict: path = ( '{parent}/tuningJobs:validateReinforcementTuningReward'.format_map( request_url_dict ) ) else: path = '{parent}/tuningJobs:validateReinforcementTuningReward' 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 = _ValidateRewardResponse_from_vertex( response_dict, None, parameter_model ) return_value = types.ValidateRewardResponse._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 {}, ) return_value.sdk_http_response = types.HttpResponse( headers=response.headers ) self._api_client._verify_response(return_value) return return_value async def get( self, *, name: str, config: Optional[types.GetTuningJobConfigOrDict] = None, ) -> types.TuningJob: job = await self._get(name=name, config=config) if ( job.experiment and self._api_client.vertexai and self._api_client.project is not None ): _IpythonUtils.display_experiment_button( experiment=job.experiment, project=self._api_client.project, ) return job @_common.experimental_warning( "The SDK's tuning implementation is experimental, " 'and may change in future versions.' ) async def tune( self, *, base_model: str, training_dataset: types.TuningDatasetOrDict, config: Optional[types.CreateTuningJobConfigOrDict] = None, ) -> types.TuningJob: if self._api_client.vertexai: if base_model.startswith('projects/'): # Pre-tuned model checkpoint_id = None if config: checkpoint_id = getattr(config, 'pre_tuned_model_checkpoint_id', None) pre_tuned_model = types.PreTunedModel( tuned_model_name=base_model, checkpoint_id=checkpoint_id ) tuning_job = await self._tune( pre_tuned_model=pre_tuned_model, training_dataset=training_dataset, config=config, ) else: if ( config is not None and getattr(config, 'evaluation_config', None) is not None ): evaluation_config = getattr(config, 'evaluation_config') if isinstance(evaluation_config, dict): evaluation_config = types.EvaluationConfig(**evaluation_config) if ( not evaluation_config.metrics or not evaluation_config.output_config ): raise ValueError( 'Evaluation config must have at least one metric and an output' ' config.' ) for i in range(len(evaluation_config.metrics)): if isinstance(evaluation_config.metrics[i], dict): evaluation_config.metrics[i] = types.Metric.model_validate( evaluation_config.metrics[i] ) if isinstance(config, dict): config['evaluation_config'] = evaluation_config else: config.evaluation_config = evaluation_config tuning_job = await self._tune( base_model=base_model, training_dataset=training_dataset, config=config, ) else: operation = await self._tune_mldev( base_model=base_model, training_dataset=training_dataset, config=config, ) if operation.metadata is not None and 'tunedModel' in operation.metadata: tuned_model_name = operation.metadata['tunedModel'] else: if operation.name is None: raise ValueError('Operation name is required.') tuned_model_name = operation.name.partition('/operations/')[0] tuning_job = types.TuningJob( name=tuned_model_name, state=types.JobState.JOB_STATE_QUEUED, ) if tuning_job.name and self._api_client.vertexai: _IpythonUtils.display_model_tuning_button( tuning_job_resource=tuning_job.name ) return tuning_job async def list( self, *, config: Optional[types.ListTuningJobsConfigOrDict] = None ) -> AsyncPager[types.TuningJob]: """Lists `TuningJob` objects asynchronously. Args: config: The configuration for the list request. Returns: A Pager object that contains one page of tuning jobs. When iterating over the pager, it automatically fetches the next page if there are more. Usage: .. code-block:: python async for tuning_job in await client.aio.tunings.list(): print(tuning_job.name) """ list_request = self._list return AsyncPager( 'tuning_jobs', list_request, await self._list(config=config), config, ) class _IpythonUtils: """Temporary class to hold the IPython related functions.""" displayed_experiments: set[str] = set() @staticmethod def _get_ipython_shell_name() -> Union[str, Any]: import sys if 'IPython' in sys.modules: from IPython import get_ipython return get_ipython().__class__.__name__ return '' @staticmethod def is_ipython_available() -> bool: return bool(_IpythonUtils._get_ipython_shell_name()) @staticmethod def _get_styles() -> str: """Returns the HTML style markup to support custom buttons.""" return """ """ @staticmethod def _parse_resource_name(marker: str, resource_parts: list[str]) -> str: """Returns the part after the marker text part.""" for i in range(len(resource_parts)): if resource_parts[i] == marker and i + 1 < len(resource_parts): return resource_parts[i + 1] return '' @staticmethod def _display_link( text: str, url: str, icon: Optional[str] = 'open_in_new' ) -> None: """Creates and displays the link to open the Gemini Enterprise Agent Platform resource. Args: text: The text displayed on the clickable button. url: The url that the button will lead to. Only cloud console URIs are allowed. icon: The icon name on the button (from material-icons library) """ CLOUD_UI_URL = 'https://console.cloud.google.com' # pylint: disable=invalid-name if not url.startswith(CLOUD_UI_URL): raise ValueError(f'Only urls starting with {CLOUD_UI_URL} are allowed.') import uuid button_id = f'view-vertex-resource-{str(uuid.uuid4())}' # Add the markup for the CSS and link component html = f""" {_IpythonUtils._get_styles()} {icon} {text} """ # Add the click handler for the link html += f""" """ from IPython.display import display from IPython.display import HTML display(HTML(html)) @staticmethod def display_experiment_button(experiment: str, project: str) -> None: """Function to generate a link bound to the Gemini Enterprise Agent Platform experiment. Args: experiment: The Gemini Enterprise Agent Platform experiment name. Example format: projects/{project_id}/locations/{location}/metadataStores/default/contexts/{experiment_name} project: The project (alphanumeric) name. """ if ( not _IpythonUtils.is_ipython_available() or experiment in _IpythonUtils.displayed_experiments ): return # Experiment gives the numeric id, but we need the alphanumeric project # name. So we get the project from the api client object as an argument. resource_parts = experiment.split('/') location = resource_parts[3] experiment_name = resource_parts[-1] uri = ( 'https://console.cloud.google.com/vertex-ai/experiments/locations/' + f'{location}/experiments/{experiment_name}/' + f'runs?project={project}' ) _IpythonUtils._display_link('View Experiment', uri, 'science') # Avoid repeatedly showing the button _IpythonUtils.displayed_experiments.add(experiment) @staticmethod def display_model_tuning_button(tuning_job_resource: str) -> None: """Function to generate a link bound to the Gemini Enterprise Agent Platform model tuning job. Args: tuning_job_resource: The Gemini Enterprise Agent Platform tuning job name. Example format: projects/{project_id}/locations/{location}/tuningJobs/{tuning_job_id} """ if not _IpythonUtils.is_ipython_available(): return resource_parts = tuning_job_resource.split('/') project = resource_parts[1] location = resource_parts[3] tuning_job_id = resource_parts[-1] uri = ( 'https://console.cloud.google.com/vertex-ai/generative/language/' + f'locations/{location}/tuning/tuningJob/{tuning_job_id}' + f'?project={project}' ) _IpythonUtils._display_link('View Tuning Job', uri, 'tune')