/
saved_model_metadata_builder.py
165 lines (141 loc) · 6.17 KB
/
saved_model_metadata_builder.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
# -*- coding: utf-8 -*-
# Copyright 2021 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.
from google.protobuf import json_format
from typing import Any, Dict, List, Optional
from google.cloud.aiplatform.compat.types import (
explanation_metadata_v1beta1 as explanation_metadata,
)
from google.cloud.aiplatform.explain.metadata import metadata_builder
class SavedModelMetadataBuilder(metadata_builder.MetadataBuilder):
"""Metadata builder class that accepts a TF1 saved model."""
def __init__(
self,
model_path: str,
tags: Optional[List[str]] = None,
signature_name: Optional[str] = None,
outputs_to_explain: Optional[List[str]] = None,
) -> None:
"""Initializes a SavedModelMetadataBuilder object.
Args:
model_path:
Required. Local or GCS path to load the saved model from.
tags:
Optional. Tags to identify the model graph. If None or empty,
TensorFlow's default serving tag will be used.
signature_name:
Optional. Name of the signature to be explained. Inputs and
outputs of this signature will be written in the metadata. If not
provided, the default signature will be used.
outputs_to_explain:
Optional. List of output names to explain. Only single output is
supported for now. Hence, the list should contain one element.
This parameter is required if the model signature (provided via
signature_name) specifies multiple outputs.
Raises:
ValueError if outputs_to_explain contains more than 1 element or
signature contains multiple outputs.
"""
if outputs_to_explain:
if len(outputs_to_explain) > 1:
raise ValueError(
"Only one output is supported at the moment. "
f"Received: {outputs_to_explain}."
)
self._output_to_explain = next(iter(outputs_to_explain))
try:
import tensorflow.compat.v1 as tf
except ImportError:
raise ImportError(
"Tensorflow is not installed and is required to load saved model. "
'Please install the SDK using "pip install "tensorflow>=1.15,<2.0""'
)
if not signature_name:
signature_name = tf.saved_model.DEFAULT_SERVING_SIGNATURE_DEF_KEY
self._tags = tags or [tf.saved_model.tag_constants.SERVING]
self._graph = tf.Graph()
with self.graph.as_default():
self._session = tf.Session(graph=self.graph)
self._metagraph_def = tf.saved_model.loader.load(
sess=self.session, tags=self._tags, export_dir=model_path
)
if signature_name not in self._metagraph_def.signature_def:
raise ValueError(
f"Serving sigdef key {signature_name} not in the signature def."
)
serving_sigdef = self._metagraph_def.signature_def[signature_name]
if not outputs_to_explain:
if len(serving_sigdef.outputs) > 1:
raise ValueError(
"The signature contains multiple outputs. Specify "
'an output via "outputs_to_explain" parameter.'
)
self._output_to_explain = next(iter(serving_sigdef.outputs.keys()))
self._inputs = _create_input_metadata_from_signature(serving_sigdef.inputs)
self._outputs = _create_output_metadata_from_signature(
serving_sigdef.outputs, self._output_to_explain
)
@property
def graph(self) -> "tf.Graph": # noqa: F821
return self._graph
@property
def session(self) -> "tf.Session": # noqa: F821
return self._session
def get_metadata(self) -> Dict[str, Any]:
"""Returns the current metadata as a dictionary.
Returns:
Json format of the explanation metadata.
"""
current_md = explanation_metadata.ExplanationMetadata(
inputs=self._inputs, outputs=self._outputs,
)
return json_format.MessageToDict(current_md._pb)
def _create_input_metadata_from_signature(
signature_inputs: Dict[str, "tf.Tensor"] # noqa: F821
) -> Dict[str, explanation_metadata.ExplanationMetadata.InputMetadata]:
"""Creates InputMetadata from signature inputs.
Args:
signature_inputs:
Required. Inputs of the signature to be explained. If not provided,
the default signature will be used.
Returns:
Inferred input metadata from the model.
"""
input_mds = {}
for key, tensor in signature_inputs.items():
input_mds[key] = explanation_metadata.ExplanationMetadata.InputMetadata(
input_tensor_name=tensor.name
)
return input_mds
def _create_output_metadata_from_signature(
signature_outputs: Dict[str, "tf.Tensor"], # noqa: F821
output_to_explain: Optional[str] = None,
) -> Dict[str, explanation_metadata.ExplanationMetadata.OutputMetadata]:
"""Creates OutputMetadata from signature inputs.
Args:
signature_outputs:
Required. Inputs of the signature to be explained. If not provided,
the default signature will be used.
output_to_explain:
Optional. Output name to explain.
Returns:
Inferred output metadata from the model.
"""
output_mds = {}
for key, tensor in signature_outputs.items():
if not output_to_explain or output_to_explain == key:
output_mds[key] = explanation_metadata.ExplanationMetadata.OutputMetadata(
output_tensor_name=tensor.name
)
return output_mds