fix: Fix the bug to support Gemini Batch inlined requests system instruction

PiperOrigin-RevId: 795249849

Author: yinghsienwu

google/genai/batches.py

@@ -2257,6 +2257,17 @@ def create(
       )
       print(batch_job.state)
     """
+    parameter_model = types._CreateBatchJobParameters(
+        model=model,
+        src=src,
+        config=config,
+    )
+    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
     if self._api_client.vertexai:
       if isinstance(src, list):
         raise ValueError(
@@ -2265,6 +2276,65 @@ def create(
         )
 
       config = _extra_utils.format_destination(src, config)
+    else:
+      if isinstance(parameter_model.src, list) or (
+          not isinstance(parameter_model.src, str)
+          and parameter_model.src
+          and parameter_model.src.inlined_requests
+      ):
+        # Handle system instruction in InlinedRequests.
+        request_url_dict: Optional[dict[str, str]]
+        request_dict: dict[str, Any] = _CreateBatchJobParameters_to_mldev(
+            self._api_client, parameter_model
+        )
+        request_url_dict = request_dict.get('_url')
+        if request_url_dict:
+          path = '{model}:batchGenerateContent'.format_map(request_url_dict)
+        else:
+          path = '{model}:batchGenerateContent'
+        query_params = request_dict.get('_query')
+        if query_params:
+          path = f'{path}?{urlencode(query_params)}'
+        request_dict.pop('config', None)
+
+        request_dict = _common.convert_to_dict(request_dict)
+        request_dict = _common.encode_unserializable_types(request_dict)
+        # Move system instruction to 'request':
+        # {'systemInstruction': system_instruction}
+        requests = []
+        batch_dict = request_dict.get('batch')
+        if batch_dict and isinstance(batch_dict, dict):
+          input_config_dict = batch_dict.get('inputConfig')
+          if input_config_dict and isinstance(input_config_dict, dict):
+            requests_dict = input_config_dict.get('requests')
+            if requests_dict and isinstance(requests_dict, dict):
+              requests = requests_dict.get('requests')
+        new_requests = []
+        if requests:
+          for req in requests:
+            if req.get('systemInstruction'):
+              value = req.pop('systemInstruction')
+              req['request'].update({'systemInstruction': value})
+            new_requests.append(req)
+        request_dict['batch']['inputConfig']['requests'][  # type: ignore
+            'requests'
+        ] = new_requests
+
+        response = self._api_client.request(
+            'post', path, request_dict, http_options
+        )
+
+        response_dict = '' if not response.body else json.loads(response.body)
+
+        response_dict = _BatchJob_from_mldev(response_dict)
+
+        return_value = types.BatchJob._from_response(
+            response=response_dict, kwargs=parameter_model.model_dump()
+        )
+
+        self._api_client._verify_response(return_value)
+        return return_value
+
     return self._create(model=model, src=src, config=config)
 
   def list(
@@ -2691,6 +2761,17 @@ async def create(
           src="gs://path/to/input/data",
       )
     """
+    parameter_model = types._CreateBatchJobParameters(
+        model=model,
+        src=src,
+        config=config,
+    )
+    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
     if self._api_client.vertexai:
       if isinstance(src, list):
         raise ValueError(
@@ -2699,6 +2780,65 @@ async def create(
         )
 
       config = _extra_utils.format_destination(src, config)
+    else:
+      if isinstance(parameter_model.src, list) or (
+          not isinstance(parameter_model.src, str)
+          and parameter_model.src
+          and parameter_model.src.inlined_requests
+      ):
+        # Handle system instruction in InlinedRequests.
+        request_url_dict: Optional[dict[str, str]]
+        request_dict: dict[str, Any] = _CreateBatchJobParameters_to_mldev(
+            self._api_client, parameter_model
+        )
+        request_url_dict = request_dict.get('_url')
+        if request_url_dict:
+          path = '{model}:batchGenerateContent'.format_map(request_url_dict)
+        else:
+          path = '{model}:batchGenerateContent'
+        query_params = request_dict.get('_query')
+        if query_params:
+          path = f'{path}?{urlencode(query_params)}'
+        request_dict.pop('config', None)
+
+        request_dict = _common.convert_to_dict(request_dict)
+        request_dict = _common.encode_unserializable_types(request_dict)
+        # Move system instruction to 'request':
+        # {'systemInstruction': system_instruction}
+        requests = []
+        batch_dict = request_dict.get('batch')
+        if batch_dict and isinstance(batch_dict, dict):
+          input_config_dict = batch_dict.get('inputConfig')
+          if input_config_dict and isinstance(input_config_dict, dict):
+            requests_dict = input_config_dict.get('requests')
+            if requests_dict and isinstance(requests_dict, dict):
+              requests = requests_dict.get('requests')
+        new_requests = []
+        if requests:
+          for req in requests:
+            if req.get('systemInstruction'):
+              value = req.pop('systemInstruction')
+              req['request'].update({'systemInstruction': value})
+            new_requests.append(req)
+        request_dict['batch']['inputConfig']['requests'][  # type: ignore
+            'requests'
+        ] = new_requests
+
+        response = await self._api_client.async_request(
+            'post', path, request_dict, http_options
+        )
+
+        response_dict = '' if not response.body else json.loads(response.body)
+
+        response_dict = _BatchJob_from_mldev(response_dict)
+
+        return_value = types.BatchJob._from_response(
+            response=response_dict, kwargs=parameter_model.model_dump()
+        )
+
+        self._api_client._verify_response(return_value)
+        return return_value
+
     return await self._create(model=model, src=src, config=config)
 
   async def list(

google/genai/tests/batches/test_create_with_inlined_requests.py

@@ -39,12 +39,17 @@
 _INLINED_TEXT_REQUEST = {
     'contents': [{
         'parts': [{
-            'text': 'What is the QQQ stock price?',
+            'text': 'high',
         }],
         'role': 'user',
     }],
     'config': {
         'response_modalities': ['TEXT'],
+        'system_instruction': 'I say high, you say low',
+        'thinking_config': {
+            'include_thoughts': True,
+            'thinking_budget': 4000,
+        },
     },
 }
 _INLINED_IMAGE_REQUEST = {
@@ -130,6 +135,17 @@
     ),
     pytest_helper.TestTableItem(
         name='test_with_inlined_request',
+        parameters=types._CreateBatchJobParameters(
+            model=_MLDEV_GEMINI_MODEL,
+            src={'inlined_requests': [_INLINED_REQUEST]},
+            config={
+                'display_name': _DISPLAY_NAME,
+            },
+        ),
+        exception_if_vertex='not supported',
+    ),
+    pytest_helper.TestTableItem(
+        name='test_union_with_inlined_request_system_instruction',
         parameters=types._CreateBatchJobParameters(
             model=_MLDEV_GEMINI_MODEL,
             src={'inlined_requests': [_INLINED_TEXT_REQUEST]},